fix: never leak or persist a user API credential

docs/WORKFLOWS.md

@@ -69,6 +69,30 @@ parent output substitution and validation. See
 `src/worker/executors/utils/graph_templates.py` for the templating
 contract.
 
+## API task
+
+`taskType: api` performs a single HTTP request. By default it routes to the Nebula endpoint and authenticates with the worker's `NEBULA_API_TOKEN`.
+
+`spec.api.url` overrides the endpoint; when absent, the executor uses `NEBULA_API_BASE_URL` (appending `/v1/chat/completions`). `spec.api.headers` may supply an `Authorization` header directly.
+
+Credential handling: a caller-supplied `Authorization` header is always used as-is and never overwritten. With no header, `NEBULA_API_TOKEN` is injected only when the call is on the Nebula url (no custom `spec.api.url`) — the Nebula token is never sent to a custom endpoint. A Nebula-path call with no token available fails closed.
+
+```yaml
+spec:
+  taskType: api
+  api:
+    method: POST
+    headers:
+      Content-Type: application/json
+    body:
+      model: gpt-4o
+      messages:
+        - role: user
+          content: Hello
+    response:
+      parse_json: true
+```
+
 ## data_retrieval: type lumid
 
 `type: lumid` routes the retrieval through lumid-data-app (HTTP). Three

examples/templates/api_two_stage.yaml

@@ -4,8 +4,9 @@
 # Stage 1 calls a chat completion endpoint and returns raw text.
 # Stage 2 sends Stage 1's returned text as the next prompt.
 #
-# NOTE: Configure NEBULA_API_BASE_URL and NEBULA_API_TOKEN on the worker before
-# submitting. APIExecutor injects the Authorization header from NEBULA_API_TOKEN.
+# Stage 1 names a custom endpoint via spec.api.url and supplies its own
+# Authorization header. Stage 2 omits url and header, so it uses the Nebula
+# defaults (NEBULA_API_BASE_URL + NEBULA_API_TOKEN).
 
 apiVersion: flowmesh/v1
 kind: APITask
@@ -19,8 +20,10 @@ spec:
     - name: stage-1
       spec:
         api:
+          url: https://api.example.com/v1/chat/completions
           method: POST
           headers:
+            Authorization: Bearer <your-token>
             Content-Type: application/json
           body:
             model: gpt-4o

src/server/dispatcher/base.py

@@ -25,6 +25,7 @@
 )
 from shared.tasks.placeholders import PLACEHOLDER_PATTERN
 from shared.tasks.specs import (
+    ApiSpecStrict,
     ConditionSpec,
     SSHSpecStrict,
     SSHSpecTemplate,
@@ -36,6 +37,7 @@
 from ..services.metrics import MetricsRecorder
 from ..task.metadata import extract_model_dataset_names
 from ..task.models import TaskRecord, TaskStatus
+from ..task.redact import REDACTED
 from ..task.runtime import TaskRuntime
 from ..utils.time import now_iso
 from .worker_selector import DEFAULT_WORKER_SELECTION, select_worker
@@ -393,6 +395,20 @@ def dispatch_once(self, task_id: str) -> bool:
             )
             return True
 
+        if self._has_redacted_credential(rendered_task.spec):
+            self._runtime.release_merge(task_id)
+            self.fail_task(
+                task_id,
+                "credential_not_retained",
+                payload={
+                    "error": (
+                        "the API credential was not retained across the server "
+                        "restart; resubmit the workflow with the credential"
+                    )
+                },
+            )
+            return True
+
         # Conditional execution: skip dispatch if condition not met
         if self._evaluate_condition_skip(task_id, rendered_task, record):
             return True
