diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index afa263a..f2de215 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/src/echo_certification_forge/subscriber.py @@ -31,6 +31,20 @@ _ZERO_HASH = "0" * 64 _TERMINAL_RUN_STATES = ("COMPLETED", "CANCELLED", "INFRASTRUCTURE_FAILURE") +# Worker exception text is untrusted; retain only documented machine reasons. +# Also redact API responses from legacy rows with previously stored raw errors. +def _bounded_dispatch_error(error: str | None) -> str | None: + if error is None: + return None + if error in ( + "subscriber_worker_claim_denied", + "dispatch_attempts_exhausted", + "dispatch_lease_expired", + ): + return error + return "redacted" + + class OrganizationStatus(StrEnum): TRIALING = "TRIALING" @@ -4441,7 +4455,7 @@ def finish_dispatch( """, ( state, - error, + _bounded_dispatch_error(error), now, dispatch.run_id, dispatch.organization_id, @@ -4465,7 +4479,9 @@ def dispatch_status(self, run_id: str, organization_id: str) -> dict[str, Any]: ).fetchone() if row is None: raise SubscriberError(404, "dispatch_not_found") - return dict(row) + result = dict(row) + result["last_error"] = _bounded_dispatch_error(result["last_error"]) + return result def _resolve_worker_governance( self, diff --git a/tests/test_dispatch_error_retention.py b/tests/test_dispatch_error_retention.py new file mode 100644 index 0000000..00d0883 --- /dev/null +++ b/tests/test_dispatch_error_retention.py @@ -0,0 +1,60 @@ +"""Untrusted dispatch worker exceptions must not persist or leak on readback.""" +from __future__ import annotations + +import sqlite3 + +import pytest + +from echo_certification_forge.models import RunState +from test_dispatch_orphaned_runs import _orphan_the_old_way, _run, _usage + + +@pytest.mark.parametrize( + ("incoming", "expected"), + [ + ("subscriber_worker_claim_denied", "subscriber_worker_claim_denied"), + ("Authorization: Bearer synthetic-do-not-retain-000", "redacted"), + ("arbitrary remote stack trace", "redacted"), + ], +) +def test_new_worker_error_storage_and_status_redacted( + tmp_path, manifest, incoming, expected +): + store, gov, org, run_id = _run(tmp_path, manifest, "error-retention", max_attempts=1) + dispatch = gov.claim_dispatch( + "dispatcher", run_id=run_id, organization_id=org.organization_id + ) + assert dispatch is not None + assert gov.finish_dispatch(dispatch, error=incoming) == "FAILED" + assert store.get_run(run_id, org.organization_id)["state"] == ( + RunState.INFRASTRUCTURE_FAILURE.value + ) + usage = _usage(gov, org) + assert usage["meters"]["certification_runs"] == 0 + assert usage["active_run_reservations"] == 0 + assert gov.dispatch_status(run_id, org.organization_id)["last_error"] == expected + with sqlite3.connect(gov.db_path) as conn: + stored = conn.execute( + "SELECT last_error FROM subscriber_run_dispatches WHERE run_id = ?", (run_id,) + ).fetchone()[0] + assert stored == expected + assert "synthetic-do-not-retain-000" not in (stored or "") + + +def test_legacy_error_read_redacted_without_mutating_db(tmp_path, manifest): + store, gov, org, run_id = _run(tmp_path, manifest, "legacy-error", max_attempts=3) + _orphan_the_old_way(gov, org, run_id) + raw = "Authorization: Bearer synthetic-legacy-no-display-000" + with sqlite3.connect(gov.db_path) as conn: + conn.execute( + "UPDATE subscriber_run_dispatches SET last_error = ? WHERE run_id = ?", + (raw, run_id), + ) + conn.commit() + assert gov.dispatch_status(run_id, org.organization_id)["last_error"] == "redacted" + with sqlite3.connect(gov.db_path) as conn: + assert conn.execute( + "SELECT last_error FROM subscriber_run_dispatches WHERE run_id = ?", + (run_id,), + ).fetchone()[0] == raw + assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value