This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7e9128a9300306db7524f108e158bc40a3bdfcb5
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Apr 16 03:58:44 2026 +0200
[v3-2-test] Enforce per-file import-error authorization using
relative_fileloc + bundle (#65329) (#65343)
The public Import Errors API used to match ParseImportError.filename
against DagModel.fileloc. In real deployments ``fileloc`` is an
absolute path while ``filename`` is relative, so the file-to-DAG
resolution often came back empty and the single endpoint fell through
to returning the raw error. The list endpoint had a related gap: its
CTE was pre-filtered by the caller-visible subset of DAGs, so the
per-file authorization check only ever saw the DAGs the caller could
already read -- a file containing a mix of readable and unreadable
DAGs passed the check on the readable subset alone.
* The single endpoint now matches ParseImportError.filename against
DagModel.relative_fileloc + DagModel.bundle_name, which is the same
key the list endpoint already uses for its join. When the resolved
DAG set is empty (parse failed before any DAG was defined, or the
name keys did not resolve), the stacktrace is now redacted rather
than returned verbatim.
* The list endpoint splits the previous ``visible_files_cte`` into
two CTEs: ``readable_files_cte`` enumerates the ``(relative_fileloc,
bundle_name)`` pairs where the caller can read at least one DAG,
and ``file_dags_cte`` enumerates the full ``(relative_fileloc,
dag_id, bundle_name)`` set for those files. The per-file
authorization check in the groupby loop now receives the complete
DAG set for each file and can correctly detect co-located DAGs
outside the caller's scope.
* The same fall-through in the list endpoint (file has no matching
DAGs in DagModel) now redacts the stacktrace before appending.
Add a test class that exercises the fix with distinct ``fileloc``
(absolute) and ``relative_fileloc`` (relative) string values, closing
the test-fixture gap where both columns previously held the same
relative string and the absolute-vs-relative mismatch could not
manifest. One existing single-endpoint test documenting the previous
fall-through behaviour is updated to assert the new redaction.
(cherry picked from commit eba9b65a587a078134f3a9808c3fd6e18f92d3d0)
Generated-by: Claude Opus 4.6 (1M context) following the guidelines at
https:
//github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../core_api/routes/public/import_error.py | 72 +++++--
.../core_api/routes/public/test_import_error.py | 211 ++++++++++++++++++++-
2 files changed, 267 insertions(+), 16 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/import_error.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/import_error.py
index 97bfc4c0691..57a937b4d45 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/import_error.py
+++
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/import_error.py
@@ -80,13 +80,28 @@ def get_import_error(
auth_manager = get_auth_manager()
readable_dag_ids = auth_manager.get_authorized_dag_ids(user=user)
- # We need file_dag_ids as a set for intersection, issubset operations
+ # ``ParseImportError.filename`` is a repository-relative path and
+ # ``DagModel.fileloc`` is typically the absolute path those files were
+ # loaded from, so matching on ``fileloc == filename`` would come back
+ # empty in most real deployments. Match on ``relative_fileloc`` (and the
+ # bundle name that scopes it) instead, which is the same key the list
+ # endpoint already uses for the join below.
file_dag_ids = set(
- session.scalars(select(DagModel.dag_id).where(DagModel.fileloc ==
error.filename)).all()
+ session.scalars(
+ select(DagModel.dag_id).where(
+ DagModel.relative_fileloc == error.filename,
+ DagModel.bundle_name == error.bundle_name,
+ )
+ ).all()
)
- # No DAGs in the file (failed to parse), nothing to check permissions
against
+ # No DAGs matched for this file -- either the file genuinely contains
+ # no DAGs (parse failed before any DAG was defined), or the name keys
+ # did not resolve. Redact the stacktrace rather than returning the raw
+ # error, so the response stays on the deny-by-default side of the
+ # authorization check.
if not file_dag_ids:
+ error.stacktrace = REDACTED_STACKTRACE
return error
# Can the user read any DAGs in the file?
@@ -138,32 +153,55 @@ def get_import_errors(
# Subquery for files that have any DAGs
files_with_any_dags =
select(DagModel.relative_fileloc).distinct().subquery()
- # CTE for DAGs the user can read
- visible_files_cte = (
- select(DagModel.relative_fileloc, DagModel.dag_id,
DagModel.bundle_name)
+ # Files (identified by ``(relative_fileloc, bundle_name)``) where the
+ # user can read at least one DAG. Used to decide which import errors
+ # the user is allowed to see at all.
+ readable_files_cte = (
+ select(DagModel.relative_fileloc, DagModel.bundle_name)
.where(DagModel.dag_id.in_(readable_dag_ids))
+ .distinct()
+ .cte()
+ )
+
+ # Full ``(relative_fileloc, dag_id, bundle_name)`` set for every file
+ # the user can see at least one DAG of. Crucially this is **not**
+ # filtered by ``readable_dag_ids`` -- the per-file authorization
+ # check in the loop below needs the complete DAG set so it can
+ # detect co-located DAGs that the caller is not authorized to read
+ # and redact the stacktrace accordingly.
+ file_dags_cte = (
+ select(DagModel.relative_fileloc, DagModel.dag_id,
DagModel.bundle_name)
+ .join(
+ readable_files_cte,
+ and_(
+ DagModel.relative_fileloc ==
readable_files_cte.c.relative_fileloc,
+ DagModel.bundle_name == readable_files_cte.c.bundle_name,
+ ),
+ )
.cte()
)
- # Prepare the import errors query by joining with the cte.
- # Each returned row will be a tuple: (ParseImportError, dag_id)
+ # Prepare the import errors query by joining with the CTE above.
+ # Each returned row will be a tuple: (ParseImportError, dag_id).
+ # ``dag_id`` is NULL for import errors whose file has no DAGs at all
+ # in ``DagModel`` (parse failed before any DAG was defined).
import_errors_stmt = (
- select(ParseImportError, visible_files_cte.c.dag_id)
+ select(ParseImportError, file_dags_cte.c.dag_id)
.outerjoin(
files_with_any_dags,
ParseImportError.filename ==
files_with_any_dags.c.relative_fileloc,
)
.outerjoin(
- visible_files_cte,
+ file_dags_cte,
and_(
- ParseImportError.filename ==
visible_files_cte.c.relative_fileloc,
- ParseImportError.bundle_name ==
visible_files_cte.c.bundle_name,
+ ParseImportError.filename == file_dags_cte.c.relative_fileloc,
+ ParseImportError.bundle_name == file_dags_cte.c.bundle_name,
),
)
.where(
or_(
files_with_any_dags.c.relative_fileloc.is_(None),
- visible_files_cte.c.dag_id.isnot(None),
+ file_dags_cte.c.dag_id.isnot(None),
)
)
.order_by(ParseImportError.id)
@@ -186,8 +224,14 @@ def get_import_errors(
for import_error, file_dag_ids_iter in import_errors_result:
dag_ids = [dag_id for _, dag_id in file_dag_ids_iter if dag_id is not
None]
- # No DAGs in the file, nothing to check permissions against
+ # No DAGs matched for this file -- either the file genuinely has
+ # no DAGs yet (parse failed before any DAG was defined), or the
+ # name keys did not resolve. Redact the stacktrace before
+ # appending so the response stays on the deny-by-default side of
+ # the authorization check.
if not dag_ids:
+ session.expunge(import_error)
+ import_error.stacktrace = REDACTED_STACKTRACE
import_errors.append(import_error)
continue
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_import_error.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_import_error.py
index 469d746fb14..ba0e0743697 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_import_error.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_import_error.py
@@ -23,6 +23,7 @@ from unittest import mock
import pytest
from airflow.api_fastapi.auth.managers.models.resource_details import
DagDetails
+from airflow.api_fastapi.core_api.routes.public.import_error import
REDACTED_STACKTRACE
from airflow.models import DagModel
from airflow.models.dagbundle import DagBundleModel
from airflow.models.errors import ParseImportError
@@ -276,7 +277,13 @@ class TestGetImportError:
@mock.patch("airflow.api_fastapi.core_api.routes.public.import_error.get_auth_manager")
def test_get_import_error__no_dag_in_dagmodel(self, mock_get_auth_manager,
test_client, import_errors):
- """Test import error is returned when no DAG exists in DagModel."""
+ """Import error is returned with a redacted stacktrace when no DAG
+ exists in ``DagModel`` for the file.
+
+ When the file-to-DAG set resolves empty the endpoint cannot tell
+ which DAGs the caller is allowed to see, so the stacktrace is
+ redacted rather than returned verbatim.
+ """
import_error_id = import_errors[0].id
set_mock_auth_manager__get_authorized_dag_ids(mock_get_auth_manager,
set())
@@ -287,7 +294,7 @@ class TestGetImportError:
"import_error_id": import_error_id,
"timestamp": from_datetime_to_zulu_without_ms(TIMESTAMP1),
"filename": FILENAME1,
- "stack_trace": STACKTRACE1,
+ "stack_trace": REDACTED_STACKTRACE,
"bundle_name": BUNDLE_NAME,
}
@@ -525,3 +532,203 @@ class TestGetImportErrors:
assert FILENAME1 in filenames
assert FILENAME2 in filenames
assert FILENAME3 in filenames
+
+
+class TestImportErrorFileAuthorization:
+ """Tests that the import error endpoints apply per-file authorization
+ using ``relative_fileloc + bundle_name`` and redact stacktraces when the
+ resolved DAG set is empty or contains co-located DAGs outside the
+ caller's scope."""
+
+ LONELY_FILE_RELATIVE = "lonely_file.py"
+ LONELY_FILE_ABSOLUTE = "/opt/airflow/dags/lonely_file.py"
+ MIXED_FILE_RELATIVE = "mixed_file.py"
+ MIXED_FILE_ABSOLUTE = "/opt/airflow/dags/mixed_file.py"
+ LONELY_STACKTRACE = "stack trace for the lonely file"
+ MIXED_STACKTRACE = "stack trace for the mixed file"
+
+ @pytest.fixture
+ @provide_session
+ def absolute_vs_relative_fileloc_dag(
+ self,
+ testing_dag_bundle,
+ session: Session = NEW_SESSION,
+ ) -> DagModel:
+ """DagModel whose ``fileloc`` is absolute and ``relative_fileloc`` is
+ the relative path that matches ``ParseImportError.filename``.
+
+ The two columns deliberately hold different string values so that a
+ ``fileloc == filename`` match (the pre-fix behaviour) comes back
+ empty and a ``relative_fileloc == filename`` match (the fix) finds
+ the row.
+ """
+ dag_model = DagModel(
+ fileloc=self.LONELY_FILE_ABSOLUTE,
+ relative_fileloc=self.LONELY_FILE_RELATIVE,
+ dag_id="lonely_dag",
+ is_paused=False,
+ bundle_name=BUNDLE_NAME,
+ )
+ session.add(dag_model)
+ session.commit()
+ return dag_model
+
+ @pytest.fixture
+ @provide_session
+ def mixed_file_dags(
+ self,
+ testing_dag_bundle,
+ session: Session = NEW_SESSION,
+ ) -> tuple[DagModel, DagModel]:
+ """Two DagModels pointing at the same ``(relative_fileloc,
+ bundle_name)`` pair so the per-file authorization check in the list
+ endpoint has a co-located DAG to redact against."""
+ readable = DagModel(
+ fileloc=self.MIXED_FILE_ABSOLUTE,
+ relative_fileloc=self.MIXED_FILE_RELATIVE,
+ dag_id="readable_mixed_dag",
+ is_paused=False,
+ bundle_name=BUNDLE_NAME,
+ )
+ colocated = DagModel(
+ fileloc=self.MIXED_FILE_ABSOLUTE,
+ relative_fileloc=self.MIXED_FILE_RELATIVE,
+ dag_id="colocated_mixed_dag",
+ is_paused=False,
+ bundle_name=BUNDLE_NAME,
+ )
+ session.add_all([readable, colocated])
+ session.commit()
+ return readable, colocated
+
+ @pytest.fixture
+ @provide_session
+ def lonely_file_import_error(
+ self,
+ session: Session = NEW_SESSION,
+ ) -> ParseImportError:
+ error = ParseImportError(
+ bundle_name=BUNDLE_NAME,
+ filename=self.LONELY_FILE_RELATIVE,
+ stacktrace=self.LONELY_STACKTRACE,
+ timestamp=TIMESTAMP1,
+ )
+ session.add(error)
+ session.commit()
+ return error
+
+ @pytest.fixture
+ @provide_session
+ def mixed_file_import_error(
+ self,
+ session: Session = NEW_SESSION,
+ ) -> ParseImportError:
+ error = ParseImportError(
+ bundle_name=BUNDLE_NAME,
+ filename=self.MIXED_FILE_RELATIVE,
+ stacktrace=self.MIXED_STACKTRACE,
+ timestamp=TIMESTAMP2,
+ )
+ session.add(error)
+ session.commit()
+ return error
+
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.import_error.get_auth_manager")
+ def test_single_endpoint_matches_file_via_relative_fileloc_not_fileloc(
+ self,
+ mock_get_auth_manager,
+ test_client,
+ absolute_vs_relative_fileloc_dag,
+ lonely_file_import_error,
+ ):
+ """Single endpoint resolves ``ParseImportError.filename`` against
+ ``DagModel.relative_fileloc`` (and ``bundle_name``), not
+ ``DagModel.fileloc``.
+
+ The DagModel's ``fileloc`` is absolute while the ParseImportError's
+ ``filename`` is relative, so a ``fileloc == filename`` match comes
+ back empty in this fixture. The endpoint must still resolve the DAG
+ set via ``relative_fileloc`` and enforce the normal authorization
+ check. Here the caller has no DAG permissions at all, so the
+ response must be 403 -- not a 200 that returns the stack trace
+ verbatim via the empty-set fall-through.
+ """
+ set_mock_auth_manager__get_authorized_dag_ids(mock_get_auth_manager,
set())
+ response =
test_client.get(f"/importErrors/{lonely_file_import_error.id}")
+ assert response.status_code == 403
+
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.import_error.get_auth_manager")
+ def test_single_endpoint_redacts_when_file_has_no_known_dags(
+ self,
+ mock_get_auth_manager,
+ test_client,
+ import_errors,
+ ):
+ """Single endpoint must redact the stacktrace when the
+ ``ParseImportError`` refers to a file with no matching ``DagModel``
+ rows at all -- for example a file that failed to parse before any
+ DAG was defined. The response must be 200 with
+ ``REDACTED_STACKTRACE``, not a 200 with the raw error body.
+ """
+ set_mock_auth_manager__get_authorized_dag_ids(mock_get_auth_manager,
set())
+ response = test_client.get(f"/importErrors/{import_errors[0].id}")
+ assert response.status_code == 200
+ body = response.json()
+ assert body["filename"] == FILENAME1
+ assert body["stack_trace"] == REDACTED_STACKTRACE
+
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.import_error.get_auth_manager")
+ def
test_list_endpoint_redacts_mixed_file_with_colocated_dag_outside_callers_scope(
+ self,
+ mock_get_auth_manager,
+ test_client,
+ mixed_file_dags,
+ mixed_file_import_error,
+ ):
+ """List endpoint must redact the stacktrace for a file that
+ contains a DAG outside the caller's scope, even when the caller can
+ read another DAG in the same file.
+
+ The ``side_effect`` below allows the call only when the request set
+ is a subset of the caller's readable set. Under the fixed code the
+ per-file authorization check receives the full DAG set for the
+ file (both ``readable_mixed_dag`` and ``colocated_mixed_dag``), so
+ the call is denied and the stacktrace is redacted. Under the
+ pre-fix code the check would only see the readable subset, the
+ call would be permitted, and the raw stacktrace would be returned.
+ """
+ readable, _ = mixed_file_dags
+ set_mock_auth_manager__get_authorized_dag_ids(mock_get_auth_manager,
{readable.dag_id})
+
+ def permit_only_readable(requests, user):
+ request_dag_ids = {req["details"].id for req in requests}
+ return request_dag_ids.issubset({readable.dag_id})
+
+ mock_get_auth_manager.return_value.batch_is_authorized_dag.side_effect
= permit_only_readable
+
+ response = test_client.get("/importErrors")
+ assert response.status_code == 200
+ body = response.json()
+ mixed_entries = [err for err in body["import_errors"] if
err["filename"] == self.MIXED_FILE_RELATIVE]
+ assert len(mixed_entries) == 1
+ assert mixed_entries[0]["stack_trace"] == REDACTED_STACKTRACE
+
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.import_error.get_auth_manager")
+ def test_list_endpoint_redacts_when_file_has_no_known_dags(
+ self,
+ mock_get_auth_manager,
+ test_client,
+ import_errors,
+ ):
+ """List endpoint must redact the stacktrace for import errors
+ whose file has no matching ``DagModel`` rows -- closing the
+ ``if not dag_ids: import_errors.append(import_error)`` fall-through
+ that previously returned the raw error.
+ """
+ set_mock_auth_manager__get_authorized_dag_ids(mock_get_auth_manager,
set())
+ response = test_client.get("/importErrors")
+ assert response.status_code == 200
+ body = response.json()
+ assert body["total_entries"] == 3
+ for entry in body["import_errors"]:
+ assert entry["stack_trace"] == REDACTED_STACKTRACE
