feat(airflow): Call OPA from FabAuthManager

Author: siegfriedweber

airflow/opa-auth-manager/opa_auth_manager/opa_fab_auth_manager.py

@@ -16,6 +16,7 @@
 )
 from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
 from airflow.utils.log.logging_mixin import LoggingMixin
+import requests
 
 class OpaFabAuthManager(FabAuthManager, LoggingMixin):
     """
@@ -42,7 +43,24 @@ def is_authorized_configuration(
 
         self.log.info("Forward is_authorized_configuration to OPA")
 
-        return True
+        if not user:
+            user = self.get_user()
+
+        input= {
+            'method': method,
+            'details': details,
+            'user': {
+                'id': user.get_id(),
+                'name': user.get_name(),
+            },
+        }
+        response = requests.post(
+            'http://opa:8081/v1/data/airflow/is_authorized_configuration',
+            json=input,
+            timeout=10
+        ).json()
+
+        return response.get("result") == "True"
 
     def is_authorized_connection(
         self,

airflow/opa-auth-manager/poetry.lock

@@ -7,12 +7,17 @@ readme = "README.md"
 
 [tool.poetry.dependencies]
 python = ">=3.9.0,<3.13"
+requests = "^2.32.3"
 
 [tool.poetry.group.dev.dependencies]
 apache-airflow = "^2.10.3"
 pylint = "^3.3.1"
 pytest = "^8.3.3"
 
+
+[tool.poetry.group.test.dependencies]
+requests-mock = "^1.12.1"
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

airflow/opa-auth-manager/pyproject.toml

@@ -1,3 +1,4 @@
+from airflow.auth.managers.models.base_user import BaseUser
 from airflow.auth.managers.models.resource_details import AccessView
 import pytest
 
@@ -6,20 +7,41 @@
 from opa_auth_manager.opa_fab_auth_manager import OpaFabAuthManager
 
 @pytest.fixture
-def flask_app():
+def flask_app() -> Flask:
     return Flask(__name__)
 
 @pytest.fixture
-def auth_manager(flask_app):
+def auth_manager(flask_app) -> OpaFabAuthManager:
     appbuilder = init_appbuilder(flask_app)
     return OpaFabAuthManager(appbuilder)
 
+class User(BaseUser):
+
+    def __init__(self, username: str) -> None:
+        self.username = username
+
+    def get_id(self) -> str:
+        return self.username
+
+    def get_name(self) -> str:
+        return self.username
+
 @pytest.mark.db_test
 class TestOpaFabAuthManager:
 
-    def test_is_authorized_configuration(self, auth_manager: OpaFabAuthManager):
-        result = auth_manager.is_authorized_configuration(
+    def test_is_authorized_configuration(
+            self,
+            auth_manager: OpaFabAuthManager,
+            requests_mock
+    ) -> None:
+        requests_mock.post(
+            'http://opa:8081/v1/data/airflow/is_authorized_configuration',
+            text='{ "result": "True" }'
+        )
+
+        result: bool = auth_manager.is_authorized_configuration(
             method="GET",
+            user=User(username='testuser'),
         )
         expected_result = True
         assert result == expected_result