You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ep...@apache.org on 2026/10/20 17:05:47 UTC
(airflow) 02/02: Rotate Dag processor session tokens in Kubernetes Secrets
This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy pushed a commit to branch dag-processor-deployment
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 531b1878a8843324c558745ac9827dcd9c854749
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Tue Oct 20 17:16:47 2026 +0100
Rotate Dag processor session tokens in Kubernetes Secrets
Dag processors using the Execution API need renewable credentials without access to signing keys. Kubernetes deployments also need to retain their session when the token publisher restarts or an API response is lost.
---
.../security/dag-processor-api/helm-values.yaml | 3 +-
.../kubernetes-token-provisioner.yaml | 141 ++++++++
.../docs/security/jwt_token_authentication.rst | 110 ++++++-
airflow-core/src/airflow/cli/cli_config.py | 18 +-
.../cli/commands/dag_processor_token_command.py | 79 ++++-
.../cli/commands/dag_processor_token_kubernetes.py | 151 +++++++++
.../commands/test_dag_processor_token_command.py | 171 ++++++++++
.../test_dag_processor_token_kubernetes.py | 361 +++++++++++++++++++++
.../helm_tests/airflow_core/test_dag_processor.py | 98 ++++++
9 files changed, 1108 insertions(+), 24 deletions(-)
diff --git a/airflow-core/docs/security/dag-processor-api/helm-values.yaml b/airflow-core/docs/security/dag-processor-api/helm-values.yaml
index a4ecfcf877d..1013e3ef5d5 100644
--- a/airflow-core/docs/security/dag-processor-api/helm-values.yaml
+++ b/airflow-core/docs/security/dag-processor-api/helm-values.yaml
@@ -15,7 +15,8 @@
# specific language governing permissions and limitations
# under the License.
---
-# Install as a separate processor-only release, using an image with API mode.
+# Install as a separate processor-only release in airflow-dag-processor.
+# Apply kubernetes-token-provisioner.yaml first, using an image with API mode.
airflowVersion: "3.5.0"
defaultAirflowTag: "3.5.0"
executor: LocalExecutor
diff --git a/airflow-core/docs/security/dag-processor-api/kubernetes-token-provisioner.yaml b/airflow-core/docs/security/dag-processor-api/kubernetes-token-provisioner.yaml
new file mode 100644
index 00000000000..cf1637aec2a
--- /dev/null
+++ b/airflow-core/docs/security/dag-processor-api/kubernetes-token-provisioner.yaml
@@ -0,0 +1,141 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+---
+apiVersion: v1
+kind: Namespace
+metadata:
+ name: airflow-dag-processor
+---
+# Omit data/stringData so applying this manifest again preserves the current session.
+apiVersion: v1
+kind: Secret
+metadata:
+ name: dag-processor-session
+ namespace: airflow-dag-processor
+type: Opaque
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+ name: dag-processor-token-provisioner
+ namespace: airflow
+automountServiceAccountToken: true
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+ name: dag-processor-token-provisioner
+ namespace: airflow-dag-processor
+rules:
+ - apiGroups: [""]
+ resources: [secrets]
+ resourceNames: [dag-processor-session]
+ verbs: [get, patch]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+ name: dag-processor-token-provisioner
+ namespace: airflow-dag-processor
+roleRef:
+ apiGroup: rbac.authorization.k8s.io
+ kind: Role
+ name: dag-processor-token-provisioner
+subjects:
+ - kind: ServiceAccount
+ name: dag-processor-token-provisioner
+ namespace: airflow
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: dag-processor-token-provisioner
+ namespace: airflow
+spec:
+ replicas: 1
+ strategy:
+ type: Recreate
+ selector:
+ matchLabels:
+ component: dag-processor-token-provisioner
+ template:
+ metadata:
+ labels:
+ component: dag-processor-token-provisioner
+ spec:
+ serviceAccountName: dag-processor-token-provisioner
+ automountServiceAccountToken: true
+ securityContext:
+ runAsNonRoot: true
+ runAsUser: 50000
+ runAsGroup: 0
+ fsGroup: 0
+ seccompProfile:
+ type: RuntimeDefault
+ containers:
+ - name: token-provisioner
+ # Use an image containing the Kubernetes token provisioner command.
+ image: apache/airflow:3.5.0
+ args:
+ - airflow
+ - dag-processor-token
+ - --kubernetes-secret
+ - dag-processor-session
+ - --kubernetes-namespace
+ - airflow-dag-processor
+ - --bundle-name
+ - dags-folder
+ - --valid-for
+ - "600"
+ - --rotate
+ env:
+ - name: AIRFLOW_HOME
+ value: /tmp/airflow
+ - name: AIRFLOW_CONFIG
+ value: /dev/null
+ - name: AIRFLOW__CORE__LOAD_EXAMPLES
+ value: "False"
+ - name: CONNECTION_CHECK_MAX_COUNT
+ value: "0"
+ - name: AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST
+ value: >-
+ [{"name":"dags-folder",
+ "classpath":"airflow.dag_processing.bundles.local.LocalDagBundle","kwargs":{}}]
+ - name: AIRFLOW__API_AUTH__JWT_SECRET
+ valueFrom:
+ secretKeyRef:
+ # Match the trusted API server's jwtSecretName (default: <release>-jwt-secret).
+ name: airflow-jwt-secret
+ key: jwt-secret
+ securityContext:
+ allowPrivilegeEscalation: false
+ readOnlyRootFilesystem: true
+ capabilities:
+ drop: [ALL]
+ resources:
+ requests:
+ cpu: 100m
+ memory: 256Mi
+ limits:
+ cpu: 500m
+ memory: 512Mi
+ volumeMounts:
+ - name: tmp
+ mountPath: /tmp
+ volumes:
+ - name: tmp
+ emptyDir: {}
diff --git a/airflow-core/docs/security/jwt_token_authentication.rst b/airflow-core/docs/security/jwt_token_authentication.rst
index d05ae02eac1..10ef8086c7a 100644
--- a/airflow-core/docs/security/jwt_token_authentication.rst
+++ b/airflow-core/docs/security/jwt_token_authentication.rst
@@ -704,18 +704,51 @@ entrypoint otherwise checks the database before starting the command. Do not set
Isolated deployment examples
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-The following examples run only the processor, alongside an existing trusted deployment.
+The following examples isolate the processor from an existing trusted deployment.
Use an image containing processor API support; this feature targets Airflow 3.5.
-Synchronize the bundle catalog and run token provisioning in the trusted deployment as
-described above. Keep the token provisioner running, and mount its directory rather than
-a single file so atomic token replacement remains visible.
+Synchronize the bundle catalog in the trusted deployment before starting the processor,
+and again whenever the bundle configuration changes::
+
+ airflow dag-processor --sync-bundles-only
+
+This step needs database access. Token provisioning itself needs only the API's signing
+configuration and the configured bundle names; it does not access the metadata database.
+
+Kubernetes token provisioner
+""""""""""""""""""""""""""""
For Kubernetes, use the :download:`processor-only Helm values <dag-processor-api/helm-values.yaml>`
in a **separate release and namespace**. These values disable other components and built-in
server credential injection. Do not apply them to the release running the scheduler/API server.
-Create the ``dag-processor-session`` Secret containing ``processor.jwt`` and the
-``processor-dags`` PVC in that namespace, then configure the API service address and image.
-The processor service account has no mounted API token; a trusted provisioner updates the Secret.
+
+The :download:`token provisioner manifest <dag-processor-api/kubernetes-token-provisioner.yaml>`
+automatically maintains the processor's ``dag-processor-session`` Secret. It runs
+``airflow dag-processor-token --kubernetes-secret dag-processor-session
+--kubernetes-namespace airflow-dag-processor --bundle-name dags-folder --valid-for 600 --rotate``
+in the trusted ``airflow`` namespace. Its service account can only get and patch that
+named Secret in the ``airflow-dag-processor`` namespace. The processor has neither the
+signing key nor a mounted Kubernetes API credential.
+
+Customize both files before applying them:
+
+* Set both images to an image containing this feature. The manifest assumes the trusted
+ release and namespace are named ``airflow``; update namespace references, the API
+ service address, and NetworkPolicy selectors together if yours differ.
+* Point the provisioner's ``AIRFLOW__API_AUTH__JWT_SECRET`` at the API server's signing
+ Secret. For the Helm chart's default configuration this is ``<release>-jwt-secret``,
+ key ``jwt-secret``; ``jwtSecretName`` overrides the Secret name. For asymmetric signing,
+ replace this environment variable with the matching private-key mount and
+ ``AIRFLOW__API_AUTH__JWT_PRIVATE_KEY_PATH``.
+* Copy any non-default signing algorithm, key ID, issuer, and execution API audience
+ settings from the API server. Do not copy database credentials or the entire server
+ configuration into the provisioner. Restart the provisioner when its signing
+ configuration changes, including a signing Secret supplied through environment variables.
+* Match ``--bundle-name`` and ``AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST`` to the
+ catalog and processor configuration. Repeat ``--bundle-name`` for additional bundles.
+ Provisioning reads their names without loading bundle classes or fetching Dag files.
+* Create the ``processor-dags`` PVC in ``airflow-dag-processor`` and populate it with
+ the local Dag files. The provisioner needs no volume for its session state: it reads
+ the current token from the Secret on startup and preserves that session across restarts.
Apply the :download:`egress policy <dag-processor-api/network-policy.yaml>` in the processor
namespace before starting its pods. It permits only DNS and the trusted API server. Adjust
@@ -724,11 +757,72 @@ requires a supporting cluster network plugin; other policies selecting these pod
grant broader egress. This example uses local Dag files. Remote bundles need additional,
explicit egress to their source service.
+For a fresh deployment, keep the empty ``Secret`` document in the provisioner manifest.
+The Secret must exist before the command runs: the provisioner has no permission to create
+or delete Secrets.
+
+To adopt an existing processor Secret while preserving its session:
+
+* Stop the previous token publisher and prevent it from restarting.
+* Keep the existing Secret and its ``processor.jwt`` value. Match the provisioner's
+ Secret name, namespace, and bundle names to that token's current grant.
+* Remove the entire ``kind: Secret`` document from your copy of
+ ``kubernetes-token-provisioner.yaml`` before applying it, and keep it excluded from
+ subsequent applies. Do not delete the live Secret. If an earlier ``kubectl apply``
+ managed its token data, applying an empty Secret document can remove that data through
+ the last-applied configuration and cause the provisioner to create a new session.
+
+After preparing the files for a fresh deployment or adoption, start the provisioner
+and then the processor::
+
+ kubectl apply -f kubernetes-token-provisioner.yaml
+ kubectl -n airflow rollout status deployment/dag-processor-token-provisioner
+ kubectl -n airflow-dag-processor wait --for=jsonpath='{.data.processor\.jwt}' secret/dag-processor-session --timeout=120s
+ kubectl -n airflow-dag-processor apply -f network-policy.yaml
+ helm upgrade --install dag-processor apache-airflow/airflow \
+ --namespace airflow-dag-processor --values helm-values.yaml
+
+For Secrets created by this example, the manifest deliberately omits ``data`` and
+``stringData`` from the first apply, so later applies leave the provisioner's token intact.
+When adopting an existing Secret, the provisioner reads its current token and reuses the
+session for the same bundle grant. Restrict other writers to this Secret; deleting and
+recreating it or changing the authorized bundle set creates a new session. Restart the
+processor when changing sessions. Each concurrently running
+processor needs a separate provisioner and Secret. Both example Deployments use one replica and
+``Recreate``; do not scale a deployment whose replicas share a session token.
+
+The provisioner writes a token immediately and rotates it every 300 seconds, with each
+token valid for 600 seconds. It updates the Secret directly; no external ``kubectl`` loop,
+sidecar publisher, or persistent token directory is needed. Kubernetes eventually projects
+the updated Secret into the processor's mounted directory, which the processor reloads
+without restarting. Keep the directory mount in the values file: a ``subPath`` mount
+does not receive `Secret updates <https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-files-from-a-pod>`_.
+Choose a token lifetime that leaves time for your cluster's projection delay and transient outages.
+
+The rollout check confirms that the provisioner process started; the Secret-key wait
+checks only that a token has been published at least once. Neither check proves that
+an existing token is still valid or that rotation continues successfully.
+
+Allow the trusted provisioner to reach the Kubernetes API over TLS if the trusted
+namespace restricts egress. It uses its mounted service-account token and cluster CA,
+and reloads that credential on each request. In rotation mode, Kubernetes read and update
+failures are retried with backoff, including during initial issuance. Initial signing or
+configuration errors exit the command and Kubernetes restarts it. Later rotation failures
+also retry; the previous token remains usable until it expires. An outage lasting beyond expiry
+prevents the processor from renewing its API credentials. Monitor provisioner failures
+and processor readiness, and restore Kubernetes API access or signing configuration
+before the token expires. Monitor the token's expiry without logging its value.
+
+Compose token provisioner
+"""""""""""""""""""""""""
+
For Compose, use the :download:`standalone processor definition <dag-processor-api/docker-compose.yaml>`
as its own project, rather than merging it with the quick-start file. It has no shared
``env_file``, server configuration mount, or dependency on a database/migration service.
Create its ``tokens``, ``dags``, and ``logs`` directories with permissions for the configured
-container user; provision ``tokens/processor.jwt``. Set ``AIRFLOW_IMAGE_NAME`` to the image
+container user; run the file-based token provisioner described above continuously to
+maintain ``tokens/processor.jwt``. Mount the directory rather than a single file so
+atomic token replacement remains visible. Set ``AIRFLOW_IMAGE_NAME`` to the image
with API support. Create an internal network and attach only the API server and processor::
docker network create --internal dag-processor-api
diff --git a/airflow-core/src/airflow/cli/cli_config.py b/airflow-core/src/airflow/cli/cli_config.py
index c75f4bd4e14..b3a95606b88 100644
--- a/airflow-core/src/airflow/cli/cli_config.py
+++ b/airflow-core/src/airflow/cli/cli_config.py
@@ -817,7 +817,14 @@ ARG_DEV = Arg(("-d", "--dev"), help="Start in development mode with hot-reload e
ARG_DAG_PROCESSOR_TOKEN_FILE = Arg(
("--token-file",),
help="File to write the token to; it is replaced in one step and readable only by the current user",
- required=True,
+)
+ARG_DAG_PROCESSOR_TOKEN_KUBERNETES_SECRET = Arg(
+ ("--kubernetes-secret",),
+ help="Existing Kubernetes Secret to update at key processor.jwt, instead of writing --token-file",
+)
+ARG_DAG_PROCESSOR_TOKEN_KUBERNETES_NAMESPACE = Arg(
+ ("--kubernetes-namespace",),
+ help="Namespace of --kubernetes-secret; uses the pod's service account and cluster CA",
)
ARG_DAG_PROCESSOR_TOKEN_VALID_FOR = Arg(
("--valid-for",),
@@ -2332,16 +2339,19 @@ core_commands: list[CLICommand] = [
),
ActionCommand(
name="dag-processor-token",
- help="Write a Dag processor session token to a file",
+ help="Write a Dag processor session token to a file or Kubernetes Secret",
description=(
"Issue a session token that lets a Dag processor reach the Execution API for the given "
"bundles, without the processor holding the signing key. Run it where the signing key is "
- "available, and give the processor read access to the token file; its mode and group are kept when "
- "rotation replaces it."
+ "available. Choose --token-file or --kubernetes-secret with --kubernetes-namespace. "
+ "File rotation preserves mode and group. Kubernetes rotation updates processor.jwt in an existing "
+ "Secret using the pod's service account; mount that Secret as a directory on the processor."
),
func=lazy_load_command("airflow.cli.commands.dag_processor_token_command.dag_processor_token"),
args=(
ARG_DAG_PROCESSOR_TOKEN_FILE,
+ ARG_DAG_PROCESSOR_TOKEN_KUBERNETES_SECRET,
+ ARG_DAG_PROCESSOR_TOKEN_KUBERNETES_NAMESPACE,
ARG_BUNDLE_NAME,
ARG_DAG_PROCESSOR_TOKEN_VALID_FOR,
ARG_DAG_PROCESSOR_TOKEN_ROTATE,
diff --git a/airflow-core/src/airflow/cli/commands/dag_processor_token_command.py b/airflow-core/src/airflow/cli/commands/dag_processor_token_command.py
index fbaaf5f6cb0..c1fdb077c90 100644
--- a/airflow-core/src/airflow/cli/commands/dag_processor_token_command.py
+++ b/airflow-core/src/airflow/cli/commands/dag_processor_token_command.py
@@ -24,7 +24,9 @@ import os
import stat
import tempfile
import time
+from functools import partial
from pathlib import Path
+from typing import TYPE_CHECKING
from uuid import UUID
import jwt
@@ -32,10 +34,17 @@ import uuid6
from airflow.api_fastapi.execution_api.app import create_jwt_generator
from airflow.api_fastapi.execution_api.dag_processor_tokens import generate_dag_processor_session_token
+from airflow.cli.commands.dag_processor_token_kubernetes import (
+ KubernetesSecretTokenError,
+ KubernetesSecretTokenStore,
+)
from airflow.configuration import conf
from airflow.dag_processing.bundles.manager import get_configured_bundle_names
from airflow.utils.providers_configuration_loader import providers_configuration_loaded
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
log = logging.getLogger(__name__)
@@ -63,9 +72,18 @@ def write_token_file(path: Path, token: str) -> None:
def get_session_id(token_file: Path, bundle_names: set[str]) -> UUID:
"""Reuse the session of the token already in the file, so a restart keeps the processors' Jobs."""
- with contextlib.suppress(OSError, KeyError, TypeError, ValueError, jwt.PyJWTError):
- # Unverified: the file is this command's own output.
- claims = jwt.decode(token_file.read_text(), options={"verify_signature": False})
+ with contextlib.suppress(OSError, ValueError):
+ return get_session_id_from_token(token_file.read_text(), bundle_names)
+ return uuid6.uuid7()
+
+
+def get_session_id_from_token(token: str | None, bundle_names: set[str]) -> UUID:
+ """Reuse the session in a previous provisioner's output for the same bundle grant."""
+ if not token:
+ return uuid6.uuid7()
+ with contextlib.suppress(KeyError, TypeError, ValueError, jwt.PyJWTError):
+ # Unverified: only trusted provisioners can write this file or Secret.
+ claims = jwt.decode(token, options={"verify_signature": False})
if claims["scope"] == "dag_processor_session" and set(claims["dag_bundles"]) == bundle_names:
return UUID(claims["sub"])
return uuid6.uuid7()
@@ -75,7 +93,11 @@ def get_session_id(token_file: Path, bundle_names: set[str]) -> UUID:
# component that only holds the signing key cannot reach.
@providers_configuration_loaded
def dag_processor_token(args) -> None:
- """Write a Dag processor session token to a file, and with ``--rotate`` keep replacing it."""
+ """Publish a Dag processor session token, and with ``--rotate`` keep replacing it."""
+ if bool(args.token_file) == bool(args.kubernetes_secret):
+ raise SystemExit("Choose exactly one of --token-file or --kubernetes-secret")
+ if bool(args.kubernetes_secret) != bool(args.kubernetes_namespace):
+ raise SystemExit("--kubernetes-secret and --kubernetes-namespace must be supplied together")
# Names only: provisioning runs where the signing key is, which may not have the bundle classes installed.
configured = get_configured_bundle_names()
bundle_names = set(args.bundle_name or configured)
@@ -83,27 +105,62 @@ def dag_processor_token(args) -> None:
raise SystemExit(f"Bundles not found: {', '.join(sorted(unknown))}")
valid_for = args.valid_for or conf.getint("execution_api", "jwt_expiration_time")
+ if args.kubernetes_secret:
+ with KubernetesSecretTokenStore(args.kubernetes_secret, args.kubernetes_namespace) as store:
+ _run_token_rotation(
+ bundle_names,
+ valid_for,
+ get_session=lambda: get_session_id_from_token(store.read_token(), bundle_names),
+ publish_token=store.write_token,
+ rotate=args.rotate,
+ retry_initial_publication=True,
+ )
+ return
+
token_file = Path(args.token_file)
session_id = get_session_id(token_file, bundle_names)
- log.info("Issuing Dag processor session %s for bundles %s", session_id, ", ".join(sorted(bundle_names)))
+ _run_token_rotation(
+ bundle_names,
+ valid_for,
+ get_session=lambda: session_id,
+ publish_token=partial(write_token_file, token_file),
+ rotate=args.rotate,
+ )
+
+
+def _run_token_rotation(
+ bundle_names: set[str],
+ valid_for: int,
+ *,
+ get_session: Callable[[], UUID],
+ publish_token: Callable[[str], None],
+ rotate: bool,
+ retry_initial_publication: bool = False,
+) -> None:
issued = False
failures = 0
while True:
try:
+ session_id = get_session()
token = generate_dag_processor_session_token(
create_jwt_generator(), session_id=session_id, bundle_names=bundle_names, valid_for=valid_for
)
- write_token_file(token_file, token)
- except Exception:
- # Without a token to keep, or without rotation, the operator must see the failure.
- if not (args.rotate and issued):
+ publish_token(token)
+ except Exception as exc:
+ # A failed first Kubernetes PATCH may already have committed; re-read its session on retry.
+ if not rotate or not (
+ issued or (retry_initial_publication and isinstance(exc, KubernetesSecretTokenError))
+ ):
raise
failures += 1
- log.exception("Could not rotate the session token; the previous one stays valid until it expires")
+ log.exception("Could not publish the session token; any previous token keeps its original expiry")
time.sleep(min(2 ** min(failures, 10), valid_for / 4))
continue
issued = True
failures = 0
- if not args.rotate:
+ log.info(
+ "Published Dag processor session %s for bundles %s", session_id, ", ".join(sorted(bundle_names))
+ )
+ if not rotate:
return
time.sleep(valid_for / 2)
diff --git a/airflow-core/src/airflow/cli/commands/dag_processor_token_kubernetes.py b/airflow-core/src/airflow/cli/commands/dag_processor_token_kubernetes.py
new file mode 100644
index 00000000000..325ee767acf
--- /dev/null
+++ b/airflow-core/src/airflow/cli/commands/dag_processor_token_kubernetes.py
@@ -0,0 +1,151 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Publish Dag processor session tokens to an existing Kubernetes Secret."""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import os
+import re
+import ssl
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import httpx2
+
+if TYPE_CHECKING:
+ from types import TracebackType
+
+_SERVICE_ACCOUNT_DIRECTORY = Path("/var/run/secrets/kubernetes.io/serviceaccount")
+_TOKEN_KEY = "processor.jwt"
+_DNS_LABEL = r"[a-z0-9](?:[-a-z0-9]*[a-z0-9])?"
+
+
+class KubernetesSecretTokenError(RuntimeError):
+ """A Secret could not be read or updated without exposing its contents."""
+
+ def __init__(self, message: str, *, status_code: int | None = None):
+ super().__init__(message)
+ self.status_code = status_code
+
+
+class KubernetesSecretTokenStore:
+ """Read and conditionally update one pre-created Secret using in-cluster credentials."""
+
+ def __init__(self, secret_name: str, namespace: str):
+ if len(namespace) > 63 or re.fullmatch(_DNS_LABEL, namespace) is None:
+ raise ValueError("Kubernetes namespace must be a DNS label of at most 63 characters")
+ if len(secret_name) > 253 or re.fullmatch(rf"{_DNS_LABEL}(?:\.{_DNS_LABEL})*", secret_name) is None:
+ raise ValueError("Kubernetes Secret name must be a DNS subdomain of at most 253 characters")
+ self._path = f"/api/v1/namespaces/{namespace}/secrets/{secret_name}"
+ self._resource_version: str | None = None
+ host = os.environ.get("KUBERNETES_SERVICE_HOST")
+ port = os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS") or os.environ.get("KUBERNETES_SERVICE_PORT")
+ if not host or not port:
+ raise ValueError(
+ "Kubernetes API access requires KUBERNETES_SERVICE_HOST and "
+ "KUBERNETES_SERVICE_PORT_HTTPS or KUBERNETES_SERVICE_PORT"
+ )
+ try:
+ port_number = int(port)
+ if not 1 <= port_number <= 65535:
+ raise ValueError
+ except ValueError:
+ raise ValueError("Kubernetes service port must be an integer between 1 and 65535") from None
+ self._client = httpx2.Client(
+ base_url=httpx2.URL(scheme="https", host=host, port=port_number),
+ verify=ssl.create_default_context(cafile=str(_SERVICE_ACCOUNT_DIRECTORY / "ca.crt")),
+ timeout=10,
+ follow_redirects=False,
+ trust_env=False,
+ )
+
+ def __enter__(self) -> KubernetesSecretTokenStore:
+ return self
+
+ def __exit__(
+ self,
+ exc_type: type[BaseException] | None,
+ exc_value: BaseException | None,
+ traceback: TracebackType | None,
+ ) -> None:
+ self._client.close()
+
+ def _request(self, method: str, *, body: dict | None = None) -> httpx2.Response:
+ # Kubelet rotates this credential independently of the session token we publish.
+ try:
+ bearer = (_SERVICE_ACCOUNT_DIRECTORY / "token").read_text().strip()
+ except OSError:
+ raise KubernetesSecretTokenError("Could not read the Kubernetes service account token") from None
+ if not bearer:
+ raise KubernetesSecretTokenError("The Kubernetes service account token is empty")
+ headers = {"Authorization": f"Bearer {bearer}"}
+ if body is not None:
+ headers["Content-Type"] = "application/merge-patch+json"
+ try:
+ response = self._client.request(method, self._path, headers=headers, json=body)
+ except httpx2.HTTPError as error:
+ raise KubernetesSecretTokenError(
+ f"Kubernetes Secret {method} request failed ({type(error).__name__})"
+ ) from None
+ if not response.is_success:
+ raise KubernetesSecretTokenError(
+ f"Kubernetes Secret {method} returned HTTP {response.status_code}",
+ status_code=response.status_code,
+ )
+ return response
+
+ def read_token(self) -> str | None:
+ """Read the current token and retain the version required for the next write."""
+ self._resource_version = None
+ response = self._request("GET")
+ try:
+ secret = response.json()
+ resource_version = secret["metadata"]["resourceVersion"]
+ if not isinstance(resource_version, str) or not resource_version:
+ raise ValueError
+ data = secret.get("data")
+ if data is None:
+ data = {}
+ if not isinstance(data, dict):
+ raise ValueError
+ encoded = data.get(_TOKEN_KEY, "")
+ if not isinstance(encoded, str):
+ raise ValueError
+ token = base64.b64decode(encoded, validate=True).decode().strip() or None
+ except (ValueError, TypeError, KeyError, AttributeError, binascii.Error):
+ raise KubernetesSecretTokenError("Kubernetes Secret has invalid token data or metadata") from None
+ self._resource_version = resource_version
+ return token
+
+ def write_token(self, token: str) -> None:
+ """Update only the token, rejecting writes based on a stale Secret version."""
+ resource_version = self._resource_version
+ self._resource_version = None
+ if resource_version is None:
+ raise KubernetesSecretTokenError("Read the Kubernetes Secret successfully before writing a token")
+ if not token.strip():
+ raise ValueError("The Dag processor session token must not be empty")
+ # Consume the version even on an uncertain response: the next attempt must read the persisted session.
+ self._request(
+ "PATCH",
+ body={
+ "metadata": {"resourceVersion": resource_version},
+ "data": {_TOKEN_KEY: base64.b64encode(token.encode()).decode("ascii")},
+ },
+ )
diff --git a/airflow-core/tests/unit/cli/commands/test_dag_processor_token_command.py b/airflow-core/tests/unit/cli/commands/test_dag_processor_token_command.py
index b29e9460337..ebace24ffb7 100644
--- a/airflow-core/tests/unit/cli/commands/test_dag_processor_token_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_dag_processor_token_command.py
@@ -27,6 +27,7 @@ import pytest
from airflow.cli import cli_parser
from airflow.cli.commands import dag_processor_token_command
+from airflow.cli.commands.dag_processor_token_kubernetes import KubernetesSecretTokenError
from airflow.dag_processing.bundles.manager import _load_bundle_config_snapshot
from tests_common.test_utils.config import conf_vars
@@ -167,6 +168,176 @@ class TestDagProcessorTokenCommand:
mock_sleep.assert_not_called()
+ @pytest.mark.parametrize(
+ ("args", "error"),
+ [
+ ([], "Choose exactly one"),
+ (["--token-file", "token", "--kubernetes-secret", "processor"], "Choose exactly one"),
+ (["--kubernetes-secret", "processor"], "must be supplied together"),
+ (["--token-file", "token", "--kubernetes-namespace", "processors"], "must be supplied together"),
+ ],
+ )
+ @mock.patch.object(dag_processor_token_command, "KubernetesSecretTokenStore", autospec=True)
+ def test_rejects_invalid_output_options(self, store, args, error):
+ with pytest.raises(SystemExit, match=error):
+ self._run(*args)
+ store.assert_not_called()
+
+
+class TestKubernetesTokenCommand:
+ @pytest.fixture
+ def secret_store(self):
+ with mock.patch.object(
+ dag_processor_token_command, "KubernetesSecretTokenStore", autospec=True
+ ) as factory:
+ store = factory.return_value
+ store.__enter__.return_value = store
+ published = []
+ store.read_token.side_effect = lambda: published[-1] if published else None
+ store.write_token.side_effect = published.append
+ yield store, published
+
+ factory.assert_called_with("processor-session", "processors")
+ store.__exit__.assert_called()
+
+ def _run(self, *args: str) -> None:
+ dag_processor_token_command.dag_processor_token(
+ cli_parser.get_parser().parse_args(
+ [
+ "dag-processor-token",
+ "--kubernetes-secret",
+ "processor-session",
+ "--kubernetes-namespace",
+ "processors",
+ *args,
+ ]
+ )
+ )
+
+ def test_bootstraps_without_a_local_token_file(self, secret_store):
+ store, published = secret_store
+
+ self._run("-B", "bundle_a")
+
+ claims = _decode(published[0])
+ assert claims["scope"] == "dag_processor_session"
+ assert claims["dag_bundles"] == ["bundle_a"]
+ assert claims["exp"] - claims["iat"] == 600
+ store.read_token.assert_called_once_with()
+
+ @pytest.mark.parametrize("bundle", ["bundle_a", "bundle_b"])
+ @pytest.mark.parametrize("elapsed", [0, 601], ids=["unexpired", "expired"])
+ def test_restart_recovers_the_session_unless_grants_change(
+ self, secret_store, bundle, elapsed, time_machine
+ ):
+ _, published = secret_store
+ time_machine.move_to("2026-10-20T12:00:00Z", tick=False)
+ self._run("-B", "bundle_a")
+ time_machine.shift(elapsed)
+
+ self._run("-B", bundle)
+
+ first, second = map(_decode, published)
+ assert (first["sub"] == second["sub"]) == (bundle == "bundle_a")
+ assert first["jti"] != second["jti"]
+ assert second["dag_bundles"] == [bundle]
+ assert second["exp"] - first["exp"] == elapsed
+
+ @mock.patch.object(
+ dag_processor_token_command.time, "sleep", autospec=True, side_effect=[None, _StopRotation]
+ )
+ def test_rotation_reloads_the_secret_and_preserves_the_session(self, mock_sleep, secret_store):
+ store, published = secret_store
+ with pytest.raises(_StopRotation):
+ self._run("--valid-for", "120", "--rotate")
+
+ first, second = map(_decode, published)
+ assert first["sub"] == second["sub"]
+ assert first["jti"] != second["jti"]
+ assert store.read_token.call_count == 2
+ assert mock_sleep.call_args_list == [mock.call(60), mock.call(60)]
+
+ @mock.patch.object(
+ dag_processor_token_command.time, "sleep", autospec=True, side_effect=[None, _StopRotation]
+ )
+ def test_first_patch_conflict_recovers_the_winning_session(self, mock_sleep, secret_store):
+ store, published = secret_store
+ self._run()
+ winner = published[0]
+ store.read_token.side_effect = [None, winner]
+ store.write_token.reset_mock()
+ store.write_token.side_effect = [KubernetesSecretTokenError("Conflict"), None]
+
+ with pytest.raises(_StopRotation):
+ self._run("--valid-for", "120", "--rotate")
+
+ rejected, retried = (_decode(call.args[0]) for call in store.write_token.call_args_list)
+ assert rejected["sub"] != _decode(winner)["sub"]
+ assert retried["sub"] == _decode(winner)["sub"]
+ assert mock_sleep.call_args_list == [mock.call(2), mock.call(60)]
+
+ @mock.patch.object(
+ dag_processor_token_command.time, "sleep", autospec=True, side_effect=[None, _StopRotation]
+ )
+ def test_lost_first_patch_response_recovers_the_published_session(self, mock_sleep, secret_store):
+ store, published = secret_store
+
+ def publish(token):
+ published.append(token)
+ if len(published) == 1:
+ raise KubernetesSecretTokenError("Response timed out")
+
+ store.write_token.side_effect = publish
+ with pytest.raises(_StopRotation):
+ self._run("--valid-for", "120", "--rotate")
+
+ first, second = map(_decode, published)
+ assert first["sub"] == second["sub"]
+ assert first["jti"] != second["jti"]
+ assert mock_sleep.call_args_list == [mock.call(2), mock.call(60)]
+
+ @mock.patch.object(
+ dag_processor_token_command.time, "sleep", autospec=True, side_effect=[None, _StopRotation]
+ )
+ def test_first_read_failure_is_retried_without_publishing(self, mock_sleep, secret_store):
+ store, published = secret_store
+ store.read_token.side_effect = [KubernetesSecretTokenError("API unavailable"), None]
+
+ with pytest.raises(_StopRotation):
+ self._run("--valid-for", "120", "--rotate")
+
+ assert len(published) == 1
+ assert mock_sleep.call_args_list == [mock.call(2), mock.call(60)]
+
+ @mock.patch.object(dag_processor_token_command.time, "sleep", autospec=True)
+ @pytest.mark.parametrize("operation", ["read_token", "write_token"])
+ def test_one_shot_failure_is_reported(self, mock_sleep, secret_store, operation):
+ store, published = secret_store
+ getattr(store, operation).side_effect = KubernetesSecretTokenError("Forbidden")
+
+ with pytest.raises(KubernetesSecretTokenError, match="Forbidden"):
+ self._run()
+
+ assert not published
+ mock_sleep.assert_not_called()
+
+ @mock.patch.object(dag_processor_token_command.time, "sleep", autospec=True)
+ @mock.patch.object(
+ dag_processor_token_command,
+ "create_jwt_generator",
+ autospec=True,
+ side_effect=ValueError("Invalid signing configuration"),
+ )
+ def test_first_signing_failure_exits_without_publishing(self, _, mock_sleep, secret_store):
+ store, published = secret_store
+
+ with pytest.raises(ValueError, match="Invalid signing configuration"):
+ self._run("--rotate")
+
+ assert not published
+ store.write_token.assert_not_called()
+ mock_sleep.assert_not_called()
+
class TestWriteTokenFile:
def test_creates_the_file_readable_only_by_its_owner(self, tmp_path):
diff --git a/airflow-core/tests/unit/cli/commands/test_dag_processor_token_kubernetes.py b/airflow-core/tests/unit/cli/commands/test_dag_processor_token_kubernetes.py
new file mode 100644
index 00000000000..544ccb0430c
--- /dev/null
+++ b/airflow-core/tests/unit/cli/commands/test_dag_processor_token_kubernetes.py
@@ -0,0 +1,361 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import base64
+import json
+import ssl
+import traceback
+from functools import partial
+from unittest import mock
+
+import httpx2
+import pytest
+
+from airflow.cli.commands import dag_processor_token_kubernetes
+from airflow.cli.commands.dag_processor_token_kubernetes import (
+ KubernetesSecretTokenError,
+ KubernetesSecretTokenStore,
+)
+
+
+def make_secret_response(token: str = "old-session", version: str = "10") -> httpx2.Response:
+ return httpx2.Response(
+ 200,
+ json={
+ "metadata": {"resourceVersion": version},
+ "data": {
+ "processor.jwt": base64.b64encode(token.encode()).decode(),
+ "other-key": base64.b64encode(b"preserved").decode(),
+ },
+ },
+ )
+
+
+@pytest.fixture
+def service_account_directory(tmp_path, monkeypatch):
+ (tmp_path / "token").write_text("service-account-1\n")
+ monkeypatch.setattr(dag_processor_token_kubernetes, "_SERVICE_ACCOUNT_DIRECTORY", tmp_path)
+ return tmp_path
+
+
+@pytest.fixture
+def make_store(service_account_directory, monkeypatch):
+ monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1")
+ monkeypatch.setenv("KUBERNETES_SERVICE_PORT", "443")
+ monkeypatch.delenv("KUBERNETES_SERVICE_PORT_HTTPS", raising=False)
+ client_class = httpx2.Client
+ context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+ create_context = mock.create_autospec(ssl.create_default_context, return_value=context)
+ monkeypatch.setattr(dag_processor_token_kubernetes.ssl, "create_default_context", create_context)
+ clients = []
+
+ def create(*responses):
+ requests = []
+ pending = iter(responses)
+
+ def handle(request):
+ requests.append(request)
+ result = next(pending)
+ if isinstance(result, Exception):
+ raise result
+ return result
+
+ def create_client(**kwargs):
+ return client_class(transport=httpx2.MockTransport(handle), **kwargs)
+
+ factory = mock.create_autospec(client_class, side_effect=create_client)
+ monkeypatch.setattr(dag_processor_token_kubernetes.httpx2, "Client", factory)
+ store = KubernetesSecretTokenStore("processor-session", "dag-processors")
+ clients.append(store)
+ return store, requests, factory, create_context
+
+ yield create
+ for store in clients:
+ store.__exit__(None, None, None)
+
+
+def test_reads_and_conditionally_updates_only_the_token(make_store):
+ store, requests, _, _ = make_store(make_secret_response(), httpx2.Response(200, json={}))
+
+ with store:
+ assert store.read_token() == "old-session"
+ store.write_token("new-session")
+
+ assert [request.method for request in requests] == ["GET", "PATCH"]
+ assert str(requests[0].url) == (
+ "https://10.96.0.1/api/v1/namespaces/dag-processors/secrets/processor-session"
+ )
+ assert requests[1].headers["content-type"] == "application/merge-patch+json"
+ assert json.loads(requests[1].content) == {
+ "metadata": {"resourceVersion": "10"},
+ "data": {"processor.jwt": base64.b64encode(b"new-session").decode()},
+ }
+ assert store._client.is_closed
+
+
+def test_uses_cluster_ca_and_bounded_direct_requests(make_store, service_account_directory):
+ store, _, factory, create_context = make_store()
+
+ create_context.assert_called_once_with(cafile=str(service_account_directory / "ca.crt"))
+ factory.assert_called_once_with(
+ base_url=httpx2.URL("https://10.96.0.1"),
+ verify=create_context.return_value,
+ timeout=10,
+ follow_redirects=False,
+ trust_env=False,
+ )
+ with pytest.raises(RuntimeError, match="stop"), store:
+ raise RuntimeError("stop")
+ assert store._client.is_closed
+
+
+@pytest.mark.parametrize(
+ ("host", "https_port", "service_port", "endpoint"),
+ [
+ ("10.96.0.2", None, "443", "https://10.96.0.2"),
+ ("10.96.0.2", "6443", "443", "https://10.96.0.2:6443"),
+ ("api.internal", "8443", None, "https://api.internal:8443"),
+ ("api.internal", "", "8443", "https://api.internal:8443"),
+ ("fd00::1", None, "6443", "https://[fd00::1]:6443"),
+ ("fd00::1", "443", "6443", "https://[fd00::1]"),
+ ],
+)
+def test_uses_advertised_kubernetes_endpoint(
+ make_store, monkeypatch, host, https_port, service_port, endpoint
+):
+ for name, value in (
+ ("KUBERNETES_SERVICE_HOST", host),
+ ("KUBERNETES_SERVICE_PORT_HTTPS", https_port),
+ ("KUBERNETES_SERVICE_PORT", service_port),
+ ):
+ if value is None:
+ monkeypatch.delenv(name, raising=False)
+ else:
+ monkeypatch.setenv(name, value)
+ store, requests, _, _ = make_store(make_secret_response(), httpx2.Response(200))
+
+ store.read_token()
+ store.write_token("new-session")
+
+ assert [str(request.url) for request in requests] == [
+ f"{endpoint}/api/v1/namespaces/dag-processors/secrets/processor-session"
+ ] * 2
+
+
+@pytest.mark.parametrize(
+ ("host", "port"),
+ [(None, "443"), ("", "443"), ("10.96.0.1", None), ("10.96.0.1", "")],
+)
+def test_requires_in_cluster_endpoint_configuration(make_store, monkeypatch, host, port):
+ for name, value in (("KUBERNETES_SERVICE_HOST", host), ("KUBERNETES_SERVICE_PORT", port)):
+ if value is None:
+ monkeypatch.delenv(name, raising=False)
+ else:
+ monkeypatch.setenv(name, value)
+
+ with pytest.raises(ValueError, match="Kubernetes API access requires KUBERNETES_SERVICE_HOST"):
+ make_store()
+
+
+@pytest.mark.parametrize("port", ["not-a-port", "0", "-1", "65536"])
+def test_rejects_invalid_kubernetes_service_port(make_store, monkeypatch, port):
+ monkeypatch.setenv("KUBERNETES_SERVICE_PORT", port)
+
+ with pytest.raises(ValueError, match="Kubernetes service port must be an integer between 1 and 65535"):
+ make_store()
+
+
+@pytest.mark.parametrize("data", [None, {}, {"processor.jwt": ""}, {"processor.jwt": "IAo="}])
+def test_accepts_an_uninitialized_secret(make_store, data):
+ body = {"metadata": {"resourceVersion": "1"}}
+ if data is not None:
+ body["data"] = data
+ store, requests, _, _ = make_store(httpx2.Response(200, json=body), httpx2.Response(200))
+
+ assert store.read_token() is None
+ store.write_token("first-session")
+
+ assert json.loads(requests[1].content)["metadata"] == {"resourceVersion": "1"}
+
+
+def test_reloads_service_account_token_for_each_request(make_store, service_account_directory):
+ store, requests, _, _ = make_store(make_secret_response(), httpx2.Response(200))
+
+ store.read_token()
+ (service_account_directory / "token").write_text("service-account-2\n")
+ store.write_token("new-session")
+
+ assert [request.headers["authorization"] for request in requests] == [
+ "Bearer service-account-1",
+ "Bearer service-account-2",
+ ]
+
+
+@pytest.mark.parametrize("missing", [False, True])
+def test_rejects_empty_or_unreadable_service_account_token(make_store, service_account_directory, missing):
+ store, requests, _, _ = make_store()
+ token_file = service_account_directory / "token"
+ if missing:
+ token_file.unlink()
+ else:
+ token_file.write_text(" \n")
+
+ with pytest.raises(KubernetesSecretTokenError, match="service account token"):
+ store.read_token()
+
+ assert not requests
+
+
+@pytest.mark.parametrize("method", ["GET", "PATCH"])
+@pytest.mark.parametrize("status", [302, 403, 404, 409, 422, 503])
+def test_reports_http_failures_without_secret_contents(make_store, method, status):
+ failure = httpx2.Response(
+ status,
+ json={"message": "sensitive-token"},
+ headers={"Location": "https://another-service.invalid/"},
+ )
+ responses = [failure] if method == "GET" else [make_secret_response(), failure]
+ store, requests, _, _ = make_store(*responses)
+ if method == "PATCH":
+ store.read_token()
+ operation = store.read_token if method == "GET" else partial(store.write_token, "new-session")
+
+ with pytest.raises(KubernetesSecretTokenError, match=f"{method} returned HTTP {status}") as raised:
+ operation()
+
+ assert raised.value.status_code == status
+ assert "sensitive-token" not in str(raised.value)
+ assert len(requests) == len(responses)
+
+
+@pytest.mark.parametrize("method", ["GET", "PATCH"])
+@pytest.mark.parametrize("error_type", [httpx2.ConnectError, httpx2.ReadTimeout, httpx2.PoolTimeout])
+def test_reports_transport_failures_without_secret_contents(make_store, method, error_type):
+ failure = error_type("sensitive-token")
+ responses = [failure] if method == "GET" else [make_secret_response(), failure]
+ store, _, _, _ = make_store(*responses)
+ if method == "PATCH":
+ store.read_token()
+ operation = store.read_token if method == "GET" else partial(store.write_token, "new-session")
+
+ with pytest.raises(
+ KubernetesSecretTokenError, match=rf"{method} request failed \({error_type.__name__}\)"
+ ) as raised:
+ operation()
+
+ assert raised.value.status_code is None
+ assert "sensitive-token" not in "".join(traceback.format_exception(raised.value))
+
+
+@pytest.mark.parametrize(
+ "body",
+ [
+ None,
+ [],
+ {},
+ {"metadata": None},
+ {"metadata": {"resourceVersion": ""}},
+ {"metadata": {"resourceVersion": 10}},
+ {"metadata": {"resourceVersion": "10"}, "data": []},
+ {"metadata": {"resourceVersion": "10"}, "data": {"processor.jwt": None}},
+ {"metadata": {"resourceVersion": "10"}, "data": {"processor.jwt": "invalid!"}},
+ {"metadata": {"resourceVersion": "10"}, "data": {"processor.jwt": "/w=="}},
+ ],
+)
+def test_rejects_malformed_secret_without_allowing_a_write(make_store, body):
+ store, requests, _, _ = make_store(httpx2.Response(200, json=body))
+
+ with pytest.raises(KubernetesSecretTokenError, match="invalid token data or metadata"):
+ store.read_token()
+ with pytest.raises(KubernetesSecretTokenError, match="Read the Kubernetes Secret successfully"):
+ store.write_token("new-session")
+
+ assert len(requests) == 1
+
+
+def test_failed_read_invalidates_the_previous_version(make_store):
+ store, requests, _, _ = make_store(make_secret_response(), httpx2.Response(503))
+ store.read_token()
+
+ with pytest.raises(KubernetesSecretTokenError, match="HTTP 503"):
+ store.read_token()
+ with pytest.raises(KubernetesSecretTokenError, match="Read the Kubernetes Secret successfully"):
+ store.write_token("new-session")
+
+ assert [request.method for request in requests] == ["GET", "GET"]
+
+
+@pytest.mark.parametrize(
+ "write_result",
+ [
+ httpx2.Response(200),
+ httpx2.Response(409),
+ httpx2.ReadTimeout("The response was lost after persistence"),
+ ],
+)
+def test_each_write_requires_a_new_read_even_after_an_uncertain_response(make_store, write_result):
+ store, requests, _, _ = make_store(
+ make_secret_response(),
+ write_result,
+ make_secret_response("persisted-session", "11"),
+ httpx2.Response(200),
+ )
+ store.read_token()
+ if isinstance(write_result, Exception) or write_result.status_code == 409:
+ with pytest.raises(KubernetesSecretTokenError):
+ store.write_token("first-attempt")
+ else:
+ store.write_token("first-attempt")
+ with pytest.raises(KubernetesSecretTokenError, match="Read the Kubernetes Secret successfully"):
+ store.write_token("must-not-overwrite")
+
+ assert store.read_token() == "persisted-session"
+ store.write_token("second-attempt")
+
+ assert [request.method for request in requests] == ["GET", "PATCH", "GET", "PATCH"]
+ assert json.loads(requests[3].content)["metadata"] == {"resourceVersion": "11"}
+
+
+def test_rejects_empty_session_token(make_store):
+ store, requests, _, _ = make_store(make_secret_response())
+ store.read_token()
+
+ with pytest.raises(ValueError, match="must not be empty"):
+ store.write_token(" \n")
+
+ assert len(requests) == 1
+
+
+@pytest.mark.parametrize(
+ ("secret_name", "namespace"),
+ [
+ ("", "processors"),
+ ("../other-secret", "processors"),
+ ("secret/another", "processors"),
+ ("secret..name", "processors"),
+ ("Secret", "processors"),
+ ("s" * 254, "processors"),
+ ("secret", ""),
+ ("secret", "another/namespace"),
+ ("secret", "processor.example"),
+ ("secret", "n" * 64),
+ ],
+)
+def test_rejects_invalid_secret_or_namespace_names(secret_name, namespace):
+ with pytest.raises(ValueError, match="Kubernetes"):
+ KubernetesSecretTokenStore(secret_name, namespace)
diff --git a/chart/tests/helm_tests/airflow_core/test_dag_processor.py b/chart/tests/helm_tests/airflow_core/test_dag_processor.py
index 7fdff7fb35a..68fa6baeb98 100644
--- a/chart/tests/helm_tests/airflow_core/test_dag_processor.py
+++ b/chart/tests/helm_tests/airflow_core/test_dag_processor.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+import json
from configparser import ConfigParser
from pathlib import Path
from subprocess import CalledProcessError
@@ -541,6 +542,103 @@ class TestDagProcessor:
assert policy["egress"][1]["to"][0]["podSelector"]["matchLabels"] == {"k8s-app": "kube-dns"}
assert {port["port"] for port in policy["egress"][1]["ports"]} == {53}
+ def test_kubernetes_token_provisioner_example(self):
+ examples = Path(__file__).resolve().parents[4] / "airflow-core/docs/security/dag-processor-api"
+ docs = list(yaml.safe_load_all((examples / "kubernetes-token-provisioner.yaml").read_text()))
+ resources = {doc["kind"]: doc for doc in docs}
+ assert len(resources) == len(docs) == 6
+ secret = resources["Secret"]
+ target_namespace = secret["metadata"]["namespace"]
+ secret_name = secret["metadata"]["name"]
+ assert resources["Namespace"]["metadata"]["name"] == target_namespace
+ assert "data" not in secret
+ assert "stringData" not in secret
+ assert not secret.get("immutable")
+
+ role = resources["Role"]
+ assert role["metadata"]["namespace"] == target_namespace
+ assert role["rules"] == [
+ {
+ "apiGroups": [""],
+ "resources": ["secrets"],
+ "resourceNames": [secret_name],
+ "verbs": ["get", "patch"],
+ }
+ ]
+ account = resources["ServiceAccount"]
+ trusted_namespace = account["metadata"]["namespace"]
+ assert trusted_namespace != target_namespace
+ assert account["automountServiceAccountToken"] is True
+ binding = resources["RoleBinding"]
+ assert binding["metadata"]["namespace"] == target_namespace
+ assert binding["roleRef"] == {
+ "apiGroup": "rbac.authorization.k8s.io",
+ "kind": "Role",
+ "name": role["metadata"]["name"],
+ }
+ assert binding["subjects"] == [
+ {
+ "kind": "ServiceAccount",
+ "name": account["metadata"]["name"],
+ "namespace": trusted_namespace,
+ }
+ ]
+
+ deployment = resources["Deployment"]
+ assert deployment["metadata"]["namespace"] == trusted_namespace
+ assert deployment["spec"]["replicas"] == 1
+ assert deployment["spec"]["strategy"] == {"type": "Recreate"}
+ pod = deployment["spec"]["template"]["spec"]
+ assert pod["serviceAccountName"] == account["metadata"]["name"]
+ assert pod["automountServiceAccountToken"] is True
+ assert pod["securityContext"]["runAsNonRoot"] is True
+ assert pod["securityContext"]["runAsUser"] != 0
+ assert pod["volumes"] == [{"name": "tmp", "emptyDir": {}}]
+ assert not pod.get("initContainers")
+ assert len(pod["containers"]) == 1
+ provisioner = pod["containers"][0]
+ assert provisioner["args"] == [
+ "airflow",
+ "dag-processor-token",
+ "--kubernetes-secret",
+ secret_name,
+ "--kubernetes-namespace",
+ target_namespace,
+ "--bundle-name",
+ "dags-folder",
+ "--valid-for",
+ "600",
+ "--rotate",
+ ]
+ assert provisioner["securityContext"] == {
+ "allowPrivilegeEscalation": False,
+ "readOnlyRootFilesystem": True,
+ "capabilities": {"drop": ["ALL"]},
+ }
+ assert provisioner["resources"]["requests"]
+ assert provisioner["resources"]["limits"]
+ assert provisioner["volumeMounts"] == [{"name": "tmp", "mountPath": "/tmp"}]
+ assert not provisioner.get("envFrom")
+ env = {entry["name"]: entry for entry in provisioner["env"]}
+ signing_secret = env.pop("AIRFLOW__API_AUTH__JWT_SECRET")["valueFrom"]["secretKeyRef"]
+ assert signing_secret == {"name": "airflow-jwt-secret", "key": "jwt-secret"}
+ bundles = json.loads(env.pop("AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST")["value"])
+ assert {bundle["name"] for bundle in bundles} == {"dags-folder"}
+ assert {name: entry["value"] for name, entry in env.items()} == {
+ "AIRFLOW_HOME": "/tmp/airflow",
+ "AIRFLOW_CONFIG": "/dev/null",
+ "AIRFLOW__CORE__LOAD_EXAMPLES": "False",
+ "CONNECTION_CHECK_MAX_COUNT": "0",
+ }
+
+ values = yaml.safe_load((examples / "helm-values.yaml").read_text())
+ token_volume = next(
+ volume for volume in values["dagProcessor"]["extraVolumes"] if volume["name"] == "processor-token"
+ )
+ assert token_volume["secret"]["secretName"] == secret_name
+ assert token_volume["secret"]["secretName"] != signing_secret["name"]
+ assert provisioner["image"] == f"apache/airflow:{values['defaultAirflowTag']}"
+
@pytest.mark.parametrize(
("log_values", "expected_volume"),
[