Fixed #31354: Properly reconstruct host header in presence of X-Forwarded-* header.

django/http/request.py

@@ -31,9 +31,7 @@ from django.utils.http import is_same_domain, parse_header_parameters
 from django.utils.regex_helper import _lazy_re_compile
 
 RAISE_ERROR = object()
-host_validation_re = _lazy_re_compile(
+host_validation_re = _lazy_re_compile(r"^([a-z0-9.-]+|\[[a-f0-9]*:[a-f0-9\.:]+\])(?::(\d+))?$")
-    r"^([a-z0-9.-]+|\[[a-f0-9]*:[a-f0-9.:]+\])(?::([0-9]+))?$"
-)
 
 
 class UnreadablePostError(OSError):
@@ -145,38 +143,48 @@ class HttpRequest:
             else:
                 self.encoding = self.content_params["charset"]
 
-    def _get_raw_host(self):
+    @cached_property
-        """
+    def _parsed_host_header(self):
-        Return the HTTP host using the environment or request headers. Skip
+        use_x_fw_host = settings.USE_X_FORWARDED_HOST
-        allowed hosts protection, so may return an insecure host.
+        use_x_fw_port = settings.USE_X_FORWARDED_PORT
-        """
+
-        # We try three options, in order of decreasing preference.
+        port_in_x_fw_host = False
-        if settings.USE_X_FORWARDED_HOST and ("HTTP_X_FORWARDED_HOST" in self.META):
+        default_port = ('443' if self.is_secure() else '80')
-            host = self.META["HTTP_X_FORWARDED_HOST"]
+
-        elif "HTTP_HOST" in self.META:
+        if use_x_fw_host and 'HTTP_X_FORWARDED_HOST' in self.META:
-            host = self.META["HTTP_HOST"]
+            host, port = _parse_host_header(self.META['HTTP_X_FORWARDED_HOST'])
+            port_in_x_fw_host = port != ''
+        elif 'HTTP_HOST' in self.META:
+            host, port = _parse_host_header(self.META['HTTP_HOST'])
         else:
             # Reconstruct the host using the algorithm from PEP 333.
-            host = self.META["SERVER_NAME"]
+            host, port = self.META['SERVER_NAME'], str(self.META['SERVER_PORT'])
-            server_port = self.get_port()
+            if port == default_port:
-            if server_port != ("443" if self.is_secure() else "80"):
+                port = ''
-                host = "%s:%s" % (host, server_port)
+
-        return host
+        if use_x_fw_port and 'HTTP_X_FORWARDED_PORT' in self.META:
+            if port_in_x_fw_host:
+                raise ImproperlyConfigured('HTTP_X_FORWARDED_HOST contains a port number '
+                                           'and USE_X_FORWARDED_PORT is set to True')
+            port = self.META['HTTP_X_FORWARDED_PORT']
+
+        reconstructed = '%s:%s' % (host, port) if port else host
+        return host, port or default_port, reconstructed
 
     def get_host(self):
         """Return the HTTP host using the environment or request headers."""
-        host = self._get_raw_host()
+        _, _, host_header = self._parsed_host_header
 
         # Allow variants of localhost if ALLOWED_HOSTS is empty and DEBUG=True.
         allowed_hosts = settings.ALLOWED_HOSTS
         if settings.DEBUG and not allowed_hosts:
             allowed_hosts = [".localhost", "127.0.0.1", "[::1]"]
 
-        domain, port = split_domain_port(host)
+        domain, port = split_domain_port(host_header)
         if domain and validate_host(domain, allowed_hosts):
-            return host
+            return host_header
         else:
-            msg = "Invalid HTTP_HOST header: %r." % host
+            msg = "Invalid HTTP_HOST header: %r." % host_header
             if domain:
                 msg += " You may need to add %r to ALLOWED_HOSTS." % domain
             else:
@@ -187,11 +195,8 @@ class HttpRequest:
 
     def get_port(self):
         """Return the port number for the request as a string."""
-        if settings.USE_X_FORWARDED_PORT and "HTTP_X_FORWARDED_PORT" in self.META:
+        _, port, _ = self._parsed_host_header
-            port = self.META["HTTP_X_FORWARDED_PORT"]
+        return port
-        else:
-            port = self.META["SERVER_PORT"]
-        return str(port)
 
     def get_full_path(self, force_append_slash=False):
         return self._get_full_path(self.path, force_append_slash)
@@ -236,6 +241,18 @@ class HttpRequest:
                 raise
         return value
 