@@ -1055,6 +1071,18 @@ def _collect_upstream_results(
             results[name] = envelope.result
         return results
 
+    def _has_redacted_credential(self, spec: TaskSpecStrict) -> bool:
+        """Whether an api spec carries a redacted credential placeholder."""
+        if not isinstance(spec, ApiSpecStrict):
+            return False
+        api = spec.api
+        if api is None:
+            return False
+        headers = api.get("headers")
+        if not isinstance(headers, dict):
+            return False
+        return any(value == REDACTED for value in headers.values())
+
     def _resolve_upstream_task_ids(
         self, record: TaskRecord, spec: TaskSpecStrict
     ) -> dict[str, str] | None:

src/server/task/models.py

@@ -1,12 +1,19 @@
 import time
 from typing import Any
 
-from pydantic import BaseModel, Field, computed_field
+from pydantic import (
+    BaseModel,
+    Field,
+    SerializerFunctionWrapHandler,
+    computed_field,
+    model_serializer,
+)
 
 from shared.tasks import TaskEnvelopeTemplate
 from shared.tasks.worker_message import HardwareUsage
 
 from ..utils.time import now_iso
+from .redact import redact_api, redact_raw_yaml
 
 TRAINING_TASK_TYPES = {
     "sft",
@@ -161,6 +168,16 @@ def last_failed_worker(self) -> str | None:
         """The most recent worker to have failed this task."""
         return self.failed_workers[-1] if self.failed_workers else None
 
+    @model_serializer(mode="wrap")
+    def _serialize(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]:
+        data = handler(self)
+        data["raw_yaml"] = redact_raw_yaml(self.raw_yaml)
+        spec = self.task.spec
+        api = getattr(spec, "api", None)
+        if isinstance(api, dict):
+            data["task"]["spec"]["api"] = redact_api(api)
+        return data
+
 
 class TaskInfo(TaskRecord):
     depends_on: list[str] = Field(description="Dependency task IDs.")

src/server/task/redact.py

@@ -0,0 +1,84 @@
+"""Redaction of API credentials before a task record is serialized.
+
+The in-memory ``TaskRecord`` keeps the real credential so dispatch works; the
+serializer applies this module so that every dump to Redis is redacted.
+
+``raw_yaml`` is re-emitted via ``yaml.safe_dump``, which does not preserve
+comments, key order or original formatting. That is an accepted cost: the field
+is a stored record and is never re-parsed, so losing formatting is fine. If the
+YAML cannot be parsed, the whole field is redacted rather than storing text
+that might contain a key.
+"""
+
+from typing import Any
+
+import yaml
+
+REDACTED = "[REDACTED]"
+
+# Whole-key matches; a bare substring test would over-redact (e.g. "monkey").
+_SENSITIVE_KEYS = frozenset(
+    {
+        "authorization",
+        "token",
+        "api-key",
+        "api_key",
+        "apikey",
+        "secret",
+        "access_token",
+        "bearer",
+        "x-api-key",
+    }
+)
+_SENSITIVE_SUFFIXES = ("_key", "-key", "_token", "-token")
+
+
+def _is_sensitive_key(name: str) -> bool:
+    lowered = name.lower()
+    if lowered in _SENSITIVE_KEYS:
+        return True
+    return any(lowered.endswith(suffix) for suffix in _SENSITIVE_SUFFIXES)
+
+
+def _redact_value(value: Any) -> Any:
+    """Recursively redact credential values by key name at any depth."""
+    if isinstance(value, dict):
+        return {
+            key: (REDACTED if _is_sensitive_key(str(key)) else _redact_value(val))
+            for key, val in value.items()
+        }
+    if isinstance(value, list):
+        return [_redact_value(item) for item in value]
+    return value
+
+
+def redact_api(api: dict[str, Any] | None) -> dict[str, Any] | None:
+    """Return a copy of an api spec with credential values replaced.
+
+    Only the five credential-bearing locations are touched; the rest of the
+    spec is returned unchanged. The original mapping is never mutated.
+    """
+    if not isinstance(api, dict):
+        return api
+    redacted = dict(api)
+    for field in ("headers", "params", "body", "json", "data"):
+        value = redacted.get(field)
+        if isinstance(value, (dict, list)):
+            redacted[field] = _redact_value(value)
+    return redacted
+
+
+def redact_raw_yaml(raw_yaml: str) -> str:
+    """Redact credential values from the original workflow YAML text.
+
+    The YAML is parsed and redacted with the same recursive key rule used for
+    the parsed spec, then re-emitted. If parsing fails, the whole field is
+    replaced with a marker rather than storing un-analysed text.
+    """
+    try:
+        tree = yaml.safe_load(raw_yaml)
+    except yaml.YAMLError:
+        return REDACTED
+    if tree is None:
+        return raw_yaml
+    return yaml.safe_dump(_redact_value(tree), sort_keys=False)

src/worker/executors/api_executor.py

@@ -19,12 +19,13 @@
 
 
 class APIExecutor(Executor):
-    """Executor that performs a single HTTP request defined by task YAML.
+    """Performs a single HTTP request defined by task YAML.
 
-    Uses a class-level connection pool keyed by (base_url, timeout, verify_tls,
-    follow_redirects) so that repeated calls to the same endpoint (e.g. a trading
-    bot hitting QuantArena every few seconds) reuse the underlying TCP/TLS
-    connection instead of paying the handshake cost on every request.
+    Defaults to the Nebula endpoint via ``NEBULA_API_BASE_URL`` and authenticates
+    with ``NEBULA_API_TOKEN``. ``spec.api.url`` overrides the endpoint and
+    ``spec.api.headers`` may supply an ``Authorization`` header directly. A
+    custom ``spec.api.url`` requires its own credential: the Nebula token is
+    never sent to an endpoint the caller chose.
     """
 
     name = "api"
@@ -99,6 +100,7 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
             raise ExecutionError("spec.api must be a mapping")
 
         url = api_cfg.get("url")
+        custom_url = url is not None
         if url is None:
             url = os.getenv("NEBULA_API_BASE_URL")
             if not url:
@@ -110,8 +112,21 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
         if not isinstance(headers, dict):
             raise ExecutionError("spec.api.headers must be a mapping")
 
-        token = os.getenv("NEBULA_API_TOKEN")
-        if token and not any(k.lower() == "authorization" for k in headers):
+        has_credential = any(k.lower() == "authorization" for k in headers)
+        if custom_url:
+            if not has_credential:
+                raise ExecutionError(
+                    "spec.api.url names a custom endpoint but no credential was "
+                    "supplied; set an Authorization header. The Nebula token is "
+                    "never sent to an endpoint the caller chose."
+                )
+        elif not has_credential:
+            token = os.getenv("NEBULA_API_TOKEN")
+            if not token:
+                raise ExecutionError(
+                    "no credential configured: set an Authorization header or "
+                    "NEBULA_API_TOKEN"
+                )
             headers["Authorization"] = f"Bearer {token}"
 
         params = api_cfg.get("params")