Run Dag processors without metadata database access

airflow-core/docs/security/dag-processor-api/docker-compose.yaml

@@ -0,0 +1,53 @@
+# 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.
+---
+# Run independently of the quick-start Compose file so its credentials are not inherited.
+services:
+  dag-processor:
+    image: ${AIRFLOW_IMAGE_NAME:?Set an Airflow image with processor API support}
+    command: [dag-processor]
+    user: "${AIRFLOW_UID:-50000}:0"
+    environment:
+      USER: airflow
+      AIRFLOW_CONFIG: /tmp/airflow.cfg
+      AIRFLOW__CORE__LOAD_EXAMPLES: "False"
+      AIRFLOW__CORE__EXECUTION_API_SERVER_URL: http://airflow-apiserver:8080/execution/
+      AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE: /run/dag-processor/processor.jwt
+      AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_STORAGE_PATH: /tmp/dag-bundles
+      AIRFLOW__DAG_PROCESSOR__HEALTH_CHECK_FILE: /tmp/dag-processor-health.json
+      CONNECTION_CHECK_MAX_COUNT: "0"
+    volumes:
+      - ./tokens:/run/dag-processor:ro
+      - ./dags:/opt/airflow/dags:ro
+      - ./logs:/opt/airflow/logs
+    read_only: true
+    tmpfs:
+      - /tmp
+    cap_drop: [ALL]
+    security_opt: [no-new-privileges:true]
+    healthcheck:
+      test: [CMD, airflow, dag-processor, --check-health]
+      interval: 30s
+      timeout: 10s
+      start_period: 30s
+      retries: 5
+    networks: [processor-api]
+    restart: unless-stopped
+networks:
+  processor-api:
+    name: ${DAG_PROCESSOR_API_NETWORK:-dag-processor-api}
+    external: true

airflow-core/docs/security/security_model.rst

@@ -499,6 +499,14 @@ potentially still executes with direct database access in the Dag File Processor
    the database directly, including the JWT signing key configuration if it is available in the
    process environment. If a Dag author obtains the JWT signing key, they could forge arbitrary tokens.
 
+   Since Airflow 3.5 the Dag File Processor can instead run without metadata database access, by
+   setting ``[dag_processor] execution_api_token_file``. It then uses short-lived Execution API
+   credentials limited to its granted bundles, and each parsing attempt gets a bundle-scoped token
+   carrying its file and attempt identity. This mode removes the processor's need for database
+   credentials, the Fernet key and signing keys, but does not remove them: Deployment Managers must
+   take them out of the processor's environment and block its database network path. The Triggerer
+   still needs database access. See :doc:`/security/jwt_token_authentication`.
+
 **Dag File Processor and Triggerer are shared across teams**
    In the default deployment, a **single Dag File Processor instance** parses all Dag files and a
    **single Triggerer instance** handles all triggers — regardless of team assignment. There is no
@@ -775,6 +783,10 @@ model — Airflow does not enforce these natively.
    value itself. However, access to the file it points to should be restricted to the Scheduler
    (which generates tokens) and the API Server (which validates them).
 
+   A Dag File Processor running with ``[dag_processor] execution_api_token_file`` needs none of the
+   metadata database variables, the Fernet key or the JWT signing secret listed for it above; it
+   authenticates with the token file instead.
+
    .. START AUTOGENERATED PROVIDER SENSITIVE VARS
 
    **Provider-specific sensitive configuration variables:**
@@ -848,7 +860,8 @@ model — Airflow does not enforce these natively.
      the metadata database exclusively through the API server (similar to how workers use the
      Execution API today). This would eliminate the need for these components to have database
      credentials at all, providing security by design rather than relying on deployment-level
-     measures.
+     measures. Airflow 3.5 offers this for the Dag File Processor as an opt-in mode
+     (``[dag_processor] execution_api_token_file``); the Triggerer still uses the database.
    * **Tactical (shorter-term)**: Native support for Unix user impersonation in the Dag File
      Processor and Triggerer child processes, so that Dag author code runs as a different, low-
      privilege user that cannot access the parent's credentials or the database.

