Add OIDC workload identity for Dag processor session credentials

airflow-core/docs/security/jwt_token_authentication.rst

@@ -519,6 +519,133 @@ API without holding its signing key. The standard ``airflow dag-processor`` comm
 not yet use this client. Provisioning a token file alone does not switch that command to
 HTTP or remove its need for database access.
 
+In-memory sessions from workload identity
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Processor integrations can use ``WorkloadIdentitySessionTokenProvider`` to obtain and renew
+``dag_processor_session`` credentials through ``POST /execution/dag-processor/session-token``.
+The provider retains these credentials only in memory. It requires a renewable, signed workload
+identity from the deployment platform; Airflow does not create that identity or require the
+processor to hold an Airflow signing key.
+
+Configure the API server's ``[execution_api] dag_processor_workload_identity`` option with JSON::
+
+    {
+      "issuer": "https://identity.example/",
+      "jwks_url": "https://identity.example/openid/v1/jwks",
+      "audience": "airflow-dag-processor-bootstrap",
+      "algorithm": "RS256",
+      "instance_id_claim": ["kubernetes.io", "pod", "uid"],
+      "grants": {
+        "system:serviceaccount:airflow:processor": ["dags-folder"]
+      }
+    }
+
+Use your identity provider's exact issuer, HTTPS JWKS URL, and signing algorithm (``RS256``,
+``ES256``, or ``EdDSA``). Use a dedicated audience, distinct from ``[execution_api] jwt_audience``.
+The API server verifies the signature, issuer, audience, issued-at time, expiry, and any not-before
+claim, using ``[api_auth] jwt_leeway`` for clock skew. Only exact subjects listed in ``grants`` can
+obtain a session. If a grant names a Dag bundle that the API server does not have configured, the
+API server logs an error at startup and refuses that subject's exchanges with HTTP 403. This option
+is empty by default, disabling the exchange.
+Apply the same trust configuration to all API-server replicas and restart them after changes.
+
+The option trusts a single issuer. Dag processors whose identities come from different issuers, such
+as separate Kubernetes clusters, cannot all obtain sessions from the same API server.
+
+The JWKS endpoint must accept unauthenticated HTTPS GET requests from the Airflow API server.
+The fetcher sends no authentication credentials and uses HTTPX's default certificate trust.
+``SSL_CERT_FILE`` or ``SSL_CERT_DIR``, when set, replace the default CA bundle rather than adding
+to it, and they apply to every HTTPX client in the API-server process. A bundle that holds only a
+private CA cuts those clients off from public roots, so include the roots they still need. There is
+no per-issuer CA or authentication option.
+Authenticated JWKS endpoints, including Kubernetes API-server discovery protected by service-account
+authentication, are not supported by this implementation. For those clusters, serve the public JWKS
+through a suitable HTTPS endpoint before enabling the exchange. See Kubernetes'
+`service account issuer discovery documentation <https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/#service-account-issuer-discovery>`_.
+The processor-to-Airflow TLS settings described below do not configure this outbound JWKS connection.
+
+``instance_id_claim`` is a path through signed claims to a nonempty string identifying one
+processor replica. A shared service-account subject alone is insufficient: replicas must have
+different instance identities. Run one processor per identity. The session ID stays stable across
+identity-token renewal and API-server restarts. Changes to the issuer, audience, subject, instance
+claim, instance value, or bundle grants produce a different session. Request bodies and caller
+hostnames cannot select the session or its grants.
+
+Changing grants while API-server replicas serve different configurations can repeatedly retire
+processor registrations, because those replicas issue different session IDs for the same identity.
+Stop the affected processors, finish deploying identical trust configuration to every API-server
+replica, and then restart the processors.
+
+For Kubernetes, the example uses the Pod UID in a
+`bound service-account token <https://kubernetes.io/docs/reference/access-authn-authz/service-accounts-admin/#bound-service-account-token-volume-mechanism>`_.
+Project a token with the dedicated audience and mount the projection directory read-only in the
+processor container. Configure the projection's ``expirationSeconds`` and allow Kubernetes to
+rotate it. See `serviceAccountToken projected volumes
+<https://kubernetes.io/docs/concepts/storage/projected-volumes/#serviceaccounttoken>`_.
+
+A container restart within the same Pod keeps the Pod UID and therefore the session ID.
+A replacement process cannot register while the previous Job is still alive. The client raises
+``DagProcessorJobAlreadyRunning``; the integration must retry registration with backoff until the
+previous Job completes or its heartbeat becomes stale under ``[dag_processor] health_check_threshold``.
+The client does not wait automatically. This prevents two processes sharing an identity from owning
+active Jobs at the same time.
+
+The processor integration supplies a callable that reads the current platform identity on each
+exchange. Any exception the callable raises is re-raised as ``WorkloadIdentityUnavailableError``, an
+``OSError``, so a failed optional renewal leaves the valid Job credential in use. For example:
+
+.. code-block:: python
+
+    from pathlib import Path
+
+    from airflow.dag_processing.api_client import DagProcessorAPIClient
+    from airflow.dag_processing.credentials import WorkloadIdentitySessionTokenProvider
+
+    identity_path = Path("/var/run/airflow-identity/token")
+    provider = WorkloadIdentitySessionTokenProvider(
+        base_url="https://airflow.example/execution/",
+        identity_token=identity_path.read_text,
+    )
+    processor = DagProcessorAPIClient(
+        base_url="https://airflow.example/execution/",
+        token_provider=provider,
+        hostname="processor-1",
+    )
+
+Use this client for the integration's Job lifecycle and close it during shutdown; closing it also
+closes the provider and clears its session cache. The file in this example contains the platform's
+projected identity, not an Airflow session credential. The provider never writes the exchanged
+session JWT to disk and does not fall back to a session-token file. ``token_provider`` and
+``token_file`` are mutually exclusive. HTTPS is required for the exchange, and redirects are rejected.
+The exchange uses the same API TLS configuration and request timeout as the Task SDK client,
+including private certificate authorities and client certificates. Registration retries transient
+exchange failures using the configured Execution API retry policy, waiting at least the
+``Retry-After`` interval of an HTTP 503 before the next attempt. Optional early renewal makes one
+bounded attempt so it does not delay requests using a valid Job credential.
+
+Session expiry is capped by both the external identity's remaining lifetime and
+``[execution_api] jwt_expiration_time``. Clock-skew leeway never extends this cap, and an identity
+with less than one second remaining cannot obtain a session. Renewal rereads the platform identity
+and retains the same session and Job. During an identity-provider or API outage, the client can continue using
+an already-issued Job credential until it expires; it cannot renew without a valid identity.
+Removing a grant stops future exchanges once the API servers restart with the new configuration. Credentials
+already issued remain subject to their expiry and Job ownership checks.
+
+Verification uses the issuer's JWKS and does not perform online token introspection. In particular,
+it does not check whether a Kubernetes Pod still exists. A previously issued identity may remain
+usable until expiry. The JWKS cache refreshes hourly and also refreshes when an identity names an
+unknown signing key. Fetches, including failed attempts, are limited to one per ten seconds per
+API-server process. Publish new keys before signing with them and retain old keys while their tokens
+remain valid.
+
+If a required signing key cannot be fetched, the exchange returns HTTP 503 with ``Retry-After``, and
+registration waits at least that long before retrying. A cached matching key remains usable during
+an outage. Invalid identities and keys absent from a successfully fetched key set return HTTP 401.
+
+File-based provisioning
+^^^^^^^^^^^^^^^^^^^^^^^
+
 On a trusted host with the API signing key, provision a session token for the bundles
 the client may access::
 
