Skip to content

Commit 87288e2

Browse files
razvanrazvan
committed
tests: airflow 3 opa auth manager (and some reformatting)
1 parent b82f3dc commit 87288e2

File tree

5 files changed

+3328
-24
lines changed

5 files changed

+3328
-24
lines changed

‎airflow/Dockerfile‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,18 +11,29 @@ FROM oci.stackable.tech/sdp/git-sync/git-sync:${GIT_SYNC} AS gitsync-image
1111

1212
FROM stackable/image/shared/statsd-exporter AS statsd_exporter-builder
1313

14-
FROM python:3.12-bookworm AS opa-auth-manager-builder
14+
FROM stackable/image/vector AS opa-auth-manager-builder
1515

1616
ARG OPA_AUTH_MANAGER
17+
ARG PYTHON
1718
ARG UV
1819

1920
COPY airflow/opa-auth-manager/${OPA_AUTH_MANAGER} /tmp/opa-auth-manager
2021

2122
WORKDIR /tmp/opa-auth-manager
2223

2324
RUN <<EOF
24-
pip install --no-cache-dir uv==${UV}
25-
uv run pytest
25+
microdnf update
26+
microdnf install python${PYTHON}-pip
27+
microdnf clean all
28+
29+
pip${PYTHON} install --no-cache-dir uv==${UV}
30+
31+
# This folder is required by the tests to set up an sqlite database
32+
mkdir /root/airflow
33+
34+
# Warnings are disabled because they come from various third party testing libraries
35+
# that we have no control over.
36+
uv run pytest --disable-warnings
2637
uv build
2738
EOF
2839

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

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,15 @@
22
Custom Auth manager for Airflow
33
"""
44

5-
from airflow.providers.fab.auth_manager.models import User
6-
from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
5+
import json
6+
from typing import Optional, Union
77

8+
import requests
89
from airflow.api_fastapi.auth.managers.base_auth_manager import ResourceMethod
9-
from airflow.configuration import conf
1010
from airflow.api_fastapi.auth.managers.models.resource_details import (
1111
AccessView,
12-
AssetDetails,
1312
AssetAliasDetails,
13+
AssetDetails,
1414
BackfillDetails,
1515
ConfigurationDetails,
1616
ConnectionDetails,
@@ -19,13 +19,13 @@
1919
PoolDetails,
2020
VariableDetails,
2121
)
22+
from airflow.configuration import conf
23+
from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
24+
from airflow.providers.fab.auth_manager.models import User
2225
from airflow.stats import Stats
2326
from airflow.utils.log.logging_mixin import LoggingMixin
2427
from cachetools import TTLCache, cachedmethod
25-
from typing import Optional, Union
2628
from overrides import override
27-
import json
28-
import requests
2929

3030
METRIC_NAME_OPA_CACHE_LIMIT_REACHED = "opa_cache_limit_reached"
3131

@@ -246,7 +246,6 @@ def is_authorized_dag(
246246
:param user: the user to perform the action on. If not provided (or None), it uses the
247247
current user
248248
"""
249-
250249
self.log.debug("Check is_authorized_dag")
251250