airflow-core/src/airflow/cli/cli_config.py

@@ -2304,6 +2304,21 @@ class GroupCommand(NamedTuple):
         help="Start a dag processor instance",
         func=lazy_load_command("airflow.cli.commands.dag_processor_command.dag_processor"),
         args=(
+            Arg(
+                ("--check-health",),
+                action="store_true",
+                help="Check the API-mode processor's local loop heartbeat and exit",
+            ),
+            Arg(
+                ("--check-ready",),
+                action="store_true",
+                help="Check the API-mode processor's loop and API heartbeat and exit",
+            ),
+            Arg(
+                ("--sync-bundles-only",),
+                action="store_true",
+                help="Provision the configured bundle catalog using trusted database and signing credentials, then exit",
+            ),
             ARG_PID,
             ARG_DAEMON,
             ARG_BUNDLE_NAME,

airflow-core/src/airflow/cli/commands/dag_processor_command.py

@@ -19,19 +19,28 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from airflow.cli.commands.daemon_utils import run_command_with_daemon_option
 from airflow.configuration import conf
-from airflow.dag_processing.bundles.manager import get_configured_bundle_team_names
+from airflow.dag_processing.bundles.manager import (
+    _load_bundle_config_snapshot,
+    get_configured_bundle_team_names,
+)
+from airflow.dag_processing.health import check_processor_health
 from airflow.dag_processing.manager import DagFileProcessorManager
+from airflow.executors.base_executor import get_execution_api_server_url
 from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
 from airflow.jobs.job import Job, run_job
 from airflow.utils import cli as cli_utils
 from airflow.utils.memray_utils import MemrayTraceComponents, enable_memray_trace
+from airflow.utils.net import get_hostname
 from airflow.utils.process_utils import set_component_mp_start_method
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
 log = logging.getLogger(__name__)
 
 
@@ -60,7 +69,12 @@ def _get_team_names(bundle_names: list[str] | None) -> list[str]:
 def _create_dag_processor_job_runner(args: Any) -> DagProcessorJobRunner:
     """Create DagFileProcessorProcess instance."""
     if args.bundle_name:
-        cli_utils.validate_dag_bundle_arg(args.bundle_name)
+        if conf.get("dag_processor", "execution_api_token_file", fallback=None):
+            # Bundle constructors can read secrets; defer them until the API client is registered.
+            if unknown := set(args.bundle_name) - _load_bundle_config_snapshot().names:
+                raise SystemExit(f"Bundles not found: {', '.join(sorted(unknown))}")
+        else:
+            cli_utils.validate_dag_bundle_arg(args.bundle_name)
     return DagProcessorJobRunner(
         job=Job(bundle_names=args.bundle_name, team_names=_get_team_names(args.bundle_name)),
         processor=DagFileProcessorManager(
@@ -72,13 +86,55 @@ def _create_dag_processor_job_runner(args: Any) -> DagProcessorJobRunner:
 
 @enable_memray_trace(component=MemrayTraceComponents.dag_processor)
 def _run_dag_processor_job(job_runner: DagProcessorJobRunner) -> None:
-    run_job(job=job_runner.job, execute_callable=job_runner._execute)
+    if not (token_file := conf.get("dag_processor", "execution_api_token_file", fallback=None)):
+        run_job(job=job_runner.job, execute_callable=job_runner._execute)
+        return
+
+    # Create the connection pool after the daemon or hot-reload fork.
+    from airflow.dag_processing.api_client import DagProcessorAPIClient
+
+    with DagProcessorAPIClient(
+        base_url=get_execution_api_server_url(),
+        token_file=token_file,
+        hostname=get_hostname(),
+        unixname=job_runner.job.unixname,
+        bundle_names=job_runner.processor.bundle_names_to_parse
+        or sorted(_load_bundle_config_snapshot().names),
+    ) as client:
+        job_runner.run_with_api(client)
 
 
-@cli_utils.action_cli
 @providers_configuration_loaded
 def dag_processor(args):
     """Start Airflow Dag Processor Job."""
+    if getattr(args, "check_health", False) or getattr(args, "check_ready", False):
+        if not conf.get("dag_processor", "execution_api_token_file", fallback=None):
+            raise SystemExit("Local health probes require Dag processor API mode")
+        return check_processor_health(readiness=getattr(args, "check_ready", False))
+    if getattr(args, "sync_bundles_only", False):
+        return _run_with_action_log(_sync_bundle_catalog, args)
+    if not conf.get("dag_processor", "execution_api_token_file", fallback=None):
+        return _run_with_action_log(_start_dag_processor, args)
+    if getattr(args, "verbose", False):
+        cli_utils.set_verbose_logging()
+    return _start_dag_processor(args)
+
+
+def _run_with_action_log(func: Callable[[Any], Any], args: Any) -> Any:
+    # The action log names its event after the decorated function; keep the command's own name.
+    def dag_processor(args):
+        return func(args)
+
+    return cli_utils.action_cli(dag_processor)(args)
+
+
+def _sync_bundle_catalog(args):
+    if args.bundle_name:
+        cli_utils.validate_dag_bundle_arg(args.bundle_name)
+    DagFileProcessorManager(max_runs=0, bundle_names_to_parse=args.bundle_name).sync_bundles()
+
+
+def _start_dag_processor(args):
     set_component_mp_start_method("dag_processor")
     job_runner = _create_dag_processor_job_runner(args)
 

airflow-core/src/airflow/config_templates/config.yml

@@ -3313,6 +3313,32 @@ dag_processor:
          - how often to refresh DAGs from those sources
          - how many files to parse concurrently
   options:
+    execution_api_token_file:
+      description: |
+        Path to a session token written by ``airflow dag-processor-token``. When set,
+        ``airflow dag-processor`` uses the authenticated Execution API for its Job lifecycle,
+        bundle metadata and secrets, parse-time requests, result publication, and callback
+        and priority-request delivery. The API URL comes from
+        ``[core] execution_api_server_url``. Each processor needs its own session token.
+
+        Provision the bundle catalog on a trusted host before starting the processor.
+        The processor does not need metadata database credentials, the Fernet key, or
+        API signing keys. Remove those credentials from its deployment and block database
+        network access to enforce isolation; setting this option does not remove them.
+        Leave unset to use the in-process API and database-backed Job lifecycle.
+      version_added: 3.5.0
+      type: string
+      default: ~
+      example: "/run/airflow/dag-processor.jwt"
+    health_check_file:
+      description: |
+        Local heartbeat file written by a Dag processor using the Execution API.
+        Use a separate writable file for each processor. ``airflow dag-processor --check-health``
+        checks its loop heartbeat; ``--check-ready`` also checks the last successful API heartbeat.
+      version_added: 3.5.0
+      type: string
+      default: "{AIRFLOW_HOME}/dag_processor_health.json"
+      example: ~
     dag_bundle_storage_path:
       description: |
         String path to folder where Airflow bundles can store files locally. Not templated.
@@ -3476,6 +3502,10 @@ dag_processor:
         dag processor is considered unhealthy.
         This is used by the health check in the **/health** endpoint and in ``airflow jobs check``
         CLI for DagProcessorJob.
+        For a Dag processor using the Execution API, the API server also reads this value to decide
+        whether a session's previous Job is still alive, and the processor reads it for its local
+        ``--check-health`` and ``--check-ready`` probes and to bound how long registration waits for
+        a previous Job to expire. Set the same value for the API server and the processor.
       version_added: 3.2.0
       type: integer
       example: ~
@@ -3484,9 +3514,17 @@ dag_processor:
       description: |
         Seconds a Dag processor using the Execution API may go without a recorded Job heartbeat.
         After this, the scheduler's next cleanup (every ``[scheduler] parsing_cleanup_interval``)
-        ends the Job and releases the callbacks and parse requests it claimed.
+        ends the Job and releases the callbacks and parse requests it claimed, and a processor that
+        could not reach the API stops. Processors do not heartbeat while initializing, refreshing,
+        or listing a single bundle, so set this above the longest of those operations; the local
+        health check must also tolerate them, through ``health_check_threshold`` and the probe's
+        failure budget. It also bounds how long a processor retries preparing the bundle
+        version of a claimed callback before acknowledging it without running it.
         A callback that ran before its acknowledgment was recorded can run again after the Job
         times out, repeating its side effects. Callbacks should tolerate repeated execution.
+        The scheduler reads this value to end Jobs and the processor reads it to stop, so set the
+        same value for both; changing it only in the processor's configuration has no effect on
+        when the scheduler ends the Job.
       version_added: 3.5.0
       type: integer
       example: ~

airflow-core/src/airflow/dag_processing/api_client.py

@@ -25,7 +25,7 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 from time import monotonic
-from typing import Any
+from typing import Any, Literal
 from urllib.parse import quote
 from uuid import UUID
 
@@ -39,6 +39,9 @@
     DagBundleStateResponse,
     DagParseResultBody,
     DagParseResultResponse,
+    ProcessorWorkAckBody,
+    ProcessorWorkClaimBody,
+    ProcessorWorkItem,
 )
 from airflow.api_fastapi.execution_api.datamodels.job import (
     DagParseTokenBody,
@@ -142,6 +145,8 @@ def __init__(
         self._job_id: int | None = None
         self._completion_state: TerminalJobState | None = None
         self._completed = False
+        self._pending_claims: dict[str, ProcessorWorkClaimBody] = {}
+        self._claim_sequences: dict[str, int] = {}
         super().__init__(base_url=base_url, token="", **kwargs)
 
     @property
@@ -398,6 +403,44 @@ def get_bundles(self) -> list[DagBundleStateResponse]:
         )
         return [DagBundleStateResponse.model_validate(item) for item in response.json()]
 
+    def claim_work(
+        self, kind: Literal["callbacks", "priority"], bundle_names: list[str], limit: int
+    ) -> list[ProcessorWorkItem]:
+        if not bundle_names:
+            return []
+        body = self._pending_claims.setdefault(
+            kind,
+            ProcessorWorkClaimBody(
+                claim_id=uuid7(),
+                claim_sequence=self._claim_sequences.get(kind, 0) + 1,
+                bundle_names=bundle_names,
+                limit=min(limit, 100),
+            ),
+        )
+        response = self.request(
+            "POST",
+            f"jobs/{self._require_job_id()}/requested-work/{kind}/claim",
+            content=body.model_dump_json(),
+            headers=_JOB_API_HEADERS,
+            retry=False,
+            timeout=self._get_bounded_timeout(),
+        )
+        work = [ProcessorWorkItem.model_validate(item) for item in response.json()]
+        self._claim_sequences[kind] = body.claim_sequence
+        del self._pending_claims[kind]
+        return work
+
+    def acknowledge_work(self, kind: Literal["callbacks", "priority"], work: ProcessorWorkItem) -> None:
+        body = ProcessorWorkAckBody(claim_id=work.claim_id)
+        self.request(
+            "POST",
+            f"jobs/{self._require_job_id()}/requested-work/{kind}/{work.id}/ack",
+            content=body.model_dump_json(),
+            headers=_JOB_API_HEADERS,
+            retry=False,
+            timeout=self._get_bounded_timeout(),
+        )
+
     def publish_inventory(self, bundle_name: str, body: DagBundleInventoryBody) -> DagBundleInventoryResponse:
         response = self.request(
             "POST",