Gate audit log rows not tied to a Dag on a dedicated AccessView

airflow-core/newsfragments/70759.significant.rst

@@ -0,0 +1,24 @@
+Audit log rows not tied to a Dag now require the ``AUDIT_LOGS_ALL`` view
+
+Rows in the audit log whose ``dag_id`` is ``NULL`` record operations that are not tied to a
+Dag, such as Connection, Variable and Pool changes. They were returned to any caller that
+could read event logs at all, because there is no per-Dag key to authorize them on and both
+read paths fell back to a Dag-level audit log check.
+
+They are now gated on a dedicated ``AccessView.AUDIT_LOGS_ALL``, in the same way
+``AccessView.IMPORT_ERRORS_ALL`` gates import errors for files with no registered Dag. Rows
+bound to a Dag keep their existing per-Dag check.
+
+**Behaviour changes:**
+
+- ``GET /eventLogs`` no longer returns rows with a ``NULL`` ``dag_id`` to callers without the
+  new view, and excludes them from ``total_entries`` and pagination.
+  ``GET /eventLogs/{event_log_id}`` returns ``403`` for such a row. An id matching no row
+  still returns ``404``.
+- With the simple auth manager the view is granted to the ``ADMIN`` role only. With the FAB
+  auth manager it maps to a new ``All Audit Logs`` resource, which ships in the ``Admin``
+  role; custom roles that need to read these rows must be granted
+  ``All Audit Logs.can_read`` explicitly.
+- An auth manager that cannot map a requested ``AccessView`` now denies that view and emits
+  a ``UserWarning`` instead of raising ``KeyError``. This affects auth managers older than
+  the core they run against, which previously failed the request with a server error.

airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py

