fix: make local Telegram verification preparation-only

This commit is contained in:
twentyOne2x 2026-07-15 06:17:20 +02:00
parent 8f87c540dc
commit 35c9531950
2 changed files with 164 additions and 83 deletions

View file

@ -1,5 +1,5 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Verify retained authenticated-Chrome proof for Leo's unseen Telegram chain.""" """Validate local Telegram capture structure without authenticating live delivery."""
from __future__ import annotations from __future__ import annotations
@ -24,12 +24,15 @@ except ImportError: # pragma: no cover - imported as scripts.* in tests
from scripts import collect_telegram_visible_unseen_chain_state as state_collector from scripts import collect_telegram_visible_unseen_chain_state as state_collector
from scripts import verify_leo_unseen_reasoning_chain as unseen from scripts import verify_leo_unseen_reasoning_chain as unseen
SCHEMA = "livingip.telegramVisibleUnseenChainReceipt.v2" SCHEMA = "livingip.telegramVisibleUnseenChainReceipt.v3"
CAPTURE_SCHEMA = "livingip.telegramVisibleUnseenChainCapture.v1" CAPTURE_SCHEMA = "livingip.telegramVisibleUnseenChainCapture.v1"
BROWSER_PROVENANCE_SCHEMA = "livingip.authenticatedChromeTelegramCaptureProvenance.v1" BROWSER_PROVENANCE_SCHEMA = "livingip.authenticatedChromeTelegramCaptureProvenance.v1"
SHA64_RE = re.compile(r"^[0-9a-f]{64}$") SHA64_RE = re.compile(r"^[0-9a-f]{64}$")
MIN_SCREENSHOT_WIDTH = 320 MIN_SCREENSHOT_WIDTH = 320
MIN_SCREENSHOT_HEIGHT = 200 MIN_SCREENSHOT_HEIGHT = 200
PREPARATION_STATUS = "prepared_external_receipt_required"
PREPARATION_TIER = "T0_local_preparation"
FAILED_PREPARATION_TIER = "T0_failed_preparation"
DEFAULT_CODEX_SESSIONS_ROOT = Path.home() / ".codex" / "sessions" DEFAULT_CODEX_SESSIONS_ROOT = Path.home() / ".codex" / "sessions"
REPORT_DIR = Path("docs/reports/leo-working-state-20260709") REPORT_DIR = Path("docs/reports/leo-working-state-20260709")
DEFAULT_PACKET = packet_builder.DEFAULT_OUT DEFAULT_PACKET = packet_builder.DEFAULT_OUT
@ -57,6 +60,24 @@ def _present(value: Any) -> bool:
return True return True
def _preparation_only_fields(*, preparation_ready: bool) -> dict[str, Any]:
return {
"verification_scope": "caller_writable_preparation_structure",
"preparation_ready": preparation_ready,
"receipt_ready": False,
"current_tier": PREPARATION_TIER if preparation_ready else FAILED_PREPARATION_TIER,
"required_tier": "T3_live_readonly",
"telegram_visible_messages_sent": False,
"live_receipt_boundary": {
"required": True,
"verified": False,
"accepted_by_this_api": False,
"authority": "independently_authenticated_platform_browser_receipt",
"local_inputs_are": "caller_writable_preparation_evidence",
},
}
def blank_capture_template(packet: dict[str, Any], preflight: dict[str, Any]) -> dict[str, Any]: def blank_capture_template(packet: dict[str, Any], preflight: dict[str, Any]) -> dict[str, Any]:
target = packet.get("target") or {} target = packet.get("target") or {}
return { return {
@ -322,7 +343,7 @@ def _browser_provenance_checks(
"browser_provenance_artifact_hashes_exact": provenance.get("artifacts") == artifact_hashes, "browser_provenance_artifact_hashes_exact": provenance.get("artifacts") == artifact_hashes,
"browser_provenance_capture_timestamp_bound": browser.get("captured_at_utc") "browser_provenance_capture_timestamp_bound": browser.get("captured_at_utc")
== capture_source.get("captured_at_utc"), == capture_source.get("captured_at_utc"),
"authorization_provenance_is_independent_codex_user_message": authorization.get("source") "authorization_provenance_claim_has_codex_user_message_shape": authorization.get("source")
== "codex_platform_user_message" == "codex_platform_user_message"
and all( and all(
len(str(authorization.get(key) or "")) >= 8 for key in ("thread_id", "user_message_id", "source_receipt_id") len(str(authorization.get(key) or "")) >= 8 for key in ("thread_id", "user_message_id", "source_receipt_id")
@ -336,7 +357,7 @@ def _browser_provenance_checks(
return checks, provenance return checks, provenance
def _codex_rollout_provenance_checks( def _local_codex_rollout_structure_checks(
rollout_path: Path | None, rollout_path: Path | None,
*, *,
sessions_root: Path, sessions_root: Path,
@ -348,6 +369,8 @@ def _codex_rollout_provenance_checks(
artifact_hashes: dict[str, Any], artifact_hashes: dict[str, Any],
messages: list[dict[str, Any]], messages: list[dict[str, Any]],
) -> tuple[dict[str, bool], dict[str, Any]]: ) -> tuple[dict[str, bool], dict[str, Any]]:
# Codex session JSONL is same-user writable. These checks establish internal
# consistency for external review, never an independently authenticated receipt.
root = sessions_root.resolve() root = sessions_root.resolve()
resolved = rollout_path.resolve() if rollout_path is not None else Path() resolved = rollout_path.resolve() if rollout_path is not None else Path()
contained = bool(rollout_path) and resolved.is_file() and root in resolved.parents contained = bool(rollout_path) and resolved.is_file() and root in resolved.parents
@ -513,9 +536,8 @@ def _codex_rollout_provenance_checks(
authorization_event_at = _parse_aware_timestamp(user_event.get("timestamp")) if user_event else None authorization_event_at = _parse_aware_timestamp(user_event.get("timestamp")) if user_event else None
capture_source = capture.get("capture_source") if isinstance(capture.get("capture_source"), dict) else {} capture_source = capture.get("capture_source") if isinstance(capture.get("capture_source"), dict) else {}
checks = { checks = {
"codex_rollout_log_is_trusted_external_file": contained "codex_rollout_log_has_expected_local_sessions_shape": contained and thread_id in resolved.name,
and outside_evidence_root "codex_rollout_log_is_separate_from_evidence_root": outside_evidence_root,
and thread_id in resolved.name,
"codex_rollout_log_jsonl_valid": jsonl_valid and bool(events), "codex_rollout_log_jsonl_valid": jsonl_valid and bool(events),
"codex_rollout_session_meta_matches_thread": session_meta_ids == [thread_id], "codex_rollout_session_meta_matches_thread": session_meta_ids == [thread_id],
"codex_rollout_has_exact_single_authorization_event": len(matching_user_events) == 1, "codex_rollout_has_exact_single_authorization_event": len(matching_user_events) == 1,
@ -726,7 +748,7 @@ def verify_capture(
packet_path: Path, packet_path: Path,
evidence_root: Path, evidence_root: Path,
codex_rollout_log: Path | None = None, codex_rollout_log: Path | None = None,
codex_sessions_root: Path = DEFAULT_CODEX_SESSIONS_ROOT, codex_sessions_root: Path | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
if capture is None: if capture is None:
packet_integrity = state_collector.packet_checks(packet) packet_integrity = state_collector.packet_checks(packet)
@ -740,12 +762,9 @@ def verify_capture(
return { return {
"schema": SCHEMA, "schema": SCHEMA,
"generated_at_utc": datetime.now(timezone.utc).isoformat(), "generated_at_utc": datetime.now(timezone.utc).isoformat(),
"mode": "telegram_visible_unseen_chain_capture_verifier", "mode": "telegram_visible_unseen_chain_preparation_verifier",
"status": "awaiting_action_time_authorization" if ready else "fail", "status": "awaiting_action_time_authorization" if ready else "fail",
"receipt_ready": False, **_preparation_only_fields(preparation_ready=ready),
"current_tier": "T0_packet_plus_T3_preflight" if ready else "T0_failed_preflight",
"required_tier": "T3_live_readonly",
"telegram_visible_messages_sent": False,
"mutates_db": False, "mutates_db": False,
"packet_path": str(packet_path), "packet_path": str(packet_path),
"preflight_path": str(DEFAULT_PREFLIGHT), "preflight_path": str(DEFAULT_PREFLIGHT),
@ -760,11 +779,12 @@ def verify_capture(
}, },
"claim_ceiling": { "claim_ceiling": {
"proven": ( "proven": (
"The exact packet and live zero-mutation preflight are ready." "The caller-supplied packet and preflight are structurally ready for external review."
if ready if ready
else "No readiness claim; packet or preflight integrity validation failed." else "No readiness claim; packet or preflight integrity validation failed."
), ),
"not_proven": [ "not_proven": [
"independently authenticated platform or browser receipt",
"Telegram-visible message delivery", "Telegram-visible message delivery",
"Telegram-visible Leo replies", "Telegram-visible Leo replies",
"post-send fingerprint equality", "post-send fingerprint equality",
@ -834,9 +854,9 @@ def verify_capture(
messages=messages, messages=messages,
artifact_hashes=artifact_hashes, artifact_hashes=artifact_hashes,
) )
codex_rollout_checks, codex_rollout_evidence = _codex_rollout_provenance_checks( codex_rollout_checks, codex_rollout_evidence = _local_codex_rollout_structure_checks(
codex_rollout_log, codex_rollout_log,
sessions_root=codex_sessions_root, sessions_root=codex_sessions_root or DEFAULT_CODEX_SESSIONS_ROOT,
evidence_root=evidence_root, evidence_root=evidence_root,
packet=packet, packet=packet,
capture=capture, capture=capture,
@ -971,8 +991,8 @@ def verify_capture(
), ),
**semantics, **semantics,
} }
outcomes = { preparation_outcomes = {
"telegram_visible_three_turn_chain": all( "caller_bundle_three_turn_chain_structurally_consistent": all(
checks[key] checks[key]
for key in ( for key in (
"authenticated_chrome_transport_declared", "authenticated_chrome_transport_declared",
@ -999,7 +1019,7 @@ def verify_capture(
"turn_accessibility_binds_exact_messages_and_replies", "turn_accessibility_binds_exact_messages_and_replies",
) )
), ),
"visible_db_grounded_challenge": all( "caller_bundle_db_grounded_challenge_structurally_consistent": all(
checks[key] checks[key]
for key in ( for key in (
"visible_db_object_selected_without_operator_id", "visible_db_object_selected_without_operator_id",
@ -1009,7 +1029,7 @@ def verify_capture(
"independent_db_object_matches_visible_subject", "independent_db_object_matches_visible_subject",
) )
), ),
"visible_review_only_revision": all( "caller_bundle_review_only_revision_structurally_consistent": all(
checks[key] checks[key]
for key in ( for key in (
"multiple_review_only_candidates_proposed", "multiple_review_only_candidates_proposed",
@ -1019,24 +1039,23 @@ def verify_capture(
"reply_claimed_no_mutation", "reply_claimed_no_mutation",
) )
), ),
"canonical_state_unchanged": checks["canonical_fingerprint_unchanged_from_preflight"], "caller_bundle_canonical_state_claim_structurally_consistent": checks[
"canonical_fingerprint_unchanged_from_preflight"
],
} }
passed = all(checks.values()) and all(outcomes.values()) preparation_ready = all(checks.values()) and all(preparation_outcomes.values())
return { return {
"schema": SCHEMA, "schema": SCHEMA,
"generated_at_utc": datetime.now(timezone.utc).isoformat(), "generated_at_utc": datetime.now(timezone.utc).isoformat(),
"mode": "telegram_visible_unseen_chain_capture_verifier", "mode": "telegram_visible_unseen_chain_preparation_verifier",
"status": "pass" if passed else "fail", "status": PREPARATION_STATUS if preparation_ready else "fail",
"receipt_ready": passed, **_preparation_only_fields(preparation_ready=preparation_ready),
"current_tier": "T3_live_readonly" if passed else "T2_or_lower_incomplete_visible_proof", "capture_bundle_present": True,
"required_tier": "T3_live_readonly",
"telegram_visible_messages_sent": passed,
"capture_claimed_messages_sent": True,
"production_apply_executed": False, "production_apply_executed": False,
"mutates_db": False, "mutates_db": False,
"packet_path": str(packet_path), "packet_path": str(packet_path),
"checks": checks, "checks": checks,
"outcomes": outcomes, "preparation_outcomes": preparation_outcomes,
"selected_subject": { "selected_subject": {
"ref": subject_ref, "ref": subject_ref,
"table": subject_match.get("table"), "table": subject_match.get("table"),
@ -1060,13 +1079,15 @@ def verify_capture(
}, },
"claim_ceiling": { "claim_ceiling": {
"proven": ( "proven": (
"One exact authenticated-Chrome Telegram chain visibly selected and challenged a DB object, " "The caller-supplied capture bundle is structurally consistent, including its chronology, evidence, "
"proposed and revised review-only knowledge, preserved the approval/apply boundary, and left the " "semantic, and unchanged-state claims. It is ready for an independently authenticated platform or "
"canonical fingerprint unchanged." "browser receipt outside this verifier; no live activity is authenticated here."
if passed if preparation_ready
else "No T3 claim; at least one visible, semantic, binding, or state check failed." else "No preparation claim; at least one structural, semantic, binding, or state check failed."
), ),
"not_proven": [ "not_proven": [
"T3 live-readonly Telegram proof",
"Telegram-visible message delivery or replies",
"canonical proposal staging or apply", "canonical proposal staging or apply",
"agent-to-agent review", "agent-to-agent review",
"GCP runtime parity for the Telegram-visible chain", "GCP runtime parity for the Telegram-visible chain",
@ -1086,10 +1107,12 @@ def write_markdown(path: Path, data: dict[str, Any]) -> None:
"", "",
f"Generated UTC: `{data['generated_at_utc']}`", f"Generated UTC: `{data['generated_at_utc']}`",
f"Status: `{data['status']}`", f"Status: `{data['status']}`",
f"Preparation ready: `{data['preparation_ready']}`",
f"Receipt ready: `{data['receipt_ready']}`", f"Receipt ready: `{data['receipt_ready']}`",
f"Current tier: `{data['current_tier']}`", f"Current tier: `{data['current_tier']}`",
f"Required tier: `{data['required_tier']}`", f"Required tier: `{data['required_tier']}`",
f"Telegram-visible messages sent: `{data['telegram_visible_messages_sent']}`", f"Telegram-visible messages sent: `{data['telegram_visible_messages_sent']}`",
f"Independent live receipt verified: `{data['live_receipt_boundary']['verified']}`",
f"Mutates DB: `{data['mutates_db']}`", f"Mutates DB: `{data['mutates_db']}`",
"", "",
"## Checks", "## Checks",
@ -1102,12 +1125,17 @@ def write_markdown(path: Path, data: dict[str, Any]) -> None:
"", "",
"## Awaiting Authorized Capture", "## Awaiting Authorized Capture",
"", "",
"The packet and preflight are ready. No Telegram message has been sent by this verifier.", "The packet and preflight are structurally ready. No Telegram message has been sent by this verifier.",
"A future T3 receipt also requires the exact platform authorization event and Chrome-backend capture receipts from the bound Codex rollout log.", "A future T3 receipt requires an independently authenticated platform/browser authority outside this verifier.",
] ]
) )
lines.extend( lines.extend(
[ [
"",
"## Trust Boundary",
"",
"Local files, capture artifacts, and Codex rollout JSONL are caller-writable preparation evidence only. "
"This verifier cannot accept or mint a live receipt.",
"", "",
"## Claim Ceiling", "## Claim Ceiling",
"", "",
@ -1129,7 +1157,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser.add_argument( parser.add_argument(
"--codex-rollout-log", "--codex-rollout-log",
type=Path, type=Path,
help="Platform rollout JSONL containing the exact user authorization and Chrome capture tool receipts", help="Local Codex rollout JSONL for preparation-structure checks; this never authenticates a live receipt",
) )
parser.add_argument("--out", type=Path, default=DEFAULT_OUT) parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT) parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT)
@ -1165,7 +1193,7 @@ def main(argv: list[str] | None = None) -> int:
sort_keys=True, sort_keys=True,
) )
) )
return 0 if receipt["status"] in {"pass", "awaiting_action_time_authorization"} else 2 return 0 if receipt["status"] in {PREPARATION_STATUS, "awaiting_action_time_authorization"} else 2
if __name__ == "__main__": if __name__ == "__main__":

View file

@ -383,9 +383,9 @@ def verified_fixture(tmp_path: Path) -> tuple[dict, Path, dict, dict, dict, Path
packet, packet_path = packet_and_path(tmp_path) packet, packet_path = packet_and_path(tmp_path)
before = preflight(packet, packet_path) before = preflight(packet, packet_path)
after = postflight(packet, packet_path, before) after = postflight(packet, packet_path, before)
capture, trusted_sha256 = good_capture(packet, tmp_path) capture, rollout = good_capture(packet, tmp_path)
capture["preflight_state_token_sha256"] = before["state_token_sha256"] capture["preflight_state_token_sha256"] = before["state_token_sha256"]
return packet, packet_path, before, after, capture, trusted_sha256 return packet, packet_path, before, after, capture, rollout
def run_verifier( def run_verifier(
@ -425,22 +425,72 @@ def test_no_capture_returns_awaiting_authorization_template(tmp_path: Path) -> N
) )
assert receipt["status"] == "awaiting_action_time_authorization" assert receipt["status"] == "awaiting_action_time_authorization"
assert receipt["preparation_ready"] is True
assert receipt["receipt_ready"] is False
assert receipt["current_tier"] == verifier.PREPARATION_TIER
assert receipt["telegram_visible_messages_sent"] is False assert receipt["telegram_visible_messages_sent"] is False
assert receipt["live_receipt_boundary"]["accepted_by_this_api"] is False
assert receipt["live_receipt_boundary"]["verified"] is False
assert receipt["capture_template"]["schema"] == verifier.CAPTURE_SCHEMA assert receipt["capture_template"]["schema"] == verifier.CAPTURE_SCHEMA
assert receipt["capture_template"]["messages"][0]["prompt_id"] == "OOS-CHAIN-01" assert receipt["capture_template"]["messages"][0]["prompt_id"] == "OOS-CHAIN-01"
def test_good_chrome_capture_and_unchanged_fingerprint_pass_t3(tmp_path: Path) -> None: def test_public_verifier_keeps_fully_well_formed_all_local_forgery_preparation_only(
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
_packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
preflight_path = tmp_path / "preflight.json"
capture_path = tmp_path / "capture.json"
postflight_path = tmp_path / "postflight.json"
out_path = tmp_path / "receipt.json"
markdown_path = tmp_path / "receipt.md"
verifier.write_json(preflight_path, before)
verifier.write_json(capture_path, capture)
verifier.write_json(postflight_path, after)
monkeypatch.setattr(verifier, "DEFAULT_CODEX_SESSIONS_ROOT", rollout.parents[3])
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) exit_code = verifier.main(
[
"--packet",
str(packet_path),
"--preflight",
str(preflight_path),
"--capture-json",
str(capture_path),
"--postflight",
str(postflight_path),
"--evidence-root",
str(tmp_path),
"--codex-rollout-log",
str(rollout),
"--out",
str(out_path),
"--markdown-out",
str(markdown_path),
]
)
receipt = json.loads(out_path.read_text(encoding="utf-8"))
assert receipt["status"] == "pass" assert exit_code == 0
assert receipt["receipt_ready"] is True assert receipt["status"] == verifier.PREPARATION_STATUS
assert receipt["current_tier"] == "T3_live_readonly" assert receipt["verification_scope"] == "caller_writable_preparation_structure"
assert receipt["preparation_ready"] is True
assert receipt["receipt_ready"] is False
assert receipt["current_tier"] == verifier.PREPARATION_TIER
assert receipt["telegram_visible_messages_sent"] is False
assert receipt["live_receipt_boundary"] == {
"accepted_by_this_api": False,
"authority": "independently_authenticated_platform_browser_receipt",
"local_inputs_are": "caller_writable_preparation_evidence",
"required": True,
"verified": False,
}
assert all(receipt["checks"].values()) assert all(receipt["checks"].values())
assert all(receipt["outcomes"].values()) assert all(receipt["preparation_outcomes"].values())
assert receipt["selected_subject"]["row_id"] == SUBJECT_ID assert receipt["selected_subject"]["row_id"] == SUBJECT_ID
assert "T3_live_readonly" not in receipt["current_tier"]
assert "caller-writable preparation evidence only" in markdown_path.read_text(encoding="utf-8")
def test_self_declared_browser_provenance_without_codex_rollout_cannot_pass_t3( def test_self_declared_browser_provenance_without_codex_rollout_cannot_pass_t3(
@ -452,7 +502,7 @@ def test_self_declared_browser_provenance_without_codex_rollout_cannot_pass_t3(
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["telegram_visible_messages_sent"] is False assert receipt["telegram_visible_messages_sent"] is False
assert receipt["checks"]["codex_rollout_log_is_trusted_external_file"] is False assert receipt["checks"]["codex_rollout_log_has_expected_local_sessions_shape"] is False
assert receipt["checks"]["codex_rollout_capture_events_are_chrome_backend"] is False assert receipt["checks"]["codex_rollout_capture_events_are_chrome_backend"] is False
@ -475,10 +525,10 @@ def test_non_chrome_rollout_capture_events_are_rejected(tmp_path: Path) -> None:
def test_bot_api_capture_is_rejected(tmp_path: Path) -> None: def test_bot_api_capture_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
capture["capture_source"]["transport"] = "telegram_bot_api" capture["capture_source"]["transport"] = "telegram_bot_api"
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["authenticated_chrome_transport_declared"] is False assert receipt["checks"]["authenticated_chrome_transport_declared"] is False
@ -486,10 +536,10 @@ def test_bot_api_capture_is_rejected(tmp_path: Path) -> None:
def test_postflight_fingerprint_drift_is_rejected(tmp_path: Path) -> None: def test_postflight_fingerprint_drift_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
after["remote"]["fingerprint_after"]["fingerprint_sha256"] = "0" * 64 after["remote"]["fingerprint_after"]["fingerprint_sha256"] = "0" * 64
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["postflight_state_token_derivation_valid"] is False assert receipt["checks"]["postflight_state_token_derivation_valid"] is False
@ -497,10 +547,10 @@ def test_postflight_fingerprint_drift_is_rejected(tmp_path: Path) -> None:
def test_message_text_drift_is_rejected(tmp_path: Path) -> None: def test_message_text_drift_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
capture["messages"][1]["sent_text"] += " extra" capture["messages"][1]["sent_text"] += " extra"
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["sent_text_exact"] is False assert receipt["checks"]["sent_text_exact"] is False
@ -518,31 +568,31 @@ def test_sent_and_reply_message_id_reuse_is_rejected(tmp_path: Path) -> None:
def test_copied_reply_not_present_in_accessibility_is_rejected(tmp_path: Path) -> None: def test_copied_reply_not_present_in_accessibility_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
capture["messages"][0]["reply"] += " copied-only suffix" capture["messages"][0]["reply"] += " copied-only suffix"
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["turn_accessibility_binds_exact_messages_and_replies"] is False assert receipt["checks"]["turn_accessibility_binds_exact_messages_and_replies"] is False
def test_all_prompts_sent_before_replies_is_rejected(tmp_path: Path) -> None: def test_all_prompts_sent_before_replies_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
capture["messages"][0]["reply_at_utc"] = "2026-07-15T00:02:30Z" capture["messages"][0]["reply_at_utc"] = "2026-07-15T00:02:30Z"
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["strict_send_reply_turn_order"] is False assert receipt["checks"]["strict_send_reply_turn_order"] is False
def test_after_the_fact_authorization_is_rejected(tmp_path: Path) -> None: def test_after_the_fact_authorization_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, _rollout = verified_fixture(tmp_path)
capture["operator_authorization_captured_at_utc"] = "2026-07-15T00:01:10Z" capture["operator_authorization_captured_at_utc"] = "2026-07-15T00:01:10Z"
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path) rollout = refresh_provenance_and_rollout(packet, capture, tmp_path)
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["captured_authorization_before_first_send"] is False assert receipt["checks"]["captured_authorization_before_first_send"] is False
@ -550,22 +600,22 @@ def test_after_the_fact_authorization_is_rejected(tmp_path: Path) -> None:
@pytest.mark.parametrize("timestamp", ["2026-07-15T00:01:00", "not-a-timestamp"]) @pytest.mark.parametrize("timestamp", ["2026-07-15T00:01:00", "not-a-timestamp"])
def test_naive_or_malformed_timestamps_are_rejected(tmp_path: Path, timestamp: str) -> None: def test_naive_or_malformed_timestamps_are_rejected(tmp_path: Path, timestamp: str) -> None:
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
capture["messages"][0]["sent_at_utc"] = timestamp capture["messages"][0]["sent_at_utc"] = timestamp
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["timestamps_valid_and_timezone_aware"] is False assert receipt["checks"]["timestamps_valid_and_timezone_aware"] is False
def test_reused_per_turn_evidence_is_rejected(tmp_path: Path) -> None: def test_reused_per_turn_evidence_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, _rollout = verified_fixture(tmp_path)
capture["messages"][1]["turn_screenshot"] = capture["messages"][0]["turn_screenshot"] capture["messages"][1]["turn_screenshot"] = capture["messages"][0]["turn_screenshot"]
capture["messages"][1]["turn_accessibility"] = capture["messages"][0]["turn_accessibility"] capture["messages"][1]["turn_accessibility"] = capture["messages"][0]["turn_accessibility"]
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path) rollout = refresh_provenance_and_rollout(packet, capture, tmp_path)
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["per_turn_screenshot_hashes_distinct_nonempty"] is False assert receipt["checks"]["per_turn_screenshot_hashes_distinct_nonempty"] is False
@ -573,56 +623,59 @@ def test_reused_per_turn_evidence_is_rejected(tmp_path: Path) -> None:
def test_final_capture_may_reuse_turn_three_when_documented(tmp_path: Path) -> None: def test_final_capture_may_reuse_turn_three_when_documented(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, _rollout = verified_fixture(tmp_path)
capture["capture_source"]["final_screenshot"] = capture["messages"][2]["turn_screenshot"] capture["capture_source"]["final_screenshot"] = capture["messages"][2]["turn_screenshot"]
capture["capture_source"]["final_accessibility"] = capture["messages"][2]["turn_accessibility"] capture["capture_source"]["final_accessibility"] = capture["messages"][2]["turn_accessibility"]
capture["capture_source"]["final_capture_reuses_turn_3"] = True capture["capture_source"]["final_capture_reuses_turn_3"] = True
capture["capture_source"]["final_capture_reuse_reason"] = ( capture["capture_source"]["final_capture_reuse_reason"] = (
"The final Chrome capture is the retained third-turn response state." "The final Chrome capture is the retained third-turn response state."
) )
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path) rollout = refresh_provenance_and_rollout(packet, capture, tmp_path)
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "pass" assert receipt["status"] == verifier.PREPARATION_STATUS
assert receipt["preparation_ready"] is True
assert receipt["receipt_ready"] is False
assert receipt["telegram_visible_messages_sent"] is False
assert receipt["checks"]["final_capture_reuse_policy_satisfied"] is True assert receipt["checks"]["final_capture_reuse_policy_satisfied"] is True
def test_undocumented_final_turn_three_reuse_is_rejected(tmp_path: Path) -> None: def test_undocumented_final_turn_three_reuse_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, _rollout = verified_fixture(tmp_path)
capture["capture_source"]["final_screenshot"] = capture["messages"][2]["turn_screenshot"] capture["capture_source"]["final_screenshot"] = capture["messages"][2]["turn_screenshot"]
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path) rollout = refresh_provenance_and_rollout(packet, capture, tmp_path)
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["final_capture_reuse_policy_satisfied"] is False assert receipt["checks"]["final_capture_reuse_policy_satisfied"] is False
def test_evidence_path_escape_is_rejected(tmp_path: Path) -> None: def test_evidence_path_escape_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
outside = tmp_path.parent / "outside-turn.png" outside = tmp_path.parent / "outside-turn.png"
write_png(outside, (1, 2, 3, 255)) write_png(outside, (1, 2, 3, 255))
capture["messages"][0]["turn_screenshot"] = str(outside) capture["messages"][0]["turn_screenshot"] = str(outside)
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["all_evidence_paths_contained_in_root"] is False assert receipt["checks"]["all_evidence_paths_contained_in_root"] is False
def test_non_png_screenshot_is_rejected(tmp_path: Path) -> None: def test_non_png_screenshot_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, _rollout = verified_fixture(tmp_path)
Path(capture["messages"][0]["turn_screenshot"]).write_bytes(b"not a png") Path(capture["messages"][0]["turn_screenshot"]).write_bytes(b"not a png")
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path) rollout = refresh_provenance_and_rollout(packet, capture, tmp_path)
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["screenshots_are_valid_png_with_metadata"] is False assert receipt["checks"]["screenshots_are_valid_png_with_metadata"] is False
def test_fabricated_local_bundle_without_supported_state_or_trusted_chrome_provenance_fails( def test_malformed_local_bundle_without_supported_state_or_rollout_structure_fails(
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
packet, packet_path = packet_and_path(tmp_path) packet, packet_path = packet_and_path(tmp_path)
@ -696,18 +749,18 @@ def test_fabricated_local_bundle_without_supported_state_or_trusted_chrome_prove
assert receipt["checks"]["capture_schema_supported"] is False assert receipt["checks"]["capture_schema_supported"] is False
assert receipt["checks"]["preflight_schema_supported"] is False assert receipt["checks"]["preflight_schema_supported"] is False
assert receipt["checks"]["preflight_state_token_derivation_valid"] is False assert receipt["checks"]["preflight_state_token_derivation_valid"] is False
assert receipt["checks"]["codex_rollout_log_is_trusted_external_file"] is False assert receipt["checks"]["codex_rollout_log_has_expected_local_sessions_shape"] is False
assert receipt["checks"]["codex_rollout_has_exact_single_authorization_event"] is False assert receipt["checks"]["codex_rollout_has_exact_single_authorization_event"] is False
assert receipt["checks"]["screenshots_are_valid_png_with_metadata"] is False assert receipt["checks"]["screenshots_are_valid_png_with_metadata"] is False
assert receipt["checks"]["per_turn_screenshot_hashes_distinct_nonempty"] is False assert receipt["checks"]["per_turn_screenshot_hashes_distinct_nonempty"] is False
def test_state_token_tampering_is_rejected(tmp_path: Path) -> None: def test_state_token_tampering_is_rejected(tmp_path: Path) -> None:
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path) packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
before = copy.deepcopy(before) before = copy.deepcopy(before)
before["generated_at_utc"] = "2026-07-14T23:58:00Z" before["generated_at_utc"] = "2026-07-14T23:58:00Z"
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path) receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
assert receipt["status"] == "fail" assert receipt["status"] == "fail"
assert receipt["checks"]["preflight_state_token_derivation_valid"] is False assert receipt["checks"]["preflight_state_token_derivation_valid"] is False