summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMichael Merickel <michael@merickel.org>2019-07-18 23:42:30 -0500
committerMichael Merickel <michael@merickel.org>2019-07-18 23:42:30 -0500
commita15155f54e8ff615fd3f1db3cc35a60127fb58f3 (patch)
treee6ebee7dea8dc6cfdef356e88b9ea3e4232ba6fc
parent9695c5d9a0eae818a4e9c2e5becb826c24e4533d (diff)
downloadwaitress-a15155f54e8ff615fd3f1db3cc35a60127fb58f3.tar.gz
move tests into test_proxy_headers
-rw-r--r--waitress/proxy_headers.py6
-rw-r--r--waitress/task.py10
-rw-r--r--waitress/tests/test_proxy_headers.py670
-rw-r--r--waitress/tests/test_task.py617
4 files changed, 691 insertions, 612 deletions
diff --git a/waitress/proxy_headers.py b/waitress/proxy_headers.py
index 501dc2b..132fea8 100644
--- a/waitress/proxy_headers.py
+++ b/waitress/proxy_headers.py
@@ -25,9 +25,9 @@ class MalformedProxyHeader(Exception):
def proxy_headers_middleware(
app,
- trusted_proxy,
- trusted_proxy_count,
- trusted_proxy_headers,
+ trusted_proxy=None,
+ trusted_proxy_count=1,
+ trusted_proxy_headers=None,
clear_untrusted=True,
log_untrusted=False,
logger=logger,
diff --git a/waitress/task.py b/waitress/task.py
index 96451a7..78d8b83 100644
--- a/waitress/task.py
+++ b/waitress/task.py
@@ -356,9 +356,19 @@ class ErrorTask(Task):
status, headers, body = e.to_response()
self.status = status
self.response_headers.extend(headers)
+ if self.version == '1.1':
+ connection = self.request.headers.get('CONNECTION', '').lower()
+ if connection == 'close':
+ self.response_headers.append(('Connection', 'close'))
+ # under HTTP 1.1 keep-alive is default, no need to set the header
+ else:
+ # HTTP 1.0
+ self.response_headers.append(('Connection', 'close'))
+ self.close_on_finish = True
self.content_length = len(body)
self.write(tobytes(body))
+
class WSGITask(Task):
"""A WSGI task produces a response from a WSGI application.
"""
diff --git a/waitress/tests/test_proxy_headers.py b/waitress/tests/test_proxy_headers.py
new file mode 100644
index 0000000..aa4c907
--- /dev/null
+++ b/waitress/tests/test_proxy_headers.py
@@ -0,0 +1,670 @@
+import unittest
+
+from waitress.compat import tobytes
+
+class TestProxyHeadersMiddleware(unittest.TestCase):
+
+ def _makeOne(self, app, **kw):
+ from waitress.proxy_headers import proxy_headers_middleware
+ return proxy_headers_middleware(app, **kw)
+
+ def _callFUT(self, app, **kw):
+ response = DummyResponse()
+ environ = DummyEnviron(**kw)
+ def start_response(status, response_headers):
+ response.status = status
+ response.headers = response_headers
+ response.steps = list(app(environ, start_response))
+ response.body = b''.join(tobytes(s) for s in response.steps)
+ return response
+
+ def test_get_environment_values_w_scheme_override_untrusted(self):
+ inner = DummyApp()
+ app = self._makeOne(inner)
+ response = self._callFUT(app, headers={
+ 'X_FOO': 'BAR',
+ 'X_FORWARDED_PROTO': 'https',
+ })
+ self.assertEqual(response.status, '200 OK')
+ self.assertEqual(inner.environ['wsgi.url_scheme'], 'http')
+
+ def test_get_environment_values_w_scheme_override_trusted(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='192.168.1.1',
+ trusted_proxy_headers={'x-forwarded-proto'},
+ )
+ response = self._callFUT(
+ app,
+ addr=['192.168.1.1', 8080],
+ headers={
+ 'X_FOO': 'BAR',
+ 'X_FORWARDED_PROTO': 'https',
+ },
+ )
+
+ environ = inner.environ
+ self.assertEqual(response.status, '200 OK')
+ self.assertEqual(environ['SERVER_PORT'], '443')
+ self.assertEqual(environ['SERVER_NAME'], 'localhost')
+ self.assertEqual(environ['REMOTE_ADDR'], '192.168.1.1')
+ self.assertEqual(environ['HTTP_X_FOO'], 'BAR')
+
+ def test_get_environment_values_w_bogus_scheme_override(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='192.168.1.1',
+ trusted_proxy_headers={'x-forwarded-proto'},
+ )
+ response = self._callFUT(
+ app,
+ addr=['192.168.1.1', 80],
+ headers={
+ 'X_FOO': 'BAR',
+ 'X_FORWARDED_PROTO': 'http://p02n3e.com?url=http',
+ },
+ )
+ self.assertEqual(response.status, '400 Bad Request')
+ self.assertIn(b'Header "X-Forwarded-Proto" malformed', response.body)
+
+ def test_get_environment_warning_other_proxy_headers(self):
+ inner = DummyApp()
+ logger = DummyLogger()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='192.168.1.1',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'forwarded'},
+ log_untrusted=True,
+ logger=logger,
+ )
+ response = self._callFUT(
+ app,
+ addr=['192.168.1.1', 80],
+ headers={
+ 'X_FORWARDED_FOR': '[2001:db8::1]',
+ 'FORWARDED': 'For=198.51.100.2;host=example.com:8080;proto=https'
+ },
+ )
+ self.assertEqual(response.status, '200 OK')
+
+ self.assertEqual(len(logger.logged), 1)
+
+ environ = inner.environ
+ self.assertNotIn('HTTP_X_FORWARDED_FOR', environ)
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+ self.assertEqual(environ['wsgi.url_scheme'], 'https')
+
+ def test_get_environment_contains_all_headers_including_untrusted(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='192.168.1.1',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'x-forwarded-by'},
+ clear_untrusted=False,
+ )
+ headers_orig = {
+ 'X_FORWARDED_FOR': '198.51.100.2',
+ 'X_FORWARDED_BY': 'Waitress',
+ 'X_FORWARDED_PROTO': 'https',
+ 'X_FORWARDED_HOST': 'example.org',
+ }
+ response = self._callFUT(
+ app,
+ addr=['192.168.1.1', 80],
+ headers=headers_orig.copy(),
+ )
+ self.assertEqual(response.status, '200 OK')
+ environ = inner.environ
+ for k, expected in headers_orig.items():
+ result = environ['HTTP_%s' % k]
+ self.assertEqual(result, expected)
+
+ def test_get_environment_contains_only_trusted_headers(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='192.168.1.1',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'x-forwarded-by'},
+ clear_untrusted=True,
+ )
+ response = self._callFUT(
+ app,
+ addr=['192.168.1.1', 80],
+ headers={
+ 'X_FORWARDED_FOR': '198.51.100.2',
+ 'X_FORWARDED_BY': 'Waitress',
+ 'X_FORWARDED_PROTO': 'https',
+ 'X_FORWARDED_HOST': 'example.org',
+ },
+ )
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['HTTP_X_FORWARDED_BY'], 'Waitress')
+ self.assertNotIn('HTTP_X_FORWARDED_FOR', environ)
+ self.assertNotIn('HTTP_X_FORWARDED_PROTO', environ)
+ self.assertNotIn('HTTP_X_FORWARDED_HOST', environ)
+
+ def test_get_environment_clears_headers_if_untrusted_proxy(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='192.168.1.1',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'x-forwarded-by'},
+ clear_untrusted=True,
+ )
+ response = self._callFUT(
+ app,
+ addr=['192.168.1.255', 80],
+ headers={
+ 'X_FORWARDED_FOR': '198.51.100.2',
+ 'X_FORWARDED_BY': 'Waitress',
+ 'X_FORWARDED_PROTO': 'https',
+ 'X_FORWARDED_HOST': 'example.org',
+ },
+ )
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertNotIn('HTTP_X_FORWARDED_BY', environ)
+ self.assertNotIn('HTTP_X_FORWARDED_FOR', environ)
+ self.assertNotIn('HTTP_X_FORWARDED_PROTO', environ)
+ self.assertNotIn('HTTP_X_FORWARDED_HOST', environ)
+
+ def test_parse_proxy_headers_forwarded_for(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_headers={'x-forwarded-for'},
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '192.0.2.1'
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '192.0.2.1')
+
+ def test_parse_proxy_headers_forwarded_for_v6_missing_brackets(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_headers={'x-forwarded-for'},
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '2001:db8::0'
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '2001:db8::0')
+
+ def test_parse_proxy_headers_forwared_for_multiple(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=2,
+ trusted_proxy_headers={'x-forwarded-for'},
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1'
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+
+ def test_parse_forwarded_multiple_proxies_trust_only_two(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=2,
+ trusted_proxy_headers={'forwarded'},
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': (
+ 'For=192.0.2.1;host=fake.com, '
+ 'For=198.51.100.2;host=example.com:8080, '
+ 'For=203.0.113.1'
+ ),
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+
+ def test_parse_forwarded_multiple_proxies(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=2,
+ trusted_proxy_headers={'forwarded'},
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': (
+ 'for="[2001:db8::1]:3821";host="example.com:8443";proto="https", '
+ 'for=192.0.2.1;host="example.internal:8080"'
+ ),
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '2001:db8::1')
+ self.assertEqual(environ['REMOTE_PORT'], '3821')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:8443')
+ self.assertEqual(environ['SERVER_PORT'], '8443')
+ self.assertEqual(environ['wsgi.url_scheme'], 'https')
+
+ def test_parse_forwarded_multiple_proxies_minimal(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=2,
+ trusted_proxy_headers={'forwarded'},
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': (
+ 'for="[2001:db8::1]";proto="https", '
+ 'for=192.0.2.1;host="example.org"'
+ ),
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '2001:db8::1')
+ self.assertEqual(environ['SERVER_NAME'], 'example.org')
+ self.assertEqual(environ['HTTP_HOST'], 'example.org')
+ self.assertEqual(environ['SERVER_PORT'], '443')
+ self.assertEqual(environ['wsgi.url_scheme'], 'https')
+
+ def test_parse_proxy_headers_forwarded_host_with_port(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=2,
+ trusted_proxy_headers={
+ 'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host',
+ },
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
+ 'X_FORWARDED_PROTO': 'http',
+ 'X_FORWARDED_HOST': 'example.com:8080',
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+
+ def test_parse_proxy_headers_forwarded_host_without_port(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=2,
+ trusted_proxy_headers={
+ 'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host',
+ },
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
+ 'X_FORWARDED_PROTO': 'http',
+ 'X_FORWARDED_HOST': 'example.com',
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com')
+ self.assertEqual(environ['SERVER_PORT'], '80')
+
+ def test_parse_proxy_headers_forwarded_host_with_forwarded_port(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=2,
+ trusted_proxy_headers={
+ 'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host',
+ 'x-forwarded-port',
+ },
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
+ 'X_FORWARDED_PROTO': 'http',
+ 'X_FORWARDED_HOST': 'example.com',
+ 'X_FORWARDED_PORT': '8080'
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+
+ def test_parse_proxy_headers_forwarded_host_multiple_with_forwarded_port(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=2,
+ trusted_proxy_headers={
+ 'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host',
+ 'x-forwarded-port',
+ },
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
+ 'X_FORWARDED_PROTO': 'http',
+ 'X_FORWARDED_HOST': 'example.com, example.org',
+ 'X_FORWARDED_PORT': '8080'
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+
+ def test_parse_proxy_headers_forwarded_host_multiple_with_forwarded_port_limit_one_trusted(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={
+ 'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host',
+ 'x-forwarded-port',
+ },
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
+ 'X_FORWARDED_PROTO': 'http',
+ 'X_FORWARDED_HOST': 'example.com, example.org',
+ 'X_FORWARDED_PORT': '8080'
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '203.0.113.1')
+ self.assertEqual(environ['SERVER_NAME'], 'example.org')
+ self.assertEqual(environ['HTTP_HOST'], 'example.org:8080')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+
+ def test_parse_forwarded(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'forwarded'},
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': 'For=198.51.100.2:5858;host=example.com:8080;proto=https',
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+ self.assertEqual(environ['REMOTE_PORT'], '5858')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+ self.assertEqual(environ['wsgi.url_scheme'], 'https')
+
+ def test_parse_forwarded_empty_pair(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'forwarded'},
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': 'For=198.51.100.2;;proto=https;by=_unused',
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+
+ def test_parse_forwarded_pair_token_whitespace(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'forwarded'},
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': 'For=198.51.100.2; proto =https',
+ })
+ self.assertEqual(response.status, '400 Bad Request')
+ self.assertIn(b'Header "Forwarded" malformed', response.body)
+
+ def test_parse_forwarded_pair_value_whitespace(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'forwarded'},
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': 'For= "198.51.100.2"; proto =https',
+ })
+ self.assertEqual(response.status, '400 Bad Request')
+ self.assertIn(b'Header "Forwarded" malformed', response.body)
+
+ def test_parse_forwarded_pair_no_equals(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'forwarded'},
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': 'For'
+ })
+ self.assertEqual(response.status, '400 Bad Request')
+ self.assertIn(b'Header "Forwarded" malformed', response.body)
+
+ def test_parse_forwarded_warning_unknown_token(self):
+ inner = DummyApp()
+ logger = DummyLogger()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'forwarded'},
+ logger=logger,
+ )
+ response = self._callFUT(app, headers={
+ 'FORWARDED': (
+ 'For=198.51.100.2;host=example.com:8080;proto=https;'
+ 'unknown="yolo"'
+ ),
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ self.assertEqual(len(logger.logged), 1)
+ self.assertIn('Unknown Forwarded token', logger.logged[0])
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+ self.assertEqual(environ['wsgi.url_scheme'], 'https')
+
+ def test_parse_no_valid_proxy_headers(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_FOR': '198.51.100.2',
+ 'FORWARDED': 'For=198.51.100.2;host=example.com:8080;proto=https'
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['REMOTE_ADDR'], '127.0.0.1')
+ self.assertEqual(environ['SERVER_NAME'], 'localhost')
+ self.assertEqual(environ['HTTP_HOST'], '192.168.1.1:80')
+ self.assertEqual(environ['SERVER_PORT'], '8080')
+ self.assertEqual(environ['wsgi.url_scheme'], 'http')
+
+ def test_parse_multiple_x_forwarded_proto(self):
+ inner = DummyApp()
+ logger = DummyLogger()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'x-forwarded-proto'},
+ logger=logger,
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_PROTO': 'http, https',
+ })
+ self.assertEqual(response.status, '400 Bad Request')
+ self.assertIn(b'Header "X-Forwarded-Proto" malformed', response.body)
+
+ def test_parse_multiple_x_forwarded_port(self):
+ inner = DummyApp()
+ logger = DummyLogger()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={'x-forwarded-port'},
+ logger=logger,
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_PORT': '443, 80',
+ })
+ self.assertEqual(response.status, '400 Bad Request')
+ self.assertIn(b'Header "X-Forwarded-Port" malformed', response.body)
+
+ def test_parse_forwarded_port_wrong_proto_port_80(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={
+ 'x-forwarded-port', 'x-forwarded-host', 'x-forwarded-proto',
+ },
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_PORT': '80',
+ 'X_FORWARDED_PROTO': 'https',
+ 'X_FORWARDED_HOST': 'example.com',
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:80')
+ self.assertEqual(environ['SERVER_PORT'], '80')
+ self.assertEqual(environ['wsgi.url_scheme'], 'https')
+
+ def test_parse_forwarded_port_wrong_proto_port_443(self):
+ inner = DummyApp()
+ app = self._makeOne(
+ inner,
+ trusted_proxy='*',
+ trusted_proxy_count=1,
+ trusted_proxy_headers={
+ 'x-forwarded-port', 'x-forwarded-host', 'x-forwarded-proto',
+ },
+ )
+ response = self._callFUT(app, headers={
+ 'X_FORWARDED_PORT': '443',
+ 'X_FORWARDED_PROTO': 'http',
+ 'X_FORWARDED_HOST': 'example.com',
+ })
+ self.assertEqual(response.status, '200 OK')
+
+ environ = inner.environ
+ self.assertEqual(environ['SERVER_NAME'], 'example.com')
+ self.assertEqual(environ['HTTP_HOST'], 'example.com:443')
+ self.assertEqual(environ['SERVER_PORT'], '443')
+ self.assertEqual(environ['wsgi.url_scheme'], 'http')
+
+
+class DummyLogger(object):
+ def __init__(self):
+ self.logged = []
+
+ def warning(self, msg, *args):
+ self.logged.append(msg % args)
+
+ def exception(self, msg, *args):
+ self.logged.append(msg % args)
+
+
+class DummyApp(object):
+ def __call__(self, environ, start_response):
+ self.environ = environ
+ start_response('200 OK', [('Content-Type', 'text/plain')])
+ yield 'hello'
+
+
+class DummyResponse(object):
+ status = None
+ headers = None
+ body = None
+
+
+def DummyEnviron(
+ addr=('127.0.0.1', 8080),
+ scheme='http',
+ server='localhost',
+ headers=None,
+):
+ environ = {
+ 'REMOTE_ADDR': addr[0],
+ 'REMOTE_HOST': addr[0],
+ 'REMOTE_PORT': addr[1],
+ 'SERVER_PORT': str(addr[1]),
+ 'SERVER_NAME': server,
+ 'wsgi.url_scheme': scheme,
+ 'HTTP_HOST': '192.168.1.1:80',
+ }
+ if headers:
+ environ.update({
+ 'HTTP_' + key.upper().replace('-', '_'): value
+ for key, value in headers.items()
+ })
+ return environ
diff --git a/waitress/tests/test_task.py b/waitress/tests/test_task.py
index ffc34b7..a75b87f 100644
--- a/waitress/tests/test_task.py
+++ b/waitress/tests/test_task.py
@@ -770,603 +770,6 @@ class TestWSGITask(unittest.TestCase):
self.assertEqual(environ['wsgi.input_terminated'], True)
self.assertEqual(inst.environ, environ)
- def test_get_environment_values_w_scheme_override_untrusted(self):
- inst = self._makeOne()
- request = DummyParser()
- request.headers = {
- 'CONTENT_TYPE': 'abc',
- 'CONTENT_LENGTH': '10',
- 'X_FOO': 'BAR',
- 'X_FORWARDED_PROTO': 'https',
- 'CONNECTION': 'close',
- }
- request.query = 'abc'
- inst.request = request
- environ = inst.get_environment()
- self.assertEqual(environ['wsgi.url_scheme'], 'http')
-
- def test_get_environment_values_w_scheme_override_trusted(self):
- import sys
- inst = self._makeOne()
- inst.channel.addr = ['192.168.1.1', 8080]
- inst.channel.server.adj.trusted_proxy = '192.168.1.1'
- inst.channel.server.adj.trusted_proxy_headers = {'x-forwarded-proto'}
- request = DummyParser()
- request.headers = {
- 'CONTENT_TYPE': 'abc',
- 'CONTENT_LENGTH': '10',
- 'X_FOO': 'BAR',
- 'X_FORWARDED_PROTO': 'https',
- 'CONNECTION': 'close',
- }
- request.query = 'abc'
- inst.request = request
- environ = inst.get_environment()
-
- # nail the keys of environ
- self.assertEqual(sorted(environ.keys()), [
- 'CONTENT_LENGTH', 'CONTENT_TYPE', 'HTTP_CONNECTION', 'HTTP_X_FOO',
- 'HTTP_X_FORWARDED_PROTO', 'PATH_INFO', 'QUERY_STRING',
- 'REMOTE_ADDR', 'REMOTE_HOST', 'REQUEST_METHOD', 'SCRIPT_NAME',
- 'SERVER_NAME', 'SERVER_PORT', 'SERVER_PROTOCOL', 'SERVER_SOFTWARE',
- 'wsgi.errors', 'wsgi.file_wrapper', 'wsgi.input',
- 'wsgi.input_terminated', 'wsgi.multiprocess', 'wsgi.multithread',
- 'wsgi.run_once', 'wsgi.url_scheme', 'wsgi.version'
- ])
-
- self.assertEqual(environ['REQUEST_METHOD'], 'GET')
- self.assertEqual(environ['SERVER_PORT'], '443')
- self.assertEqual(environ['SERVER_NAME'], 'localhost')
- self.assertEqual(environ['SERVER_SOFTWARE'], 'waitress')
- self.assertEqual(environ['SERVER_PROTOCOL'], 'HTTP/1.0')
- self.assertEqual(environ['SCRIPT_NAME'], '')
- self.assertEqual(environ['HTTP_CONNECTION'], 'close')
- self.assertEqual(environ['PATH_INFO'], '/')
- self.assertEqual(environ['QUERY_STRING'], 'abc')
- self.assertEqual(environ['REMOTE_ADDR'], '192.168.1.1')
- self.assertEqual(environ['CONTENT_TYPE'], 'abc')
- self.assertEqual(environ['CONTENT_LENGTH'], '10')
- self.assertEqual(environ['HTTP_X_FOO'], 'BAR')
- self.assertEqual(environ['wsgi.version'], (1, 0))
- self.assertEqual(environ['wsgi.url_scheme'], 'https')
- self.assertEqual(environ['wsgi.errors'], sys.stderr)
- self.assertEqual(environ['wsgi.multithread'], True)
- self.assertEqual(environ['wsgi.multiprocess'], False)
- self.assertEqual(environ['wsgi.run_once'], False)
- self.assertEqual(environ['wsgi.input'], 'stream')
- self.assertEqual(environ['wsgi.input_terminated'], True)
- self.assertEqual(inst.environ, environ)
-
- def test_get_environment_values_w_bogus_scheme_override(self):
- inst = self._makeOne()
- inst.channel.addr = ['192.168.1.1', 80]
- inst.channel.server.adj.trusted_proxy = '192.168.1.1'
- inst.channel.server.adj.trusted_proxy_headers = {'x-forwarded-proto'}
- request = DummyParser()
- request.headers = {
- 'CONTENT_TYPE': 'abc',
- 'CONTENT_LENGTH': '10',
- 'X_FOO': 'BAR',
- 'X_FORWARDED_PROTO': 'http://p02n3e.com?url=http',
- 'CONNECTION': 'close',
- }
- request.query = 'abc'
- inst.request = request
- self.assertRaises(ValueError, inst.get_environment)
-
- def test_get_environment_warning_other_proxy_headers(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- inst.request.headers = {
- 'X_FORWARDED_FOR': '[2001:db8::1]',
- 'FORWARDED': 'For=198.51.100.2;host=example.com:8080;proto=https'
- }
- inst.channel.addr = ['192.168.1.1', 80]
- inst.channel.server.adj.trusted_proxy = '192.168.1.1'
- inst.channel.server.adj.trusted_proxy_count = 1
- inst.channel.server.adj.trusted_proxy_headers = {'forwarded'}
- inst.channel.server.adj.log_untrusted_proxy_headers = True
- environ = inst.get_environment()
-
- self.assertEqual(len(inst.logger.logged), 1)
- self.assertNotIn('HTTP_X_FORWARDED_FOR', environ)
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
- self.assertEqual(environ['SERVER_PORT'], '8080')
- self.assertEqual(environ['wsgi.url_scheme'], 'https')
-
- def test_get_environment_contains_all_headers_including_untrusted(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- inst.request.headers = {
- 'X_FORWARDED_FOR': '198.51.100.2',
- 'X_FORWARDED_BY': 'Waitress',
- 'X_FORWARDED_PROTO': 'https',
- 'X_FORWARDED_HOST': 'example.org',
- }
- headers_orig = inst.request.headers.copy()
- inst.channel.addr = ['192.168.1.1', 80]
- inst.channel.server.adj.trusted_proxy = '192.168.1.1'
- inst.channel.server.adj.trusted_proxy_count = 1
- inst.channel.server.adj.trusted_proxy_headers = {'x-forwarded-by'}
- inst.channel.server.adj.clear_untrusted_proxy_headers = False
- environ = inst.get_environment()
-
- for k, expected in headers_orig.items():
- result = environ['HTTP_%s' % k]
- self.assertEqual(result, expected)
-
- def test_get_environment_contains_only_trusted_headers(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- inst.request.headers = {
- 'X_FORWARDED_FOR': '198.51.100.2',
- 'X_FORWARDED_BY': 'Waitress',
- 'X_FORWARDED_PROTO': 'https',
- 'X_FORWARDED_HOST': 'example.org',
- }
- inst.channel.addr = ['192.168.1.1', 80]
- inst.channel.server.adj.trusted_proxy = '192.168.1.1'
- inst.channel.server.adj.trusted_proxy_count = 1
- inst.channel.server.adj.trusted_proxy_headers = {'x-forwarded-by'}
- inst.channel.server.adj.clear_untrusted_proxy_headers = True
- environ = inst.get_environment()
-
- self.assertEqual(environ['HTTP_X_FORWARDED_BY'], 'Waitress')
- self.assertNotIn('HTTP_X_FORWARDED_FOR', environ)
- self.assertNotIn('HTTP_X_FORWARDED_PROTO', environ)
- self.assertNotIn('HTTP_X_FORWARDED_HOST', environ)
-
- def test_get_environment_clears_headers_if_untrusted_proxy(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- inst.request.headers = {
- 'X_FORWARDED_FOR': '198.51.100.2',
- 'X_FORWARDED_BY': 'Waitress',
- 'X_FORWARDED_PROTO': 'https',
- 'X_FORWARDED_HOST': 'example.org',
- }
- inst.channel.addr = ['192.168.1.255', 80]
- inst.channel.server.adj.trusted_proxy = '192.168.1.1'
- inst.channel.server.adj.trusted_proxy_count = 1
- inst.channel.server.adj.trusted_proxy_headers = {'x-forwarded-by'}
- inst.channel.server.adj.clear_untrusted_proxy_headers = True
- environ = inst.get_environment()
-
- self.assertNotIn('HTTP_X_FORWARDED_BY', environ)
- self.assertNotIn('HTTP_X_FORWARDED_FOR', environ)
- self.assertNotIn('HTTP_X_FORWARDED_PROTO', environ)
- self.assertNotIn('HTTP_X_FORWARDED_HOST', environ)
-
- def test_parse_proxy_headers_forwarded_for(self):
- inst = self._makeOne()
-
- headers = {
- 'X_FORWARDED_FOR': '192.0.2.1'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'x-forwarded-for'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '192.0.2.1')
-
- def test_parse_proxy_headers_forwarded_for_v6_missing_brackets(self):
- inst = self._makeOne()
-
- headers = {
- 'X_FORWARDED_FOR': '2001:db8::0'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'x-forwarded-for'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '2001:db8::0')
-
- def test_parse_proxy_headers_forwared_for_multiple(self):
- inst = self._makeOne()
-
- headers = {
- 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=2,
- trusted_proxy_headers={'x-forwarded-for'}
- )
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
-
- def test_parse_forwarded_multiple_proxies_trust_only_two(self):
- inst = self._makeOne()
-
- headers = {
- 'FORWARDED': 'For=192.0.2.1;host=fake.com, For=198.51.100.2;host=example.com:8080, For=203.0.113.1'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=2,
- trusted_proxy_headers={'forwarded'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
- self.assertEqual(environ['SERVER_PORT'], '8080')
-
- def test_parse_forwarded_multiple_proxies(self):
- inst = self._makeOne()
-
- headers = {
- 'FORWARDED': 'for="[2001:db8::1]:3821";host="example.com:8443";proto="https", for=192.0.2.1;host="example.internal:8080"'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=2,
- trusted_proxy_headers={'forwarded'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '2001:db8::1')
- self.assertEqual(environ['REMOTE_PORT'], '3821')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:8443')
- self.assertEqual(environ['SERVER_PORT'], '8443')
- self.assertEqual(environ['wsgi.url_scheme'], 'https')
-
- def test_parse_forwarded_multiple_proxies_minimal(self):
- inst = self._makeOne()
-
- headers = {
- 'FORWARDED': 'for="[2001:db8::1]";proto="https", for=192.0.2.1;host="example.org"'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=2,
- trusted_proxy_headers={'forwarded'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '2001:db8::1')
- self.assertEqual(environ['SERVER_NAME'], 'example.org')
- self.assertEqual(environ['HTTP_HOST'], 'example.org')
- self.assertEqual(environ['SERVER_PORT'], '443')
- self.assertEqual(environ['wsgi.url_scheme'], 'https')
-
- def test_parse_proxy_headers_forwarded_host_with_port(self):
- inst = self._makeOne()
-
- headers = {
- 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
- 'X_FORWARDED_PROTO': 'http',
- 'X_FORWARDED_HOST': 'example.com:8080',
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=2,
- trusted_proxy_headers={'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
- self.assertEqual(environ['SERVER_PORT'], '8080')
-
- def test_parse_proxy_headers_forwarded_host_without_port(self):
- inst = self._makeOne()
-
- headers = {
- 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
- 'X_FORWARDED_PROTO': 'http',
- 'X_FORWARDED_HOST': 'example.com',
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=2,
- trusted_proxy_headers={'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com')
- self.assertEqual(environ['SERVER_PORT'], '80')
-
- def test_parse_proxy_headers_forwarded_host_with_forwarded_port(self):
- inst = self._makeOne()
-
- headers = {
- 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
- 'X_FORWARDED_PROTO': 'http',
- 'X_FORWARDED_HOST': 'example.com',
- 'X_FORWARDED_PORT': '8080'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=2,
- trusted_proxy_headers={'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host', 'x-forwarded-port'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
- self.assertEqual(environ['SERVER_PORT'], '8080')
-
- def test_parse_proxy_headers_forwarded_host_multiple_with_forwarded_port(self):
- inst = self._makeOne()
-
- headers = {
- 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
- 'X_FORWARDED_PROTO': 'http',
- 'X_FORWARDED_HOST': 'example.com, example.org',
- 'X_FORWARDED_PORT': '8080'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=2,
- trusted_proxy_headers={'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host', 'x-forwarded-port'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
- self.assertEqual(environ['SERVER_PORT'], '8080')
-
- def test_parse_proxy_headers_forwarded_host_multiple_with_forwarded_port_limit_one_trusted(self):
- inst = self._makeOne()
-
- headers = {
- 'X_FORWARDED_FOR': '192.0.2.1, 198.51.100.2, 203.0.113.1',
- 'X_FORWARDED_PROTO': 'http',
- 'X_FORWARDED_HOST': 'example.com, example.org',
- 'X_FORWARDED_PORT': '8080'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'x-forwarded-for', 'x-forwarded-proto', 'x-forwarded-host', 'x-forwarded-port'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '203.0.113.1')
- self.assertEqual(environ['SERVER_NAME'], 'example.org')
- self.assertEqual(environ['HTTP_HOST'], 'example.org:8080')
- self.assertEqual(environ['SERVER_PORT'], '8080')
-
- def test_parse_forwarded(self):
- inst = self._makeOne()
-
- headers = {
- 'FORWARDED': 'For=198.51.100.2:5858;host=example.com:8080;proto=https'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'forwarded'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
- self.assertEqual(environ['REMOTE_PORT'], '5858')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
- self.assertEqual(environ['SERVER_PORT'], '8080')
- self.assertEqual(environ['wsgi.url_scheme'], 'https')
-
- def test_parse_forwarded_empty_pair(self):
- inst = self._makeOne()
-
- headers = {
- 'FORWARDED': 'For=198.51.100.2;;proto=https;by=_unused'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'forwarded'}
- )
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
-
- def test_parse_forwarded_pair_token_whitespace(self):
- inst = self._makeOne()
-
- headers = {
- 'FORWARDED': 'For=198.51.100.2; proto =https'
- }
- environ = {}
-
- with self.assertRaises(ValueError):
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'forwarded'}
- )
-
- def test_parse_forwarded_pair_value_whitespace(self):
- inst = self._makeOne()
-
- headers = {
- 'FORWARDED': 'For= "198.51.100.2"; proto =https'
- }
- environ = {}
-
- with self.assertRaises(ValueError):
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'forwarded'}
- )
-
- def test_parse_forwarded_pair_no_equals(self):
- inst = self._makeOne()
-
- headers = {
- 'FORWARDED': 'For'
- }
- environ = {}
-
- with self.assertRaises(ValueError):
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'forwarded'}
- )
-
- def test_parse_forwarded_warning_unknown_token(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- headers = {
- 'FORWARDED': 'For=198.51.100.2;host=example.com:8080;proto=https;unknown="yolo"'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'forwarded'}
- )
-
- self.assertEqual(len(inst.logger.logged), 1)
- self.assertIn('Unknown Forwarded token', inst.logger.logged[0])
-
- self.assertEqual(environ['REMOTE_ADDR'], '198.51.100.2')
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:8080')
- self.assertEqual(environ['SERVER_PORT'], '8080')
- self.assertEqual(environ['wsgi.url_scheme'], 'https')
-
- def test_parse_no_valid_proxy_headers(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- headers = {
- 'X_FORWARDED_FOR': '198.51.100.2',
- 'FORWARDED': 'For=198.51.100.2;host=example.com:8080;proto=https'
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- )
-
- self.assertEqual(environ, {})
-
- def test_parse_multiple_x_forwarded_proto(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- headers = {
- 'X_FORWARDED_PROTO': 'http, https',
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'x-forwarded-proto'}
- )
-
- self.assertEqual(environ, {})
- self.assertEqual(len(inst.logger.logged), 1)
- self.assertIn("Found multiple values in X-Forwarded-Proto", inst.logger.logged[0])
-
- def test_parse_multiple_x_forwarded_port(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- headers = {
- 'X_FORWARDED_PORT': '443, 80',
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'x-forwarded-port'}
- )
-
- self.assertEqual(environ, {})
- self.assertEqual(len(inst.logger.logged), 1)
- self.assertIn("Found multiple values in X-Forwarded-Port", inst.logger.logged[0])
-
- def test_parse_forwarded_port_wrong_proto_port_80(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- headers = {
- 'X_FORWARDED_PORT': '80',
- 'X_FORWARDED_PROTO': 'https',
- 'X_FORWARDED_HOST': 'example.com',
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'x-forwarded-port', 'x-forwarded-host', 'x-forwarded-proto'}
- )
-
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:80')
- self.assertEqual(environ['SERVER_PORT'], '80')
- self.assertEqual(environ['wsgi.url_scheme'], 'https')
-
- def test_parse_forwarded_port_wrong_proto_port_443(self):
- inst = self._makeOne()
- inst.logger = DummyLogger()
-
- headers = {
- 'X_FORWARDED_PORT': '443',
- 'X_FORWARDED_PROTO': 'http',
- 'X_FORWARDED_HOST': 'example.com',
- }
- environ = {}
- inst.parse_proxy_headers(
- environ,
- headers,
- trusted_proxy_count=1,
- trusted_proxy_headers={'x-forwarded-port', 'x-forwarded-host', 'x-forwarded-proto'}
- )
-
- self.assertEqual(environ['SERVER_NAME'], 'example.com')
- self.assertEqual(environ['HTTP_HOST'], 'example.com:443')
- self.assertEqual(environ['SERVER_PORT'], '443')
- self.assertEqual(environ['wsgi.url_scheme'], 'http')
-
class TestErrorTask(unittest.TestCase):
@@ -1375,10 +778,17 @@ class TestErrorTask(unittest.TestCase):
channel = DummyChannel()
if request is None:
request = DummyParser()
- request.error = DummyError()
+ request.error = self._makeDummyError()
from waitress.task import ErrorTask
return ErrorTask(channel, request)
+ def _makeDummyError(self):
+ from waitress.utilities import Error
+ e = Error('body')
+ e.code = 432
+ e.reason = 'Too Ugly'
+ return e
+
def test_execute_http_10(self):
inst = self._makeOne()
inst.execute()
@@ -1442,12 +852,6 @@ class TestErrorTask(unittest.TestCase):
self.assertEqual(lines[6], b'body')
self.assertEqual(lines[7], b'(generated by waitress)')
-
-class DummyError(object):
- code = '432'
- reason = 'Too Ugly'
- body = 'body'
-
class DummyTask(object):
serviced = False
cancelled = False
@@ -1464,11 +868,6 @@ class DummyAdj(object):
host = '127.0.0.1'
port = 80
url_prefix = ''
- trusted_proxy = None
- trusted_proxy_count = 1
- trusted_proxy_headers = set()
- log_untrusted_proxy_headers = True
- clear_untrusted_proxy_headers = True
class DummyServer(object):
server_name = 'localhost'