This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 779efc6983b Require existing-connection read access when testing an
existing connection (#67620)
779efc6983b is described below
commit 779efc6983b2b2497cd8f9c2d7bb376bae196825
Author: Jarek Potiuk <[email protected]>
AuthorDate: Thu Jun 25 14:35:57 2026 -0400
Require existing-connection read access when testing an existing connection
(#67620)
* Require existing-connection read access when testing an existing
connection
The `POST /api/v2/connections/test` route was authorizing the caller only
as a connection POST operation (i.e. "can create a connection"). When the
request body referenced an existing `connection_id`, the route then loaded that
connection from the configured secrets backend and merged its hidden fields
(`login`, `password`, parts of `extra`) into the test object. The route did not
check whether the caller was authorized to read that existing connection — so a
caller authorized to crea [...]
This change adds a `GET` authorization check on the existing connection
before its secrets are merged into the test object. A caller authorized to
create connections but not to read the existing `connection_id` now gets a 403.
Reference: airflow-s/airflow-s#444
Generated-by: Claude Opus 4.7 (1M context) following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Treat unreadable existing connection as missing in /connections/test
Codex's adversarial review of the previous commit on this branch pointed
out that returning 403 only when the existing connection is found AND
unreadable creates an existence oracle: callers with route-level POST
permission but no read permission could distinguish a protected connection_id
(which returns 403) from a non-existent id (which falls through to the
body-only test path).
This commit removes the oracle by collapsing both cases — "connection not
found" and "connection found but caller lacks read access" — into the same
body-only test path. The hidden-field borrow is gated on `(existing_conn is not
None and auth_manager.is_authorized_connection("GET", ...))` and the response
shape is identical for any unreadable connection_id, regardless of whether it
exists in the secrets backend.
The regression test is updated to assert that the response to an unreadable
existing connection_id is indistinguishable (status code + response body keys)
from the response to a non-existent connection_id, under the same submitted
body fields.
Generated-by: Claude Opus 4.7 (1M context) following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Gate /connections/test secrets lookup on read authorization
A second Codex adversarial-review pass on this branch pointed out that the
previous follow-up commit still performed the secrets-backend lookup
(`Connection.get_connection_from_secrets`) before the GET authorization check.
Although the response shape was normalized so the caller could not distinguish
unreadable-existing from missing, the lookup itself still ran for arbitrary
connection ids — meaning an unauthorized caller could still: trigger backend
queries against every configured s [...]
This commit moves the GET authorization boundary *ahead of* the
secrets-backend lookup. Team scope is resolved via `Connection.get_team_name`,
which is a metadata-only DB lookup against the `connection` table and does not
touch any secrets backend. Only when the caller is authorized to read the
requested `connection_id` does the route call
`Connection.get_connection_from_secrets`. Unauthorized callers fall through to
the body-only test path without any secrets-backend interaction.
A new regression test spies on `Connection.get_connection_from_secrets`
with `wraps=` and asserts it is never called when the caller lacks GET access —
locking in the gate at the test level.
Generated-by: Claude Opus 4.7 (1M context) following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Preserve team_name through /connections/test secrets lookup
A third Codex adversarial-review pass on this branch flagged that the route
resolves and authorizes against `team_name` from `Connection.get_team_name`,
but then calls `Connection.get_connection_from_secrets(connection_id)` without
forwarding that team. `Connection.get_connection_from_secrets` accepts a
`team_name` kwarg and forwards it to team-aware secrets backends (Vault,
Akeyless, …); when omitted, those backends fall back to global secrets. In
multi-team deployments this could ei [...]
This commit threads `team_name` through to the secrets lookup, keeping the
read scope consistent end-to-end. A new regression test enables `[core]
multi_team=true`, creates a team-owned `TEST_CONN_ID`, and asserts the spy on
`get_connection_from_secrets` is called with the correct `team_name` kwarg —
locking in the team-scope propagation at the test level.
Generated-by: Claude Opus 4.7 (1M context) following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Fall back to body team_name for secrets-only connections in
/connections/test
A fourth Codex adversarial-review pass on this branch flagged that for
connections that live *only* in a team-aware secrets backend (Vault,
Kubernetes, Akeyless, …) and have no metadata-DB row,
`Connection.get_team_name(connection_id)` returns None. The route was then
authorizing GET access and calling `get_connection_from_secrets(...,
team_name=None)`, dropping the caller's already-validated body `team_name`.
Team-aware backends only consult team-scoped paths when `team_name is not N
[...]
This commit falls back to `test_body.team_name` whenever the DB metadata
lookup returns None. The body's `team_name` is already validated by
`ConnectionBody.validate_team_name` (which rejects a non-None value unless
`[core] multi_team` is enabled), so a non-None body value here is always
already gated by that validator.
A new regression test enables `[core] multi_team=true`, posts to
`/connections/test` with a `team_name` body field, does *not* create a
metadata-DB row, and asserts the secrets-backend spy is called with the body's
`team_name`. This locks in the body-fallback path at the test level so future
refactors can't silently drop team scope for secrets-only connections.
Generated-by: Claude Opus 4.7 (1M context) following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
---
.../core_api/routes/public/connections.py | 51 ++++++++-
.../core_api/routes/public/test_connections.py | 127 +++++++++++++++++++++
2 files changed, 173 insertions(+), 5 deletions(-)
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 a1827253180..0d122946208 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.datamodels.connections
import (
)
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,
@@ -292,7 +293,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.
@@ -305,13 +310,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 74920b11d51..7c5ce03654e 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
@@ -1069,6 +1069,133 @@ class TestConnection(TestConnectionEndpoint):
)
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(
