From f4ba205c9f755d817e2206c7a9cb390a3e61fd2f Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 22 Sep 2026 17:52:51 +0000 Subject: [PATCH] fix(issue-23): the remote audit record names the dispatch it vouches for; a record without identity fails closed Review round 4 (planner-reviewer-claude-01 B-1, planner-reviewer-codex-01 P2, reached independently): dispatch_verified is a boolean. It says the subscriber's token check ran when the record was written, but names no dispatch, the token being masked by the audit redactor and attempt_id not written. events.ndjson is appended to across a job's attempts, so a started verified under attempt 1 satisfied every later attempt's wait - the very misattribution R1 forbids, alive on the remote path while the local path (data.dispatch_token in events.log) was already immune. codex reproduced it with the real signed producer, the real _Watcher, the real audit writer and the real operator re-dispatch; both waits answered 0 where 1 was due, even when attempt 2's subscriber had correctly dropped attempt 1's replayed packet. The subscriber now writes dispatch_fingerprint = SHA-256 of the token it verified, and attempt_id for readers. With a token demanded, the predicate accepts an audit record only if it is dispatch_verified AND its fingerprint equals SHA-256 of that token; a record with no fingerprint - written before the field existed - names nothing and is rejected on that path. Without a token, any received started counts, as before. Fingerprint over attempt_id as the identity, weighing the two proposals: the local path already compares the token, so hashing it keeps one identity axis for both sources; the token is a uuid4 nonce (registry.py:141), 128 bits, so its SHA-256 is not recoverable and it is a CAS fence rather than the HMAC secret in the first place; and attempt_id cannot tell apart a token re-minted within one attempt, where the token can. attempt_id is recorded because it is cheap and readable, not relied on. W7 pins codex's five cases end to end - current dispatch accepted, an old attempt's record rejected after a real re-dispatch, still rejected when the stale packet is replayed and dropped, the new dispatch's own started accepted, a bad HMAC never recorded - with create's wait and the wrapper's wait required to agree on each. W8 pins the fail-closed legacy record and the untokened compatibility path. W5 gains the no-fingerprint and other-dispatch cases; W6 checks the real _Watcher writes the fingerprint and never the token. S7e's status reset had no reason left and is gone. codex's own reproduction file, unmodified, passes 5/5 against this tree. Co-Authored-By: Claude Fable 5.1 --- .agents/skills/lib.sh | 18 ++- .../lib_py/delegation/job_subscriber.py | 25 +++- tests/test_issue23_24_startup_path.py | 120 +++++++++++++++++- 3 files changed, 148 insertions(+), 15 deletions(-) diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index e8782a3..94559a6 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2233,8 +2233,15 @@ except Exception: # `started` it received, the only evidence a REMOTE agent (no shared disk) # can leave. The subscriber verifies the HMAC and drops any event whose # data.dispatch_token differs from the job's expected token before it -# writes this record, and marks it dispatch_verified when that check ran; -# with given, only a dispatch_verified record counts. +# writes this record (dispatch_verified), and writes dispatch_fingerprint +# = SHA-256 of that token. The file is appended to across a job's +# attempts, so the flag alone would let an earlier attempt's record +# acknowledge a later dispatch (review round 4): with +# given, a record counts only if it is dispatch_verified AND its +# fingerprint equals SHA-256(). A record with no +# fingerprint (written before this field existed) names no dispatch and +# is rejected on that path - fail closed. Without a token, any received +# started counts, as before. # # Never the registry status: a record leaves `pending` when another writer # closes it or another dispatch claims it. Lines are parsed as JSON, never @@ -2247,8 +2254,9 @@ delegate_started_recorded() { local audit="$logs_dir/$job_id/events.ndjson" [ -f "$ev_log" ] || [ -f "$audit" ] || return 1 EV_LOG="$ev_log" EV_AUDIT="$audit" EV_TOK="$tok" python3 -c ' -import json, os, sys +import hashlib, json, os, sys tok = os.environ.get("EV_TOK", "") +fp = hashlib.sha256(tok.encode("utf-8")).hexdigest() if tok else "" def lines(path): if not os.path.isfile(path): @@ -2271,7 +2279,9 @@ for e in lines(os.environ["EV_LOG"]): for e in lines(os.environ["EV_AUDIT"]): if e.get("event") != "received" or e.get("source_event") != "started": continue - if not tok or e.get("dispatch_verified") is True: + if not tok: + sys.exit(0) + if e.get("dispatch_verified") is True and e.get("dispatch_fingerprint") == fp: sys.exit(0) sys.exit(1) ' 2>/dev/null diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index 0fd9998..fa5eea9 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -24,6 +24,7 @@ Exit codes: from __future__ import annotations import argparse +import hashlib import json import logging import queue @@ -279,13 +280,19 @@ class _Watcher: # that survives defensive parsing is recorded here, including ones a # different host published. This is the external-observer record that # backstops the publisher's own "published" line if it never wrote one. - # `dispatch_verified` says this record is attributable to the job's - # current dispatch: the token check above ran (the registry record had - # a dispatch_token and the event's data.dispatch_token matched it) and - # HMAC verification passed before that. The token itself is never - # written - the audit redactor masks token-named keys - so this flag is - # what a local reader (delegate_started_recorded) can trust for a - # remote agent's `started`, which has no line in events.log. + # Which dispatch this record vouches for. `dispatch_verified` says the + # token check above ran (the record had a dispatch_token and the + # event's data.dispatch_token matched it) after HMAC verification; on + # its own it names no dispatch, and this file is appended to across a + # job's attempts, so a reader waiting for a later dispatch would take + # an earlier attempt's record as its acknowledgement (review round 4). + # `dispatch_fingerprint` is SHA-256 of the token that was verified - + # the same identity the local path compares in events.log, minus the + # token itself, which the audit redactor masks by key name and which a + # 128-bit uuid4 nonce is not recoverable from. `attempt_id` is kept for + # readers; it is not the identity, since a token can be re-minted + # within an attempt. A reader that demands a token must match the + # fingerprint; a record without one proves nothing about identity. mqtt_common.append_event(jid, { "event": "received", "source_event": payload.get("event"), @@ -294,6 +301,10 @@ class _Watcher: "timestamp": payload.get("timestamp"), "detail": payload.get("detail", ""), "dispatch_verified": expected_disp_tok is not None, + "dispatch_fingerprint": (hashlib.sha256(expected_disp_tok.encode("utf-8")).hexdigest() + if expected_disp_tok else None), + "attempt_id": (data.get("attempt_id") if isinstance(data, dict) and data.get("attempt_id") is not None + else self.attempts.get(jid)), }) self.events.put((msg.topic, payload)) diff --git a/tests/test_issue23_24_startup_path.py b/tests/test_issue23_24_startup_path.py index 31a1242..e75ac47 100644 --- a/tests/test_issue23_24_startup_path.py +++ b/tests/test_issue23_24_startup_path.py @@ -956,11 +956,17 @@ def test_w5_a_remote_agents_started_counts_only_through_the_subscribers_verified base.update(kw) return json.dumps(base) + "\n" - audit.write_text(rec(dispatch_verified=False)) # untokened record: not attributed + import hashlib + fp = lambda t: hashlib.sha256(t.encode()).hexdigest() + audit.write_text(rec(dispatch_verified=False, dispatch_fingerprint=None)) # untokened: not attributed assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout - audit.write_text(rec(source_event="progress", dispatch_verified=True)) # not a started + audit.write_text(rec(source_event="progress", dispatch_verified=True, dispatch_fingerprint=fp("tok-1"))) + assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout # not a started + audit.write_text(rec(dispatch_verified=True)) # verified, no identity: fail closed assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout - audit.write_text(rec(dispatch_verified=True)) + audit.write_text(rec(dispatch_verified=True, dispatch_fingerprint=fp("tok-OLD"))) # another dispatch's + assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout + audit.write_text(rec(dispatch_verified=True, dispatch_fingerprint=fp("tok-1"))) assert "rc=0" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout @@ -982,6 +988,112 @@ def test_w6_the_subscriber_stamps_dispatch_verified_on_what_it_records(mam_sandb recs = list(mqtt_common.iter_logged_events("job1", str(logs))) started = [r for r in recs if r.get("source_event") == "started"] assert len(started) == 1 and started[0]["dispatch_verified"] is True, recs + import hashlib + assert started[0]["dispatch_fingerprint"] == hashlib.sha256(b"tok-1").hexdigest(), started[0] + assert started[0]["attempt_id"] == 1, started[0] + assert "tok-1" not in json.dumps(started[0]), "the token itself must never reach the audit log" + + +# -------------------------------------------------------------------------- +# review round 4 (planner-reviewer-claude-01 B-1, planner-reviewer-codex-01): +# the remote audit record must name the dispatch it vouches for. Real signed +# producer, real _Watcher.on_message, real audit writer, real re-dispatch +# (registry operator write minting a new token) - no record is edited by hand. +# -------------------------------------------------------------------------- + +def _remote_fixture(tmp_path, monkeypatch): + import test_issue18_liveness_contract as live + jobs = tmp_path / "jobs" + env = live._env(tmp_path, dict(live.UNREACHABLE, MAM_DELEGATE_JOB_NO_MAIN="1", + MAM_JOBS_DIR=str(jobs))) + for key in ("MAM_ENV_FILE", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT"): + monkeypatch.setenv(key, env[key]) + sys.path.insert(0, str(live.SKILLS)) + from lib_py.delegation import job_subscriber, mqtt_common, publish_event, registry + return live, env, jobs, job_subscriber, mqtt_common, publish_event, registry + + +def _receive(publish_event, watcher, jid, token, attempt, seq, secret): + """A signed `started` from the real producer, handed to the real subscriber.""" + import types + payload = publish_event.build_payload(jid, seq, "started", "", + {"dispatch_token": token, "attempt_id": attempt}, secret) + watcher.on_message(None, None, types.SimpleNamespace( + topic="python/mqtt/jobs/%s/events" % jid, payload=json.dumps(payload).encode())) + + +def _both_waits(live, env, tmp_path, jobs, jid, token): + """create's delegate_wait_started and the wrapper's _wait_started, same inputs.""" + import shlex + script = ( + 'source %s\nset +e\nPY=%s\nREGISTRY_DIR=%s\nACK_TIMEOUT=0\nTIMEOUT=60\nSTART_TIME=$(date +%%s)\n' + 'LOGS_DIR_ABS=%s\n_disp_tok=%s\n' + 'delegate_wait_started %s 0 "$_disp_tok"; echo CREATE_RC=$?\n' + '_wait_started %s 0; echo WRAPPER_RC=$?\n' + ) % (shlex.quote(str(live.WRAPPER)), shlex.quote(sys.executable), shlex.quote(str(jobs)), + shlex.quote(env["DELEGATE_JOB_LOGS_DIR"]), shlex.quote(token), jid, jid) + res = subprocess.run(["bash", "-c", script], env=env, cwd=str(tmp_path), + capture_output=True, text=True, timeout=60) + out = res.stdout + res.stderr + c = int(out.split("CREATE_RC=")[1].split()[0]); w = int(out.split("WRAPPER_RC=")[1].split()[0]) + return c, w, out + + +@pytest.mark.parametrize("scenario,expected", [ + ("current", 0), ("old_audit", 1), ("old_audit_and_stale_wire", 1), + ("new_dispatch_received", 0), ("invalid_hmac", 1), +]) +def test_w7_remote_audit_is_bound_to_the_dispatch_being_waited_for(tmp_path, monkeypatch, scenario, expected): + """W7 (codex's five cases, real paths throughout). Attempt 1 under token A + leaves a verified audit record; the operator re-dispatches (attempt 2, + new token B, nothing edited by hand); waiting for B must not be answered + by A's record, even when B's subscriber correctly dropped A's replayed + packet - and B's own verified started must still be accepted. Both waits + must agree.""" + live, env, jobs, job_subscriber, mqtt_common, publish_event, registry = _remote_fixture(tmp_path, monkeypatch) + jid, secret = "auditjob", "test-only-hmac-secret" + registry.register_job("remote audit", job_id=jid, registry_dir=str(jobs), auth_token=secret, + dispatch_token="dispatch-A", transport="remote", + broker={"host": "127.0.0.1", "port": 9, "tls": False}) + w1 = job_subscriber._Watcher({jid}, {jid: secret}, {jid: 0}, {jid: "dispatch-A"}, {jid: 1}) + _receive(publish_event, w1, jid, "dispatch-A", 1, 1, "wrong-secret" if scenario == "invalid_hmac" else secret) + recs = list(mqtt_common.iter_logged_events(jid, env["DELEGATE_JOB_LOGS_DIR"])) + assert sum(r.get("source_event") == "started" for r in recs) == (0 if scenario == "invalid_hmac" else 1), recs + token = "dispatch-A" + if scenario not in ("current", "invalid_hmac"): + job_subscriber._sync_remote_status(jid, "completed", str(jobs), token) + assert registry._operator_write(jid, str(jobs), status="pending", attempt_id=2, iteration=2) == 0 + record = mqtt_common.load_job(jid, str(jobs)) + token = record["dispatch_token"] + assert token != "dispatch-A" and record["attempt_id"] == 2, record + w2 = job_subscriber._Watcher({jid}, {jid: secret}, {jid: 0}, {jid: token}, {jid: 2}) + if scenario == "old_audit_and_stale_wire": + _receive(publish_event, w2, jid, "dispatch-A", 1, 2, secret) # A's packet replayed to B's subscriber + assert w2.events.empty(), "the subscriber must drop the stale dispatch" + elif scenario == "new_dispatch_received": + _receive(publish_event, w2, jid, token, 2, 2, secret) + assert not w2.events.empty() + assert not (jobs / ("%s.events.log" % jid)).exists(), "remote evidence comes only from the subscriber's audit" + c, w, out = _both_waits(live, env, tmp_path, jobs, jid, token) + assert (c, w) == (expected, expected), (scenario, c, w, out) + + +def test_w8_a_record_without_identity_fails_closed_when_a_token_is_demanded(tmp_path, monkeypatch): + """W8: an audit record from before dispatch_fingerprint existed (verified, + no fingerprint) names no dispatch: rejected when a token is demanded, + accepted when none is (an untokened legacy job), exactly as before.""" + live, env, jobs, job_subscriber, mqtt_common, publish_event, registry = _remote_fixture(tmp_path, monkeypatch) + jid = "legacyjob" + registry.register_job("legacy", job_id=jid, registry_dir=str(jobs), auth_token=None, + dispatch_token="dispatch-L", transport="remote", + broker={"host": "127.0.0.1", "port": 9, "tls": False}) + mqtt_common.append_event(jid, {"event": "received", "source_event": "started", "seq": 1, + "topic": "t", "timestamp": "t", "detail": "", + "dispatch_verified": True}, env["DELEGATE_JOB_LOGS_DIR"]) + c, w, out = _both_waits(live, env, tmp_path, jobs, jid, "dispatch-L") + assert (c, w) == (1, 1), ("token demanded: a record with no identity must not count", out) + c, w, out = _both_waits(live, env, tmp_path, jobs, jid, "") + assert (c, w) == (0, 0), ("no token demanded: compatibility path", out) def test_s7n_the_wrappers_wait_started_accepts_the_real_producers_line(mam_sandbox): @@ -1037,7 +1149,7 @@ def test_s7e_a_started_from_another_dispatch_is_not_evidence(mam_sandbox): "--event", "started", "--expect-dispatch-token", "tok-OLD", "--attempt-id", "1"], check=True, capture_output=True, text=True, env=env, timeout=60) rec = jobs / "job1.json" - d = json.loads(rec.read_text()); d["dispatch_token"] = "tok-1"; d["status"] = "pending" + d = json.loads(rec.read_text()); d["dispatch_token"] = "tok-1" # the re-dispatch; status untouched rec.write_text(json.dumps(d)) res, published = _run_delivery(mam_sandbox, 0, seed_events=False) assert published == ["error"], (published, res.stderr)