|
|
|
|
@@ -413,15 +413,15 @@ def test_standard_forwarded(app):
|
|
|
|
|
"Forwarded": (
|
|
|
|
|
'for=1.1.1.1, for=injected;host="'
|
|
|
|
|
', for="[::2]";proto=https;host=me.tld;path="/app/";secret=mySecret'
|
|
|
|
|
',for=broken;;secret=b0rked'
|
|
|
|
|
', for=127.0.0.3;scheme=http;port=1234'
|
|
|
|
|
",for=broken;;secret=b0rked"
|
|
|
|
|
", for=127.0.0.3;scheme=http;port=1234"
|
|
|
|
|
),
|
|
|
|
|
"X-Real-IP": "127.0.0.2",
|
|
|
|
|
"X-Forwarded-For": "127.0.1.1",
|
|
|
|
|
"X-Scheme": "ws",
|
|
|
|
|
}
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == { "for": "127.0.0.2", "proto": "ws" }
|
|
|
|
|
assert response.json == {"for": "127.0.0.2", "proto": "ws"}
|
|
|
|
|
assert request.remote_addr == "127.0.0.2"
|
|
|
|
|
assert request.scheme == "ws"
|
|
|
|
|
assert request.server_port == 80
|
|
|
|
|
@@ -433,7 +433,7 @@ def test_standard_forwarded(app):
|
|
|
|
|
"proto": "https",
|
|
|
|
|
"host": "me.tld",
|
|
|
|
|
"path": "/app/",
|
|
|
|
|
"secret": "mySecret"
|
|
|
|
|
"secret": "mySecret",
|
|
|
|
|
}
|
|
|
|
|
assert request.remote_addr == "[::2]"
|
|
|
|
|
assert request.server_name == "me.tld"
|
|
|
|
|
@@ -443,7 +443,7 @@ def test_standard_forwarded(app):
|
|
|
|
|
# Empty Forwarded header -> use X-headers
|
|
|
|
|
headers["Forwarded"] = ""
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == { "for": "127.0.0.2", "proto": "ws" }
|
|
|
|
|
assert response.json == {"for": "127.0.0.2", "proto": "ws"}
|
|
|
|
|
|
|
|
|
|
# Header present but not matching anything
|
|
|
|
|
request, response = app.test_client.get("/", headers={"Forwarded": "."})
|
|
|
|
|
@@ -451,8 +451,8 @@ def test_standard_forwarded(app):
|
|
|
|
|
|
|
|
|
|
# Forwarded header present but no matching secret -> use X-headers
|
|
|
|
|
headers = {
|
|
|
|
|
"Forwarded": 'for=1.1.1.1;secret=x, for=127.0.0.1',
|
|
|
|
|
"X-Real-IP": "127.0.0.2"
|
|
|
|
|
"Forwarded": "for=1.1.1.1;secret=x, for=127.0.0.1",
|
|
|
|
|
"X-Real-IP": "127.0.0.2",
|
|
|
|
|
}
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == {"for": "127.0.0.2"}
|
|
|
|
|
@@ -464,7 +464,7 @@ def test_standard_forwarded(app):
|
|
|
|
|
assert response.json == {
|
|
|
|
|
"for": "127.0.0.4",
|
|
|
|
|
"port": 1234,
|
|
|
|
|
"secret": "mySecret"
|
|
|
|
|
"secret": "mySecret",
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
# Test escapes (modify this if you see anyone implementing quoted-pairs)
|
|
|
|
|
@@ -472,29 +472,29 @@ def test_standard_forwarded(app):
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == {
|
|
|
|
|
"for": "test",
|
|
|
|
|
"quoted": '\\,x=x;y=\\',
|
|
|
|
|
"secret": "mySecret"
|
|
|
|
|
"quoted": "\\,x=x;y=\\",
|
|
|
|
|
"secret": "mySecret",
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
# Secret insulated by malformed field #1
|
|
|
|
|
headers = {"Forwarded": 'for=test;secret=mySecret;b0rked;proto=wss;'}
|
|
|
|
|
headers = {"Forwarded": "for=test;secret=mySecret;b0rked;proto=wss;"}
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == {"for": "test", "secret": "mySecret"}
|
|
|
|
|
|
|
|
|
|
# Secret insulated by malformed field #2
|
|
|
|
|
headers = {"Forwarded": 'for=test;b0rked;secret=mySecret;proto=wss'}
|
|
|
|
|
headers = {"Forwarded": "for=test;b0rked;secret=mySecret;proto=wss"}
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == {"proto": "wss", "secret": "mySecret"}
|
|
|
|
|
|
|
|
|
|
# Unexpected termination should not lose existing acceptable values
|
|
|
|
|
headers = {"Forwarded": 'b0rked;secret=mySecret;proto=wss'}
|
|
|
|
|
headers = {"Forwarded": "b0rked;secret=mySecret;proto=wss"}
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == {"proto": "wss", "secret": "mySecret"}
|
|
|
|
|
|
|
|
|
|
# Field normalization
|
|
|
|
|
headers = {
|
|
|
|
|
"Forwarded": 'PROTO=WSS;BY="CAFE::8000";FOR=unknown;PORT=X;HOST="A:2";'
|
|
|
|
|
'PATH="/With%20Spaces%22Quoted%22/sanicApp?key=val";SECRET=mySecret'
|
|
|
|
|
'PATH="/With%20Spaces%22Quoted%22/sanicApp?key=val";SECRET=mySecret'
|
|
|
|
|
}
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == {
|
|
|
|
|
@@ -507,7 +507,7 @@ def test_standard_forwarded(app):
|
|
|
|
|
|
|
|
|
|
# Using "by" field as secret
|
|
|
|
|
app.config.FORWARDED_SECRET = "_proxySecret"
|
|
|
|
|
headers = {"Forwarded": 'for=1.2.3.4; by=_proxySecret'}
|
|
|
|
|
headers = {"Forwarded": "for=1.2.3.4; by=_proxySecret"}
|
|
|
|
|
request, response = app.test_client.get("/", headers=headers)
|
|
|
|
|
assert response.json == {"for": "1.2.3.4", "by": "_proxySecret"}
|
|
|
|
|
|
|
|
|
|
@@ -525,15 +525,15 @@ async def test_standard_forwarded_asgi(app):
|
|
|
|
|
"Forwarded": (
|
|
|
|
|
'for=1.1.1.1, for=injected;host="'
|
|
|
|
|
', for="[::2]";proto=https;host=me.tld;path="/app/";secret=mySecret'
|
|
|
|
|
',for=broken;;secret=b0rked'
|
|
|
|
|
', for=127.0.0.3;scheme=http;port=1234'
|
|
|
|
|
",for=broken;;secret=b0rked"
|
|
|
|
|
", for=127.0.0.3;scheme=http;port=1234"
|
|
|
|
|
),
|
|
|
|
|
"X-Real-IP": "127.0.0.2",
|
|
|
|
|
"X-Forwarded-For": "127.0.1.1",
|
|
|
|
|
"X-Scheme": "ws",
|
|
|
|
|
}
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == { "for": "127.0.0.2", "proto": "ws" }
|
|
|
|
|
assert response.json() == {"for": "127.0.0.2", "proto": "ws"}
|
|
|
|
|
assert request.remote_addr == "127.0.0.2"
|
|
|
|
|
assert request.scheme == "ws"
|
|
|
|
|
assert request.server_port == 80
|
|
|
|
|
@@ -545,7 +545,7 @@ async def test_standard_forwarded_asgi(app):
|
|
|
|
|
"proto": "https",
|
|
|
|
|
"host": "me.tld",
|
|
|
|
|
"path": "/app/",
|
|
|
|
|
"secret": "mySecret"
|
|
|
|
|
"secret": "mySecret",
|
|
|
|
|
}
|
|
|
|
|
assert request.remote_addr == "[::2]"
|
|
|
|
|
assert request.server_name == "me.tld"
|
|
|
|
|
@@ -555,16 +555,18 @@ async def test_standard_forwarded_asgi(app):
|
|
|
|
|
# Empty Forwarded header -> use X-headers
|
|
|
|
|
headers["Forwarded"] = ""
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == { "for": "127.0.0.2", "proto": "ws" }
|
|
|
|
|
assert response.json() == {"for": "127.0.0.2", "proto": "ws"}
|
|
|
|
|
|
|
|
|
|
# Header present but not matching anything
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers={"Forwarded": "."})
|
|
|
|
|
request, response = await app.asgi_client.get(
|
|
|
|
|
"/", headers={"Forwarded": "."}
|
|
|
|
|
)
|
|
|
|
|
assert response.json() == {}
|
|
|
|
|
|
|
|
|
|
# Forwarded header present but no matching secret -> use X-headers
|
|
|
|
|
headers = {
|
|
|
|
|
"Forwarded": 'for=1.1.1.1;secret=x, for=127.0.0.1',
|
|
|
|
|
"X-Real-IP": "127.0.0.2"
|
|
|
|
|
"Forwarded": "for=1.1.1.1;secret=x, for=127.0.0.1",
|
|
|
|
|
"X-Real-IP": "127.0.0.2",
|
|
|
|
|
}
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == {"for": "127.0.0.2"}
|
|
|
|
|
@@ -576,7 +578,7 @@ async def test_standard_forwarded_asgi(app):
|
|
|
|
|
assert response.json() == {
|
|
|
|
|
"for": "127.0.0.4",
|
|
|
|
|
"port": 1234,
|
|
|
|
|
"secret": "mySecret"
|
|
|
|
|
"secret": "mySecret",
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
# Test escapes (modify this if you see anyone implementing quoted-pairs)
|
|
|
|
|
@@ -584,29 +586,29 @@ async def test_standard_forwarded_asgi(app):
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == {
|
|
|
|
|
"for": "test",
|
|
|
|
|
"quoted": '\\,x=x;y=\\',
|
|
|
|
|
"secret": "mySecret"
|
|
|
|
|
"quoted": "\\,x=x;y=\\",
|
|
|
|
|
"secret": "mySecret",
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
# Secret insulated by malformed field #1
|
|
|
|
|
headers = {"Forwarded": 'for=test;secret=mySecret;b0rked;proto=wss;'}
|
|
|
|
|
headers = {"Forwarded": "for=test;secret=mySecret;b0rked;proto=wss;"}
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == {"for": "test", "secret": "mySecret"}
|
|
|
|
|
|
|
|
|
|
# Secret insulated by malformed field #2
|
|
|
|
|
headers = {"Forwarded": 'for=test;b0rked;secret=mySecret;proto=wss'}
|
|
|
|
|
headers = {"Forwarded": "for=test;b0rked;secret=mySecret;proto=wss"}
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == {"proto": "wss", "secret": "mySecret"}
|
|
|
|
|
|
|
|
|
|
# Unexpected termination should not lose existing acceptable values
|
|
|
|
|
headers = {"Forwarded": 'b0rked;secret=mySecret;proto=wss'}
|
|
|
|
|
headers = {"Forwarded": "b0rked;secret=mySecret;proto=wss"}
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == {"proto": "wss", "secret": "mySecret"}
|
|
|
|
|
|
|
|
|
|
# Field normalization
|
|
|
|
|
headers = {
|
|
|
|
|
"Forwarded": 'PROTO=WSS;BY="CAFE::8000";FOR=unknown;PORT=X;HOST="A:2";'
|
|
|
|
|
'PATH="/With%20Spaces%22Quoted%22/sanicApp?key=val";SECRET=mySecret'
|
|
|
|
|
'PATH="/With%20Spaces%22Quoted%22/sanicApp?key=val";SECRET=mySecret'
|
|
|
|
|
}
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == {
|
|
|
|
|
@@ -619,7 +621,7 @@ async def test_standard_forwarded_asgi(app):
|
|
|
|
|
|
|
|
|
|
# Using "by" field as secret
|
|
|
|
|
app.config.FORWARDED_SECRET = "_proxySecret"
|
|
|
|
|
headers = {"Forwarded": 'for=1.2.3.4; by=_proxySecret'}
|
|
|
|
|
headers = {"Forwarded": "for=1.2.3.4; by=_proxySecret"}
|
|
|
|
|
request, response = await app.asgi_client.get("/", headers=headers)
|
|
|
|
|
assert response.json() == {"for": "1.2.3.4", "by": "_proxySecret"}
|
|
|
|
|
|
|
|
|
|
@@ -813,11 +815,14 @@ def test_forwarded_scheme(app):
|
|
|
|
|
assert request.scheme == "http"
|
|
|
|
|
|
|
|
|
|
request, response = app.test_client.get(
|
|
|
|
|
"/", headers={"X-Forwarded-For": "127.1.2.3", "X-Forwarded-Proto": "https"}
|
|
|
|
|
"/",
|
|
|
|
|
headers={"X-Forwarded-For": "127.1.2.3", "X-Forwarded-Proto": "https"},
|
|
|
|
|
)
|
|
|
|
|
assert request.scheme == "https"
|
|
|
|
|
|
|
|
|
|
request, response = app.test_client.get("/", headers={"X-Forwarded-For": "127.1.2.3", "X-Scheme": "https"})
|
|
|
|
|
request, response = app.test_client.get(
|
|
|
|
|
"/", headers={"X-Forwarded-For": "127.1.2.3", "X-Scheme": "https"}
|
|
|
|
|
)
|
|
|
|
|
assert request.scheme == "https"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@@ -1872,7 +1877,7 @@ def test_request_server_name_in_host_header(app):
|
|
|
|
|
request, response = app.test_client.get(
|
|
|
|
|
"/", headers={"Host": "mal_formed"}
|
|
|
|
|
)
|
|
|
|
|
assert request.server_name == None # For now (later maybe 127.0.0.1)
|
|
|
|
|
assert request.server_name == None # For now (later maybe 127.0.0.1)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_request_server_name_forwarded(app):
|
|
|
|
|
@@ -1883,7 +1888,11 @@ def test_request_server_name_forwarded(app):
|
|
|
|
|
app.config.PROXIES_COUNT = 1
|
|
|
|
|
request, response = app.test_client.get(
|
|
|
|
|
"/",
|
|
|
|
|
headers={"Host": "my-server:5555", "X-Forwarded-For": "127.1.2.3", "X-Forwarded-Host": "your-server"},
|
|
|
|
|
headers={
|
|
|
|
|
"Host": "my-server:5555",
|
|
|
|
|
"X-Forwarded-For": "127.1.2.3",
|
|
|
|
|
"X-Forwarded-Host": "your-server",
|
|
|
|
|
},
|
|
|
|
|
)
|
|
|
|
|
assert request.server_name == "your-server"
|
|
|
|
|
|
|
|
|
|
@@ -1925,7 +1934,12 @@ def test_request_server_port_forwarded(app):
|
|
|
|
|
|
|
|
|
|
app.config.PROXIES_COUNT = 1
|
|
|
|
|
request, response = app.test_client.get(
|
|
|
|
|
"/", headers={"Host": "my-server:5555", "X-Forwarded-For": "127.1.2.3", "X-Forwarded-Port": "4444"}
|
|
|
|
|
"/",
|
|
|
|
|
headers={
|
|
|
|
|
"Host": "my-server:5555",
|
|
|
|
|
"X-Forwarded-For": "127.1.2.3",
|
|
|
|
|
"X-Forwarded-Port": "4444",
|
|
|
|
|
},
|
|
|
|
|
)
|
|
|
|
|
assert request.server_port == 4444
|
|
|
|
|
|
|
|
|
|
@@ -1948,7 +1962,10 @@ def test_server_name_and_url_for(app):
|
|
|
|
|
app.config.SERVER_NAME = "my-server"
|
|
|
|
|
assert app.url_for("handler", _external=True) == "http://my-server/foo"
|
|
|
|
|
request, response = app.test_client.get("/foo")
|
|
|
|
|
assert request.url_for("handler") == f"http://my-server:{app.test_client.port}/foo"
|
|
|
|
|
assert (
|
|
|
|
|
request.url_for("handler")
|
|
|
|
|
== f"http://my-server:{app.test_client.port}/foo"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
app.config.SERVER_NAME = "https://my-server/path"
|
|
|
|
|
request, response = app.test_client.get("/foo")
|
|
|
|
|
@@ -1969,7 +1986,12 @@ def test_url_for_with_forwarded_request(app):
|
|
|
|
|
app.config.SERVER_NAME = "my-server"
|
|
|
|
|
app.config.PROXIES_COUNT = 1
|
|
|
|
|
request, response = app.test_client.get(
|
|
|
|
|
"/", headers={"X-Forwarded-For": "127.1.2.3", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "6789"}
|
|
|
|
|
"/",
|
|
|
|
|
headers={
|
|
|
|
|
"X-Forwarded-For": "127.1.2.3",
|
|
|
|
|
"X-Forwarded-Proto": "https",
|
|
|
|
|
"X-Forwarded-Port": "6789",
|
|
|
|
|
},
|
|
|
|
|
)
|
|
|
|
|
assert app.url_for("view_name") == "/another_view"
|
|
|
|
|
assert (
|
|
|
|
|
@@ -1981,7 +2003,12 @@ def test_url_for_with_forwarded_request(app):
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
request, response = app.test_client.get(
|
|
|
|
|
"/", headers={"X-Forwarded-For": "127.1.2.3", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "443"}
|
|
|
|
|
"/",
|
|
|
|
|
headers={
|
|
|
|
|
"X-Forwarded-For": "127.1.2.3",
|
|
|
|
|
"X-Forwarded-Proto": "https",
|
|
|
|
|
"X-Forwarded-Port": "443",
|
|
|
|
|
},
|
|
|
|
|
)
|
|
|
|
|
assert request.url_for("view_name") == "https://my-server/another_view"
|
|
|
|
|
|
|
|
|
|
|