Fix Dag processor token rotation and registration failures

Provisioner restarts and transient write failures must not interrupt running processors. Failed registration must not leave an incomplete Job that a retry can adopt.

Author: ephraimbuddy

airflow-core/docs/security/jwt_token_authentication.rst

@@ -337,7 +337,7 @@ The token flows through the execution stack as follows:
 Even if a workload token is intercepted in transit, it can only call ``/run``. That endpoint
 rejects re-runs (``409 Conflict`` unless the task instance is in ``QUEUED`` or ``RESTARTING``),
 so the attack surface for the longer-lived token is bounded to "start a task that is already
-queued". All other endpoints require ``scope=execution`` and reject workload tokens.
+queued". Every other endpoint rejects workload tokens.
 
 Token validation (Execution API)
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -358,7 +358,7 @@ The ``JWTBearer`` security dependency validates the token once per request:
    ``403 Forbidden`` even when cryptographic validation succeeds. All claim types allow extra
    fields so auth managers can attach deployment-specific claims.
 6. Constructs the corresponding token object: ``TIToken``, ``DagProcessorSessionToken``,
-   ``DagProcessorToken``, or an ``ExecutionToken`` carrying ``DagParseClaims``.
+   ``DagProcessorToken``, or ``DagParseToken``.
 7. Caches the validated token on the ASGI request scope for the duration of the request.
 
 Route-level enforcement is handled by ``require_auth``:
@@ -508,9 +508,12 @@ the client may access::
     airflow dag-processor-token --token-file /run/airflow/processor.jwt \
         --bundle-name dags-folder --rotate
 
-The command writes the token atomically with owner-only permissions. Keep the provisioner
-running and mount its directory rather than a single file so token replacement stays
-visible. Each processor process needs its own session.
+The command writes the token atomically. A new file is readable only by its owner, and replacing
+an existing file keeps its mode and group, so grant the processor read access once. ACLs are not
+carried over. Keep the provisioner running and mount its directory rather than a single file so
+token replacement stays visible. Each processor process needs its own session. Restarting the
+provisioner with the same token file and bundles keeps that session, so running processors keep
+their Jobs.
 
 The three credentials have different issuers and permissions:
 
@@ -530,7 +533,12 @@ invalidates its runtime credentials.
 
 The client selects the bundle with ``use_bundle`` and a parsing attempt with ``use_parse``.
 A request header cannot override the parsing token's signed bundle.
-Connection and Variable access uses the bundle's current team.
+Connection and Variable access uses the bundle's current team. A registered Job's teams come from
+``[dag_processor] dag_bundle_config_list`` on the API server, so it needs the same bundle-to-team
+configuration as the processor. A ``dag_parse`` token can also write and delete Variables, as a
+parsing subprocess can without this client. ``Variable.set`` upserts on the key alone, so in
+multi-team mode a write can overwrite another team's Variable with the same key, as it can with task
+tokens.
 
 The client attempts renewal at 80% of token lifetime. Transient early-renewal failures leave
 the existing token usable until expiry. A retired registration signals that its owner must

airflow-core/src/airflow/api/common/airflow_health.py

@@ -23,7 +23,7 @@
 from sqlalchemy import select
 
 from airflow.configuration import conf
-from airflow.dag_processing.bundles.manager import _get_configured_bundle_team_names
+from airflow.dag_processing.bundles.manager import get_configured_bundle_team_names
 from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
 from airflow.jobs.job import Job
 from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
@@ -121,7 +121,7 @@ def _dag_processor_instance_health(job: Job) -> dict[str, Any]:
 def _configured_bundle_teams() -> dict[str, str | None]:
     """Map every configured Dag bundle to the team owning it, empty when the config is unreadable."""
     try:
-        return _get_configured_bundle_team_names()
+        return get_configured_bundle_team_names()
     except Exception:
         # A health probe must not fail on malformed bundle config; callers fall back to liveness only.
         log.warning("Could not read the Dag bundle configuration", exc_info=True)

