This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun 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 c5faeb1555b Authorize Dag reparse against the file's Dags, not the
query-string dag_id (#69471)
c5faeb1555b is described below
commit c5faeb1555b4076739665b9434f616db37559727
Author: Pierre Jeambrun <[email protected]>
AuthorDate: Fri Jul 17 15:26:47 2026 +0200
Authorize Dag reparse against the file's Dags, not the query-string dag_id
(#69471)
* Authorize Dag reparse against the file's Dags, not the query-string dag_id
The PUT /parseDagFile/{file_token} endpoint authorized the caller through a
route dependency that, with no dag_id path parameter, resolves the target from
the query string, which is decoupled from the Dag file the signed file_token
actually resolves to.
Add a requires_access_dag_from_file_token dependency that decodes the
token, resolves the Dags defined in that file, and authorizes the caller
against exactly those Dags, so authorization always matches the Dag being
reparsed regardless of any request parameter.
* Update airflow-core/src/airflow/api_fastapi/core_api/security.py
Co-authored-by: Amogh Desai <[email protected]>
* Fix CI
---------
Co-authored-by: Amogh Desai <[email protected]>
---
.../core_api/routes/public/dag_parsing.py | 49 ++++------------------
.../src/airflow/api_fastapi/core_api/security.py | 44 +++++++++++++++++++
.../core_api/routes/public/test_dag_parsing.py | 11 ++++-
scripts/ci/prek/extract_permissions.py | 1 +
4 files changed, 63 insertions(+), 42 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_parsing.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_parsing.py
index 5a9a5d7e397..1a901537dce 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_parsing.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_parsing.py
@@ -16,25 +16,16 @@
# under the License.
from __future__ import annotations
-from collections.abc import Sequence
-from typing import TYPE_CHECKING
+from fastapi import Depends, Request, status
+from itsdangerous import URLSafeSerializer
-from fastapi import Depends, HTTPException, Request, status
-from itsdangerous import BadSignature, URLSafeSerializer
-from sqlalchemy import select
-
-from airflow.api_fastapi.auth.managers.models.resource_details import
DagDetails
from airflow.api_fastapi.common.db.common import SessionDep
from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
-from airflow.api_fastapi.core_api.security import requires_access_dag
+from airflow.api_fastapi.core_api.security import
requires_access_dag_from_file_token
from airflow.api_fastapi.logging.decorators import action_logging
-from airflow.models.dag import DagModel
from airflow.models.dagbag import DagPriorityParsingRequest
-if TYPE_CHECKING:
- from airflow.api_fastapi.auth.managers.models.batch_apis import
IsAuthorizedDagRequest
-
dag_parsing_router = AirflowRouter(tags=["DAG Parsing"],
prefix="/parseDagFile/{file_token}")
@@ -42,34 +33,12 @@ dag_parsing_router = AirflowRouter(tags=["DAG Parsing"],
prefix="/parseDagFile/{
"",
responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
status_code=status.HTTP_201_CREATED,
- dependencies=[Depends(requires_access_dag(method="PUT")),
Depends(action_logging())],
+ dependencies=[Depends(requires_access_dag_from_file_token(method="PUT")),
Depends(action_logging())],
)
-def reparse_dag_file(
- file_token: str,
- session: SessionDep,
- request: Request,
-) -> None:
+def reparse_dag_file(file_token: str, session: SessionDep, request: Request)
-> None:
"""Request re-parsing a Dag file."""
- secret_key = request.app.state.secret_key
- auth_s = URLSafeSerializer(secret_key)
- try:
- payload = auth_s.loads(file_token)
- except BadSignature:
- raise HTTPException(status.HTTP_404_NOT_FOUND, "File not found")
-
- bundle_name = payload["bundle_name"]
- relative_fileloc = payload["relative_fileloc"]
-
- requests: Sequence[IsAuthorizedDagRequest] = [
- {"method": "PUT", "details": DagDetails(id=dag_id)}
- for dag_id in session.scalars(
- select(DagModel.dag_id).where(
- DagModel.bundle_name == bundle_name, DagModel.relative_fileloc
== relative_fileloc
- )
- )
- ]
- if not requests:
- raise HTTPException(status.HTTP_404_NOT_FOUND, "File not found")
-
- parsing_request = DagPriorityParsingRequest(bundle_name=bundle_name,
relative_fileloc=relative_fileloc)
+ payload = URLSafeSerializer(request.app.state.secret_key).loads(file_token)
+ parsing_request = DagPriorityParsingRequest(
+ bundle_name=payload["bundle_name"],
relative_fileloc=payload["relative_fileloc"]
+ )
session.add(parsing_request)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/security.py
b/airflow-core/src/airflow/api_fastapi/core_api/security.py
index 724a6967f74..5f85b68e6b1 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/security.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/security.py
@@ -25,6 +25,7 @@ from urllib.parse import ParseResult, unquote, urljoin,
urlparse
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer,
OAuth2PasswordBearer
+from itsdangerous import BadSignature, URLSafeSerializer
from jwt import ExpiredSignatureError, InvalidTokenError
from sqlalchemy import or_, select
from sqlalchemy.orm import Session
@@ -198,6 +199,49 @@ def requires_access_dag(
return inner
+def requires_access_dag_from_file_token(
+ method: ResourceMethod,
+) -> Callable[[str, Request, BaseUser, Session], None]:
+ """
+ Authorize the caller against the DAGs referenced by a signed
``file_token``.
+
+ For ``file_token`` based endpoints (such as ``reparse``), the token is
resolved to its referenced file, and authorization is performed against exactly
the DAGs defined in that file, never against a request parameter.
+ """
+
+ def inner(
+ file_token: str,
+ request: Request,
+ user: GetUserDep,
+ session: SessionDep,
+ ) -> None:
+ try:
+ payload =
URLSafeSerializer(request.app.state.secret_key).loads(file_token)
+ except BadSignature:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, "File not found")
+
+ dag_ids = list(
+ session.scalars(
+ select(DagModel.dag_id).where(
+ DagModel.bundle_name == payload["bundle_name"],
+ DagModel.relative_fileloc == payload["relative_fileloc"],
+ )
+ )
+ )
+ if not dag_ids:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, "File not found")
+
+ dag_id_to_team = DagModel.get_dag_id_to_team_name_mapping(dag_ids,
session=session)
+ requests: list[IsAuthorizedDagRequest] = [
+ {"method": method, "details": DagDetails(id=dag_id,
team_name=dag_id_to_team.get(dag_id))}
+ for dag_id in dag_ids
+ ]
+ _requires_access(
+ is_authorized_callback=lambda:
get_auth_manager().batch_is_authorized_dag(requests, user=user),
+ )
+
+ return inner
+
+
class PermittedDagFilter(OrmClause[set[str]]):
"""A parameter that filters the permitted dags for the user."""
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_parsing.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_parsing.py
index 763e5e88669..b60b7011a9e 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_parsing.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_parsing.py
@@ -75,8 +75,15 @@ class TestDagParsingEndpoint:
)
assert response.status_code == 401
- def test_should_respond_403(self, unauthorized_test_client):
- response = unauthorized_test_client.put("/parseDagFile/token",
headers={"Accept": "application/json"})
+ def test_should_respond_403(self, unauthorized_test_client,
url_safe_serializer, session):
+ parse_and_sync_to_db(EXAMPLE_DAG_FILE)
+ test_dag =
DBDagBag(load_op_links=False).get_latest_version_of_dag(TEST_DAG_ID,
session=session)
+ token = url_safe_serializer.dumps(
+ {"bundle_name": "example_dags", "relative_fileloc":
test_dag.relative_fileloc}
+ )
+ response = unauthorized_test_client.put(
+ f"/parseDagFile/{token}", headers={"Accept": "application/json"}
+ )
assert response.status_code == 403
def test_bad_file_request(self, url_safe_serializer, session, test_client):
diff --git a/scripts/ci/prek/extract_permissions.py
b/scripts/ci/prek/extract_permissions.py
index e9e454fe299..1f0c68d313e 100644
--- a/scripts/ci/prek/extract_permissions.py
+++ b/scripts/ci/prek/extract_permissions.py
@@ -225,6 +225,7 @@ def _extract_entity_arg(call_node: ast.Call) -> str | None:
# Map from requires_access_* function name → (resource base name, forced
entity or None)
_FN_TO_RESOURCE_INFO: dict[str, tuple[str, str | None]] = {
"requires_access_dag": ("DAG", None),
+ "requires_access_dag_from_file_token": ("DAG", None), # reparse
authorizes the file_token's Dags
"requires_access_backfill": ("DAG", "RUN"), # backfill is a DAG.RUN alias
"requires_access_dag_run_bulk": ("DAG", "RUN"), # dag_run bulk is a
DAG.RUN alias
"requires_access_dag_run_clear_bulk": ("DAG", "RUN"), # dag_run clear
bulk is a DAG.RUN alias
