Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions src/echo_certification_forge/subscriber.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -4441,7 +4455,7 @@ def finish_dispatch(
""",
(
state,
error,
_bounded_dispatch_error(error),
now,
dispatch.run_id,
dispatch.organization_id,
Expand All @@ -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,
Expand Down
60 changes: 60 additions & 0 deletions tests/test_dispatch_error_retention.py
Original file line number Diff line number Diff line change
@@ -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
Loading