airflow-core/src/airflow/api_fastapi/execution_api/AGENTS.md

@@ -77,4 +77,4 @@ Adding a new Execution API feature touches multiple packages. All of these must
 
 ## Token Scope Infrastructure
 
-Token types (`"execution"`, `"workload"`, `"callback"`, `"dag_processor_session"`, `"dag_processor"`), route-level enforcement via `ExecutionAPIRoute` + `require_auth`, and the `ti:self` path-parameter validation are documented in the module docstring of `security.py`. A route that admits `dag_processor` tokens must also bind the request to a granted Dag bundle; `test_token_scope_boundaries.py` lists those routes and the parse-time messages that use them.
+Token types (`"execution"`, `"workload"`, `"callback"`, `"dag_processor_session"`, `"dag_processor"`, `"dag_parse"`), route-level enforcement via `ExecutionAPIRoute` + `require_auth`, and the `ti:self` path-parameter validation are documented in the module docstring of `security.py`. A route that admits `dag_processor` or `dag_parse` tokens for Connection, Variable, XCom, Dag run or task-instance reads must also bind the request to a granted Dag bundle (`SelectedDagBundle` or `DagInGrantedBundle`); the `/jobs` routes bind to the `job_id` claim instead. `test_token_scope_boundaries.py` lists the bundle-bound routes and the parse-time messages that use them.

airflow-core/src/airflow/api_fastapi/execution_api/app.py

@@ -153,8 +153,9 @@ async def dispatch(self, request: Request, call_next):
 
                     # Only short-lived execution tokens are renewed here. Any other type has a
                     # lifetime set by its issuer (workload and callback tokens outlive queue waits,
-                    # dag_processor tokens are rotated by provisioning), so a new type must not
-                    # become renewable by default. Tokens without a scope are legacy execution tokens.
+                    # the dag_processor_session token is rotated by provisioning, and Job and parsing
+                    # tokens are renewed by re-registering and re-exchanging), so a new type must
+                    # not become renewable by default. Tokens without a scope are legacy execution tokens.
                     if claims.get("scope", "execution") != "execution":
                         return response
 

airflow-core/src/airflow/api_fastapi/execution_api/dag_processor_tokens.py

@@ -23,7 +23,8 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import math
+from typing import TYPE_CHECKING, NamedTuple
 
 from airflow._shared.timezones import timezone
 
@@ -34,10 +35,25 @@
     from airflow.api_fastapi.auth.tokens import JWTGenerator
 
 
+class IssuedToken(NamedTuple):
+    """A token with the seconds left until it expires."""
+
+    token: str
+    expires_in: int
+
+
 class ExpiredDagProcessorToken(ValueError):
     """The parent credential expired before a Job or parsing credential could be issued."""
 
 
+def _get_lifetime(generator: JWTGenerator, parent_expiry: float, message: str) -> int:
+    """Return whole seconds a child token may live: the signer's lifetime, capped by its parent's expiry."""
+    lifetime = math.floor(min(generator.valid_for, parent_expiry - timezone.utcnow().timestamp()))
+    if lifetime < 1:
+        raise ExpiredDagProcessorToken(message)
+    return lifetime
+
+
 def generate_dag_processor_session_token(
     generator: JWTGenerator, *, session_id: UUID, bundle_names: Collection[str], valid_for: float
 ) -> str:
@@ -59,20 +75,19 @@ def generate_dag_processor_token(
     job_id: int,
     bundle_names: Collection[str],
     session_expiry: float,
-) -> str:
+) -> IssuedToken:
     """Issue a Job credential that cannot outlive its provisioned session credential."""