@@ -383,18 +383,48 @@ def authorize_view(self, *, access_view: AccessView, user: T, team_name: str | N
         Core calls this instead of :meth:`is_authorized_view` on team-scoped paths: an
         override still on the old ``(access_view, user)`` signature would otherwise raise
         ``TypeError``. Removed in Airflow 4.
+
+        A manager that does not recognise ``access_view`` at all is also tolerated, and
+        denied -- see :meth:`_authorize_view_unmapped_denied`.
+        """
+        try:
+            if self._is_authorized_view_team_aware:
+                return self.is_authorized_view(access_view=access_view, user=user, team_name=team_name)
+            warnings.warn(
+                f"The '{type(self).__name__}' auth manager is not team-aware, so team-scoped views are "
+                "authorized across all teams and may be visible to users of other teams. Add the "
+                "'team_name' argument to its is_authorized_view (or upgrade the provider). Airflow 4 "
+                "will require team-aware auth managers.",
+                RemovedInAirflow4Warning,
+                stacklevel=2,
+            )
+            return self.is_authorized_view(access_view=access_view, user=user)
+        except KeyError:
+            return self._authorize_view_unmapped_denied(access_view)
+
+    def _authorize_view_unmapped_denied(self, access_view: AccessView) -> bool:
+        """
+        Deny an ``AccessView`` the auth manager cannot map, instead of raising.
+
+        ``AccessView`` members are added by core, but auth managers ship as separately
+        released providers, so a core newer than the installed auth manager can name a
+        view the manager has never heard of. Managers that translate the enum through a
+        lookup table (the FAB auth manager, for one) raise ``KeyError`` on such a member,
+        which would surface as a 500 on the endpoint that consults it.
+
+        Denying keeps the endpoint working and fails closed: an unmappable view means the
+        manager cannot express who may see the records, and these views gate records with
+        no other authorization key. Upgrading the auth manager provider to a version that
+        maps the view restores access.
         """
-        if self._is_authorized_view_team_aware:
-            return self.is_authorized_view(access_view=access_view, user=user, team_name=team_name)
         warnings.warn(
-            f"The '{type(self).__name__}' auth manager is not team-aware, so team-scoped views are "
-            "authorized across all teams and may be visible to users of other teams. Add the "
-            "'team_name' argument to its is_authorized_view (or upgrade the provider). Airflow 4 "
-            "will require team-aware auth managers.",
-            RemovedInAirflow4Warning,
-            stacklevel=2,
+            f"The '{type(self).__name__}' auth manager cannot map the '{access_view.name}' view, so "
+            "access to it is denied. This usually means the auth manager provider is older than "
+            "Airflow core; upgrade it to a version that supports this view.",
+            UserWarning,
+            stacklevel=3,
         )
-        return self.is_authorized_view(access_view=access_view, user=user)
+        return False
 
     @abstractmethod
     def is_authorized_custom_view(self, *, method: ResourceMethod, resource_name: str, user: T) -> bool:

airflow-core/src/airflow/api_fastapi/auth/managers/models/resource_details.py

@@ -98,6 +98,10 @@ class VariableDetails:
 class AccessView(Enum):
     """Enum of specific views the user tries to access."""
 
+    # Audit log rows not tied to a Dag -- Connection, Variable, Pool, … operations:
+    # there is no per-Dag key to authorize on, so they get their own admin-by-default
+    # view rather than riding on Dag-level ``DagAccessEntity.AUDIT_LOG`` access.
+    AUDIT_LOGS_ALL = "AUDIT_LOGS_ALL"
     CLUSTER_ACTIVITY = "CLUSTER_ACTIVITY"
     DOCS = "DOCS"
     IMPORT_ERRORS = "IMPORT_ERRORS"

airflow-core/src/airflow/api_fastapi/auth/managers/simple/simple_auth_manager.py

@@ -354,11 +354,12 @@ def is_authorized_variable(
     def is_authorized_view(
         self, *, access_view: AccessView, user: SimpleAuthManagerUser, team_name: str | None = None
     ) -> bool:
-        # Import errors for files with no registered Dag are admin-only; every other
-        # view stays readable by viewers.
+        # Views covering records that have no per-Dag key to authorize on are admin-only --
+        # import errors for files with no registered Dag, and audit log rows not tied to a
+        # Dag. Every other view stays readable by viewers.
         allow_role = (
             SimpleAuthManagerRole.ADMIN
-            if access_view == AccessView.IMPORT_ERRORS_ALL
+            if access_view in (AccessView.IMPORT_ERRORS_ALL, AccessView.AUDIT_LOGS_ALL)
             else SimpleAuthManagerRole.VIEWER
         )
         return self._is_authorized(method="GET", allow_role=allow_role, user=user, team_name=team_name)

airflow-core/src/airflow/api_fastapi/core_api/security.py

@@ -265,12 +265,22 @@ def to_orm(self, select: Select) -> Select:
 
 
 class PermittedEventLogFilter(PermittedDagFilter):
-    """A parameter that filters the permitted even logs for the user."""
+    """A parameter that filters the permitted event logs for the user."""
+
+    def __init__(self, value: set[str] | None = None, *, include_non_dag_logs: bool = False):
+        super().__init__(value)
+        self.include_non_dag_logs = include_non_dag_logs
 
     def to_orm(self, select: Select) -> Select:
-        # Event Logs not related to Dags have dag_id as None and are always returned.
-        # return select.where(Log.dag_id.in_(self.value or set()) or Log.dag_id.is_(None))
-        return select.where(or_(Log.dag_id.in_(self.value or set()), Log.dag_id.is_(None)))
+        permitted_dag_logs = Log.dag_id.in_(self.value or set())
+        if not self.include_non_dag_logs:
+            return select.where(permitted_dag_logs)
+        # Event logs not related to a Dag have dag_id as None. They record Connection,
+        # Variable, Pool, … operations, so they carry no per-Dag key to authorize on and
+        # are gated on ``AccessView.AUDIT_LOGS_ALL`` instead of Dag-level audit access.
+        # Filtering here rather than after the fact keeps unauthorized rows out of the
+        # count and pagination too, so their existence does not leak either.
+        return select.where(or_(permitted_dag_logs, Log.dag_id.is_(None)))
 
 
 class PermittedTIFilter(PermittedDagFilter):
@@ -339,9 +349,32 @@ def depends_permitted_dags_filter(
 ReadableTIFilterDep = Annotated[
     PermittedTIFilter, Depends(permitted_dag_filter_factory("GET", PermittedTIFilter))
 ]
-ReadableEventLogsFilterDep = Annotated[
-    PermittedTIFilter, Depends(permitted_dag_filter_factory("GET", PermittedEventLogFilter))
-]
+
+
+def readable_event_logs_filter_factory() -> Callable[[BaseUser, BaseAuthManager], PermittedEventLogFilter]:
+    """
+    Create a callable for Depends in FastAPI that returns the event-log filter for the user.
+
+    Event logs need their own factory rather than ``permitted_dag_filter_factory``: besides
+    the readable Dag ids, the filter needs to know whether the user may read audit rows that
+    are not tied to a Dag, which is a separate authorization decision.
+    """
+
+    def depends_readable_event_logs_filter(
+        user: GetUserDep,
+        auth_manager: AuthManagerDep,
+    ) -> PermittedEventLogFilter:
+        return PermittedEventLogFilter(
+            auth_manager.get_authorized_dag_ids(user=user, method="GET"),
+            include_non_dag_logs=auth_manager.authorize_view(
+                access_view=AccessView.AUDIT_LOGS_ALL, user=user
+            ),
+        )
+
+    return depends_readable_event_logs_filter
+
+
+ReadableEventLogsFilterDep = Annotated[PermittedEventLogFilter, Depends(readable_event_logs_filter_factory())]
 ReadableXComFilterDep = Annotated[
     PermittedXComFilter, Depends(permitted_dag_filter_factory("GET", PermittedXComFilter))
 ]
@@ -423,7 +456,22 @@ async def inner(
                     status_code=status.HTTP_400_BAD_REQUEST,
                     detail="'event_log_id' must be an integer",
                 )
-            dag_id = session.scalar(select(Log.dag_id).where(Log.id == event_log_id))
+            # Select the id alongside dag_id: a NULL dag_id and a missing row both come back
+            # as None from a bare ``Log.dag_id`` scalar, and they authorize differently.
+            row = session.execute(select(Log.id, Log.dag_id).where(Log.id == event_log_id)).one_or_none()
+            if row is not None and row.dag_id is None:
+                # The row records an operation that is not tied to a Dag, so there is no
+                # per-Dag key to authorize on: gate it on ``AccessView.AUDIT_LOGS_ALL``,
+                # the same view the list endpoint's filter uses for these rows.
+                _requires_access(
+                    is_authorized_callback=lambda: get_auth_manager().authorize_view(
+                        access_view=AccessView.AUDIT_LOGS_ALL, user=user
+                    ),
+                )
+                return
+            # A missing row keeps the Dag-level check so the route can answer 404 rather
+            # than turning an unknown id into a permission error.
+            dag_id = row.dag_id if row is not None else None
 
         requires_access_dag(method, DagAccessEntity.AUDIT_LOG, dag_id)(
             request,

airflow-core/tests/unit/api_fastapi/auth/managers/simple/test_simple_auth_manager.py

@@ -347,6 +347,33 @@ def test_is_authorized_view_methods(self, auth_manager, api, kwargs, role, resul
             is result
         )
 
+    @pytest.mark.parametrize(
+        "access_view",
+        [AccessView.IMPORT_ERRORS_ALL, AccessView.AUDIT_LOGS_ALL],
+    )
+    @pytest.mark.parametrize(
+        ("role", "result"),
+        [
+            ("ADMIN", True),
+            ("OP", False),
+            ("USER", False),
+            ("VIEWER", False),
+        ],
+    )
+    def test_is_authorized_view_admin_only_views(self, auth_manager, access_view, role, result):
+        """The views covering records with no per-Dag key to authorize on are admin-only.
+
+        Every other view is readable by a viewer (asserted above); these two gate records
+        that carry no other authorization key -- import errors for files with no registered
+        Dag, and audit log rows not tied to a Dag -- so they must not ride on viewer access.
+        """
+        assert (
+            auth_manager.is_authorized_view(
+                access_view=access_view, user=SimpleAuthManagerUser(username="test", role=role)
+            )
+            is result
+        )
+
     @pytest.mark.parametrize(
         "api",
         [

airflow-core/tests/unit/api_fastapi/auth/managers/test_base_auth_manager.py

@@ -190,6 +190,26 @@ def is_authorized_view(self, *, access_view, user=None):  # old signature, no te
 
         assert result is True
 
+    def test_authorize_view_denies_a_view_the_manager_cannot_map(self):
+        # AccessView members are added by core, but auth managers ship as separately
+        # released providers, so a core newer than the installed manager can name a view
+        # the manager has never heard of. Managers that translate the enum through a lookup
+        # table raise KeyError on such a member, which would surface as a 500 on the
+        # endpoint. authorize_view denies instead -- the endpoint keeps working and the
+        # records these views gate stay closed.
+        class LookupTableAuthManager(EmptyAuthManager):
+            def is_authorized_view(self, *, access_view, user=None, team_name=None):
+                return {AccessView.WEBSITE: True}[access_view]
+
+        manager = LookupTableAuthManager()
+
+        with pytest.warns(UserWarning, match="cannot map the 'DOCS' view"):
+            result = manager.authorize_view(access_view=AccessView.DOCS, user=None)
+
+        assert result is False
+        # A view the manager does map is unaffected.
+        assert manager.authorize_view(access_view=AccessView.WEBSITE, user=None) is True
+
     def test_authorize_view_treats_kwargs_override_as_team_aware(self):
         class KwargsAuthManager(EmptyAuthManager):
             def is_authorized_view(self, *, access_view, user=None, **kwargs):

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

@@ -22,7 +22,11 @@
 import pytest
 from sqlalchemy.orm import Session
 
-from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity, DagDetails
+from airflow.api_fastapi.auth.managers.models.resource_details import (
+    AccessView,
+    DagAccessEntity,
+    DagDetails,
+)
 from airflow.models.log import Log
 from airflow.utils.session import NEW_SESSION, provide_session
 
@@ -236,6 +240,49 @@ def test_should_authorize_with_event_log_dag_id(self, test_client, setup):
             user=mock.ANY,
         )
 
+    @pytest.mark.parametrize(
+        ("can_view_all_audit_logs", "expected_status_code"),
+        [
+            pytest.param(True, 200, id="with-AUDIT_LOGS_ALL-sees-row"),
+            pytest.param(False, 403, id="without-AUDIT_LOGS_ALL-forbidden"),
+        ],
+    )
+    def test_non_dag_row_is_gated_on_audit_logs_all(
+        self, test_client, setup, can_view_all_audit_logs, expected_status_code
+    ):
+        """A row with a NULL dag_id records an operation that is not tied to a Dag -- a
+        Connection, Variable or Pool change -- so it has no per-Dag key to authorize on.
+        Visibility is gated on the dedicated ``AUDIT_LOGS_ALL`` view rather than riding on
+        Dag-level audit log access, which every viewer holds.
+        """
+        event_log_id = setup[EVENT_NORMAL].id
+        with mock.patch(
+            "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.is_authorized_view",
+            return_value=can_view_all_audit_logs,
+        ) as mock_is_authorized_view:
+            response = test_client.get(f"/eventLogs/{event_log_id}")
+
+        assert response.status_code == expected_status_code
+        mock_is_authorized_view.assert_called_once_with(
+            access_view=AccessView.AUDIT_LOGS_ALL, user=mock.ANY, team_name=None
+        )
+
+    def test_unknown_id_stays_404_and_does_not_consult_audit_logs_all(self, test_client, setup):
+        """An id that matches no row must answer 404, not 403.
+
+        A missing row and a NULL dag_id both read back as ``None``, so the guard has to tell
+        them apart: turning an unknown id into a permission error would change the documented
+        contract of the endpoint for callers that are allowed to use it.
+        """
+        with mock.patch(
+            "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.is_authorized_view",
+            return_value=False,
+        ) as mock_is_authorized_view:
+            response = test_client.get(f"/eventLogs/{EVENT_NON_EXISTED_ID}")
+
+        assert response.status_code == 404
+        mock_is_authorized_view.assert_not_called()
+
     @provide_session
     def test_should_return_404_for_log_without_dttm(self, test_client, *, session: Session = NEW_SESSION):  # noqa: PT028
         event_log = Log(event=EVENT_WITHOUT_DTTM)
@@ -463,3 +510,43 @@ def test_should_raises_401_unauthenticated(self, unauthenticated_test_client):
     def test_should_raises_403_forbidden(self, unauthorized_test_client):
         response = unauthorized_test_client.get("/eventLogs")
         assert response.status_code == 403
+
+    @pytest.mark.parametrize(
+        ("can_view_all_audit_logs", "expected_events"),
+        [
+            pytest.param(
+                True,
+                [EVENT_NORMAL, EVENT_WITH_OWNER, TASK_INSTANCE_EVENT, EVENT_WITH_OWNER_AND_TASK_INSTANCE],
+                id="with-AUDIT_LOGS_ALL-sees-non-dag-rows",
+            ),
+            pytest.param(
+                False,
+                [TASK_INSTANCE_EVENT, EVENT_WITH_OWNER_AND_TASK_INSTANCE],
+                id="without-AUDIT_LOGS_ALL-only-dag-rows",
+            ),
+        ],
+    )
+    def test_non_dag_rows_are_gated_on_audit_logs_all(
+        self, test_client, can_view_all_audit_logs, expected_events
+    ):
+        """Rows with a NULL dag_id are returned only to callers holding ``AUDIT_LOGS_ALL``.
+
+        Before this gate every caller that could read event logs at all received them, which
+        for the default auth manager is any viewer. ``EVENT_NORMAL`` and ``EVENT_WITH_OWNER``
+        carry no dag_id; the other two are bound to a Dag and stay visible either way.
+        """
+        with mock.patch(
+            "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.is_authorized_view",
+            return_value=can_view_all_audit_logs,
+        ) as mock_is_authorized_view:
+            response = test_client.get("/eventLogs")
+
+        assert response.status_code == 200
+        resp_json = response.json()
+        # Filtered in the query, so the excluded rows are absent from the count and
+        # pagination too -- their existence does not leak through total_entries.
+        assert resp_json["total_entries"] == len(expected_events)
+        assert {event_log["event"] for event_log in resp_json["event_logs"]} == set(expected_events)
+        mock_is_authorized_view.assert_called_once_with(
+            access_view=AccessView.AUDIT_LOGS_ALL, user=mock.ANY, team_name=None
+        )