[v3-3-test] Require existing-connection read access when testing an existing connection

airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py

@@ -53,6 +53,7 @@
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.api_fastapi.core_api.security import (
+    AuthManagerDep,
     GetUserDep,
     ReadableConnectionsFilterDep,
     requires_access_connection,
@@ -293,7 +294,11 @@ def patch_connection(
 
 
 @connections_router.post("/test", dependencies=[Depends(requires_access_connection(method="POST"))])
-def test_connection(test_body: ConnectionBody) -> ConnectionTestResponse:
+def test_connection(
+    test_body: ConnectionBody,
+    user: GetUserDep,
+    auth_manager: AuthManagerDep,
+) -> ConnectionTestResponse:
     """
     Test an API connection.
 
@@ -306,11 +311,43 @@ def test_connection(test_body: ConnectionBody) -> ConnectionTestResponse:
     transient_conn_id = get_random_string()
     conn_env_var = f"{CONN_ENV_PREFIX}{transient_conn_id.upper()}"
     try:
-        # Try to get existing connection and merge with provided values
-        try:
-            existing_conn: Connection | None = Connection.get_connection_from_secrets(test_body.connection_id)
-        except AirflowNotFoundException:
-            existing_conn = None
+        # Authorize read access on the requested ``connection_id`` *before*
+        # touching the secrets backends. The route-level POST dependency only
+        # verifies the caller can create connections; merging the existing
+        # connection's hidden fields also requires read access to that
+        # specific connection. Gating the backend lookup itself (rather than
+        # the post-load merge) prevents an unauthorized caller from using
+        # this endpoint to enumerate protected connection ids, generate
+        # access-log entries in audited backends, or impose backend load for
+        # arbitrary ids. ``get_team_name`` is a metadata-only DB lookup and
+        # does not touch the configured secrets backends.
+        #
+        # When the connection has no metadata-DB row (e.g. it lives only in
+        # a team-aware secrets backend like Vault or Kubernetes), fall back
+        # to the request body's validated ``team_name`` so the GET
+        # authorization and the secrets lookup both run in the right team
+        # scope. ``ConnectionBody.validate_team_name`` already rejects
+        # ``team_name`` from clients when ``[core] multi_team`` is off, so
+        # a non-None body value here is always already gated by that
+        # validator.
+        team_name = Connection.get_team_name(test_body.connection_id)
+        if team_name is None:
+            team_name = test_body.team_name
+        existing_conn: Connection | None = None
+        if auth_manager.is_authorized_connection(
+            method="GET",
+            details=ConnectionDetails(
+                conn_id=test_body.connection_id,
+                team_name=team_name,
+            ),
+            user=user,
+        ):
+            try:
+                existing_conn = Connection.get_connection_from_secrets(
+                    test_body.connection_id, team_name=team_name
+                )
+            except AirflowNotFoundException:
+                existing_conn = None
 
         if existing_conn is not None:
             # Stored credentials are only reused to test the connection's own

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py

@@ -1069,6 +1069,133 @@ def test_should_respond_403(self, unauthorized_test_client):
         )
         assert response.status_code == 403
 
+    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
+    def test_unreadable_existing_connection_indistinguishable_from_missing(self, test_client):
+        """Route-level POST authorization is not enough on its own — when the
+        request references an existing connection_id, the caller must also be
+        authorized to read that specific connection before its hidden fields
+        are merged into the test object. A caller lacking read access must
+        get the same response shape as for a non-existent connection_id, so
+        the route cannot be used to enumerate protected connection ids."""
+        self.create_connection()
+
+        from airflow.api_fastapi.auth.managers.simple.simple_auth_manager import SimpleAuthManager
+
+        real_method = SimpleAuthManager.is_authorized_connection
+
+        def gated_authz(self, *, method, details=None, user=None):
+            if method == "GET":
+                return False
+            return real_method(self, method=method, details=details, user=user)
+
+        with mock.patch.object(SimpleAuthManager, "is_authorized_connection", gated_authz):
+            existing_response = test_client.post(
+                "/connections/test",
+                json={"connection_id": TEST_CONN_ID, "conn_type": "sqlite"},
+            )
+            missing_response = test_client.post(
+                "/connections/test",
+                json={"connection_id": "this_connection_does_not_exist", "conn_type": "sqlite"},
+            )
+
+        # Both calls reach the body-only test path (the unreadable existing
+        # connection is treated as if it did not exist), so status + body
+        # shape must be identical — no existence oracle for protected ids.
+        assert existing_response.status_code == missing_response.status_code
+        assert set(existing_response.json().keys()) == set(missing_response.json().keys())
+
+    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
+    def test_unreadable_existing_connection_does_not_trigger_secrets_lookup(self, test_client):
+        """The route must gate the secrets-backend lookup on the GET
+        authorization check, not perform the lookup and then suppress the
+        result. Otherwise an unauthorized caller can still force
+        ``Connection.get_connection_from_secrets`` to query every configured
+        secrets backend for arbitrary connection ids — leaking timing /
+        existence signals, generating access-log entries in audited
+        backends, and imposing backend load."""
+        self.create_connection()
+
+        from airflow.api_fastapi.auth.managers.simple.simple_auth_manager import SimpleAuthManager
+
+        real_method = SimpleAuthManager.is_authorized_connection
+
+        def gated_authz(self, *, method, details=None, user=None):
+            if method == "GET":
+                return False
+            return real_method(self, method=method, details=details, user=user)
+
+        with (
+            mock.patch.object(SimpleAuthManager, "is_authorized_connection", gated_authz),
+            mock.patch.object(
+                Connection,
+                "get_connection_from_secrets",
+                wraps=Connection.get_connection_from_secrets,
+            ) as spy_secrets,
+        ):
+            response = test_client.post(
+                "/connections/test",
+                json={"connection_id": TEST_CONN_ID, "conn_type": "sqlite"},
+            )
+
+        assert response.status_code == 200
+        spy_secrets.assert_not_called()
+
+    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
+    @conf_vars({("core", "multi_team"): "true"})
+    def test_existing_connection_lookup_preserves_team_scope(self, test_client, testing_team, session):
+        """The secrets-backend lookup must propagate the authorized
+        ``team_name`` to ``Connection.get_connection_from_secrets``.
+        Otherwise the call falls back to global / wrong-team paths in
+        team-aware backends (Vault, Akeyless, …) and can return a
+        cross-scope secret with the same ``conn_id`` — exactly what the
+        team-scoped authorization check above is supposed to prevent."""
+        self.create_connection(team_name=testing_team.name)
+        session.commit()
+
+        with mock.patch.object(
+            Connection,
+            "get_connection_from_secrets",
+            wraps=Connection.get_connection_from_secrets,
+        ) as spy_secrets:
+            response = test_client.post(
+                "/connections/test",
+                json={"connection_id": TEST_CONN_ID, "conn_type": "sqlite"},
+            )
+
+        assert response.status_code == 200
+        spy_secrets.assert_called_once_with(TEST_CONN_ID, team_name=testing_team.name)
+
+    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
+    @conf_vars({("core", "multi_team"): "true"})
+    def test_secrets_only_team_connection_uses_body_team_scope(self, test_client, testing_team):
+        """When the connection exists only in a team-aware secrets backend
+        (no metadata-DB row), ``Connection.get_team_name`` returns None.
+        The route must then fall back to the body's validated ``team_name``
+        so the GET authorization and the subsequent secrets lookup both
+        run in the right team scope — otherwise a team-scoped existing
+        connection would be authorized and looked up as global, losing
+        the multi-team isolation guarantee in deployments that keep
+        connections in Vault / Kubernetes / Akeyless rather than the DB."""
+        # No ``self.create_connection()`` — TEST_CONN_ID lives only in a
+        # secrets backend in this scenario.
+
+        with mock.patch.object(
+            Connection,
+            "get_connection_from_secrets",
+            wraps=Connection.get_connection_from_secrets,
+        ) as spy_secrets:
+            response = test_client.post(
+                "/connections/test",
+                json={
+                    "connection_id": TEST_CONN_ID,
+                    "conn_type": "sqlite",
+                    "team_name": testing_team.name,
+                },
+            )
+
+        assert response.status_code == 200
+        spy_secrets.assert_called_once_with(TEST_CONN_ID, team_name=testing_team.name)
+
     @skip_if_force_lowest_dependencies_marker
     @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
     @pytest.mark.parametrize(