Add scoped Execution API access for Dag processors (#74370)

* Add scoped Execution API access for Dag processors

Dag processors need credentials tied to their registered Job and granted bundles before they can communicate with a remote Execution API. Keep this foundation independently reviewable while the processor command retains its existing database-backed behavior.

* Fix documentation build for task-token revocation

The heading warning prevents the security documentation from building successfully.

* Allow parsing-time queries before a Dag is first stored

New Dags can query execution state while parsing, before they have a DagModel row. Rejecting those requests prevents their initial parse from completing.

* Enforce scoped Dag processor token claims and expiry

Task and processor credentials need distinct validation so task-only routes cannot receive processor identities. Expired provisioning credentials must not leave a registered Job behind, while trusted in-process task identities must remain usable without JWT expiry.

* 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.

* Fix Dag processor token docs and renewal test flakiness

Slow CI responses can exceed the best-effort renewal deadline, while the authentication guidance must reflect the scoped Execution API contract.

Author: ephraimbuddy

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``a5d7b9c13e40`` (head) | ``c3e7a9182f64`` | ``3.4.0``         | Preserve exact task execution attribution in asset state.    |
+| ``c71065a1b14f`` (head) | ``a5d7b9c13e40`` | ``3.5.0``         | Add session_id and registration_id to Job.                   |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``a5d7b9c13e40``        | ``c3e7a9182f64`` | ``3.4.0``         | Preserve exact task execution attribution in asset state.    |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``c3e7a9182f64``        | ``7f8c9a2d410e`` | ``3.4.0``         | Attribute log and asset events to exact task tries.          |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

airflow-core/docs/security/jwt_token_authentication.rst

@@ -254,20 +254,21 @@ Token structure (Execution API)
    * - ``aud``
      - Audience (from ``[execution_api] jwt_audience``, default: ``urn:airflow.apache.org:task``).
    * - ``sub``
-     - Task instance UUID — the identity of the workload.
+     - UUID identifying the task instance, callback, processor session, or parsing attempt, depending on the scope.
    * - ``scope``
-     - Token scope: ``"execution"`` or ``"workload"``.
+     - Token scope: ``"execution"``, ``"workload"``, ``"callback"``, ``"dag_processor_session"``,
+       ``"dag_processor"``, or ``"dag_parse"``.
    * - ``iat``
      - Issued-at timestamp.
    * - ``nbf``
      - Not-before timestamp.
    * - ``exp``
-     - Expiration timestamp (``iat + [execution_api] jwt_expiration_time``).
+     - Expiration timestamp, determined by the scope's lifetime and any parent credential's expiry.
 
 Token scopes (Execution API)
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-The Execution API defines two token scopes with different lifetimes:
+The Execution API uses the following token scopes. Each endpoint accepts only its declared scopes:
 
 **workload**
    A token embedded in the workload JSON payload when the Scheduler
@@ -281,12 +282,28 @@ The Execution API defines two token scopes with different lifetimes:
    backed-up queue before its workload token expires.
 
 **execution**
-   A short-lived token (default 10 minutes) accepted by all Execution API endpoints.
+   A short-lived token (default 10 minutes) accepted by endpoints that allow task or callback execution.
    This is the standard scope for worker communication during task execution. Issued
    by the server when the worker transitions to running via the ``/run`` endpoint.
    The ``JWTReissueMiddleware`` refreshes ``execution`` tokens transparently,
    so the worker maintains access for the duration of the task.
 
+**callback**
+   A single-use token exchanged at ``/callbacks/{callback_id}/run`` for an ``execution``
+   token when a queued callback starts.
+
+**dag_processor_session**
+   A provisioned credential accepted only by ``POST /jobs`` for Job registration and renewal.
+
+**dag_processor**
+   A Job credential for heartbeats, completion, Connection and Variable reads for granted bundles,
+   and parsing-token exchange.
+
+**dag_parse**
+   A credential for one parsing attempt, granting parse-time access within its signed bundle.
+
+See `Dag processor HTTP client credentials`_ for provisioning, lifetime limits, and authorization details.
+
 Tokens without a ``scope`` claim default to ``"execution"`` for backwards compatibility.
 
 Token delivery to workers
@@ -337,7 +354,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)
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -348,15 +365,17 @@ The ``JWTBearer`` security dependency validates the token once per request:
 2. Performs cryptographic signature validation via ``JWTValidator``.
 3. Verifies standard claims (``exp``, ``iat``, ``aud`` — ``nbf`` and ``iss`` if configured).
 4. Defaults the ``scope`` claim to ``"execution"`` if absent.