@@ -532,9 +659,12 @@ token replacement stays visible. Each processor process needs its own session. R
 provisioner with the same token file and bundles keeps that session, so running processors keep
 their Jobs.
 
+Credential permissions and renewal
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
 The three credentials have different issuers and permissions:
 
-- ``dag_processor_session`` is issued by trusted provisioning. It can only register a
+- ``dag_processor_session`` is issued by the identity exchange or trusted file 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
@@ -625,6 +755,9 @@ All JWT-related configuration parameters:
    * - ``[execution_api] jwt_audience``
      - ``urn:airflow.apache.org:task``
      - Audience claim for Execution API tokens.
+   * - ``[execution_api] dag_processor_workload_identity``
+     - Empty (disabled)
+     - JSON trust settings and subject-to-bundle grants for workload identity session provisioning.
 
 .. important::
 

airflow-core/docs/security/security_model.rst

@@ -540,6 +540,12 @@ potentially still executes with direct database access in the Dag File Processor
      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.
+   * **Workload identity provisioning.** The optional Dag processor session exchange validates
+     deployment-issued JWTs against separately configured issuer, audience, and JWKS settings.
+     Bundle grants come from the API server configuration. Exchanged sessions stay in the client's
+     memory, but workload identity and runtime credentials remain accessible to code with access
+     to that process. This does not add isolation between the processor and user code; see
+     :doc:`jwt_token_authentication` for configuration and expiry limitations.
 
 **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_fastapi/auth/tokens.py

