This is an automated email from the ASF dual-hosted git repository.
ferruzzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 82215cf97c7 Add single-use callback token for deadline callback
context fetch (#71192)
82215cf97c7 is described below
commit 82215cf97c7c96fd3cdc413be17f9692513b8374
Author: Sean Ghaeli <[email protected]>
AuthorDate: Fri Aug 28 21:02:59 2026 -0700
Add single-use callback token for deadline callback context fetch (#71192)
Deadline callbacks run in a subprocess that needs to read the DagRun
context (and connections/variables/xcoms) from the Execution API. PR
also accept the long-lived ``workload`` token via ``token:workload``
opt-ins. That over-broadened the workload token's reach (scope creep):
a long-lived token could read arbitrary DagRun/connection/variable/xcom
data for the whole queue-wait lifetime, and Ash asked for a single-use
credential instead.
This re-lands the security core of #66608 with a tighter design:
---------
Co-authored-by: Ash Berlin-Taylor <[email protected]>
---
airflow-core/newsfragments/71192.significant.rst | 12 ++
.../src/airflow/api_fastapi/execution_api/app.py | 8 +-
.../api_fastapi/execution_api/datamodels/token.py | 2 +-
.../api_fastapi/execution_api/routes/__init__.py | 2 +
.../api_fastapi/execution_api/routes/callbacks.py | 85 +++++++++++++
.../execution_api/routes/task_instances.py | 6 +-
.../airflow/api_fastapi/execution_api/security.py | 18 ++-
.../api_fastapi/execution_api/versions/__init__.py | 7 +-
.../execution_api/versions/v2026_10_30.py | 12 ++
.../src/airflow/executors/workloads/base.py | 11 +-
.../src/airflow/executors/workloads/callback.py | 4 +-
.../src/airflow/jobs/scheduler_job_runner.py | 8 +-
.../unit/api_fastapi/execution_api/conftest.py | 5 +-
.../api_fastapi/execution_api/test_security.py | 10 ++
.../execution_api/test_token_scope_boundaries.py | 2 +
.../execution_api/versions/head/test_callbacks.py | 131 +++++++++++++++++++++
.../versions/v2026_10_30/test_callbacks.py | 43 +++++++
.../tests/unit/executors/test_workloads.py | 19 +++
task-sdk/src/airflow/sdk/api/client.py | 17 +++
.../sdk/execution_time/callback_supervisor.py | 3 +
task-sdk/tests/task_sdk/api/test_client.py | 33 ++++++
.../execution_time/test_callback_supervisor.py | 50 +++++++-
22 files changed, 463 insertions(+), 25 deletions(-)
diff --git a/airflow-core/newsfragments/71192.significant.rst
b/airflow-core/newsfragments/71192.significant.rst
new file mode 100644
index 00000000000..6d3119377d6
--- /dev/null
+++ b/airflow-core/newsfragments/71192.significant.rst
@@ -0,0 +1,12 @@
+Callbacks now redeem a single-use token before they run
+
+A worker now redeems a callback's single-use token with the API server
+(``PATCH /execution/callbacks/{callback_id}/run``) before importing and
running the callback.
+Redeeming the token moves the callback from queued to running and swaps it for
a short-lived
+execution-scoped token. Before this change a callback ran without the worker
making any
+authenticated call, so the token minted for it was never checked.
+
+Because the token is single-use, a redelivered or replayed message that
reaches a callback that is
+already running or finished is refused rather than run again. A worker on 3.4
needs an API server
+that serves this endpoint; a worker talking to an older API server negotiates
the API version and
+behaves as though the endpoint is not there.
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/app.py
b/airflow-core/src/airflow/api_fastapi/execution_api/app.py
index f9e7cb17250..53be7e5af67 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/app.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/app.py
@@ -148,10 +148,10 @@ class JWTReissueMiddleware(BaseHTTPMiddleware):
validator: JWTValidator = await services.aget(JWTValidator)
claims = await validator.avalidated_claims(token, {})
- # Workload tokens are long-lived and meant to survive queue
- # wait times so avoid refreshing them. If avalidated_claims
- # raises for a workload token, the outer except handles it.
- if claims.get("scope") == "workload":
+ # Workload and callback tokens are long-lived and meant to
survive
+ # queue wait times so avoid refreshing them. If
avalidated_claims
+ # raises for such a token, the outer except handles it.
+ if claims.get("scope") in ("workload", "callback"):
return response
now = int(time.time())
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/token.py
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/token.py
index 4c3b935f5aa..eb858fac1df 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/token.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/token.py
@@ -24,7 +24,7 @@ from pydantic import ConfigDict
from airflow.api_fastapi.core_api.base import BaseModel
-TokenScope = Literal["execution", "workload"]
+TokenScope = Literal["execution", "workload", "callback"]
class TIClaims(BaseModel):
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py
index 7b19f3ddd30..5662378c870 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py
@@ -23,6 +23,7 @@ from airflow.api_fastapi.execution_api.routes import (
asset_events,
asset_state_store,
assets,
+ callbacks,
connection_tests,
connections,
dag_runs,
@@ -53,6 +54,7 @@ authenticated_router.include_router(
connection_tests.router, prefix="/connection-tests", tags=["Connection
Tests"]
)
authenticated_router.include_router(connections.router, prefix="/connections",
tags=["Connections"])
+authenticated_router.include_router(callbacks.router, prefix="/callbacks",
tags=["Callbacks"])
authenticated_router.include_router(dag_runs.router, prefix="/dag-runs",
tags=["Dag Runs"])
authenticated_router.include_router(dags.router, prefix="/dags", tags=["Dags"])
authenticated_router.include_router(task_instances.router,
prefix="/task-instances", tags=["Task Instances"])
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/callbacks.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/callbacks.py
new file mode 100644
index 00000000000..2f6e386b0b3
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/callbacks.py
@@ -0,0 +1,85 @@
+# 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
+
+from uuid import UUID
+
+from cadwyn import VersionedAPIRouter
+from fastapi import HTTPException, Response, Security, status
+
+from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
+from airflow.api_fastapi.execution_api.deps import DepContainer
+from airflow.api_fastapi.execution_api.security import (
+ ExecutionAPIRoute,
+ issue_execution_token,
+ require_auth,
+)
+from airflow.models.callback import Callback
+from airflow.utils.state import CallbackState
+
+router = VersionedAPIRouter(
+ route_class=ExecutionAPIRoute,
+ dependencies=[
+ Security(require_auth, scopes=["cb:self", "token:callback"]),
+ ],
+)
+
+
[email protected](
+ "/{callback_id}/run",
+ status_code=status.HTTP_204_NO_CONTENT,
+ responses=create_openapi_http_exception_doc(
+ [
+ (status.HTTP_404_NOT_FOUND, "Callback not found"),
+ (status.HTTP_409_CONFLICT, "The callback token was already
exchanged"),
+ ]
+ ),
+)
+def run_callback(
+ callback_id: UUID,
+ response: Response,
+ session: SessionDep,
+ services=DepContainer,
+) -> None:
+ """Exchange a single-use callback token for a short-lived execution
token."""
+ callback = session.get(Callback, callback_id, with_for_update=True)
+ if callback is None:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail={
+ "reason": "not_found",
+ "message": f"Callback {callback_id} not found",
+ },
+ )
+
+ if callback.state != CallbackState.QUEUED:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail={
+ "reason": "invalid_state",
+ "message": (
+ f"Callback {callback_id} is in state {callback.state}; its
token can only be "
+ "exchanged once while QUEUED."
+ ),
+ "previous_state": callback.state,
+ },
+ )
+
+ callback.state = CallbackState.RUNNING
+
+ issue_execution_token(services, response, sub=str(callback_id))
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 123f56c1f10..625e2ee9116 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -43,7 +43,6 @@ from structlog.contextvars import bind_contextvars
from airflow._shared.observability.traces import override_ids
from airflow._shared.state import TaskScope
from airflow._shared.timezones import timezone
-from airflow.api_fastapi.auth.tokens import JWTGenerator
from airflow.api_fastapi.common.dagbag import DagBagDep,
get_latest_version_of_dag
from airflow.api_fastapi.common.db.common import SessionDep
from airflow.api_fastapi.common.db.dags import eager_load_teams
@@ -75,6 +74,7 @@ from airflow.api_fastapi.execution_api.security import (
CurrentTIToken,
ExecutionAPIRoute,
get_team_name_for_ti,
+ issue_execution_token,
require_auth,
)
from airflow.api_fastapi.execution_api.services.task_instances import (
@@ -346,9 +346,7 @@ def ti_run(
# JWTReissueMiddleware also writes Refreshed-API-Token but skips workload
tokens, so we set it here for the workload→execution swap.
if token.claims.scope == "workload":
- generator: JWTGenerator = services.get(JWTGenerator)
- execution_token = generator.generate(extras={"sub":
str(task_instance_id), "scope": "execution"})
- response.headers["Refreshed-API-Token"] = execution_token
+ issue_execution_token(services, response, sub=str(task_instance_id))
return context
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/security.py
b/airflow-core/src/airflow/api_fastapi/execution_api/security.py
index 9de3493061f..c85980db08b 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/security.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/security.py
@@ -70,14 +70,15 @@ Why ``ExecutionAPIRoute`` is needed:
from typing import Any, get_args
import structlog
-from fastapi import Depends, HTTPException, Request, status
+import svcs
+from fastapi import Depends, HTTPException, Request, Response, status
from fastapi.params import Security as SecurityParam
from fastapi.routing import APIRoute
from fastapi.security import HTTPBearer, SecurityScopes
from pydantic import ValidationError
from sqlalchemy import select
-from airflow.api_fastapi.auth.tokens import JWTValidator
+from airflow.api_fastapi.auth.tokens import JWTGenerator, JWTValidator
from airflow.api_fastapi.execution_api.datamodels.token import TIClaims,
TIToken, TokenScope
from airflow.api_fastapi.execution_api.deps import DepContainer
@@ -196,6 +197,13 @@ async def require_auth(
status_code=status.HTTP_403_FORBIDDEN,
detail="Token subject does not match connection test ID",
)
+ elif "cb:self" in security_scopes.scopes:
+ cb_self_id = str(request.path_params["callback_id"])
+ if str(token.id) != cb_self_id:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="Token subject does not match callback ID",
+ )
return token
@@ -203,6 +211,12 @@ async def require_auth(
CurrentTIToken: TIToken = Depends(require_auth)
+def issue_execution_token(services: svcs.Container, response: Response, sub:
str) -> None:
+ """Mint an ``execution``-scoped token and set it on the
``Refreshed-API-Token`` header."""
+ generator: JWTGenerator = services.get(JWTGenerator)
+ response.headers["Refreshed-API-Token"] =
generator.generate(extras={"sub": sub, "scope": "execution"})
+
+
class ExecutionAPIRoute(APIRoute):
"""
Custom route class that precomputes allowed token types from Security
scopes.
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
index d56ec735c8f..d055131a510 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
@@ -51,11 +51,14 @@ from airflow.api_fastapi.execution_api.versions.v2026_06_30
import (
AddTeamNameField,
AddVariableKeysEndpoint,
)
-from airflow.api_fastapi.execution_api.versions.v2026_10_30 import
AddArgBindingsToTIRunContext
+from airflow.api_fastapi.execution_api.versions.v2026_10_30 import (
+ AddArgBindingsToTIRunContext,
+ AddCallbackRunEndpoint,
+)
bundle = VersionBundle(
HeadVersion(),
- Version("2026-10-30", AddArgBindingsToTIRunContext),
+ Version("2026-10-30", AddArgBindingsToTIRunContext,
AddCallbackRunEndpoint),
Version(
"2026-06-30",
AddVariableKeysEndpoint,
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
index 1c85aed252c..0620053b3e2 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
@@ -19,8 +19,10 @@ from __future__ import annotations
from cadwyn import (
ResponseInfo,
+ VersionChange,
VersionChangeWithSideEffects,
convert_response_to_previous_version_for,
+ endpoint,
schema,
)
@@ -40,3 +42,13 @@ class
AddArgBindingsToTIRunContext(VersionChangeWithSideEffects):
def remove_arg_bindings_field(response: ResponseInfo) -> None: # type:
ignore[misc]
"""Strip ``arg_bindings`` from the run context for older clients."""
response.body.pop("arg_bindings", None)
+
+
+class AddCallbackRunEndpoint(VersionChange):
+ """Add the callbacks/{callback_id}/run endpoint a worker uses to exchange
its single-use callback token."""
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version = (
+ endpoint("/callbacks/{callback_id}/run", ["PATCH"]).didnt_exist,
+ )
diff --git a/airflow-core/src/airflow/executors/workloads/base.py
b/airflow-core/src/airflow/executors/workloads/base.py
index 41334d68f30..43e47065b66 100644
--- a/airflow-core/src/airflow/executors/workloads/base.py
+++ b/airflow-core/src/airflow/executors/workloads/base.py
@@ -21,7 +21,7 @@ from __future__ import annotations
import os
from abc import ABC, abstractmethod
from collections.abc import Hashable
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
from pydantic import BaseModel, ConfigDict, Field
@@ -83,13 +83,16 @@ class BaseWorkloadSchema(BaseModel):
token: str = Field(repr=False)
"""The identity token for this workload"""
- @staticmethod
- def generate_token(sub_id: str, generator: JWTGenerator | None = None) ->
str:
+ token_scope: ClassVar[str] = "workload"
+ """Scope claim stamped into tokens minted for this workload type."""
+
+ @classmethod
+ def generate_token(cls, sub_id: str, generator: JWTGenerator | None =
None) -> str:
if not generator:
return ""
valid_for = conf.getfloat("scheduler", "task_queued_timeout")
return generator.generate(
- extras={"sub": sub_id, "scope": "workload"},
+ extras={"sub": sub_id, "scope": cls.token_scope},
valid_for=valid_for,
)
diff --git a/airflow-core/src/airflow/executors/workloads/callback.py
b/airflow-core/src/airflow/executors/workloads/callback.py
index c1842a59006..29d6234123a 100644
--- a/airflow-core/src/airflow/executors/workloads/callback.py
+++ b/airflow-core/src/airflow/executors/workloads/callback.py
@@ -20,7 +20,7 @@ from __future__ import annotations
from enum import Enum
from pathlib import Path
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, ClassVar, Literal
from uuid import UUID
import structlog
@@ -77,6 +77,8 @@ class ExecuteCallback(BaseDagBundleWorkload):
type: Literal["ExecuteCallback"] = Field(init=False,
default="ExecuteCallback")
+ token_scope: ClassVar[str] = "callback"
+
@property
def key(self) -> CallbackKey:
"""Return the callback key for this workload."""
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index f8b2df4aac2..127f385f0e0 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -1439,7 +1439,8 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
cls.logger().debug("Draining executor event with state %s for
connection test %s", state, key)
elif isinstance(key, CallbackKey):
cls.logger().info("Received executor event with state %s for
callback %s", state, key)
- if state in (CallbackState.RUNNING, CallbackState.FAILED,
CallbackState.SUCCESS):
+ # Skip RUNNING: the callback token endpoint owns that
transition, so persisting it here races.
+ if state in (CallbackState.FAILED, CallbackState.SUCCESS):
callback_keys_with_events.append(key)
else:
cls.logger().error("Unknown workload key type in event buffer:
%r", key)
@@ -1456,10 +1457,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
)
continue
- if state == CallbackState.RUNNING:
- callback.state = CallbackState.RUNNING
- cls.logger().info("Callback %s is currently running",
callback_id)
- elif state == CallbackState.SUCCESS:
+ if state == CallbackState.SUCCESS:
callback.state = CallbackState.SUCCESS
cls.logger().info("Callback %s completed successfully",
callback_id)
elif state == CallbackState.FAILED:
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/conftest.py
b/airflow-core/tests/unit/api_fastapi/execution_api/conftest.py
index fda87d21e21..be6e7f259c5 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/conftest.py
+++ b/airflow-core/tests/unit/api_fastapi/execution_api/conftest.py
@@ -58,7 +58,10 @@ def client(request: pytest.FixtureRequest):
raw_id = request.path_params.get(
"task_instance_id",
- request.path_params.get("connection_test_id",
"00000000-0000-0000-0000-000000000000"),
+ request.path_params.get(
+ "connection_test_id",
+ request.path_params.get("callback_id",
"00000000-0000-0000-0000-000000000000"),
+ ),
)
try:
ti_id = UUID(raw_id)
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/test_security.py
b/airflow-core/tests/unit/api_fastapi/execution_api/test_security.py
index e98d9183859..1f2bb1825fa 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/test_security.py
+++ b/airflow-core/tests/unit/api_fastapi/execution_api/test_security.py
@@ -80,6 +80,16 @@ class TestExecutionAPIRoute:
)
assert route.allowed_token_types == frozenset({"execution"})
+ def test_extracts_callback_token_scope(self):
+ route = ExecutionAPIRoute(
+ path="/test",
+ endpoint=lambda: None,
+ dependencies=[
+ Security(require_auth, scopes=["cb:self", "token:callback"]),
+ ],
+ )
+ assert route.allowed_token_types == frozenset({"callback"})
+
def test_rejects_invalid_token_types(self):
with pytest.raises(ValueError, match="Invalid token types"):
ExecutionAPIRoute(
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/test_token_scope_boundaries.py
b/airflow-core/tests/unit/api_fastapi/execution_api/test_token_scope_boundaries.py
index bbf2be8704f..195503f2bf0 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/test_token_scope_boundaries.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/test_token_scope_boundaries.py
@@ -49,6 +49,8 @@ NON_DEFAULT_TOKEN_POLICY: dict[str, set[str]] = {
# Connection test routes run from a queued worker context (workload-only).
"PATCH /connection-tests/{connection_test_id}": {"workload"},
"GET /connection-tests/{connection_test_id}/connection": {"workload"},
+ # Callback /run exchanges a single-use callback token for an execution
token.
+ "PATCH /callbacks/{callback_id}/run": {"callback"},
}
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_callbacks.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_callbacks.py
new file mode 100644
index 00000000000..da24c5c5f9f
--- /dev/null
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_callbacks.py
@@ -0,0 +1,131 @@
+# 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
+
+from unittest import mock
+
+import jwt
+import pytest
+
+from airflow.api_fastapi.auth.tokens import JWTValidator
+from airflow.api_fastapi.execution_api.app import lifespan
+from airflow.api_fastapi.execution_api.security import require_auth
+from airflow.models.callback import CallbackFetchMethod, CallbackState,
ExecutorCallback
+from airflow.sdk.definitions.callback import SyncCallback
+
+from tests_common.test_utils.db import clear_db_callbacks
+
+pytestmark = pytest.mark.db_test
+
+
+def sync_callback():
+ """Empty (sync) callable used for unit tests"""
+ pass
+
+
[email protected](autouse=True)
+def clean_callbacks():
+ clear_db_callbacks()
+ yield
+ clear_db_callbacks()
+
+
[email protected]
+def queued_callback(session):
+ callback = ExecutorCallback(
+ callback_def=SyncCallback(sync_callback, kwargs={}),
+ fetch_method=CallbackFetchMethod.IMPORT_PATH,
+ )
+ callback.data["dag_id"] = "test_dag"
+ callback.state = CallbackState.QUEUED
+ session.add(callback)
+ session.commit()
+ return callback
+
+
+class TestRunCallback:
+ def test_run_is_single_use(self, client, session, queued_callback):
+ """First call moves QUEUED -> RUNNING and swaps the token; a replay is
rejected with 409."""
+ response =
client.patch(f"/execution/callbacks/{queued_callback.id}/run")
+
+ assert response.status_code == 204
+ payload = jwt.decode(response.headers["Refreshed-API-Token"],
options={"verify_signature": False})
+ assert payload["scope"] == "execution"
+ assert payload["sub"] == str(queued_callback.id)
+
+ session.expire_all()
+ session.refresh(queued_callback)
+ assert queued_callback.state == CallbackState.RUNNING
+
+ second = client.patch(f"/execution/callbacks/{queued_callback.id}/run")
+ assert second.status_code == 409
+ detail = second.json()["detail"]
+ assert detail["reason"] == "invalid_state"
+ assert detail["previous_state"] == CallbackState.RUNNING
+
+ @pytest.mark.parametrize("state", [CallbackState.SUCCESS,
CallbackState.FAILED, CallbackState.PENDING])
+ def test_run_rejects_non_queued_state(self, client, session,
queued_callback, state):
+ """The token can only be exchanged while the callback is QUEUED."""
+ queued_callback.state = state
+ session.commit()
+
+ response =
client.patch(f"/execution/callbacks/{queued_callback.id}/run")
+ assert response.status_code == 409
+ assert response.json()["detail"]["reason"] == "invalid_state"
+
+ def test_run_returns_404_for_nonexistent(self, client):
+ """Exchanging a token for an unknown callback returns 404."""
+ response =
client.patch("/execution/callbacks/00000000-0000-0000-0000-000000000000/run")
+ assert response.status_code == 404
+ assert response.json()["detail"]["reason"] == "not_found"
+
+
[email protected]
+def _use_real_jwt_bearer(exec_app):
+ """Remove the mock require_auth override so the real JWT validation runs
end-to-end."""
+ exec_app.dependency_overrides.pop(require_auth, None)
+
+
[email protected]("_use_real_jwt_bearer")
[email protected](
+ ("token_sub", "token_scope", "expected_status"),
+ [
+ pytest.param("self", "callback", 204,
id="callback-token-matching-sub-accepted"),
+ pytest.param(
+ "11111111-1111-1111-1111-111111111111",
+ "callback",
+ 403,
+ id="callback-token-for-other-callback-rejected",
+ ),
+ pytest.param("self", "execution", 403,
id="execution-scope-token-rejected"),
+ ],
+)
+def test_run_validates_token_scope_and_sub(client, queued_callback, token_sub,
token_scope, expected_status):
+ """The exchange endpoint only accepts a callback-scope JWT whose sub is
this callback's id."""
+ sub = str(queued_callback.id) if token_sub == "self" else token_sub
+ validator = mock.AsyncMock(spec=JWTValidator)
+ validator.avalidated_claims.return_value = {
+ "sub": sub,
+ "scope": token_scope,
+ "exp": 9999999999,
+ "iat": 1000000000,
+ "nbf": 1000000000,
+ }
+ lifespan.registry.register_value(JWTValidator, validator)
+
+ resp = client.patch(f"/execution/callbacks/{queued_callback.id}/run")
+ assert resp.status_code == expected_status, resp.json()
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_callbacks.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_callbacks.py
new file mode 100644
index 00000000000..c36f682a22f
--- /dev/null
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_callbacks.py
@@ -0,0 +1,43 @@
+# 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 pytest
+
+pytestmark = pytest.mark.db_test
+
+MISSING_CALLBACK_URL =
"/execution/callbacks/00000000-0000-0000-0000-000000000000/run"
+
+
+class TestRunCallbackEndpointVersioning:
+ """The callbacks/{callback_id}/run endpoint didn't exist before the
2026-10-30 API version."""
+
+ def test_old_version_returns_404(self, client):
+ """Before 2026-10-30 the route is absent, so routing itself 404s (no
endpoint-shaped detail)."""
+ client.headers["Airflow-API-Version"] = "2026-06-30"
+
+ response = client.patch(MISSING_CALLBACK_URL)
+
+ assert response.status_code == 404
+ assert response.json() == {"detail": "Not Found"}
+
+ def test_head_version_routes_to_endpoint(self, client):
+ """At head the route exists: the same request reaches the endpoint's
own 404 handling."""
+ response = client.patch(MISSING_CALLBACK_URL)
+
+ assert response.status_code == 404
+ assert response.json()["detail"]["reason"] == "not_found"
diff --git a/airflow-core/tests/unit/executors/test_workloads.py
b/airflow-core/tests/unit/executors/test_workloads.py
index 37fbcd96ce9..7c639c43594 100644
--- a/airflow-core/tests/unit/executors/test_workloads.py
+++ b/airflow-core/tests/unit/executors/test_workloads.py
@@ -90,6 +90,25 @@ def test_generate_token_without_generator():
assert BaseWorkloadSchema.generate_token("ti-123", None) == ""
+def test_token_scope_is_a_class_level_invariant():
+ """Token scope is fixed per workload type, not caller-suppliable."""
+ assert BaseWorkloadSchema.token_scope == "workload"
+ assert ExecuteTask.token_scope == "workload"
+ assert ExecuteCallback.token_scope == "callback"
+
+
+def test_generate_token_uses_subclass_token_scope(monkeypatch):
+ """ExecuteCallback.generate_token should stamp its own 'callback' scope."""
+ monkeypatch.setattr(workloads_base.conf, "getfloat", lambda section, key:
86400.0)
+
+ generator = JWTGenerator(secret_key="test-secret", audience="test",
valid_for=60)
+ token = ExecuteCallback.generate_token("cb-123", generator)
+
+ claims = jwt.decode(token, "test-secret", algorithms=["HS512"],
audience="test")
+ assert claims["sub"] == "cb-123"
+ assert claims["scope"] == "callback"
+
+
def test_callback_key_is_frozen_and_hashable():
"""CallbackKey must be usable as a dict key (hashable) and immutable
(frozen)."""
cid = "some-uuid-value"
diff --git a/task-sdk/src/airflow/sdk/api/client.py
b/task-sdk/src/airflow/sdk/api/client.py
index a0c0a30de08..ea220c68fde 100644
--- a/task-sdk/src/airflow/sdk/api/client.py
+++ b/task-sdk/src/airflow/sdk/api/client.py
@@ -1097,6 +1097,17 @@ class ConnectionTestOperations:
self.client.patch(f"connection-tests/{id}",
content=body.model_dump_json())
+class CallbackOperations:
+ __slots__ = ("client",)
+
+ def __init__(self, client: Client):
+ self.client = client
+
+ def run(self, callback_id: uuid.UUID) -> None:
+ """Exchange the single-use callback token for an execution token."""
+ self.client.patch(f"callbacks/{callback_id}/run")
+
+
class BearerAuth(httpx.Auth):
def __init__(self, token: str):
self.token: str = token
@@ -1304,6 +1315,12 @@ class Client(httpx.Client):
"""Operations related to Connection Tests."""
return ConnectionTestOperations(self)
+ @lru_cache() # type: ignore[misc]
+ @property
+ def callbacks(self) -> CallbackOperations:
+ """Operations related to Callbacks."""
+ return CallbackOperations(self)
+
@lru_cache() # type: ignore[misc]
@property
def dags(self) -> DagsOperations:
diff --git a/task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py
index 98307717012..6b7650b9e4c 100644
--- a/task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py
@@ -433,6 +433,9 @@ def supervise_callback(
else:
logger = structlog.get_logger(logger_name="callback").bind()
+ # Swap the single-use callback token for an execution token before any
context read.
+ client.callbacks.run(UUID(id))
+
try:
process = CallbackSubprocess.start(
id=id,
diff --git a/task-sdk/tests/task_sdk/api/test_client.py
b/task-sdk/tests/task_sdk/api/test_client.py
index 8bd558b461b..684d4e7dcbb 100644
--- a/task-sdk/tests/task_sdk/api/test_client.py
+++ b/task-sdk/tests/task_sdk/api/test_client.py
@@ -2280,3 +2280,36 @@ class TestAssetStateOperations:
client = make_client(transport=httpx.MockTransport(handle_request))
result = client.asset_state_store.clear(uri="s3://bucket/key")
assert result == OKResponse(ok=True)
+
+
+class TestCallbackOperations:
+ def test_run_exchanges_token(self):
+ callback_id = uuid7()
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.method == "PATCH"
+ assert request.url.path == f"/callbacks/{callback_id}/run"
+ return httpx.Response(
+ status_code=204,
+ headers={"Refreshed-API-Token": "execution-token"},
+ )
+
+ client = make_client(transport=httpx.MockTransport(handle_request))
+ client.callbacks.run(callback_id)
+
+ assert client.auth is not None
+ assert client.auth.token == "execution-token"
+
+ def test_run_raises_on_conflict(self):
+ callback_id = uuid7()
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(
+ status_code=409,
+ json={"detail": {"reason": "invalid_state", "previous_state":
"running"}},
+ )
+
+ client = make_client(transport=httpx.MockTransport(handle_request))
+ with pytest.raises(ServerResponseError) as err:
+ client.callbacks.run(callback_id)
+ assert err.value.response.status_code == 409
diff --git a/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
index a6858abcfcd..7ccbb8c0dfa 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
@@ -30,7 +30,12 @@ from unittest.mock import ANY, Mock, patch
import pytest
import structlog
-from airflow.sdk.execution_time.callback_supervisor import CallbackSubprocess,
Path, execute_callback
+from airflow.sdk.execution_time.callback_supervisor import (
+ CallbackSubprocess,
+ Path,
+ execute_callback,
+ supervise_callback,
+)
from airflow.sdk.execution_time.comms import (
BundleInfo,
ConnectionResult,
@@ -547,3 +552,46 @@ class TestCallbackSubprocessStart:
self.mock_super_start.call_args.kwargs["target"]()
assert exc_info.value.code == 1
+
+
+class TestSuperviseCallbackExchangesTokenFirst:
+ """No callback code may run until the single-use token has been
exchanged."""
+
+ CALLBACK_ID = "01890a5d-ac70-7a5b-b7d5-0dd5b1c7be47"
+
+ def _supervise(self, client):
+ return supervise_callback(
+ id=self.CALLBACK_ID,
+ callback_path="does.not.matter",
+ callback_kwargs={},
+ dag_rel_path=Path("dag.py"),
+ client=client,
+ )
+
+
@patch("airflow.sdk.execution_time.callback_supervisor._make_process_nondumpable")
+ @patch.object(CallbackSubprocess, "start")
+ def test_exchange_happens_before_the_subprocess_starts(self, mock_start,
_nondumpable):
+ calls: list[str] = []
+ client = Mock()
+ client.callbacks.run.side_effect = lambda callback_id:
calls.append("exchange")
+
+ def record_start(**kwargs):
+ calls.append("start")
+ return Mock(wait=Mock(return_value=0))
+
+ mock_start.side_effect = record_start
+
+ assert self._supervise(client) == 0
+
client.callbacks.run.assert_called_once_with(uuid.UUID(self.CALLBACK_ID))
+ assert calls == ["exchange", "start"]
+
+
@patch("airflow.sdk.execution_time.callback_supervisor._make_process_nondumpable")
+ @patch.object(CallbackSubprocess, "start")
+ def test_rejected_exchange_prevents_the_subprocess_from_starting(self,
mock_start, _nondumpable):
+ client = Mock()
+ client.callbacks.run.side_effect = RuntimeError("409 Conflict: token
already exchanged")
+
+ with pytest.raises(RuntimeError, match="already exchanged"):
+ self._supervise(client)
+
+ mock_start.assert_not_called()