-5. **Validates the task identity claims against a typed Pydantic schema.** ``TIClaims``
-   (in ``airflow.api_fastapi.execution_api.datamodels.token``) enforces that ``scope`` is one
-   of the declared ``TokenScope`` literals (``"execution"`` or ``"workload"``); ``TIToken``
-   then parses the ``sub`` claim through a ``UUID`` field, which rejects non-UUID values.
-   A token whose ``scope`` is unknown, or whose ``sub`` is not a valid UUID, is rejected with
-   ``403 Forbidden`` even when the cryptographic signature checks pass. ``TIClaims`` keeps
-   ``extra="allow"`` so auth managers can attach additional, deployment-specific claims
-   without modifying the core schema; only the security-critical fields are typed.
-6. Creates a ``TIToken`` object with the task instance ID and validated claims.
+5. **Validates the identity and its scope-specific claims against typed Pydantic schemas.**
+   ``ExecutionToken`` (in ``airflow.api_fastapi.execution_api.datamodels.token``) parses ``sub``
+   as a UUID. Its claims form a discriminated union selected by ``scope``: ``TIClaims`` for
+   ``"execution"``, ``"workload"``, and ``"callback"``; ``DagProcessorSessionClaims`` for
+   provisioning; ``DagProcessorClaims`` for a registered Job; and ``DagParseClaims`` for a
+   parsing attempt. Each processor claim type requires expiry and its scope-specific grants
+   and identity fields. Unknown scopes, invalid subjects, or missing required claims return
+   ``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 ``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``:
@@ -367,26 +386,32 @@ Route-level enforcement is handled by ``require_auth``:
   ``{task_instance_id}`` path parameter, preventing a worker from accessing another task's
   endpoints.
 
+The task-only, processor-session, and processor-Job dependencies additionally reject tokens of
+the wrong principal type. Endpoints shared by tasks and parsing attempts use the general
+``CurrentExecutionToken`` dependency with their declared route scopes.
+
 .. mermaid::
 
     flowchart TD
         REQ([Incoming request<br/>Authorization: Bearer ...])
         REQ --> CACHE{Cached on<br/>request.scope?}
-        CACHE -->|yes| RET([Return cached TIToken])
+        CACHE -->|yes| RET([Return cached token])
         CACHE -->|no| SIG[JWTValidator:<br/>verify signature]
         SIG -->|fail| F1([403 Forbidden])
         SIG -->|ok| STD[Verify exp / iat / nbf<br/>aud / iss]
         STD -->|fail| F1
         STD -->|ok| SCOPE[Default scope to<br/>'execution' if absent]
-        SCOPE --> SCHEMA[TIClaims:<br/>typed Pydantic schema]
+        SCOPE --> SCHEMA[ExecutionToken:<br/>UUID identity and<br/>scope-discriminated claims]
         SCHEMA -->|ValidationError| F1
         SCHEMA -->|ok| TYP{require_auth:<br/>scope in<br/>route.allowed_token_types?}
         TYP -->|no| F1
         TYP -->|yes| SELF{ti:self scope<br/>declared?}
-        SELF -->|no| OK([Return TIToken])
+        SELF -->|no| PRINCIPAL{Typed dependency:<br/>principal type matches?}
         SELF -->|yes| MATCH{token.sub ==<br/>task_instance_id?}
         MATCH -->|no| F1
-        MATCH -->|yes| OK
+        MATCH -->|yes| PRINCIPAL
+        PRINCIPAL -->|no| F1
+        PRINCIPAL -->|yes or general dependency| OK([Return validated token])
 
         classDef fail fill:#ffcdd2,stroke:#c62828,stroke-width:2px,color:#000
         classDef pass fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px,color:#000
@@ -412,14 +437,14 @@ they are designed to survive executor queue wait times without needing refresh.
 ensures long-running tasks do not lose API access without requiring the worker to
 re-authenticate.
 
-No token revocation (Execution API)
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Task-token revocation (Execution API)
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-Execution API tokens are not subject to revocation. ``execution``-scoped tokens are short-lived
+Task Execution API tokens are not subject to revocation. ``execution``-scoped tokens are short-lived
 (default 10 minutes) and automatically refreshed by the ``JWTReissueMiddleware``.
 ``workload``-scoped tokens (tracking ``[scheduler] task_queued_timeout``) are not refreshed —
-they expire naturally after their validity period. Revocation is not part of the Execution API
-security model.
+they expire naturally after their validity period. Processor and parsing credentials additionally
+require an open, owning Job, as described under "Dag processor HTTP client credentials" below.
 
 
 
@@ -486,6 +511,62 @@ See :doc:`/security/security_model` for the full security implications, deployme
 guidance, and the planned strategic and tactical improvements.
 
 