-    remaining = session_expiry - timezone.utcnow().timestamp()
-    if remaining <= 0:
-        raise ExpiredDagProcessorToken("Session credential has expired")
-    return generator.generate(
+    lifetime = _get_lifetime(generator, session_expiry, "Session credential has expired")
+    token = generator.generate(
         extras={
             "sub": str(session_id),
             "scope": "dag_processor",
             "dag_bundles": sorted(bundle_names),
             "job_id": job_id,
         },
-        valid_for=min(generator.valid_for, remaining),
+        valid_for=lifetime,
     )
+    return IssuedToken(token, lifetime)
 
 
 def generate_dag_parse_token(
@@ -84,12 +99,10 @@ def generate_dag_parse_token(
     bundle_name: str,
     relative_fileloc: str,
     processor_expiry: float,
-) -> str:
+) -> IssuedToken:
     """Issue a parsing credential that cannot outlive its parent Job credential."""
-    remaining = processor_expiry - timezone.utcnow().timestamp()
-    if remaining <= 0:
-        raise ExpiredDagProcessorToken("Processor credential has expired")
-    return generator.generate(
+    lifetime = _get_lifetime(generator, processor_expiry, "Processor credential has expired")
+    token = generator.generate(
         extras={
             "sub": str(attempt_id),
             "scope": "dag_parse",
@@ -98,5 +111,6 @@ def generate_dag_parse_token(
             "dag_bundles": [bundle_name],
             "relative_fileloc": relative_fileloc,
         },
-        valid_for=min(generator.valid_for, remaining),
+        valid_for=lifetime,
     )
+    return IssuedToken(token, lifetime)

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/job.py

@@ -56,7 +56,8 @@ class JobRegisterResponse(StrictBaseModel):
     """The registered Job and its management credential."""
 
     job_id: int
-    token: str = Field(description="A ``dag_processor`` token valid while the Job is open.")
+    token: str = Field(description="A ``dag_processor`` token, valid until it expires or the Job ends.")
+    expires_in: int = Field(gt=0, description="Seconds until the token expires.")
 
 
 class JobHeartbeatResponse(StrictBaseModel):
@@ -90,3 +91,4 @@ class DagParseTokenResponse(StrictBaseModel):
     """Short-lived parsing credential; cannot register, heartbeat, or complete Jobs."""
 
     token: str
+    expires_in: int = Field(gt=0, description="Seconds until the token expires.")

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/token.py

@@ -109,3 +109,9 @@ class DagProcessorToken(ExecutionToken):
     """Authenticated Dag processor Job."""
 
     claims: DagProcessorClaims
+
+
+class DagParseToken(ExecutionToken):
+    """Authenticated parsing attempt."""
+
+    claims: DagParseClaims

airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py

@@ -65,7 +65,7 @@
 authenticated_router.include_router(variables.router, prefix="/variables", tags=["Variables"])
 authenticated_router.include_router(xcoms.router, prefix="/xcoms", tags=["XComs"])
 authenticated_router.include_router(hitl.router, prefix="/hitlDetails", tags=["Human in the Loop"])
-authenticated_router.include_router(jobs.router, prefix="/jobs", tags=["Jobs"])
+authenticated_router.include_router(jobs.router, tags=["Jobs"])
 authenticated_router.include_router(task_state_store.router, prefix="/store/ti", tags=["Task State Store"])
 authenticated_router.include_router(
     asset_state_store.router, prefix="/store/asset", tags=["Asset State Store"]

airflow-core/src/airflow/api_fastapi/execution_api/routes/connections.py

@@ -67,7 +67,8 @@ async def has_connection_access(
     dependencies=[ExecutionOrProcessorSecretsToken],
     responses={
         status.HTTP_401_UNAUTHORIZED: {"description": "Unauthorized"},
-        status.HTTP_403_FORBIDDEN: {"description": "Task does not have access to the connection"},
+        status.HTTP_400_BAD_REQUEST: {"description": "A Dag processor token did not name its Dag bundle"},
+        status.HTTP_403_FORBIDDEN: {"description": "The caller does not have access to the connection"},
     },
 )
 def get_connection(