Fixed logic for handling headers on redirect

scrapy/downloadermiddlewares/redirect.py

@@ -20,14 +20,37 @@ def _build_redirect_request(source_request, *, url, **kwargs):
     has_cookie_header = "Cookie" in redirect_request.headers
     has_authorization_header = "Authorization" in redirect_request.headers
     if has_cookie_header or has_authorization_header:
-        source_request_netloc = urlparse_cached(source_request).netloc
-        redirect_request_netloc = urlparse_cached(redirect_request).netloc
-        if source_request_netloc != redirect_request_netloc:
+        default_ports = {"http": 80, "https": 443}
+
+        parsed_source_request = urlparse_cached(source_request)
+        source_scheme, source_host, source_port = (
+            parsed_source_request.scheme,
+            parsed_source_request.hostname,
+            parsed_source_request.port
+            or default_ports.get(parsed_source_request.scheme),
+        )
+
+        parsed_redirect_request = urlparse_cached(redirect_request)
+        redirect_scheme, redirect_host, redirect_port = (
+            parsed_redirect_request.scheme,
+            parsed_redirect_request.hostname,
+            parsed_redirect_request.port
+            or default_ports.get(parsed_redirect_request.scheme),
+        )
+
+        if source_scheme != redirect_scheme or source_host != redirect_host:
             if has_cookie_header:
                 del redirect_request.headers["Cookie"]
+
+        if (
+            source_scheme != redirect_scheme
+            or source_host != redirect_host
+            or source_port != redirect_port
+        ):
             # https://fetch.spec.whatwg.org/#ref-for-cors-non-wildcard-request-header-name
             if has_authorization_header:
                 del redirect_request.headers["Authorization"]
+
     return redirect_request
 
 

tests/test_downloadermiddleware_redirect.py

@@ -247,11 +247,14 @@ class RedirectMiddlewareTest(unittest.TestCase):
         perc_encoded_utf8_url = "http://scrapytest.org/a%C3%A7%C3%A3o"
         self.assertEqual(perc_encoded_utf8_url, req_result.url)
 
-    def test_cross_domain_header_dropping(self):
+    def test_cross_origin_header_dropping(self):
         safe_headers = {"A": "B"}
+        cookie_header = {"Cookie": "a=b"}
+        authorization_header = {"Authorization": "Bearer 123456"}
+
         original_request = Request(
             "https://example.com",
-            headers={"Cookie": "a=b", "Authorization": "a", **safe_headers},
+            headers={**safe_headers, **cookie_header, **authorization_header},
         )
 
         internal_response = Response(
@@ -265,6 +268,33 @@ class RedirectMiddlewareTest(unittest.TestCase):
         self.assertIsInstance(internal_redirect_request, Request)
         self.assertEqual(original_request.headers, internal_redirect_request.headers)
 
+        default_port_response = Response(
+            "https://example.com",
+            headers={"Location": "https://example.com:443/a"},
+            status=301,
+        )
+        default_port_redirect_request = self.mw.process_response(
+            original_request, default_port_response, self.spider
+        )
+        self.assertIsInstance(default_port_redirect_request, Request)
+        self.assertEqual(
+            original_request.headers, default_port_redirect_request.headers
+        )
+
+        different_port_response = Response(
+            "https://example.com",
+            headers={"Location": "https://example.com:8080/a"},
+            status=301,
+        )
+        different_port_redirect_request = self.mw.process_response(
+            original_request, different_port_response, self.spider
+        )
+        self.assertIsInstance(different_port_redirect_request, Request)
+        self.assertEqual(
+            {**safe_headers, **cookie_header},
+            different_port_redirect_request.headers.to_unicode_dict(),
+        )
+
         external_response = Response(
             "https://example.com",
             headers={"Location": "https://example.org/a"},
@@ -278,6 +308,19 @@ class RedirectMiddlewareTest(unittest.TestCase):
             safe_headers, external_redirect_request.headers.to_unicode_dict()
         )
 
+        downgrade_response = Response(
+            "https://example.com",
+            headers={"Location": "http://example.com/a"},
+            status=301,
+        )
+        downgrade_redirect_request = self.mw.process_response(
+            original_request, downgrade_response, self.spider
+        )
+        self.assertIsInstance(downgrade_redirect_request, Request)
+        self.assertEqual(
+            safe_headers, downgrade_redirect_request.headers.to_unicode_dict()
+        )
+
 
 class MetaRefreshMiddlewareTest(unittest.TestCase):
     def setUp(self):