Strip connection headers on cross-host redirect in HttpHook

providers/http/docs/changelog.rst

@@ -27,6 +27,13 @@
 Changelog
 ---------
 
+Breaking changes
+~~~~~~~~~~~~~~~~
+
+* ``HttpHook no longer forwards Connection extra headers across a cross-host redirect (#70000)``
+
+  Headers defined in an HTTP Connection's ``extra`` field are documented as a way to carry credentials, and ``requests`` only strips ``Authorization`` when a redirect leaves the original host. ``HttpHook`` now gives those Connection-supplied headers the same treatment, so a secret held under any header name (``X-API-Key``, ``Private-Token``, ...) is no longer replayed to the redirect target. Same-host redirects are unaffected. If you relied on those headers reaching a different host - for example an API that redirects downloads to a CDN or object store that needs the same key - either configure a Connection whose host matches the final destination, or follow the redirect explicitly by passing ``allow_redirects=False`` in ``extra_options`` and issuing the second request yourself. ``HttpAsyncHook`` still forwards the headers; that is tracked in https://github.com/apache/airflow/issues/70164.
+
 6.0.4
 .....
 

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

@@ -120,6 +120,25 @@ def _retryable_error_async(exception: BaseException) -> bool:
     return exception.status >= 500
 
 
+class _ConnectionSession(Session):
+    """
+    Session that drops Connection-supplied headers when a redirect leaves the original host.
+
+    ``requests`` only strips ``Authorization``, but an HTTP Connection may carry its
+    credentials under any header name, so those need the same treatment.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.connection_headers: set[str] = set()
+
+    def rebuild_auth(self, prepared_request: PreparedRequest, response: Response) -> None:
+        super().rebuild_auth(prepared_request, response)
+        if self.should_strip_auth(response.request.url, prepared_request.url):
+            for header in self.connection_headers:
+                prepared_request.headers.pop(header, None)
+
+
 class HttpHook(BaseHook):
     """
     Interact with HTTP servers.
@@ -196,7 +215,7 @@ def get_conn(
         :param extra_options: additional options to be used when executing the request
         :return: A configured requests.Session object.
         """
-        session = Session()
+        session: Session = _ConnectionSession()
         connection = self.get_connection(self.http_conn_id)
         self._set_base_url(connection)
         session = self._configure_session_from_auth(session, connection)  # type: ignore[arg-type]
@@ -268,6 +287,9 @@ def _configure_session_from_extra(
             session.headers.update(conn_extra_options)
         except TypeError:
             self.log.warning("Connection to %s has invalid extra field.", connection.host)
+        else:
+            if isinstance(session, _ConnectionSession):
+                session.connection_headers.update(conn_extra_options)
 
         return session
 
@@ -630,6 +652,9 @@ async def config(self) -> SessionConfig:
                     auth = self.auth_type(conn.login, conn.password)
 
                 if conn.extra:
+                    # Unlike HttpHook these headers survive a cross-host redirect, because aiohttp
+                    # has no per-redirect hook to drop them from;
+                    # tracked at https://github.com/apache/airflow/issues/70164
                     conn_extra_options, extra_options = _process_extra_options_from_connection(
                         conn=conn, extra_options={}
                     )

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

@@ -337,6 +337,34 @@ def test_post_request_raises_error_when_redirects_with_max_redirects_set_to_0(se
         assert history[0].url == "http://test:8080/v1/test"
         assert history[0].method == "POST"
 
+    @pytest.mark.parametrize(
+        ("location", "expected_forwarded_key"),
+        [
+            pytest.param("http://another.host/v1/redirected", None, id="cross-host"),
+            pytest.param("http://test:8080/v1/redirected", "secret-key", id="same-host"),
+        ],
+    )
+    def test_connection_header_is_only_forwarded_on_a_same_host_redirect(
+        self, requests_mock, create_connection_without_db, location, expected_forwarded_key
+    ):
+        create_connection_without_db(
+            Connection(
+                conn_id="http_conn_with_api_key",
+                conn_type="http",
+                host="test:8080/",
+                extra='{"X-API-Key": "secret-key"}',
+            )
+        )
+        requests_mock.get("http://test:8080/v1/test", status_code=302, headers={"Location": location})
+        requests_mock.get(location, status_code=200, text='{"message": "OK"}')
+
+        HttpHook(method="GET", http_conn_id="http_conn_with_api_key").run("v1/test")
+
+        history = requests_mock.request_history
+        assert len(history) == 2
+        assert history[0].headers["X-API-Key"] == "secret-key"
+        assert history[1].headers.get("X-API-Key") == expected_forwarded_key
+
     @pytest.mark.parametrize(
         "setup_connections_with_extras", [{"bearer": "test", "check_response": False}], indirect=True
     )

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

@@ -286,7 +286,7 @@ def mount(self, prefix, adapter):
 
 
 class TestHttpOpSensor:
-    @mock.patch("airflow.providers.http.hooks.http.Session", FakeSession)
+    @mock.patch("airflow.providers.http.hooks.http._ConnectionSession", FakeSession)
     def test_get(self):
         op = HttpOperator(
             task_id="get_op",
@@ -297,7 +297,7 @@ def test_get(self):
         )
         op.execute({})
 
-    @mock.patch("airflow.providers.http.hooks.http.Session", FakeSession)
+    @mock.patch("airflow.providers.http.hooks.http._ConnectionSession", FakeSession)
     def test_get_response_check(self):
         op = HttpOperator(
             task_id="get_op",
@@ -310,7 +310,7 @@ def test_get_response_check(self):
         op.execute({})
 
     @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow 3.0+")
-    @mock.patch("airflow.providers.http.hooks.http.Session", FakeSession)
+    @mock.patch("airflow.providers.http.hooks.http._ConnectionSession", FakeSession)
     def test_sensor(self, run_task):
         sensor = HttpSensor(
             task_id="http_sensor_check",
@@ -328,7 +328,7 @@ def test_sensor(self, run_task):
         assert run_task.error is None
 
     @pytest.mark.skipif(AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow < 3.0")
-    @mock.patch("airflow.providers.http.hooks.http.Session", FakeSession)
+    @mock.patch("airflow.providers.http.hooks.http._ConnectionSession", FakeSession)
     def test_sensor_af2(self):
         dag = DAG(TEST_DAG_ID, schedule=None)
         sensor = HttpSensor(