diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py index 7df7dd91908a7..43de733006bf8 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py @@ -52,6 +52,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, @@ -289,7 +290,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. @@ -302,13 +307,49 @@ 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.get_connection_from_secrets(test_body.connection_id) + # 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: existing_conn.conn_id = transient_conn_id update_orm_from_pydantic(existing_conn, test_body) conn = existing_conn - except AirflowNotFoundException: + else: data = test_body.model_dump(by_alias=True) data["conn_id"] = transient_conn_id conn = Connection(**data) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py index 4dd16d3b30bad..7478deca62a44 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py @@ -1068,6 +1068,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(