@@ -16,12 +16,14 @@
 # under the License.
 from __future__ import annotations
 
+import asyncio
 import json
 import os
 import time
 import uuid
 from base64 import urlsafe_b64encode
 from collections.abc import Callable, Sequence
+from contextlib import nullcontext
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, Literal, overload
 
@@ -45,6 +47,7 @@
 __all__ = [
     "InvalidClaimError",
     "JWKS",
+    "JWKSUnavailableError",
     "JWTGenerator",
     "JWTValidator",
     "generate_private_key",
@@ -99,9 +102,19 @@ def _guess_best_algorithm(key: AllowedPrivateKeys):
     raise ValueError(f"Unknown key object {type(key)}")
 
 
+class JWKSUnavailableError(RuntimeError):
+    """The required signing key could not be loaded because a JWKS fetch failed."""
+
+
 @attrs.define(repr=False)
 class JWKS:
-    """A class to fetch and sync a set of JSON Web Keys."""
+    """
+    Fetch and cache a set of JSON Web Keys.
+
+    With ``raise_on_fetch_error``, missing keys raise ``JWKSUnavailableError`` after a failed
+    fetch. Cached matching keys remain usable, and a successfully fetched set with no matching key
+    still raises ``KeyError``.
+    """
 
     url: str
     fetched_at: float = 0
@@ -113,6 +126,10 @@ class JWKS:
     refresh_jwks: bool = True
     refresh_interval_secs: int = 3600
     refresh_retry_interval_secs: int = 10
+    refresh_on_unknown_kid: bool = False
+    raise_on_fetch_error: bool = False
+    _fetch_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, init=False)
+    _fetch_failed: bool = attrs.field(default=False, init=False)
 
     def __repr__(self) -> str:
         return f"JWKS(url={self.url}, fetched_at={self.fetched_at})"
