This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 aef881370ea Sync Dag permissions for CustomAuthManager (#74022)
aef881370ea is described below
commit aef881370eaa836609b8ce6ca8c5094a56e33f6e
Author: AutomationDev85 <[email protected]>
AuthorDate: Fri Oct 9 20:16:28 2026 +0200
Sync Dag permissions for CustomAuthManager (#74022)
---
.../src/airflow/dag_processing/collection.py | 18 ++----
.../tests/unit/dag_processing/test_collection.py | 64 ++++++++++++++++------
.../providers/fab/auth_manager/fab_auth_manager.py | 9 +++
.../unit/fab/auth_manager/test_fab_auth_manager.py | 18 ++++++
4 files changed, 80 insertions(+), 29 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/collection.py
b/airflow-core/src/airflow/dag_processing/collection.py
index d6fcbfd5646..3e20722c6c2 100644
--- a/airflow-core/src/airflow/dag_processing/collection.py
+++ b/airflow-core/src/airflow/dag_processing/collection.py
@@ -267,6 +267,7 @@ def _serialize_dag_capturing_errors(
We can't place them directly in import_errors, as this may be retried, and
work the next time
"""
+ from airflow.api_fastapi.app import get_auth_manager_cls
from airflow.models.dagcode import DagCode
# Updating serialized DAG can not be faster than a minimum interval to
reduce database write rate.
@@ -291,8 +292,10 @@ def _serialize_dag_capturing_errors(
DagCode.update_source_code(
dag.dag_id, dag.fileloc, dag_source_code=dag_source_code,
session=session
)
- if "FabAuthManager" in conf.get("core", "auth_manager"):
- _sync_dag_perms(dag, session=session)
+
+ sync_dag_perms = getattr(get_auth_manager_cls(), "sync_dag_perms",
None)
+ if sync_dag_perms:
+ sync_dag_perms(dag, session=session)
return []
except OperationalError:
@@ -308,17 +311,6 @@ def _serialize_dag_capturing_errors(
]
-def _sync_dag_perms(dag: LazyDeserializedDAG, session: Session):
- """Sync DAG specific permissions."""
- dag_id = dag.dag_id
-
- log.debug("Syncing DAG permissions: %s to the DB", dag_id)
- from airflow.providers.fab.www.security_appless import
ApplessAirflowSecurityManager
-
- security_manager = ApplessAirflowSecurityManager(session=session)
- security_manager.sync_perm_for_dag(dag_id, dag.access_control)
-
-
def _build_duplicate_dag_id_warnings(
dags: Collection[LazyDeserializedDAG],
bundle_name: str,
diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py
b/airflow-core/tests/unit/dag_processing/test_collection.py
index 0f66f067b83..8a7fe712bb5 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -26,7 +26,6 @@ import textwrap
import warnings
from collections.abc import Generator
from datetime import timedelta
-from typing import TYPE_CHECKING
from unittest import mock
from unittest.mock import patch
@@ -106,9 +105,6 @@ from tests_common.test_utils.db import (
from tests_common.test_utils.mock_plugins import mock_plugin_manager
from unit.plugins.priority_weight_strategy import
StaticTestPriorityWeightStrategy
-if TYPE_CHECKING:
- from kgb import SpyAgency
-
mark_fab_auth_manager_test = pytest.mark.skipif(
condition="FabAuthManager" not in conf.get("core", "auth_manager"),
reason="This is only for FabAuthManager. Please set the environment
variable `AIRFLOW__CORE__AUTH_MANAGER` to
`airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager` in
`files/airflow-breeze-config/environment_variables.env` before running breeze
shell. To run the test, add the flag `--keep-env-variables` to the pytest
command.",
@@ -712,11 +708,50 @@ class TestUpdateDagParsingResults:
yield dag_import_error_listener
dag_import_error_listener.clear()
+ @patch("airflow.api_fastapi.app._AuthManagerState.instance", None)
+ @patch.object(SerializedDagModel, "write_dag", return_value=True,
autospec=True)
+ @patch("airflow.api_fastapi.app.get_auth_manager_cls", autospec=True)
+ def test_sync_perms_for_fab_auth_manager_subclass(
+ self, mock_get_auth_manager_cls, mock_write_dag, session
+ ):
+ FabAuthManager = pytest.importorskip(
+ "airflow.providers.fab.auth_manager.fab_auth_manager"
+ ).FabAuthManager
+
+ class CustomAuthManager(FabAuthManager):
+ pass
+
+ mock_get_auth_manager_cls.return_value = CustomAuthManager
+ dag = DAG(dag_id="test")
+
+ with patch.object(FabAuthManager, "sync_dag_perms", autospec=True) as
mock_sync_dag_perms:
+ assert (
+
airflow.dag_processing.collection._serialize_dag_capturing_errors(
+ dag, "testing", session, None
+ )
+ == []
+ )
+ mock_sync_dag_perms.assert_called_once_with(dag, session=session)
+
+ @patch.object(SerializedDagModel, "write_dag", return_value=True,
autospec=True)
+ @patch("airflow.api_fastapi.app.get_auth_manager_cls", autospec=True)
+ def test_serialize_without_auth_manager_sync_hook(
+ self, mock_get_auth_manager_cls, mock_write_dag, session
+ ):
+ mock_get_auth_manager_cls.return_value = object
+ dag = DAG(dag_id="test")
+
+ assert (
+
airflow.dag_processing.collection._serialize_dag_capturing_errors(dag,
"testing", session, None)
+ == []
+ )
+
@mark_fab_auth_manager_test
@conf_vars({("core", "min_serialized_dag_update_interval"): "5"})
@pytest.mark.usefixtures("clean_db") # sync_perms in fab has bad session
commit hygiene
+
@patch("airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager.sync_dag_perms",
autospec=True)
def test_sync_perms_syncs_dag_specific_perms_on_update(
- self, monkeypatch, spy_agency: SpyAgency, session, time_machine,
testing_dag_bundle
+ self, mock_sync_dag_perms, monkeypatch, session, time_machine,
testing_dag_bundle
):
"""Test DAG-specific permissions are synced when a DAG is new or
updated"""
serialized_dags_count =
session.scalar(select(func.count(SerializedDagModel.dag_id)))
@@ -726,29 +761,26 @@ class TestUpdateDagParsingResults:
dag = DAG(dag_id="test")
- sync_perms_spy = spy_agency.spy_on(
- airflow.dag_processing.collection._sync_dag_perms,
- call_original=False,
- )
-
def _sync_to_db():
- sync_perms_spy.reset_calls()
+ mock_sync_dag_perms.reset_mock()
time_machine.shift(20)
- update_dag_parsing_results_in_db("testing", None, [dag], dict(),
None, set(), session)
+ import_errors = {}
+ update_dag_parsing_results_in_db("testing", None, [dag],
import_errors, None, set(), session)
+ assert not import_errors
_sync_to_db()
- spy_agency.assert_spy_called_with(sync_perms_spy, dag, session=session)
+ mock_sync_dag_perms.assert_called_once_with(dag, session=session)
# DAG isn't updated
_sync_to_db()
- # `_sync_dag_perms` should be called even the DAG isn't updated.
Otherwise, any import error will not show up until DAG is updated.
- spy_agency.assert_spy_called_with(sync_perms_spy, dag, session=session)
+ # `sync_dag_perms` should be called even the DAG isn't updated.
Otherwise, any import error will not show up until DAG is updated.
+ mock_sync_dag_perms.assert_called_once_with(dag, session=session)
# DAG is updated
dag.tags = {"new_tag"}
_sync_to_db()
- spy_agency.assert_spy_called_with(sync_perms_spy, dag, session=session)
+ mock_sync_dag_perms.assert_called_once_with(dag, session=session)
serialized_dags_count =
session.scalar(select(func.count(SerializedDagModel.dag_id)))
diff --git
a/providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py
b/providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py
index 01cc28d5670..4a0ed10b73d 100644
--- a/providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py
+++ b/providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py
@@ -97,6 +97,7 @@ from airflow.providers.fab.www.security.permissions import (
RESOURCE_WEBSITE,
RESOURCE_XCOM,
)
+from airflow.providers.fab.www.security_appless import
ApplessAirflowSecurityManager
from airflow.providers.fab.www.utils import get_fab_action_from_method_map
from airflow.utils.session import NEW_SESSION, create_session, provide_session
@@ -117,6 +118,7 @@ if TYPE_CHECKING:
RESOURCE_ASSET,
RESOURCE_ASSET_ALIAS,
)
+ from airflow.serialization.serialized_objects import LazyDeserializedDAG
else:
from airflow.providers.common.compat.security.permissions import (
RESOURCE_ASSET,
@@ -214,6 +216,13 @@ class FabAuthManager(BaseAuthManager[User]):
def init_flask_resources(self) -> None:
self._sync_appbuilder_roles()
+ @staticmethod
+ def sync_dag_perms(dag: LazyDeserializedDAG, *, session: Session) -> None:
+ """Sync Dag-specific permissions without initializing the auth
manager."""
+ log.debug("Syncing DAG permissions: %s to the DB", dag.dag_id)
+ security_manager = ApplessAirflowSecurityManager(session=session)
+ security_manager.sync_perm_for_dag(dag.dag_id, dag.access_control)
+
@cached_property
def apiserver_endpoint(self) -> str:
return conf.get("api", "base_url", fallback="/")
diff --git a/providers/fab/tests/unit/fab/auth_manager/test_fab_auth_manager.py
b/providers/fab/tests/unit/fab/auth_manager/test_fab_auth_manager.py
index 2d79c00512e..34d5bfa35fe 100644
--- a/providers/fab/tests/unit/fab/auth_manager/test_fab_auth_manager.py
+++ b/providers/fab/tests/unit/fab/auth_manager/test_fab_auth_manager.py
@@ -28,6 +28,7 @@ import pytest
from flask import g
from flask_appbuilder.const import AUTH_DB, AUTH_LDAP
from sqlalchemy.exc import OperationalError, PendingRollbackError
+from sqlalchemy.orm import Session
from airflow import settings
from airflow.api_fastapi.app import AUTH_MANAGER_FASTAPI_APP_PREFIX
@@ -138,6 +139,23 @@ if TYPE_CHECKING:
from airflow.api_fastapi.auth.managers.base_auth_manager import
ResourceMethod
[email protected](
+
"airflow.providers.fab.auth_manager.fab_auth_manager.ApplessAirflowSecurityManager",
autospec=True
+)
+def
test_sync_dag_perms_without_auth_manager_initialization(mock_security_manager):
+ dag = Mock(spec=["dag_id", "access_control"])
+ dag.dag_id = "test"
+ dag.access_control = {"Viewer": {"can_read"}}
+ session = Mock(spec=Session)
+
+ FabAuthManager.sync_dag_perms(dag, session=session)
+
+ mock_security_manager.assert_called_once_with(session=session)
+
mock_security_manager.return_value.sync_perm_for_dag.assert_called_once_with(
+ dag.dag_id, dag.access_control
+ )
+
+
IS_AUTHORIZED_METHODS_SIMPLE = {
"is_authorized_configuration": RESOURCE_CONFIG,
"is_authorized_connection": RESOURCE_CONNECTION,
