test_proxy_headers.py

import unittest


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(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")

    def test_parse_forwarded_for_bad_quote(self):
        inner = DummyApp()
        app = self._makeOne(
            inner,
            trusted_proxy="*",
            trusted_proxy_count=1,
            trusted_proxy_headers={"x-forwarded-for"},
        )
        response = self._callFUT(app, headers={"X_FORWARDED_FOR": '"foo'})
        self.assertEqual(response.status, "400 Bad Request")
        self.assertIn(b'Header "X-Forwarded-For" malformed', response.body)

    def test_parse_forwarded_host_bad_quote(self):
        inner = DummyApp()
        app = self._makeOne(
            inner,
            trusted_proxy="*",
            trusted_proxy_count=1,
            trusted_proxy_headers={"x-forwarded-host"},
        )
        response = self._callFUT(app, headers={"X_FORWARDED_HOST": '"foo'})
        self.assertEqual(response.status, "400 Bad Request")
        self.assertIn(b'Header "X-Forwarded-Host" malformed', response.body)


class DummyLogger:
    def __init__(self):
        self.logged = []

    def warning(self, msg, *args):
        self.logged.append(msg % args)


class DummyApp:
    def __call__(self, environ, start_response):
        self.environ = environ
        start_response("200 OK", [("Content-Type", "text/plain")])
        yield b"hello"


class DummyResponse:
    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