252251
if not access_entity:
@@ -259,6 +258,23 @@ def is_authorized_dag(
259258
else:
260259
dag_id = details.id
261260

261+
print(
262+
OpaInput(
263+
{
264+
"input": {
265+
"method": method,
266+
"access_entity": entity,
267+
"details": {
268+
"id": dag_id,
269+
},
270+
"user": {
271+
"id": user.get_id(),
272+
"name": user.get_name(),
273+
},
274+
}
275+
}
276+
).to_dict()
277+
)
262278
return self._is_authorized_in_opa(
263279
"is_authorized_dag",
264280
OpaInput(

‎airflow/opa-auth-manager/airflow-3/pyproject.toml‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,24 +3,20 @@ name = "opa-auth-manager"
33
version = "0.1.0"
44
description = "Auth manager for Airflow 3 which delegates the authorization to an Open Policy Agent"
55
authors = [
6-
{ name = "Siegfried Weber", email="[email protected]"},
7-
{ name = "Razvan Daniel Mihai", email="[email protected]"}
6+
{ name = "Siegfried Weber", email = "[email protected]" },
7+
{ name = "Razvan Daniel Mihai", email = "[email protected]" },
88
]
99
readme = "README.md"
1010
requires-python = ">=3.12,<3.13"
1111

12-
dependencies = [
13-
"requests==2.32.*",
14-
"cachetools==5.5.*",
15-
"overrides==7.7.*",
16-
]
12+
dependencies = ["requests==2.32.*", "cachetools==5.5.*", "overrides==7.7.*"]
1713

1814
[dependency-groups]
1915
dev = [
2016
"apache-airflow~=3.0.1",
21-
"pylint==3.3.*",
17+
"apache-airflow-devel-common<=0.1.1",
2218
"apache-airflow-providers-fab==2.0.*",
23-
"pytest~=8.3.3"
19+
"pytest~=8.3.3",
2420
]
2521

2622
[build-system]

‎airflow/opa-auth-manager/airflow-3/tests/test_opa_fab_auth_manager.py‎

Lines changed: 230 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,230 @@
1-
def test_todo():
2-
# This is a placeholder test function.
3-
# Replace with actual test logic.
4-
assert True
1+
#
2+
# To make these tests relevant we would have to package the OPA auth manager as
3+
# an Airflow provider and use `breeze` to set up a docker environment with Airflow
4+
# and an SQLite database.
5+
#
6+
# Then we could run these tests against the Airflow instance and use the Airflow API to
7+
# actually test the effect of Rego policies on user authorization.
8+
#
9+
from unittest import mock
10+
from unittest.mock import Mock
11+
12+
import pytest
13+
from airflow.api_fastapi.auth.managers.models.resource_details import (
14+
DagAccessEntity,
15+
DagDetails,
16+
)
17+
from airflow.providers.fab.www.extensions.init_appbuilder import init_appbuilder
18+
from airflow.providers.fab.www.security.permissions import (
19+
ACTION_CAN_CREATE,
20+
ACTION_CAN_DELETE,
21+
ACTION_CAN_EDIT,
22+
ACTION_CAN_READ,
23+
RESOURCE_DAG,
24+
RESOURCE_DAG_RUN,
25+
RESOURCE_TASK_INSTANCE,
26+
)
27+
from flask import Flask
28+
from tests_common.test_utils.config import conf_vars
29+
30+
from opa_auth_manager.opa_fab_auth_manager import OpaFabAuthManager
31+
32+
33+
@pytest.fixture
34+
def flask_app():
35+
with conf_vars(
36+
{
37+
(
38+
"core",
39+
"auth_manager",
40+
): "opa_auth_manager.opa_fab_auth_manager.OpaFabAuthManager",
41+
}
42+
):
43+
yield Flask(__name__)
44+
45+
46+
@pytest.fixture
47+
def auth_manager_with_appbuilder(flask_app):
48+
flask_app.config["AUTH_RATE_LIMITED"] = False
49+
flask_app.config["SERVER_NAME"] = "localhost"
50+
51+
appbuilder = init_appbuilder(flask_app, enable_plugins=False)
52+
auth_manager = OpaFabAuthManager()
53+
auth_manager.appbuilder = appbuilder
54+
auth_manager.init_flask_resources()
55+
return auth_manager
56+
57+
58+
class TestOpaFabAuthManager:
59+
@pytest.mark.parametrize(
60+
"method, dag_access_entity, dag_details, user_permissions, expected_opa_result, expected_result",
61+
[
62+
# Scenario 1 #
63+
# With global permissions on Dags
64+
(
65+
"GET",
66+
None,
67+
None,
68+
[(ACTION_CAN_READ, RESOURCE_DAG)],
69+
True,
70+
True,
71+
),
72+
# On specific DAG with global permissions on Dags
73+
(
74+
"GET",
75+
None,
76+
DagDetails(id="test_dag_id"),
77+
[(ACTION_CAN_READ, RESOURCE_DAG)],
78+
True,
79+
True,
80+
),
81+
# With permission on a specific DAG
82+
(
83+
"GET",
84+
None,
85+
DagDetails(id="test_dag_id"),
86+
[
87+
(ACTION_CAN_READ, "DAG:test_dag_id"),
88+
(ACTION_CAN_READ, "DAG:test_dag_id2"),
89+
],
90+
True,
91+
True,
92+
),
93+
# Without permission on a specific DAG (wrong method)
94+
(
95+
"POST",
96+
None,
97+
DagDetails(id="test_dag_id"),
98+
[(ACTION_CAN_READ, "DAG:test_dag_id")],
99+
False,
100+
False,
101+
),
102+
# Without permission on a specific DAG
103+
(
104+
"GET",
105+
None,
106+
DagDetails(id="test_dag_id2"),
107+
[(ACTION_CAN_READ, "DAG:test_dag_id")],
108+
False,
109+
False,
110+
),
111+
# Without permission on DAGs
112+
(
113+
"GET",
114+
None,
115+
None,
116+
[(ACTION_CAN_READ, "resource_test")],
117+
False,
118+
False,
119+
),
120+
# Scenario 2 #
121+
# With global permissions on DAGs
122+
(
123+
"GET",
124+
DagAccessEntity.RUN,
125+
DagDetails(id="test_dag_id"),
126+
[(ACTION_CAN_READ, RESOURCE_DAG), (ACTION_CAN_READ, RESOURCE_DAG_RUN)],
127+
True,
128+
True,
129+
),
130+
# Without read permissions on a specific DAG
131+
(
132+
"GET",
133+
DagAccessEntity.TASK_INSTANCE,
134+
DagDetails(id="test_dag_id"),
135+
[(ACTION_CAN_READ, RESOURCE_TASK_INSTANCE)],
136+
False,
137+
False,
138+
),
139+
# With read permissions on a specific DAG but not on the DAG run
140+
(
141+
"GET",
142+
DagAccessEntity.TASK_INSTANCE,
143+
DagDetails(id="test_dag_id"),
144+
[
145+
(ACTION_CAN_READ, "DAG:test_dag_id"),
146+
(ACTION_CAN_READ, RESOURCE_TASK_INSTANCE),
147+
],
148+
False,
149+
False,
150+
),
151+
# With read permissions on a specific DAG but not on the DAG run
152+
(
153+
"GET",
154+
DagAccessEntity.TASK_INSTANCE,
155+
DagDetails(id="test_dag_id"),
156+
[
157+
(ACTION_CAN_READ, "DAG:test_dag_id"),
158+
(ACTION_CAN_READ, RESOURCE_TASK_INSTANCE),
159+
(ACTION_CAN_READ, RESOURCE_DAG_RUN),
160+
],
161+
True,
162+
True,
163+
),
164+
# With edit permissions on a specific DAG and read on the DAG access entity
165+
(
166+
"DELETE",
167+
DagAccessEntity.TASK,
168+
DagDetails(id="test_dag_id"),
169+
[
170+
(ACTION_CAN_EDIT, "DAG:test_dag_id"),
171+
(ACTION_CAN_DELETE, RESOURCE_TASK_INSTANCE),
172+
],
173+
True,
174+
True,
175+
),
176+
# With edit permissions on a specific DAG and read on the DAG access entity
177+
(
178+
"POST",
179+
DagAccessEntity.RUN,
180+
DagDetails(id="test_dag_id"),
181+
[
182+
(ACTION_CAN_EDIT, "DAG:test_dag_id"),
183+
(ACTION_CAN_CREATE, RESOURCE_DAG_RUN),
184+
],
185+
True,
186+
True,
187+
),
188+
# Without permissions to edit the DAG
189+
(
190+
"POST",
191+
DagAccessEntity.RUN,
192+
DagDetails(id="test_dag_id"),
193+
[(ACTION_CAN_CREATE, RESOURCE_DAG_RUN)],
194+
False,
195+
False,
196+
),
197+
# Without read permissions on a specific DAG
198+
(
199+
"GET",
200+
DagAccessEntity.TASK_LOGS,
201+
DagDetails(id="test_dag_id"),
202+
[(ACTION_CAN_READ, RESOURCE_TASK_INSTANCE)],
203+
False,
204+
False,
205+
),
206+
],
207+
)
208+
def test_is_authorized_dag(
209+
self,
210+
method,
211+
dag_access_entity,
212+
dag_details,
213+
user_permissions,
214+
expected_opa_result,
215+
expected_result,
216+
auth_manager_with_appbuilder,
217+
):
218+
user = Mock()
219+
user.perms = user_permissions
220+
with mock.patch.object(
221+
OpaFabAuthManager, "_is_authorized_in_opa"
222+
) as mock_is_authorized_in_opa:
223+
mock_is_authorized_in_opa.return_value = expected_opa_result
224+
result = auth_manager_with_appbuilder.is_authorized_dag(
225+
method=method,
226+
access_entity=dag_access_entity,
227+
details=dag_details,
228+
user=user,
229+
)
230+
assert result == expected_result

0 commit comments

Comments
 (0)