@@ -128,19 +145,40 @@ def from_private_key(cls, *keys: AllowedPrivateKeys | tuple[AllowedPrivateKeys,
         obj._jwks = jwt.PyJWKSet(keyset)
         return obj
 
-    async def fetch_jwks(self) -> None:
-        if not self._should_fetch_jwks():
+    async def fetch_jwks(self, *, force: bool = False) -> None:
+        # A usable cache skips the lock, so a slow refresh for an unknown key does not delay known keys.
+        if not force and self._jwks and not self._should_fetch_jwks():
             return
-        if self.url.startswith("http"):
-            data = await self._fetch_remote_jwks()
-        else:
-            data = self._fetch_local_jwks()
-
-        if not data:
-            return
-
-        self._jwks = jwt.PyJWKSet.from_dict(data)
-        log.debug("Fetched JWKS", url=self.url, keys=len(self._jwks.keys))
+        lock = self._fetch_lock if self.refresh_on_unknown_kid else nullcontext()
+        async with lock:
+            if not self._should_fetch_jwks(force=force):
+                return
+            self._fetch_failed = True
+            if self.url.startswith("http"):
+                data = await self._fetch_remote_jwks()
+            else:
+                data = self._fetch_local_jwks()
+
+            if not data:
+                return
+
+            try:
+                # Older PyJWT versions assume the body and every key are objects.
+                if not isinstance(data, dict):
+                    raise TypeError(f"JWKS must be a JSON object, not {type(data).__name__}")
+                keys = data.get("keys")
+                if not isinstance(keys, list) or any(not isinstance(key, dict) for key in keys):
+                    raise TypeError("JWKS keys must be an array of JSON objects")
+                # PyJWT rejects empty sets, but an issuer can remove all keys with an empty array.
+                keyset = jwt.PyJWKSet.from_dict(data) if keys else None
+            except (jwt.PyJWKError, jwt.PyJWKSetError, KeyError, TypeError, ValueError):
+                if not self.raise_on_fetch_error:
+                    raise
+                log.warning("Unable to load JWKS response", url=self.url, exc_info=True)
+                return
+            self._jwks = keyset
+            self._fetch_failed = False
+            log.debug("Fetched JWKS", url=self.url, keys=len(keyset.keys) if keyset is not None else 0)
 
     async def _fetch_remote_jwks(self) -> dict[str, Any] | None:
         try:
@@ -151,7 +189,7 @@ async def _fetch_remote_jwks(self) -> dict[str, Any] | None:
             )
             if TYPE_CHECKING:
                 assert self.url
-            self.last_fetch_attempt_at = int(time.monotonic())
+            self.last_fetch_attempt_at = time.monotonic()
             response = await self.client.get(self.url)
             response.raise_for_status()
             self.fetched_at = int(time.monotonic())
@@ -172,7 +210,7 @@ def _fetch_local_jwks(self) -> dict[str, Any] | None:
             log.exception("Failed to read local JWKS", url=self.url)
             return None
 
-    def _should_fetch_jwks(self) -> bool:
+    def _should_fetch_jwks(self, *, force: bool = False) -> bool:
         """
         Check if we need to fetch the JWKS based on the last fetch time and the refresh interval.
 
@@ -188,8 +226,16 @@ def _should_fetch_jwks(self) -> bool:
         # For remote fetches we check if the JWKS is not loaded (fetched_at = 0) or if the last fetch was more than
         # refresh_interval_secs ago and the last fetch attempt was more than refresh_retry_interval_secs ago
         now = time.monotonic()
+        # Unknown keys are untrusted input; bound fetches even for an empty cache or a failed refresh.
+        if (
+            self.refresh_on_unknown_kid
+            and self.last_fetch_attempt_at
+            and now - self.last_fetch_attempt_at < self.refresh_retry_interval_secs
+        ):
+            return False
         return self.refresh_jwks and (
-            not self._jwks
+            force
+            or not self._jwks
             or (
                 self.fetched_at == 0
                 or (
@@ -204,9 +250,20 @@ async def get_key(self, kid: str) -> jwt.PyJWK:
         await self.fetch_jwks()
 
         if self._jwks:
-            return self._jwks[kid]
+            try:
+                return self._jwks[kid]
+            except KeyError:
+                if self.refresh_on_unknown_kid:
+                    await self.fetch_jwks(force=True)
+                    if self._jwks is not None:
+                        try:
+                            return self._jwks[kid]
+                        except KeyError:
+                            pass
+
+        if self.raise_on_fetch_error and self._fetch_failed:
+            raise JWKSUnavailableError("Unable to load the required signing key")
 
-        # It didn't load!
         raise KeyError(f"Key ID {kid} not found in keyset")
 
     def status(self):

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

@@ -43,6 +43,10 @@
 from airflow.api_fastapi.common.telemetry import NATIVE_TELEMETRY_DISABLED
 from airflow.api_fastapi.execution_api.in_process import InProcessExecutionAPI
 from airflow.api_fastapi.execution_api.security import _REQUEST_SCOPE_TOKEN_KEY, require_auth
+from airflow.api_fastapi.execution_api.workload_identity import (
+    DagProcessorIdentityVerifier,
+    create_dag_processor_identity_verifier,
+)
 
 if TYPE_CHECKING:
     from airflow.models.dagbag import DBDagBag
@@ -107,7 +111,15 @@ async def lifespan(app: FastAPI, registry: svcs.Registry):
         # Create an app scoped validator, so that we don't have to fetch it every time
         registry.register_value(JWTValidator, _jwt_validator(), ping=JWTValidator.status)
 
-    yield
+    # InProcessExecutionAPI stubs out the identity verifier: don't create one in that case.
+    if DagProcessorIdentityVerifier in registry:
+        yield
+        return
+
+    async with create_dag_processor_identity_verifier() as verifier:
+        # None when the exchange is disabled; the route answers 404 then.
+        registry.register_value(DagProcessorIdentityVerifier, verifier)
+        yield
 
 
 class CorrelationIdMiddleware(BaseHTTPMiddleware):
@@ -277,6 +289,8 @@ def _inject_trace_context_dep(routes, mode: str) -> None:
     ``unsafe-always`` mode is handled with an app-level dependency instead, so extraction runs
     before authentication declared on any enclosing router.
     """
+    from airflow.api_fastapi.execution_api.routes.session_tokens import require_workload_identity
+
     for context in iter_route_contexts(routes):
         route = context.original_route
         if not isinstance(route, APIRoute):
@@ -288,7 +302,10 @@ def _inject_trace_context_dep(routes, mode: str) -> None:
         if mode != "only-authenticated":
             continue
         # context.dependencies also holds those of the enclosing routers and include_router() calls.
-        if any(getattr(d, "dependency", None) is require_auth for d in context.dependencies):
+        if any(
+            getattr(d, "dependency", None) in (require_auth, require_workload_identity)
+            for d in context.dependencies
+        ):
             route.dependencies.append(Depends(_extract_w3c_trace_context))