This is an automated email from the ASF dual-hosted git repository.
potiuk 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 b94f36f2d01 Add tests for ti:self JWT scope enforcement on execution
API (#66071)
b94f36f2d01 is described below
commit b94f36f2d019ea60dd4f5a91494d3139b4e5dbee
Author: Amogh Desai <[email protected]>
AuthorDate: Sun May 10 07:47:13 2026 +0530
Add tests for ti:self JWT scope enforcement on execution API (#66071)
* Add tests for ti:self JWT scope enforcement on Execution API
The ti:self scope check at security.py rejects requests where the
JWT subject does not match the task_instance_id in the path. This is
the cross-task isolation guarantee for any router that opts into
ti:self (e.g. /task-instances, /hitl, and the upcoming task state
endpoints from AIP-103). Until now the check was uncovered.
Adds three tests mirroring the existing TestTokenTypeScopeEnforcement
pattern: matching subject is accepted, mismatched subject returns 403,
and routes without ti:self ignore subject mismatches.
* update
---
.../api_fastapi/execution_api/test_security.py | 57 ++++++++++++++++++++++
1 file changed, 57 insertions(+)
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/test_security.py
b/airflow-core/tests/unit/api_fastapi/execution_api/test_security.py
index 985dfd32599..22b7f58a959 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/test_security.py
+++ b/airflow-core/tests/unit/api_fastapi/execution_api/test_security.py
@@ -158,6 +158,63 @@ class TestTokenTypeScopeEnforcement:
assert run.status_code == 200
+class TestTiSelfScopeEnforcement:
+ """Routes with the ``ti:self`` scope reject mismatched JWT subjects."""
+
+ PATH_TI_ID = "00000000-0000-0000-0000-000000000001"
+ OTHER_TI_ID = "00000000-0000-0000-0000-000000000002"
+
+ @pytest.fixture
+ def app(self):
+ """One router enforces ti:self, another doesn't — to confirm
enforcement is opt-in."""
+ app = FastAPI()
+
+ authenticated_router = APIRouter(dependencies=[Security(require_auth)])
+ ti_self_router = APIRouter(dependencies=[Security(require_auth,
scopes=["ti:self"])])
+
+ @ti_self_router.get("/{task_instance_id}/state")
+ def state_endpoint(task_instance_id: str):
+ return {"ok": True}
+
+ @authenticated_router.get("/no-scope/{task_instance_id}")
+ def no_scope_endpoint(task_instance_id: str):
+ return {"ok": True}
+
+ authenticated_router.include_router(ti_self_router, prefix="/ti")
+ app.include_router(authenticated_router)
+ return app
+
+ def _override_jwt(self, app: FastAPI, token_ti_id: UUID):
+ async def mock_jwt(request: Request):
+ return TIToken(id=token_ti_id, claims=TIClaims(scope="execution"))
+
+ app.dependency_overrides[_jwt_bearer] = mock_jwt
+
+ def test_matching_subject_is_accepted(self, app):
+ self._override_jwt(app, self.PATH_TI_ID)
+ client = TestClient(app)
+
+ resp = client.get(
+ f"/ti/{self.PATH_TI_ID}/state",
+ headers={"Authorization": "Bearer fake"},
+ )
+
+ assert resp.status_code == 200
+
+ def test_mismatched_subject_is_rejected(self, app):
+ """A task cannot read or write another task's resources."""
+ self._override_jwt(app, self.OTHER_TI_ID)
+ client = TestClient(app)
+
+ resp = client.get(
+ f"/ti/{self.PATH_TI_ID}/state",
+ headers={"Authorization": "Bearer fake"},
+ )
+
+ assert resp.status_code == 403
+ assert "does not match" in resp.json()["detail"]
+
+
class TestGetTeamNameDep:
"""Tests for get_team_name_dep avoiding unnecessary async sessions."""
