fix: Support OPA for Airflow when user role is not 'Admin'

sweb · sweb · commit 46ef412799a7 · 2026-05-28T12:48:40.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -24,6 +24,7 @@ All notable changes to this project will be documented in this file.
 - vector: Look for SBOM in correct location ([#1471]).
 - vector: Use correct license ([#1476]).
 - trino: Build a patched Airlift from source and depend on it to backport [airlift/airlift#1943](https://github.com/airlift/airlift/pull/1943), applying the configured max response header size to Jetty's `maxResponseHeaderSize` ([#1510]).
+- airflow: Route DAG listings and menu items through OPA in the Airflow 3 OPA auth manager, and wire the OPA cache on the FastAPI api-server init path ([#XXXX]).
 
 ### Removed
 
diff --git a/airflow/opa-auth-manager/airflow-3/opa_auth_manager/opa_fab_auth_manager.py b/airflow/opa-auth-manager/airflow-3/opa_auth_manager/opa_fab_auth_manager.py
@@ -19,13 +19,18 @@
     PoolDetails,
     VariableDetails,
 )
+from airflow.api_fastapi.common.types import MenuItem
 from airflow.configuration import conf
+from airflow.models.dag import DagModel
 from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
 from airflow.providers.fab.auth_manager.models import User
 from airflow.stats import Stats
 from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.session import NEW_SESSION, provide_session
 from cachetools import TTLCache, cachedmethod
 from overrides import override
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 METRIC_NAME_OPA_CACHE_LIMIT_REACHED = "opa_cache_limit_reached"
 
@@ -75,12 +80,36 @@ class OpaFabAuthManager(FabAuthManager, LoggingMixin):
     AUTH_OPA_REQUEST_TIMEOUT_DEFAULT = 10
 
     @override
-    def init_flask_resources(self) -> None:
+    def init(self) -> None:
         """
         Run operations when Airflow is initializing.
+
+        Called by the FastAPI api-server during startup.
+        """
+
+        super().init()
+        self._init_opa_resources()
+
+    @override
+    def init_flask_resources(self) -> None:
+        """
+        Run operations when the Flask app (FAB UI) is initializing.
         """
 
         super().init_flask_resources()
+        self._init_opa_resources()
+
+    def _init_opa_resources(self) -> None:
+        """
+        Set up the OPA cache and HTTP session.
+
+        Idempotent so it can be called from both ``init`` (FastAPI api-server)
+        and ``init_flask_resources`` (Flask AppBuilder) — only one of the two
+        runs depending on the Airflow component.
+        """
+
+        if getattr(self, "opa_cache", None) is not None:
+            return
 
         Stats.incr(METRIC_NAME_OPA_CACHE_LIMIT_REACHED, count=0)
 
@@ -121,9 +150,6 @@ def _is_authorized_in_opa(self, endpoint: str, input: OpaInput) -> bool:
 
         self.log.debug("Forward authorization request to OPA")
 
-        opa_url = conf.get(
-            "core", "AUTH_OPA_REQUEST_URL", fallback=self.AUTH_OPA_REQUEST_URL_DEFAULT
-        )
         opa_url = conf.get(
             "core", "AUTH_OPA_REQUEST_URL", fallback=self.AUTH_OPA_REQUEST_URL_DEFAULT
         )
@@ -551,3 +577,63 @@ def is_authorized_custom_view(
                 }
             ),
         )
+
+    @provide_session
+    @override
+    def get_authorized_dag_ids(
+        self,
+        *,
+        user: User,
+        method: ResourceMethod = "GET",
+        session: Session = NEW_SESSION,
+    ) -> set[str]:
+        # FabAuthManager's implementation consults the user's FAB DB role
+        # permissions and bypasses is_authorized_dag entirely, which makes any
+        # user without a FAB role (e.g. the default Public role) see an empty
+        # DAG list even when OPA would allow them. Re-implement the
+        # BaseAuthManager default: list all DAG ids and filter them via
+        # is_authorized_dag → OPA.
+        dag_ids = {dag.dag_id for dag in session.execute(select(DagModel.dag_id))}
+        return self.filter_authorized_dag_ids(dag_ids=dag_ids, method=method, user=user)
+
+    @override
+    def filter_authorized_menu_items(
+        self, menu_items: list[MenuItem], user: User
+    ) -> list[MenuItem]:
+        # FabAuthManager filters menu items via the FAB role permissions in the
+        # DB, which yields an empty menu for users without FAB perms even when
+        # OPA grants access. Route each menu item through the matching
+        # OPA-backed is_authorized_* call instead.
+        return [
+            item for item in menu_items if self._is_menu_item_authorized(item, user)
+        ]
+
+    def _is_menu_item_authorized(self, menu_item: MenuItem, user: User) -> bool:
+        if menu_item == MenuItem.ASSETS:
+            return self.is_authorized_asset(method="GET", user=user)
+        if menu_item == MenuItem.AUDIT_LOG:
+            return self.is_authorized_dag(
+                method="GET", access_entity=DagAccessEntity.AUDIT_LOG, user=user
+            )
+        if menu_item == MenuItem.CONFIG:
+            return self.is_authorized_configuration(method="GET", user=user)
+        if menu_item == MenuItem.CONNECTIONS:
+            return self.is_authorized_connection(method="GET", user=user)
+        if menu_item == MenuItem.DAGS:
+            return self.is_authorized_dag(method="GET", user=user)
+        if menu_item == MenuItem.DOCS:
+            return self.is_authorized_view(access_view=AccessView.DOCS, user=user)
+        if menu_item == MenuItem.PLUGINS:
+            return self.is_authorized_view(access_view=AccessView.PLUGINS, user=user)
+        if menu_item == MenuItem.POOLS:
+            return self.is_authorized_pool(method="GET", user=user)
+        if menu_item == MenuItem.PROVIDERS:
+            return self.is_authorized_view(access_view=AccessView.PROVIDERS, user=user)
+        if menu_item == MenuItem.VARIABLES:
+            return self.is_authorized_variable(method="GET", user=user)
+        if menu_item == MenuItem.XCOMS:
+            return self.is_authorized_dag(
+                method="GET", access_entity=DagAccessEntity.XCOM, user=user
+            )
+        self.log.warning("Unknown menu item %s — denying", menu_item)
+        return False
diff --git a/airflow/opa-auth-manager/airflow-3/tests/test_opa_fab_auth_manager.py b/airflow/opa-auth-manager/airflow-3/tests/test_opa_fab_auth_manager.py
@@ -11,9 +11,11 @@
 
 import pytest
 from airflow.api_fastapi.auth.managers.models.resource_details import (
+    AccessView,
     DagAccessEntity,
     DagDetails,
 )
+from airflow.api_fastapi.common.types import MenuItem
 from airflow.providers.fab.www.extensions.init_appbuilder import init_appbuilder
 from airflow.providers.fab.www.security.permissions import (
     ACTION_CAN_CREATE,
@@ -56,6 +58,17 @@ def auth_manager_with_appbuilder(flask_app):
 
 
 class TestOpaFabAuthManager:
+    def test_init_wires_opa_cache_for_fastapi_apiserver(self):
+        # The FastAPI api-server calls auth_manager.init() instead of
+        # init_flask_resources(). Without wiring the cache from init() too,
+        # any is_authorized_* call from a REST handler crashes with
+        # AttributeError: 'OpaFabAuthManager' object has no attribute 'opa_cache'.
+        auth_manager = OpaFabAuthManager()
+        auth_manager.init()
+
+        assert auth_manager.opa_cache is not None
+        assert auth_manager.opa_session is not None
+
     @pytest.mark.parametrize(
         "method, dag_access_entity, dag_details, user_permissions, expected_opa_result, expected_result",
         [
@@ -228,3 +241,153 @@ def test_is_authorized_dag(
                 user=user,
             )
             assert result == expected_result
+
+    def test_get_authorized_dag_ids_uses_opa_not_fab_db(
+        self, auth_manager_with_appbuilder
+    ):
+        # Repro for the OPA listing bug: a user with no FAB permissions
+        # (e.g. the default Public role) must still see the DAGs that OPA
+        # allows. The FabAuthManager base override would return set() here
+        # because it reads roles from the metadata DB.
+        user = Mock()
+        user.id = 1
+        user.perms = []
+
+        all_dag_ids = {"allowed_dag", "denied_dag"}
+        session = Mock()
+        session.execute.return_value = [Mock(dag_id=dag_id) for dag_id in all_dag_ids]
+
+        def fake_is_authorized_dag(*, method, details=None, access_entity=None, user):
+            return details is not None and details.id == "allowed_dag"
+
+        with mock.patch.object(
+            auth_manager_with_appbuilder,
+            "is_authorized_dag",
+            side_effect=fake_is_authorized_dag,
+        ):
+            result = auth_manager_with_appbuilder.get_authorized_dag_ids(
+                user=user, method="GET", session=session
+            )
+
+        assert result == {"allowed_dag"}
+
+    def test_get_authorized_dag_ids_provides_session_when_caller_omits_it(
+        self, auth_manager_with_appbuilder
+    ):
+        # Real callers (api_fastapi/core_api/security.py) don't pass `session`.
+        # Our override must rely on @provide_session to inject one; previously
+        # it forwarded the default NEW_SESSION (None) and crashed with
+        # 'NoneType' has no attribute 'execute'.
+        user = Mock()
+        user.perms = []
+
+        session = Mock()
+        session.execute.return_value = [Mock(dag_id="allowed_dag")]
+
+        with (
+            mock.patch("airflow.utils.session.create_session") as mock_create_session,
+            mock.patch.object(
+                auth_manager_with_appbuilder, "is_authorized_dag", return_value=True
+            ),
+        ):
+            mock_create_session.return_value.__enter__.return_value = session
+            result = auth_manager_with_appbuilder.get_authorized_dag_ids(
+                user=user, method="GET"
+            )
+
+        assert result == {"allowed_dag"}
+        mock_create_session.assert_called_once()
+
+    @pytest.mark.parametrize(
+        "menu_item, expected_method, expected_kwargs",
+        [
+            (MenuItem.ASSETS, "is_authorized_asset", {"method": "GET"}),
+            (
+                MenuItem.AUDIT_LOG,
+                "is_authorized_dag",
+                {"method": "GET", "access_entity": DagAccessEntity.AUDIT_LOG},
+            ),
+            (MenuItem.CONFIG, "is_authorized_configuration", {"method": "GET"}),
+            (MenuItem.CONNECTIONS, "is_authorized_connection", {"method": "GET"}),
+            (MenuItem.DAGS, "is_authorized_dag", {"method": "GET"}),
+            (MenuItem.DOCS, "is_authorized_view", {"access_view": AccessView.DOCS}),
+            (
+                MenuItem.PLUGINS,
+                "is_authorized_view",
+                {"access_view": AccessView.PLUGINS},
+            ),
+            (MenuItem.POOLS, "is_authorized_pool", {"method": "GET"}),
+            (
+                MenuItem.PROVIDERS,
+                "is_authorized_view",
+                {"access_view": AccessView.PROVIDERS},
+            ),
+            (MenuItem.VARIABLES, "is_authorized_variable", {"method": "GET"}),
+            (
+                MenuItem.XCOMS,
+                "is_authorized_dag",
+                {"method": "GET", "access_entity": DagAccessEntity.XCOM},
+            ),
+        ],
+    )
+    def test_filter_authorized_menu_items_routes_through_opa(
+        self,
+        menu_item,
+        expected_method,
+        expected_kwargs,
+        auth_manager_with_appbuilder,
+    ):
+        # Each MenuItem must trigger the matching OPA-backed is_authorized_*
+        # call, so menu visibility is OPA-driven rather than FAB-DB-driven.
+        user = Mock()
+        user.perms = []
+
+        with mock.patch.object(
+            auth_manager_with_appbuilder, expected_method, return_value=True
+        ) as mocked:
+            allowed = auth_manager_with_appbuilder.filter_authorized_menu_items(
+                [menu_item], user=user
+            )
+        assert allowed == [menu_item]
+        mocked.assert_called_once_with(user=user, **expected_kwargs)
+
+        with mock.patch.object(
+            auth_manager_with_appbuilder, expected_method, return_value=False
+        ):
+            denied = auth_manager_with_appbuilder.filter_authorized_menu_items(
+                [menu_item], user=user
+            )
+        assert denied == []
+
+    def test_filter_authorized_menu_items_preserves_order_and_filters(
+        self, auth_manager_with_appbuilder
+    ):
+        user = Mock()
+        user.perms = []
+
+        def fake_is_authorized_dag(*, method, details=None, access_entity=None, user):
+            return access_entity is None
+
+        with (
+            mock.patch.object(
+                auth_manager_with_appbuilder,
+                "is_authorized_dag",
+                side_effect=fake_is_authorized_dag,
+            ),
+            mock.patch.object(
+                auth_manager_with_appbuilder,
+                "is_authorized_connection",
+                return_value=True,
+            ),
+            mock.patch.object(
+                auth_manager_with_appbuilder,
+                "is_authorized_view",
+                return_value=False,
+            ),
+        ):
+            result = auth_manager_with_appbuilder.filter_authorized_menu_items(
+                [MenuItem.DAGS, MenuItem.DOCS, MenuItem.CONNECTIONS, MenuItem.XCOMS],
+                user=user,
+            )
+
+        assert result == [MenuItem.DAGS, MenuItem.CONNECTIONS]
