timkpaine opened a new pull request, #68874: URL: https://github.com/apache/airflow/pull/68874
See discussion: https://github.com/apache/airflow/discussions/61800 This PR adds a native local REST client for "secure" contexts, e.g. plugins and inside DAG processing, to be able to access metadata that was previously available via direct DB access in Airflow 2. An example of this is e.g. a DAG that ensures a Pool exists before using it. This allows one to implement things dropped for Airflow 3 like e.g. the [Astronmer DAG cleanup workflow](https://www.astronomer.io/docs/learn/2.x/cleanup-dag-tutorial/). For example here is my variant now for Airflow 3 using the rest client: ```python def clean_dag_runs(delete_successful, delete_failed, mark_failed_as_successful, max_dagruns, days_to_keep, **kwargs): from airflow.api.client import get_local_rest_client client = get_local_rest_client(process_type="cleanup") utc_now = datetime.now(tz=UTC) cutoff_date = utc_now - timedelta(days=days_to_keep) _log.info(f"Cutoff date for clean: {cutoff_date}") all_dags = _list_all_dags(client) dag_ids = [d.dag_id for d in all_dags] _log.info(f"Found DAGs to clean up: {dag_ids}") deleted = 0 for dag_id in dag_ids: _log.info(f"Cleaning up DAG: {dag_id}") runs = _list_all_dag_runs(client, dag_id) # Filter by state if delete_successful is False: _log.info(f"Not deleting successful DAG runs for DAG: {dag_id}") runs = [r for r in runs if str(r.state) != "success"] if delete_failed is False: _log.info(f"Not deleting failed DAG runs for DAG: {dag_id}") runs = [r for r in runs if str(r.state) != "failed"] # Sort by logical_date ascending (use datetime.min as fallback for None) _min_dt = datetime.min.replace(tzinfo=UTC) runs.sort(key=lambda r: r.logical_date or _min_dt) total_runs = len(runs) _log.info(f"Found {total_runs} DAG runs to clean up for DAG: {dag_id}") for dr in runs: logical_date = dr.logical_date if (logical_date and logical_date < cutoff_date) or total_runs > max_dagruns: _log.info(f"Deleting DAG run: {dr.dag_run_id}") client.dag_runs.delete(dag_id, dr.dag_run_id) deleted += 1 total_runs -= 1 elif mark_failed_as_successful: if str(dr.state) == "failed": _log.info(f"Marking failed DAG run as successful: {dr.dag_run_id}") client.dag_runs.update(dag_id, dr.dag_run_id, state="success") elif not mark_failed_as_successful: break _log.info(f"Total DAG runs deleted: {deleted}") ``` There are a bunch of followup questions, principal amongst them is the questions of the API client itself and its distinction (?) from the public airflow rest client. ##### Was generative AI tooling used to co-author this PR? - [x] Yes (please specify the tool below) Copilot in VS Code, completions via GPT 4.1, my own background information, knowledge discovery, and planning via Claude Sonnet 4.6, rest client generated via Claude Opus 4.6. See below for roadmap, with HITL validation after each phase and a local all-in-one-dev docker instance against all code changes, also validating against my [airflow-laminar](https://github.com/airflow-laminar) project and validation DAGs. <details> <summary>Roadmap planning</summary> # Local REST Client & Built-in Process Authentication — Roadmap **Context:** [Discussion #61800](https://github.com/apache/airflow/discussions/61800) ## Problem Statement In Airflow 3, tasks and plugins can no longer access the metadata database directly. All mutations (creating pools, triggering DAGs, managing connections, etc.) must go through the REST API. However, code running inside trusted Airflow processes — the DAG processor, scheduler, workers via plugins, and user-authored DAG code — has **no built-in way to obtain an authenticated API client**. Today, a plugin author or DAG author who wants to call `POST /api/v2/pools` must: 1. Manually provision a username/password or API token. 2. Store those credentials in an Airflow Connection or environment variable. 3. Configure each plugin/DAG to read those credentials. 4. Repeat for every plugin that needs API access. This is a significant usability regression from Airflow 2, where code in the same process could call helpers like `Pool.create_or_update_pool()` directly. ### Existing Patterns | Component | How it talks to the API today | |---|---| | **Workers / Supervisors** | Receive a short-lived JWT *workload token* generated by the scheduler's `JWTGenerator`. The supervisor uses `httpx.Client` with this token against the Execution API (`/execution/`). | | **DAG Processor / Triggerer** | Use `InProcessExecutionAPI`, which creates an in-process ASGI transport with **all auth checks bypassed** via FastAPI dependency overrides. Only accesses `/execution/` routes. | | **airflow-ctl** | User logs in with username/password → receives a JWT → stored in system keyring. Uses `httpx.Client` against `/api/v2/`. | | **Python client SDK** | OpenAPI-generated client. User provides `access_token` in `Configuration`. Uses `/api/v2/`. | | **`local_client.py`** | Calls `api/common/` helpers that use `@provide_session` — **direct DB access**, not REST. | **Key observation:** The Execution API already has a JWT-based auth model for machine-to-machine communication, and `InProcessExecutionAPI` already demonstrates the pattern of running the API in-process with auth bypassed. But neither mechanism is available for the **Core API** (`/api/v2/`) which contains the full CRUD surface that plugin and DAG authors need. --- ## Goals 1. **Provide a built-in, zero-configuration way** for code running inside trusted Airflow processes (DAG processor, scheduler, workers, triggerer) to call any Core API endpoint without needing user credentials. 2. **Preserve the architecture boundary** — no direct DB access from tasks. All mutations must go through the API server's validation, authorization, and audit-logging layers. 3. **Support both in-process and remote API server topologies** — the client must work whether the API server is co-located (same process / same host) or remote (separate pod in Kubernetes). 4. **Re-use existing infrastructure** — JWT generation, `httpx` transport, `InProcessExecutionAPI` patterns — rather than inventing new auth mechanisms. 5. **Provide an ergonomic Python API** for DAG/plugin authors that feels as simple as the old `Pool.create_or_update_pool()` call. --- ## Non-Goals - Replacing the Execution API's workload-token system for task ↔ supervisor communication. - Implementing full RBAC scoping for process-level tokens (a process token gets system-level access; fine-grained user-level authz is a follow-up). - Supporting external (non-Airflow-process) callers — they should continue using `airflow-ctl` or the Python client with user credentials. --- ## Architecture Overview ``` ┌─────────────────────────────────────────────────────────┐ │ Trusted Airflow Process │ │ (DAG Processor / Scheduler / Worker / Triggerer) │ │ │ │ ┌──────────────────────┐ │ │ │ User DAG / Plugin │ │ │ │ │ │ │ │ client = get_local_ │ │ │ │ rest_client() │ │ │ │ client.pools.create │ │ │ │ _pool(...) │ │ │ └──────────┬───────────┘ │ │ │ │ │ ┌──────────▼───────────┐ │ │ │ LocalRESTClient │ │ │ │ │ │ │ │ - In-process mode: │ ┌────────────────────────┐ │ │ │ ASGI transport ───┼──►│ Core API App (in-proc) │ │ │ │ │ │ Auth: SystemUser bypass │ │ │ │ - Remote mode: │ └────────────────────────┘ │ │ │ httpx + JWT ──────┼──────────────────────────────► │ │ └──────────────────────┘ │ │ │ │ │ └────────────────────────────────────────┼─────────────────┘ │ ┌──────────▼──────────┐ │ API Server (remote) │ │ /api/v2/ │ │ Auth: JWT with │ │ "process" scope │ └─────────────────────┘ ``` --- ## Detailed Design ### Phase 1: Process-Scoped JWT Token ("process" scope) #### 1.1 New JWT Scope: `token:process` Extend the existing JWT token type system with a new scope: | Scope | Existing? | Used by | Purpose | |---|---|---|---| | `execution` | Yes | Execution API general calls | Standard task-lifecycle operations | | `workload` | Yes | Workers for task-specific operations | Tied to a specific task instance | | **`process`** | **New** | Trusted Airflow processes | System-level access to Core API | The `process` scope JWT carries claims identifying the process type: ```json { "sub": "airflow:process:dag_processor", "scope": "process", "aud": "urn:airflow.apache.org:core-api", "process_type": "dag_processor", "iat": 1700000000, "exp": 1700000600 } ``` **Files to modify:** - `airflow-core/src/airflow/api_fastapi/auth/tokens.py` — add `process` scope constant - `airflow-core/src/airflow/config_templates/config.yml` — add `[api] process_jwt_expiration_time` setting #### 1.2 SystemUser Model Create a `SystemUser` that implements `BaseUser` for process-level authentication: ```python # airflow-core/src/airflow/api_fastapi/auth/managers/models/system_user.py @attrs.define class SystemUser: """Represents a trusted Airflow system process (not a human user).""" process_type: str # "dag_processor", "scheduler", "worker", "triggerer" @property def id(self) -> str: return f"airflow:process:{self.process_type}" ``` **Files to create:** - `airflow-core/src/airflow/api_fastapi/auth/managers/models/system_user.py` **Files to modify:** - `airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py` — `deserialize_user()` must recognize `process` scope and return `SystemUser` — All `is_authorized_*()` methods must grant full access to `SystemUser` #### 1.3 Process Token Generation Each Airflow component already knows its own identity. Add a utility to generate a process-scoped JWT: ```python # airflow-core/src/airflow/api_fastapi/auth/process_token.py def generate_process_token(process_type: str) -> str: """Generate a short-lived JWT for a trusted Airflow process.""" generator = JWTGenerator( valid_for=conf.getint("api", "process_jwt_expiration_time", fallback=600), audience=conf.get("api", "process_jwt_audience", fallback="urn:airflow.apache.org:core-api"), issuer=conf.get("api_auth", "jwt_issuer", fallback=None), ) return generator.generate( { "sub": f"airflow:process:{process_type}", "scope": "process", "process_type": process_type, } ) ``` **Files to create:** - `airflow-core/src/airflow/api_fastapi/auth/process_token.py` --- ### Phase 2: LocalRESTClient #### 2.1 Core Client Class A thin wrapper around `httpx.Client` that targets the Core API (`/api/v2/`): ```python # airflow-core/src/airflow/api/client/local_rest_client.py class LocalRESTClient: """ REST API client for use inside trusted Airflow processes. Supports two transport modes: - In-process: Uses ASGI transport (no network, auth bypassed) - Remote: Uses HTTP transport with auto-refreshing process JWT """ def __init__(self, *, mode: Literal["in_process", "remote"] | None = None): ... # Resource-oriented sub-clients (matching Core API structure) @cached_property def pools(self) -> PoolsClient: ... @cached_property def dags(self) -> DagsClient: ... @cached_property def connections(self) -> ConnectionsClient: ... @cached_property def variables(self) -> VariablesClient: ... @cached_property def dag_runs(self) -> DagRunsClient: ... @cached_property def config(self) -> ConfigClient: ... ``` #### 2.2 In-Process Transport (Same Pattern as InProcessExecutionAPI) For co-located deployments (scheduler + API server in same process, or the standalone mode), run the Core API app inside the process with auth bypassed: ```python class InProcessCoreAPI: """Runs the Core API in-process with auth bypassed for system calls.""" @cached_property def app(self) -> FastAPI: from airflow.api_fastapi.core_api.app import create_core_api_app app = create_core_api_app() # Override auth dependency to return SystemUser async def system_user_override(): return SystemUser(process_type=_get_current_process_type()) app.dependency_overrides[get_user] = system_user_override return app @cached_property def transport(self) -> httpx.WSGITransport: from a2wsgi import ASGIMiddleware return httpx.WSGITransport(app=ASGIMiddleware(self.app)) ``` #### 2.3 Remote Transport (JWT-Authenticated httpx) For distributed deployments where the API server runs in a separate process/pod: ```python class ProcessAuthTransport(httpx.BaseTransport): """httpx transport that auto-manages process JWT tokens.""" def __init__(self, base_transport: httpx.HTTPTransport, process_type: str): self._transport = base_transport self._process_type = process_type self._token: str | None = None self._token_expiry: float = 0 def handle_request(self, request: httpx.Request) -> httpx.Response: if time.monotonic() >= self._token_expiry: self._refresh_token() request.headers["Authorization"] = f"Bearer {self._token}" return self._transport.handle_request(request) def _refresh_token(self) -> None: self._token = generate_process_token(self._process_type) # Refresh at 80% of validity (same pattern as JWTReissueMiddleware) self._token_expiry = time.monotonic() + (expiration_time * 0.8) ``` #### 2.4 Mode Auto-Detection The client should automatically pick the right transport: ```python def _detect_mode() -> Literal["in_process", "remote"]: """Detect whether we can use in-process transport.""" execution_api_url = conf.get("core", "execution_api_server_url", fallback="") # If execution_api_server_url is empty or points to localhost with same process, # and we're running in a process that has DB access (scheduler, dag_processor), # use in-process mode. if not execution_api_url or _is_api_server_colocated(): return "in_process" return "remote" ``` **Files to create:** - `airflow-core/src/airflow/api/client/local_rest_client.py` --- ### Phase 3: Resource Sub-Clients Typed, ergonomic wrappers around each API resource. These mirror the Core API routes and use Pydantic models for request/response serialization. #### 3.1 Pools Sub-Client (First Implementation) ```python # airflow-core/src/airflow/api/client/resources/pools.py class PoolsClient: """Client for /api/v2/pools endpoints.""" def __init__(self, http: httpx.Client): self._http = http def get(self, name: str) -> PoolResponse: resp = self._http.get(f"/api/v2/pools/{name}") resp.raise_for_status() return PoolResponse.model_validate(resp.json()) def list(self, *, limit: int = 100, offset: int = 0) -> PoolCollectionResponse: resp = self._http.get("/api/v2/pools", params={"limit": limit, "offset": offset}) resp.raise_for_status() return PoolCollectionResponse.model_validate(resp.json()) def create(self, *, name: str, slots: int, description: str = "", include_deferred: bool = False) -> PoolResponse: resp = self._http.post("/api/v2/pools", json={ "name": name, "slots": slots, "description": description, "include_deferred": include_deferred, }) resp.raise_for_status() return PoolResponse.model_validate(resp.json()) def update(self, name: str, *, slots: int | None = None, description: str | None = None, include_deferred: bool | None = None) -> PoolResponse: body = {k: v for k, v in { "slots": slots, "description": description, "include_deferred": include_deferred, }.items() if v is not None} resp = self._http.patch(f"/api/v2/pools/{name}", json=body) resp.raise_for_status() return PoolResponse.model_validate(resp.json()) def delete(self, name: str) -> None: resp = self._http.delete(f"/api/v2/pools/{name}") resp.raise_for_status() ``` #### 3.2 Additional Resource Clients (Incremental) Each follows the same pattern — thin wrapper around the corresponding REST endpoints: | Sub-client | Core API routes | Priority | |---|---|---| | `PoolsClient` | `POST/GET/PATCH/DELETE /api/v2/pools` | P0 — solves the original request | | `DagsClient` | `GET/PATCH/DELETE /api/v2/dags` | P1 | | `DagRunsClient` | `POST/GET/PATCH/DELETE /api/v2/dagRuns` | P1 | | `ConnectionsClient` | `POST/GET/PATCH/DELETE /api/v2/connections` | P1 | | `VariablesClient` | `POST/GET/PATCH/DELETE /api/v2/variables` | P1 | | `TaskInstancesClient` | `GET/PATCH /api/v2/taskInstances` | P2 | | `ConfigClient` | `GET /api/v2/config` | P2 | | `AssetsClient` | `GET/POST /api/v2/assets` | P2 | **Files to create:** - `airflow-core/src/airflow/api/client/resources/__init__.py` - `airflow-core/src/airflow/api/client/resources/pools.py` - `airflow-core/src/airflow/api/client/resources/dags.py` - ... (one per resource, added incrementally) --- ### Phase 4: Convenience Entry Point #### 4.1 Module-Level Factory ```python # airflow-core/src/airflow/api/client/__init__.py def get_local_rest_client() -> LocalRESTClient: """ Get a REST API client for use inside trusted Airflow processes. This client automatically authenticates using process-level JWT tokens and requires no user credentials. It is intended for DAG code, plugins, and Airflow component code that needs to interact with the Core API. Example:: from airflow.api.client import get_local_rest_client client = get_local_rest_client() client.pools.create(name="my_pool", slots=5) pools = client.pools.list() """ return LocalRESTClient() ``` #### 4.2 Backward Compatibility The existing `local_client.py` and `get_current_api_client()` should be deprecated with a warning pointing users to `get_local_rest_client()`. --- ### Phase 5: Core API Auth Integration #### 5.1 Recognize Process Tokens in Core API Security The Core API's `get_user()` dependency must be extended to recognize process-scope JWTs and return a `SystemUser`: ```python # Modification to airflow-core/src/airflow/api_fastapi/core_api/security.py async def get_user(request, oauth_token, bearer_credentials) -> BaseUser: # ... existing token resolution ... token_str = ... # Try process token first (JWT with scope=process) user = _try_resolve_process_token(token_str) if user: return user # Fall back to normal user resolution return await resolve_user_from_token(token_str) ``` #### 5.2 Auth Manager Authorization for SystemUser Each auth manager implementation must handle `SystemUser`: ```python # In base_auth_manager.py or as a mixin def is_authorized_pool(self, *, method, user, details=None) -> bool: if isinstance(user, SystemUser): return True # System processes have full access return self._is_authorized_pool_for_user(method=method, user=user, details=details) ``` This keeps the authorization check in the standard code path (not bypassed), ensuring audit logging and validation still apply — only the authorization *decision* is different for system users. #### 5.3 Audit Trail System user actions should be logged with clear attribution: ``` action=create_pool, user=airflow:process:dag_processor, pool=my_pool ``` **Files to modify:** - `airflow-core/src/airflow/api_fastapi/core_api/security.py` - `airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py` - `airflow-core/src/airflow/api_fastapi/auth/managers/simple/simple_auth_manager.py` --- ## Implementation Plan ### Milestone 1: Foundation (Minimal Viable Feature) — ✅ DONE **Goal:** A working `LocalRESTClient` with in-process transport and pools support. **Status:** Complete. All 23 tests pass. | Step | Description | Files | Status | |---|---|---|---| | 1a | Create `SystemUser` model | `auth/managers/models/system_user.py` | ✅ | | 1b | Create `InProcessCoreAPI` helper (following `InProcessExecutionAPI` pattern) | `api/client/in_process_core_api.py` | ✅ | | 1c | Create `LocalRESTClient` with in-process transport only | `api/client/local_rest_client.py` | ✅ | | 1d | Create `PoolsClient` resource sub-client | `api/client/resources/pools.py` | ✅ | | 1e | Update `__init__.py` with `get_local_rest_client()` | `api/client/__init__.py` | ✅ | | 1f | Add unit tests | `tests/unit/api/client/test_system_user.py`, `tests/unit/api/client/test_local_rest_client.py` | ✅ | | 1g | Format and lint | — | ✅ | **Implementation notes:** - Auth bypass required a `_SystemAccessAuthManagerProxy` that wraps the real auth manager and returns `True` for all `is_authorized_*` checks. Simply overriding `get_user` was insufficient because `requires_access_*` dependencies call `get_auth_manager()` (the global singleton) which then delegates to the concrete auth manager (e.g., SimpleAuthManager) which expects user-specific attributes like `role`. - Pool update uses a fetch-modify-update pattern because the Core API's PATCH endpoint validates the body against `BasePool` which has required fields (`pool`, `include_deferred`). **Deliverable:** DAG authors can write: ```python from airflow.api.client import get_local_rest_client client = get_local_rest_client() client.pools.create(name="my_pool", slots=5) ``` ### Milestone 2: Remote Transport **Goal:** Support distributed deployments where API server is in a separate process. | Step | Description | Files | Est. Size | |---|---|---|---| | 2a | Add `process` scope to JWT token system | `auth/tokens.py` | S | | 2b | Create `generate_process_token()` | `auth/process_token.py` | S | | 2c | Create `ProcessAuthTransport` | `api/client/local_rest_client.py` | M | | 2d | Add process token recognition to Core API `get_user()` | `core_api/security.py` | M | | 2e | Handle `SystemUser` in auth managers | `base_auth_manager.py`, `simple_auth_manager.py` | M | | 2f | Add auto-detection of in-process vs remote mode | `api/client/local_rest_client.py` | S | | 2g | Add configuration options | `config.yml` | S | | 2h | Add integration tests | `tests/api/client/test_local_rest_client_remote.py` | L | **Deliverable:** The same client code works in Kubernetes deployments where the API server is a separate pod, transparently authenticating with process JWTs. ### Milestone 3: Full Resource Coverage — ✅ DONE **Goal:** Sub-clients for all major Core API resources. **Status:** Complete. All 68 tests pass (23 from M1 + 32 P1 + 13 P2). | Step | Description | Priority | Status | |---|---|---|---| | 3a | `DagsClient` — GET, GET details, list, pause/unpause, DELETE | P1 | ✅ | | 3b | `DagRunsClient` — trigger, GET, list, update (note), DELETE, clear | P1 | ✅ | | 3c | `ConnectionsClient` — CRUD with `update_mask` for partial updates | P1 | ✅ | | 3d | `VariablesClient` — CRUD with `update_mask` for partial updates | P1 | ✅ | | 3e | `TaskInstancesClient` — GET, list, update state, clear | P2 | ✅ | | 3f | `ConfigClient` — GET full config, section, individual value | P2 | ✅ | | 3g | `AssetsClient` — GET, list, create event, list events, queued events | P2 | ✅ | **Implementation notes:** - Connections and Variables PATCH endpoints use `update_mask` query parameter for partial updates, avoiding the need for all required fields in the request body. - DAG Run trigger requires `logical_date` in the body (it's a required field with no default in `TriggerDAGRunPostBody`), so the client always includes it (defaults to `None`). - Auth manager proxy needed `get_authorized_connections()` and `get_authorized_variables()` methods (returning all IDs from DB) because the `ReadableConnectionsFilterDep` and `ReadableVariablesFilterDep` dependencies call these methods for list/filter operations. - Also fixed `filter_authorized_connections` parameter name from `connection_ids` to `conn_ids` to match `base_auth_manager.py::get_authorized_connections()`'s internal call signature. **P2 implementation notes:** - `TaskInstancesClient` supports wildcard `~` for `dag_id` and `dag_run_id` in list operations to query across all DAGs/runs. Update uses `PatchTaskInstanceBody` with `new_state` and `note`. Clear takes a `ClearTaskInstancesBody` to bulk-clear task instances by task IDs. - `ConfigClient` is read-only. The Core API's `/api/v2/config` endpoint requires `("api", "expose_config"): "True"` or it returns 403. Tests use `conf_vars` context manager. - `AssetsClient` covers get, list, create_event, list_events, get/delete queued events, and materialize. Test setup requires both `AssetModel` and `AssetActive` records for assets to appear in active queries. ### Milestone 4: Polish & Migration — ✅ DONE | Step | Description | Status | |---|---|---| | 4a | Deprecate `local_client.py` and `get_current_api_client()` | ✅ | | 4b | Add migration guide in documentation | ✅ | | 4c | Update plugin authoring docs to show new pattern | ✅ | | 4d | Add newsfragment | ✅ | **Implementation notes:** - Added "Recommended Approach: Use Built-in Local REST Client (In-Process)" section to `airflow-core/docs/installation/upgrading_to_airflow3.rst` between the Python Client approach and the DbApiHook workaround, with full usage examples. - Added "Using the REST API from Plugins" section to `airflow-core/docs/administration-and-deployment/plugins.rst` with plugin-specific examples. - Created `airflow-core/newsfragments/local-rest-client.feature.rst` newsfragment. --- ## File Map (New & Modified) ### New Files ``` airflow-core/src/airflow/api/client/ ├── local_rest_client.py # Main client class + transports ├── in_process_core_api.py # In-process Core API app helper └── resources/ ├── __init__.py ├── pools.py # PoolsClient ├── dags.py # DagsClient ├── dag_runs.py # DagRunsClient ├── connections.py # ConnectionsClient ├── variables.py # VariablesClient ├── task_instances.py # TaskInstancesClient ├── config.py # ConfigClient └── assets.py # AssetsClient airflow-core/src/airflow/api_fastapi/auth/ ├── managers/models/ │ └── system_user.py # SystemUser model └── process_token.py # Process JWT generation airflow-core/tests/api/client/ ├── test_local_rest_client.py # Unit tests └── test_local_rest_client_remote.py # Integration tests with remote transport ``` ### Modified Files ``` airflow-core/src/airflow/api/client/__init__.py — Add get_local_rest_client(), deprecate get_current_api_client() airflow-core/src/airflow/api_fastapi/auth/tokens.py — Add "process" scope constant airflow-core/src/airflow/api_fastapi/core_api/security.py — Extend get_user() to recognize process-scope JWTs airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py — Handle SystemUser in is_authorized_*() methods airflow-core/src/airflow/api_fastapi/auth/managers/simple/simple_auth_manager.py — Handle SystemUser in concrete is_authorized_*() implementations airflow-core/src/airflow/config_templates/config.yml — Add process_jwt_expiration_time, process_jwt_audience settings ``` --- ## Security Considerations 1. **Process tokens are not user tokens.** They grant system-level access and must only be generated by code running inside a trusted Airflow process. The JWT signing key is the same `secret_key` / private key used for Execution API tokens, so only components with access to the Airflow configuration can generate them. 2. **Scope separation.** The `process` scope is distinct from `execution` and `workload` scopes. The Core API validates that only `process`-scoped tokens are accepted (not execution/workload tokens, which are for the Execution API). 3. **Short-lived tokens.** Process tokens follow the same short-lived pattern as execution API tokens (default 600s), with automatic refresh handled by `ProcessAuthTransport`. 4. **Audit trail preserved.** Even though `SystemUser` gets full authorization, all actions are logged with the process identity, maintaining audit traceability. 5. **In-process mode security.** The in-process transport bypasses network-level auth but still runs through the API's validation and business logic layers (e.g., "cannot delete default_pool"). This mirrors the existing `InProcessExecutionAPI` pattern used by the DAG processor and triggerer. 6. **No credential storage.** Unlike `airflow-ctl` which stores tokens in the keyring, process tokens are ephemeral — generated on-demand from the signing key that the process already has access to via configuration. --- ## Open Questions 1. **Should `SystemUser` have configurable permission scopes?** The initial design grants full access. A follow-up could allow restricting process tokens to specific resource types (e.g., a worker can only manage pools, not delete DAGs). 2. **Should the in-process transport share the Core API app instance with the actual API server when co-located?** This would reduce memory usage but adds complexity around concurrent access. 3. **Should process tokens be accepted by the Execution API too?** This would allow a single client to call both `/api/v2/` and `/execution/` endpoints. The Execution API currently only accepts `execution` and `workload` scopes. 4. **How should this interact with external auth managers (e.g., OAuth2/OIDC)?** External auth managers may not know about `SystemUser`. The `process` token validation may need to be handled at a layer before the auth manager. 5. **What process types should be allowed?** The initial set is `dag_processor`, `scheduler`, `worker`, `triggerer`. Should there be a generic `plugin` type for code that runs outside the standard component lifecycle? 6. **Naming:** `LocalRESTClient` vs `AirflowClient` vs `SystemClient`? The name should convey that this is for in-process/trusted use, not for external callers. --- ## References - [Discussion #61800](https://github.com/apache/airflow/discussions/61800) — original feature request - [InProcessExecutionAPI](../../../api_fastapi/execution_api/app.py) — existing in-process API pattern - [Execution API security](../../../api_fastapi/execution_api/security.py) — JWT scope model - [Core API security](../../../api_fastapi/core_api/security.py) — user resolution flow - [BaseAuthManager](../../../api_fastapi/auth/managers/base_auth_manager.py) — authorization interface - [airflow-ctl client](../../../../airflow-ctl/src/airflowctl/api/client.py) — httpx-based API client pattern - [local_client.py](local_client.py) — current direct-DB client (to be deprecated) </details> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
