fix: exclude JWT token from workload repr to prevent log exposure

airflow-core/newsfragments/62964.bugfix.rst

@@ -0,0 +1 @@
+Prevent JWT tokens from appearing in task logs by excluding the token field from workload object representations.

airflow-core/src/airflow/executors/workloads/base.py

@@ -22,7 +22,7 @@
 from abc import ABC
 from typing import TYPE_CHECKING
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 if TYPE_CHECKING:
     from airflow.api_fastapi.auth.tokens import JWTGenerator
@@ -69,7 +69,7 @@ class BaseWorkloadSchema(BaseModel):
 
     model_config = ConfigDict(populate_by_name=True)
 
-    token: str
+    token: str = Field(repr=False)
     """The identity token for this workload"""
 
     @staticmethod

airflow-core/tests/unit/executors/test_workloads.py

@@ -17,11 +17,47 @@
 # under the License.
 from __future__ import annotations
 
+from pathlib import PurePosixPath
+from uuid import uuid4
+
 from airflow.executors import workloads
 from airflow.executors.workloads import TaskInstance, TaskInstanceDTO
+from airflow.executors.workloads.base import BundleInfo
+from airflow.executors.workloads.task import ExecuteTask
 
 
 def test_task_instance_alias_keeps_backwards_compat():
     assert TaskInstance is TaskInstanceDTO
     assert workloads.TaskInstance is TaskInstanceDTO
     assert workloads.TaskInstanceDTO is TaskInstanceDTO
+
+
+def test_token_excluded_from_workload_repr():
+    """Ensure JWT tokens do not leak into log output via repr()."""
+    fake_token = "eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9.secret_payload.signature"
+    ti = TaskInstanceDTO(
+        id=uuid4(),
+        dag_version_id=uuid4(),
+        task_id="test_task",
+        dag_id="test_dag",
+        run_id="test_run",
+        try_number=1,
+        map_index=-1,
+        pool_slots=1,
+        queue="default",
+        priority_weight=1,
+    )
+    workload = ExecuteTask(
+        ti=ti,
+        dag_rel_path=PurePosixPath("test_dag.py"),
+        token=fake_token,
+        bundle_info=BundleInfo(name="dags-folder", version=None),
+        log_path="test.log",
+    )
+
+    workload_repr = repr(workload)
+
+    # Token MUST NOT appear in repr (prevents leaking into logs)
+    assert fake_token not in workload_repr, f"JWT token leaked into repr! Found token in: {workload_repr}"
+    # But token should still be accessible as an attribute
+    assert workload.token == fake_token