+Dag processor HTTP client credentials
+-------------------------------------
+
+The core ``DagProcessorAPIClient`` lets processor integrations authenticate to the Execution
+API without holding its signing key. The standard ``airflow dag-processor`` command does
+not yet use this client. Provisioning a token file alone does not switch that command to
+HTTP or remove its need for database access.
+
+On a trusted host with the API signing key, provision a session token for the bundles
+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. 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:
+
+- ``dag_processor_session`` is issued by trusted provisioning. It can only register a
+  Job or renew that Job's credential through ``POST /jobs``.
+- ``dag_processor`` is returned by Job registration. It can heartbeat and complete that
+  Job, read Connections and Variables for a granted bundle, and exchange a parsing
+  credential through ``POST /jobs/{job_id}/parse-token``. It cannot outlive the session
+  token used to obtain it.
+- ``dag_parse`` is returned by the parsing-token exchange for one attempt, bundle, and
+  relative file location. It permits parse-time requests, including callback-context reads,
+  but cannot manage Jobs or exchange tokens. It cannot outlive its parent Job credential.
+
+Registration retries keep the same registration UUID. A new process uses a new UUID;
+it cannot replace a session's Job while that Job is alive. Ending or replacing the Job
+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. 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
+restart. Completion retains its original identity and outcome across retries.
+
+File and attempt identity are carried in the token but are not currently used for access
+checks or request attribution. Archives retain one file identity under the current processor
+model. These credentials do not isolate hostile code from other credentials accessible on
+the processor host; the process and deployment limitations above still apply.
+
+
 Workload Isolation and Current Limitations
 ------------------------------------------
 

airflow-core/docs/security/security_model.rst

@@ -532,10 +532,11 @@ potentially still executes with direct database access in the Dag File Processor
    * **Tightened deserialization allowlist.** The serializer's class-name allowlist regex is
      anchored to require a full-string match, so attacker-controlled values cannot smuggle
      untrusted class names by appending an allowed suffix to a non-allowed name.
-   * **Typed JWT claims schema.** Even after cryptographic validation, claims are run through a
-     typed Pydantic schema (``TIClaims``) that enforces the ``scope`` literal, and then through
-     ``TIToken`` which parses ``sub`` as a UUID. A token whose ``scope`` is unknown or whose
-     ``sub`` is not a valid UUID is rejected before any route handler runs.
+   * **Typed JWT claims schemas.** Even after cryptographic validation, ``ExecutionToken`` validates
+     ``sub`` as a UUID and selects a claims schema by ``scope``: ``TIClaims`` for task, workload,
+     and callback credentials, or ``DagProcessorSessionClaims``, ``DagProcessorClaims``, or
+     ``DagParseClaims`` for processor credentials. Unknown scopes, invalid subjects, and missing
+     required scope-specific claims are rejected before any route handler runs.
 
 **Token signing key might be a shared secret**
    In symmetric key mode (``[api_auth] jwt_secret``), the same secret key is used to both generate and

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"`), route-level enforcement via `ExecutionAPIRoute` + `require_auth`, and the `ti:self` path-parameter validation are documented in the module docstring of `security.py`.
+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

@@ -53,6 +53,7 @@
 __all__ = [
     "InProcessExecutionAPI",
     "create_task_execution_api_app",
+    "create_jwt_generator",
     "lifespan",
     "CorrelationIdMiddleware",
 ]
@@ -74,7 +75,8 @@ def _jwt_validator():
     return validator
 
 
-def _jwt_generator():
+def create_jwt_generator() -> JWTGenerator:
+    """Create a signer using the Execution API's configured key, audience, and lifetime."""
     from airflow.configuration import conf
 
     generator = JWTGenerator(
@@ -96,7 +98,7 @@ async def lifespan(app: FastAPI, registry: svcs.Registry):
     # record this here
     app.state.svcs_registry = registry
 
-    registry.register_factory(JWTGenerator, _jwt_generator)
+    registry.register_factory(JWTGenerator, create_jwt_generator)
 
     # InProcessExecutionAPI stubs out JWTValidator: don't re-register in that case.
     if JWTValidator not in registry:
@@ -146,10 +148,12 @@ async def dispatch(self, request: Request, call_next):
                     validator: JWTValidator = await services.aget(JWTValidator)
                     claims = await validator.avalidated_claims(token, {})
 
-                    # Workload and callback tokens are long-lived and meant to survive
-                    # queue wait times so avoid refreshing them. If avalidated_claims
-                    # raises for such a token, the outer except handles it.
-                    if claims.get("scope") in ("workload", "callback"):
+                    # 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,
+                    # 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
 
                     now = int(time.time())