+    def get_raw_uri(self):
+        """
+        Return an absolute URI from variables available in this request. Skip
+        allowed hosts protection, so may return insecure URI.
+        """
+        _, _, host_header = self._parsed_host_header
+        return '{scheme}://{host}{path}'.format(
+            scheme=self.scheme,
+            host=host_header,
+            path=self.get_full_path(),
+        )
+
     def build_absolute_uri(self, location=None):
         """
         Build an absolute URI from the location and the variables available in
@@ -763,6 +780,20 @@ def bytes_to_text(s, encoding):
         return s
 
 
+def _parse_host_header(host_header):
+    """
+    Returns a (domain, port) tuple for a given host.
+
+    Neither domain name nor port are validated.
+    """
+
+    if host_header[-1] == ']':
+        # It's an IPv6 address without a port.
+        return host_header, ''
+    bits = host_header.rsplit(':', 1)
+    return tuple(bits) if len(bits) == 2 else (bits[0], '')
+
+
 def split_domain_port(host):
     """
     Return a (domain, port) tuple from a given host.
@@ -770,11 +801,17 @@ def split_domain_port(host):
     Returned domain is lowercased. If the host is invalid, the domain will be
     empty.
     """
-    if match := host_validation_re.fullmatch(host.lower()):
+    host = host.lower()
-        domain, port = match.groups(default="")
+
+    host_match = host_validation_re.match(host)
+    if not host_match:
+        return '', ''
+
+    domain, port = host_match.groups()
+    port = port or ''
     # Remove a trailing dot (if present) from the domain.
-        return domain.removesuffix("."), port
+    domain = domain[:-1] if domain.endswith('.') else domain
-    return "", ""
+    return domain, port
 
 
 def validate_host(host, allowed_hosts):

tests/csrf_tests/tests.py

@@ -831,8 +831,8 @@ class CsrfViewMiddlewareTestMixin(CsrfFunctionTestMixin):
     def _test_https_good_referer_matches_cookie_domain(self):
         req = self._get_POST_request_with_token()
         req._is_secure_override = True
-        req.META["HTTP_REFERER"] = "https://foo.example.com/"
+        req.META['HTTP_HOST'] = 'www.example.com'
-        req.META["SERVER_PORT"] = "443"
+        req.META['HTTP_REFERER'] = 'https://foo.example.com/'
         mw = CsrfViewMiddleware(post_form_view)
         mw.process_request(req)
         response = mw.process_view(req, post_form_view, (), {})
@@ -841,9 +841,8 @@ class CsrfViewMiddlewareTestMixin(CsrfFunctionTestMixin):
     def _test_https_good_referer_matches_cookie_domain_with_different_port(self):
         req = self._get_POST_request_with_token()
         req._is_secure_override = True
-        req.META["HTTP_HOST"] = "www.example.com"
+        req.META['HTTP_HOST'] = 'www.example.com:4443'
-        req.META["HTTP_REFERER"] = "https://foo.example.com:4443/"
+        req.META['HTTP_REFERER'] = 'https://foo.example.com:4443/'
-        req.META["SERVER_PORT"] = "4443"
         mw = CsrfViewMiddleware(post_form_view)
         mw.process_request(req)
         response = mw.process_view(req, post_form_view, (), {})

tests/sites_tests/tests.py

@@ -106,37 +106,42 @@ class SitesFrameworkTests(TestCase):
 
     @override_settings(SITE_ID=None, ALLOWED_HOSTS=["example.com", "example.net"])
     def test_get_current_site_no_site_id_and_handle_port_fallback(self):
-        request = HttpRequest()
         s1 = self.site
         s2 = Site.objects.create(domain="example.com:80", name="example.com:80")
 
         # Host header without port
-        request.META = {"HTTP_HOST": "example.com"}
+        request = HttpRequest()
+        request.META = {'HTTP_HOST': 'example.com'}
         site = get_current_site(request)
         self.assertEqual(site, s1)
 
         # Host header with port - match, no fallback without port
-        request.META = {"HTTP_HOST": "example.com:80"}
+        request = HttpRequest()
+        request.META = {'HTTP_HOST': 'example.com:80'}
         site = get_current_site(request)
         self.assertEqual(site, s2)
 
         # Host header with port - no match, fallback without port
-        request.META = {"HTTP_HOST": "example.com:81"}
+        request = HttpRequest()
+        request.META = {'HTTP_HOST': 'example.com:81'}
         site = get_current_site(request)
         self.assertEqual(site, s1)
 
         # Host header with non-matching domain
-        request.META = {"HTTP_HOST": "example.net"}
+        request = HttpRequest()
+        request.META = {'HTTP_HOST': 'example.net'}
         with self.assertRaises(ObjectDoesNotExist):
             get_current_site(request)
 
         # Ensure domain for RequestSite always matches host header
-        with self.modify_settings(INSTALLED_APPS={"remove": "django.contrib.sites"}):
+        with self.modify_settings(INSTALLED_APPS={'remove': 'django.contrib.sites'}):
-            request.META = {"HTTP_HOST": "example.com"}
+            request = HttpRequest()
+            request.META = {'HTTP_HOST': 'example.com'}
             site = get_current_site(request)
             self.assertEqual(site.name, "example.com")
 
-            request.META = {"HTTP_HOST": "example.com:80"}
+            request = HttpRequest()
+            request.META = {'HTTP_HOST': 'example.com:80'}
             site = get_current_site(request)
             self.assertEqual(site.name, "example.com:80")