Drop connection headers on cross-host redirect in HttpAsyncHook (#70164)

devel-common/src/tests_common/test_utils/aiohttp.py

@@ -40,13 +40,19 @@ def __init__(
         method: str = "GET",
         url: str = "http://example.com",
         reason: str | None = None,
+        headers: dict[str, str] | None = None,
     ) -> None:
         self.status = status
         self._payload = payload
         self._text_payload = text_payload
         self._method = method
         self._url = url
         self._reason = reason
+        self.headers = headers or {}
+
+    async def release(self) -> None:
+        """No-op for unit tests; aiohttp frees the underlying connection."""
+        pass
 
     @property
     def reason(self) -> str:

providers/http/src/airflow/providers/http/hooks/http.py

@@ -512,19 +512,63 @@ async def run(
         url = _url_from_endpoint(self.base_url, endpoint)
         merged_headers = {**(self.headers or {}), **(headers or {})}
         extra_options = {**(self.extra_options or {}), **(extra_options or {})}
+        # Track which headers came from the connection so they can be dropped
+        # on a cross-host redirect (aiohttp only strips Authorization).
+        connection_header_names = set(self.headers.keys()) if self.headers else set()
 
         async def request_func() -> ClientResponse:
-            response = await self._request(
-                url,
-                params=data if self.method == "GET" else None,
-                data=data if self.method in {"POST", "PUT", "PATCH"} else None,
-                json=json,
-                headers=merged_headers,
-                auth=self.auth,
-                **extra_options,
+            nonlocal url
+            # aiohttp does not drop custom headers on cross-host redirects,
+            # so we disable automatic redirect following and walk the chain
+            # manually, dropping connection-supplied headers when the target
+            # changes host.
+            max_redirects = 5
+            current_headers = merged_headers
+            remaining_connection_headers = connection_header_names.copy()
+
+            for _ in range(max_redirects):
+                response = await self._request(
+                    url,
+                    params=data if self.method == "GET" else None,
+                    data=data if self.method in {"POST", "PUT", "PATCH"} else None,
+                    json=json,
+                    headers=current_headers,
+                    auth=self.auth,
+                    allow_redirects=False,
+                    **extra_options,
+                )
+
+                if response.status in (301, 302, 303, 307, 308):
+                    redirect_url = response.headers.get("Location")
+                    if not redirect_url:
+                        response.raise_for_status()
+                        return response
+
+                    # Resolve relative Location URLs against the current URL.
+                    redirect_url = _url_from_endpoint(url, redirect_url)
+                    redirect_parsed = urlparse(redirect_url)
+                    current_parsed = urlparse(url)
+
+                    if current_parsed.netloc != redirect_parsed.netloc:
+                        # Cross-host redirect: drop connection-supplied headers,
+                        # keeping only headers the caller explicitly passed in.
+                        current_headers = {k: v for k, v in merged_headers.items() if k not in remaining_connection_headers}
+                        remaining_connection_headers.clear()
+
+                    url = redirect_url
+                    await response.release()
+                    continue
+
+                response.raise_for_status()
+                return response
+
+            raise ClientResponseError(
+                request_info=response.request_info,
+                history=(),
+                status=599,
+                message=f"Too many redirects ({max_redirects})",
+                headers=response.headers,
             )
-            response.raise_for_status()
-            return response
 
         async for attempt in AsyncRetrying(
             stop=stop_after_attempt(self.retry_limit),

providers/http/tests/unit/http/hooks/test_http.py

@@ -898,3 +898,52 @@ async def test_build_request_url_from_endpoint_param(self):
             async with aiohttp.ClientSession() as session:
                 await hook.run(session=session, endpoint="test.com:8080/v1/test")
                 assert mocked_function.call_args.args[0] == "http://test.com:8080/v1/test"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("location", "expected_forwarded_key"),
+        [
+            pytest.param("http://other.host/v1/redirected", None, id="cross-host"),
+            pytest.param("http://test:8080/v1/redirected", "secret-key", id="same-host"),
+        ],
+    )
+    async def test_async_connection_header_is_only_forwarded_on_a_same_host_redirect(
+        self, create_connection_without_db, location, expected_forwarded_key
+    ):
+        """Connection extra headers must not be forwarded on a cross-host redirect (async)."""
+        create_connection_without_db(
+            Connection(
+                conn_id="http_conn_with_api_key",
+                conn_type="http",
+                host="test:8080/",
+                extra='{"X-API-Key": "secret-key"}',
+            )
+        )
+        hook = HttpAsyncHook(http_conn_id="http_conn_with_api_key")
+
+        with mock.patch("aiohttp.ClientSession.get", new_callable=mock.AsyncMock) as mocked_get:
+            redirect_response = MockAiohttpClientResponse(
+                status=302,
+                method="GET",
+                url="http://test:8080/v1/test",
+                headers={"Location": location},
+            )
+            final_response = MockAiohttpClientResponse(
+                status=200,
+                payload={"message": "OK"},
+                method="GET",
+                url=location,
+            )
+            mocked_get.side_effect = [redirect_response, final_response]
+
+            async with aiohttp.ClientSession() as session:
+                resp = await hook.run(session=session, endpoint="v1/test")
+                assert resp.status == 200
+
+            # First request should carry the connection header
+            first_headers = mocked_get.call_args_list[0].kwargs["headers"]
+            assert first_headers.get("X-API-Key") == "secret-key"
+
+            # Second request should only carry it on a same-host redirect
+            second_headers = mocked_get.call_args_list[1].kwargs["headers"]
+            assert second_headers.get("X-API-Key") == expected_forwarded_key