713 lines
30 KiB
Python
713 lines
30 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import struct
|
|
import zlib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts import build_telegram_visible_unseen_chain_packet as packet_builder
|
|
from scripts import collect_telegram_visible_unseen_chain_state as state_collector
|
|
from scripts import verify_telegram_visible_unseen_chain as verifier
|
|
|
|
SUBJECT_ID = "c0000000-0000-4000-8000-000000000001"
|
|
CHALLENGE_NONCE = "a" * 64
|
|
TURN_CAPTURE_CALL_IDS = ["call_chrome_turn_1", "call_chrome_turn_2", "call_chrome_turn_3"]
|
|
FINAL_CAPTURE_CALL_ID = "call_chrome_final"
|
|
|
|
|
|
def packet_and_path(tmp_path: Path) -> tuple[dict, Path]:
|
|
packet = packet_builder.build_packet(
|
|
handler_receipt_path=packet_builder.DEFAULT_HANDLER_RECEIPT.resolve(),
|
|
git_sha="a" * 40,
|
|
)
|
|
path = tmp_path / "packet.json"
|
|
packet_builder.write_json(path, packet)
|
|
return packet, path
|
|
|
|
|
|
def _finish_state_token(value: dict) -> dict:
|
|
value["state_token_sha256"] = state_collector.derive_state_token(value)
|
|
return value
|
|
|
|
|
|
def preflight(packet: dict, packet_path: Path) -> dict:
|
|
return _finish_state_token(
|
|
{
|
|
"schema": state_collector.SCHEMA,
|
|
"generated_at_utc": "2026-07-14T23:59:00+00:00",
|
|
"phase": "preflight",
|
|
"status": "pass",
|
|
"protocol_sha256": packet["protocol_sha256"],
|
|
"packet_file_sha256": verifier._sha256_file(packet_path),
|
|
"manifest_binding": packet["manifest_binding"],
|
|
"capture_challenge_nonce": CHALLENGE_NONCE,
|
|
"baseline_state_token_sha256": "",
|
|
"subject_ref": "",
|
|
"packet_checks": {"fixture_packet_checks_pass": True},
|
|
"remote_checks": {"fixture_remote_checks_pass": True},
|
|
"baseline_checks": {},
|
|
"remote": {
|
|
"deploy_head": "d" * 40,
|
|
"last_deploy_sha": "d" * 40,
|
|
"fingerprint_after": {"fingerprint_sha256": "f" * 64},
|
|
"service_after": {
|
|
"MainPID": "123",
|
|
"NRestarts": "0",
|
|
"ExecMainStartTimestamp": "Wed 2026-07-15 00:00:00 UTC",
|
|
},
|
|
"subject_readback": None,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
def postflight(packet: dict, packet_path: Path, before: dict) -> dict:
|
|
return _finish_state_token(
|
|
{
|
|
"schema": state_collector.SCHEMA,
|
|
"generated_at_utc": "2026-07-15T00:04:00Z",
|
|
"phase": "postflight",
|
|
"status": "pass",
|
|
"protocol_sha256": packet["protocol_sha256"],
|
|
"packet_file_sha256": verifier._sha256_file(packet_path),
|
|
"manifest_binding": packet["manifest_binding"],
|
|
"capture_challenge_nonce": CHALLENGE_NONCE,
|
|
"baseline_state_token_sha256": before["state_token_sha256"],
|
|
"subject_ref": "c0000000",
|
|
"packet_checks": {"fixture_packet_checks_pass": True},
|
|
"remote_checks": {"fixture_remote_checks_pass": True},
|
|
"baseline_checks": {"fixture_baseline_checks_pass": True},
|
|
"remote": {
|
|
"deploy_head": "d" * 40,
|
|
"last_deploy_sha": "d" * 40,
|
|
"fingerprint_after": {"fingerprint_sha256": "f" * 64},
|
|
"service_after": {
|
|
"MainPID": "123",
|
|
"NRestarts": "0",
|
|
"ExecMainStartTimestamp": "Wed 2026-07-15 00:00:00 UTC",
|
|
},
|
|
"subject_readback": {
|
|
"subject_ref": "c0000000",
|
|
"matches": [
|
|
{
|
|
"table": "public.beliefs",
|
|
"row": {"id": SUBJECT_ID, "statement": "Coordination is the bottleneck."},
|
|
}
|
|
],
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
def write_png(path: Path, rgba: tuple[int, int, int, int]) -> None:
|
|
width = verifier.MIN_SCREENSHOT_WIDTH
|
|
height = verifier.MIN_SCREENSHOT_HEIGHT
|
|
scanline = b"\x00" + bytes(rgba) * width
|
|
pixels = scanline * height
|
|
|
|
def chunk(kind: bytes, payload: bytes) -> bytes:
|
|
return (
|
|
struct.pack(">I", len(payload))
|
|
+ kind
|
|
+ payload
|
|
+ struct.pack(">I", zlib.crc32(kind + payload) & 0xFFFFFFFF)
|
|
)
|
|
|
|
ihdr = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0)
|
|
path.write_bytes(
|
|
b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr) + chunk(b"IDAT", zlib.compress(pixels)) + chunk(b"IEND", b"")
|
|
)
|
|
|
|
|
|
def _artifact_hashes(capture: dict) -> dict:
|
|
source = capture["capture_source"]
|
|
return {
|
|
"final_screenshot": verifier._sha256_file(Path(source["final_screenshot"])),
|
|
"final_accessibility": verifier._sha256_file(Path(source["final_accessibility"])),
|
|
"turn_screenshots": [verifier._sha256_file(Path(item["turn_screenshot"])) for item in capture["messages"]],
|
|
"turn_accessibility": [verifier._sha256_file(Path(item["turn_accessibility"])) for item in capture["messages"]],
|
|
}
|
|
|
|
|
|
def authorization_event(packet: dict, capture: dict) -> dict:
|
|
return {
|
|
"timestamp": capture["operator_authorization_captured_at_utc"],
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [{"type": "input_text", "text": capture["operator_authorization_text"]}],
|
|
"internal_chat_message_metadata_passthrough": {"turn_id": "user-message-20260715-authorization"},
|
|
},
|
|
}
|
|
|
|
|
|
def write_browser_provenance(packet: dict, capture: dict, path: Path) -> str:
|
|
source = capture["capture_source"]
|
|
auth_event = authorization_event(packet, capture)
|
|
auth_event_sha256 = hashlib.sha256(
|
|
json.dumps(auth_event, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
).hexdigest()
|
|
final_call_id = TURN_CAPTURE_CALL_IDS[2] if source["final_capture_reuses_turn_3"] else FINAL_CAPTURE_CALL_ID
|
|
provenance = {
|
|
"schema": verifier.BROWSER_PROVENANCE_SCHEMA,
|
|
"status": "pass",
|
|
"capture_channel": "codex_chrome_extension",
|
|
"capture_challenge_nonce": CHALLENGE_NONCE,
|
|
"protocol_sha256": packet["protocol_sha256"],
|
|
"turn_capture_call_ids": TURN_CAPTURE_CALL_IDS,
|
|
"final_capture_call_id": final_call_id,
|
|
"browser": {
|
|
"browser_type": "extension",
|
|
"browser_id": "chrome-backend-20260715",
|
|
"session_id": "chrome-session-20260715",
|
|
"tab_id": "telegram-tab-20260715",
|
|
"provider_capture_id": "chrome-capture-20260715",
|
|
"page_url": "https://web.telegram.org/k/#-5146042086",
|
|
"page_title": "Leo - Telegram",
|
|
"captured_at_utc": source["captured_at_utc"],
|
|
},
|
|
"authorization": {
|
|
"source": "codex_platform_user_message",
|
|
"thread_id": packet["operator_context"]["codex_thread_id"],
|
|
"user_message_id": "user-message-20260715-authorization",
|
|
"source_receipt_id": auth_event_sha256,
|
|
"text_sha256": hashlib.sha256(capture["operator_authorization_text"].encode("utf-8")).hexdigest(),
|
|
"captured_at_utc": capture["operator_authorization_captured_at_utc"],
|
|
},
|
|
"telegram": {
|
|
"chat_label": "Leo",
|
|
"chat_id": "-5146042086",
|
|
"message_ids": [item["telegram_message_id"] for item in capture["messages"]],
|
|
"reply_ids": [item["reply_message_id"] for item in capture["messages"]],
|
|
},
|
|
"artifacts": _artifact_hashes(capture),
|
|
"checks": {
|
|
"authenticated_chrome_extension_session": True,
|
|
"telegram_web_origin_observed": True,
|
|
"chat_identity_observed_in_browser": True,
|
|
"message_ids_observed_in_browser": True,
|
|
"accessibility_captured_from_browser": True,
|
|
"screenshots_captured_from_browser": True,
|
|
"authorization_read_from_codex_user_message": True,
|
|
},
|
|
}
|
|
path.write_text(json.dumps(provenance, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
return verifier._sha256_file(path)
|
|
|
|
|
|
def write_codex_rollout(packet: dict, capture: dict, tmp_path: Path, provenance_sha256: str) -> Path:
|
|
provenance_path = Path(capture["capture_source"]["browser_provenance"])
|
|
provenance = json.loads(provenance_path.read_text(encoding="utf-8"))
|
|
browser = provenance["browser"]
|
|
artifacts = provenance["artifacts"]
|
|
thread_id = packet["operator_context"]["codex_thread_id"]
|
|
sessions_root = tmp_path.parent / f"{tmp_path.name}-codex-sessions"
|
|
rollout_dir = sessions_root / "2026" / "07" / "15"
|
|
rollout_dir.mkdir(parents=True, exist_ok=True)
|
|
rollout_path = rollout_dir / f"rollout-2026-07-15T00-00-00-{thread_id}.jsonl"
|
|
events = [
|
|
{
|
|
"timestamp": "2026-07-14T23:58:00Z",
|
|
"type": "session_meta",
|
|
"payload": {"id": thread_id},
|
|
},
|
|
authorization_event(packet, capture),
|
|
]
|
|
|
|
def chrome_event(call_id: str, timestamp: str, values: list[str]) -> dict:
|
|
return {
|
|
"timestamp": timestamp,
|
|
"type": "event_msg",
|
|
"payload": {
|
|
"type": "mcp_tool_call_end",
|
|
"call_id": call_id,
|
|
"invocation": {
|
|
"server": "node_repl",
|
|
"tool": "js",
|
|
"arguments": {
|
|
"code": (
|
|
"await tab.screenshot({fullPage:false}); "
|
|
"await tab.playwright.domSnapshot(); await tab.url(); await tab.title();"
|
|
)
|
|
},
|
|
},
|
|
"result": {
|
|
"Ok": {
|
|
"content": [{"type": "text", "text": json.dumps(values)}],
|
|
"isError": False,
|
|
"_meta": {
|
|
"codex/toolSurface": {
|
|
"backend": "chrome",
|
|
"browserId": browser["browser_id"],
|
|
"kind": "browserUse",
|
|
"openTabIds": [browser["tab_id"]],
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
turn_timestamps = ["2026-07-15T00:01:30Z", "2026-07-15T00:02:30Z", "2026-07-15T00:03:30Z"]
|
|
for index, (call_id, timestamp, message) in enumerate(
|
|
zip(TURN_CAPTURE_CALL_IDS, turn_timestamps, capture["messages"], strict=True)
|
|
):
|
|
values = [
|
|
artifacts["turn_screenshots"][index],
|
|
artifacts["turn_accessibility"][index],
|
|
message["telegram_message_id"],
|
|
message["reply_message_id"],
|
|
browser["page_url"],
|
|
browser["page_title"],
|
|
packet["target"]["chat_id"],
|
|
]
|
|
if provenance["final_capture_call_id"] == call_id:
|
|
values.extend(
|
|
[
|
|
artifacts["final_screenshot"],
|
|
artifacts["final_accessibility"],
|
|
provenance_sha256,
|
|
*[item["telegram_message_id"] for item in capture["messages"]],
|
|
*[item["reply_message_id"] for item in capture["messages"]],
|
|
]
|
|
)
|
|
timestamp = capture["capture_source"]["captured_at_utc"]
|
|
events.append(chrome_event(call_id, timestamp, values))
|
|
if provenance["final_capture_call_id"] == FINAL_CAPTURE_CALL_ID:
|
|
events.append(
|
|
chrome_event(
|
|
FINAL_CAPTURE_CALL_ID,
|
|
capture["capture_source"]["captured_at_utc"],
|
|
[
|
|
artifacts["final_screenshot"],
|
|
artifacts["final_accessibility"],
|
|
provenance_sha256,
|
|
*[item["telegram_message_id"] for item in capture["messages"]],
|
|
*[item["reply_message_id"] for item in capture["messages"]],
|
|
],
|
|
)
|
|
)
|
|
rollout_path.write_text(
|
|
"".join(json.dumps(event, sort_keys=True) + "\n" for event in events),
|
|
encoding="utf-8",
|
|
)
|
|
return rollout_path
|
|
|
|
|
|
def refresh_provenance_and_rollout(packet: dict, capture: dict, tmp_path: Path) -> Path:
|
|
provenance_path = Path(capture["capture_source"]["browser_provenance"])
|
|
provenance_sha256 = write_browser_provenance(packet, capture, provenance_path)
|
|
return write_codex_rollout(packet, capture, tmp_path, provenance_sha256)
|
|
|
|
|
|
def good_capture(packet: dict, tmp_path: Path) -> tuple[dict, Path]:
|
|
screenshot = tmp_path / "final.png"
|
|
accessibility = tmp_path / "final.txt"
|
|
write_png(screenshot, (10, 20, 30, 255))
|
|
accessibility.write_text("telegram final accessibility proof", encoding="utf-8")
|
|
replies = [
|
|
(
|
|
"The database public.beliefs object c0000000 is an axiom with confidence 0.85. What supports it in "
|
|
"the DB: nothing supports it; there are zero evidence rows and no external primary source."
|
|
),
|
|
(
|
|
"Claim c0000000 is still the coordination bottleneck belief. Assumption versus observed support: the "
|
|
"priority leap is assumed and the evidence is absent. Falsifier: a contrary case. Proposal A and "
|
|
"Proposal B are staged only for review; nothing applied and neither is live knowledge."
|
|
),
|
|
(
|
|
"Revised proposal: separate the mechanism from the business outcome. What would resolve the "
|
|
"disagreement: a randomized study and its primary source dataset. Keep it pending_review for human "
|
|
"review; approved is not apply and applied_at remains null. This is not an apply."
|
|
),
|
|
]
|
|
colors = [(40, 50, 60, 255), (70, 80, 90, 255), (100, 110, 120, 255)]
|
|
messages = []
|
|
for item, reply, color in zip(packet["exact_messages"], replies, colors, strict=True):
|
|
sequence = item["sequence"]
|
|
turn_screenshot = tmp_path / f"turn-{sequence}.png"
|
|
turn_accessibility = tmp_path / f"turn-{sequence}.txt"
|
|
write_png(turn_screenshot, color)
|
|
turn_accessibility.write_text(
|
|
f"Telegram message: {item['message']}\nLeo reply: {reply}\n",
|
|
encoding="utf-8",
|
|
)
|
|
messages.append(
|
|
{
|
|
"sequence": sequence,
|
|
"prompt_id": item["prompt_id"],
|
|
"sent_text": item["message"],
|
|
"reply": reply,
|
|
"sent_at_utc": f"2026-07-15T00:0{sequence}:00Z",
|
|
"reply_at_utc": f"2026-07-15T00:0{sequence}:20Z",
|
|
"telegram_message_id": f"sent-{sequence}",
|
|
"reply_message_id": f"reply-{sequence}",
|
|
"turn_screenshot": str(turn_screenshot),
|
|
"turn_accessibility": str(turn_accessibility),
|
|
}
|
|
)
|
|
provenance_path = tmp_path / "browser-provenance.json"
|
|
capture = {
|
|
"schema": verifier.CAPTURE_SCHEMA,
|
|
"protocol_sha256": packet["protocol_sha256"],
|
|
"preflight_state_token_sha256": "",
|
|
"operator_authorization_text": packet["explicit_operator_authorization_text"],
|
|
"operator_authorization_captured_at_utc": "2026-07-15T00:00:00Z",
|
|
"extra_messages_sent": 0,
|
|
"capture_source": {
|
|
"platform": "telegram",
|
|
"transport": "authenticated_chrome_telegram_ui",
|
|
"chat_label": "Leo",
|
|
"chat_id": "-5146042086",
|
|
"captured_at_utc": "2026-07-15T00:03:40Z",
|
|
"captured_by": "codex_chrome_extension",
|
|
"final_screenshot": str(screenshot),
|
|
"final_accessibility": str(accessibility),
|
|
"browser_provenance": str(provenance_path),
|
|
"final_capture_reuses_turn_3": False,
|
|
"final_capture_reuse_reason": "",
|
|
},
|
|
"messages": messages,
|
|
}
|
|
rollout_path = refresh_provenance_and_rollout(packet, capture, tmp_path)
|
|
return capture, rollout_path
|
|
|
|
|
|
def verified_fixture(tmp_path: Path) -> tuple[dict, Path, dict, dict, dict, Path]:
|
|
packet, packet_path = packet_and_path(tmp_path)
|
|
before = preflight(packet, packet_path)
|
|
after = postflight(packet, packet_path, before)
|
|
capture, trusted_sha256 = good_capture(packet, tmp_path)
|
|
capture["preflight_state_token_sha256"] = before["state_token_sha256"]
|
|
return packet, packet_path, before, after, capture, trusted_sha256
|
|
|
|
|
|
def run_verifier(
|
|
packet: dict,
|
|
packet_path: Path,
|
|
before: dict,
|
|
after: dict,
|
|
capture: dict,
|
|
rollout_log: Path | None,
|
|
tmp_path: Path,
|
|
) -> dict:
|
|
return verifier.verify_capture(
|
|
packet,
|
|
before,
|
|
capture,
|
|
after,
|
|
packet_path=packet_path,
|
|
evidence_root=tmp_path,
|
|
codex_rollout_log=rollout_log,
|
|
codex_sessions_root=(
|
|
rollout_log.parents[3] if rollout_log is not None else tmp_path.parent / f"{tmp_path.name}-codex-sessions"
|
|
),
|
|
)
|
|
|
|
|
|
def test_no_capture_returns_awaiting_authorization_template(tmp_path: Path) -> None:
|
|
packet, packet_path = packet_and_path(tmp_path)
|
|
before = preflight(packet, packet_path)
|
|
|
|
receipt = verifier.verify_capture(
|
|
packet,
|
|
before,
|
|
None,
|
|
None,
|
|
packet_path=packet_path,
|
|
evidence_root=tmp_path,
|
|
)
|
|
|
|
assert receipt["status"] == "awaiting_action_time_authorization"
|
|
assert receipt["telegram_visible_messages_sent"] is False
|
|
assert receipt["capture_template"]["schema"] == verifier.CAPTURE_SCHEMA
|
|
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:
|
|
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "pass"
|
|
assert receipt["receipt_ready"] is True
|
|
assert receipt["current_tier"] == "T3_live_readonly"
|
|
assert all(receipt["checks"].values())
|
|
assert all(receipt["outcomes"].values())
|
|
assert receipt["selected_subject"]["row_id"] == SUBJECT_ID
|
|
|
|
|
|
def test_self_declared_browser_provenance_without_codex_rollout_cannot_pass_t3(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
packet, packet_path, before, after, capture, _rollout = verified_fixture(tmp_path)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, None, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
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_capture_events_are_chrome_backend"] is False
|
|
|
|
|
|
def test_non_chrome_rollout_capture_events_are_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, rollout = verified_fixture(tmp_path)
|
|
events = [json.loads(line) for line in rollout.read_text(encoding="utf-8").splitlines()]
|
|
for event in events:
|
|
payload = event.get("payload") or {}
|
|
if payload.get("type") == "mcp_tool_call_end":
|
|
payload["result"]["Ok"]["_meta"]["codex/toolSurface"]["backend"] = "computerUse"
|
|
rollout.write_text(
|
|
"".join(json.dumps(event, sort_keys=True) + "\n" for event in events),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["codex_rollout_capture_events_are_chrome_backend"] is False
|
|
|
|
|
|
def test_bot_api_capture_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path)
|
|
capture["capture_source"]["transport"] = "telegram_bot_api"
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["authenticated_chrome_transport_declared"] is False
|
|
assert receipt["checks"]["bot_api_not_substituted"] is False
|
|
|
|
|
|
def test_postflight_fingerprint_drift_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path)
|
|
after["remote"]["fingerprint_after"]["fingerprint_sha256"] = "0" * 64
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["postflight_state_token_derivation_valid"] is False
|
|
assert receipt["checks"]["canonical_fingerprint_unchanged_from_preflight"] is False
|
|
|
|
|
|
def test_message_text_drift_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path)
|
|
capture["messages"][1]["sent_text"] += " extra"
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["sent_text_exact"] is False
|
|
|
|
|
|
def test_sent_and_reply_message_id_reuse_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, _rollout = verified_fixture(tmp_path)
|
|
capture["messages"][0]["reply_message_id"] = capture["messages"][0]["telegram_message_id"]
|
|
rollout = refresh_provenance_and_rollout(packet, capture, tmp_path)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, rollout, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["message_and_reply_ids_present_unique"] is False
|
|
|
|
|
|
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)
|
|
capture["messages"][0]["reply"] += " copied-only suffix"
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
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:
|
|
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path)
|
|
capture["messages"][0]["reply_at_utc"] = "2026-07-15T00:02:30Z"
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["strict_send_reply_turn_order"] is False
|
|
|
|
|
|
def test_after_the_fact_authorization_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path)
|
|
capture["operator_authorization_captured_at_utc"] = "2026-07-15T00:01:10Z"
|
|
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["captured_authorization_before_first_send"] is False
|
|
|
|
|
|
@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:
|
|
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path)
|
|
capture["messages"][0]["sent_at_utc"] = timestamp
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["timestamps_valid_and_timezone_aware"] is False
|
|
|
|
|
|
def test_reused_per_turn_evidence_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path)
|
|
capture["messages"][1]["turn_screenshot"] = capture["messages"][0]["turn_screenshot"]
|
|
capture["messages"][1]["turn_accessibility"] = capture["messages"][0]["turn_accessibility"]
|
|
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["per_turn_screenshot_hashes_distinct_nonempty"] is False
|
|
assert receipt["checks"]["per_turn_accessibility_hashes_distinct_nonempty"] is False
|
|
|
|
|
|
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)
|
|
capture["capture_source"]["final_screenshot"] = capture["messages"][2]["turn_screenshot"]
|
|
capture["capture_source"]["final_accessibility"] = capture["messages"][2]["turn_accessibility"]
|
|
capture["capture_source"]["final_capture_reuses_turn_3"] = True
|
|
capture["capture_source"]["final_capture_reuse_reason"] = (
|
|
"The final Chrome capture is the retained third-turn response state."
|
|
)
|
|
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "pass"
|
|
assert receipt["checks"]["final_capture_reuse_policy_satisfied"] is True
|
|
|
|
|
|
def test_undocumented_final_turn_three_reuse_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path)
|
|
capture["capture_source"]["final_screenshot"] = capture["messages"][2]["turn_screenshot"]
|
|
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["final_capture_reuse_policy_satisfied"] is False
|
|
|
|
|
|
def test_evidence_path_escape_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path)
|
|
outside = tmp_path.parent / "outside-turn.png"
|
|
write_png(outside, (1, 2, 3, 255))
|
|
capture["messages"][0]["turn_screenshot"] = str(outside)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["all_evidence_paths_contained_in_root"] is False
|
|
|
|
|
|
def test_non_png_screenshot_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, _trusted = verified_fixture(tmp_path)
|
|
Path(capture["messages"][0]["turn_screenshot"]).write_bytes(b"not a png")
|
|
trusted = refresh_provenance_and_rollout(packet, capture, tmp_path)
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["screenshots_are_valid_png_with_metadata"] is False
|
|
|
|
|
|
def test_fabricated_local_bundle_without_supported_state_or_trusted_chrome_provenance_fails(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
packet, packet_path = packet_and_path(tmp_path)
|
|
before = {
|
|
"phase": "preflight",
|
|
"status": "pass",
|
|
"generated_at_utc": "2026-07-14T23:59:00Z",
|
|
"protocol_sha256": packet["protocol_sha256"],
|
|
"packet_file_sha256": verifier._sha256_file(packet_path),
|
|
"state_token_sha256": "1" * 64,
|
|
"remote": {"fingerprint_after": {"fingerprint_sha256": "f" * 64}, "service_after": {}},
|
|
}
|
|
after = {
|
|
"phase": "postflight",
|
|
"status": "pass",
|
|
"generated_at_utc": "2026-07-15T00:04:00Z",
|
|
"protocol_sha256": packet["protocol_sha256"],
|
|
"state_token_sha256": "2" * 64,
|
|
"subject_ref": "c0000000",
|
|
"remote": {
|
|
"fingerprint_after": {"fingerprint_sha256": "f" * 64},
|
|
"service_after": {},
|
|
"subject_readback": {"subject_ref": "c0000000", "matches": []},
|
|
},
|
|
}
|
|
fake_png = tmp_path / "fake.png"
|
|
fake_text = tmp_path / "fake.txt"
|
|
fake_png.write_bytes(b"arbitrary non-PNG bytes")
|
|
fake_text.write_text("copied prompts and replies", encoding="utf-8")
|
|
capture = {
|
|
"protocol_sha256": packet["protocol_sha256"],
|
|
"preflight_state_token_sha256": "1" * 64,
|
|
"operator_authorization_text": packet["explicit_operator_authorization_text"],
|
|
"operator_authorization_captured_at_utc": "2026-07-15T00:00:00Z",
|
|
"extra_messages_sent": 0,
|
|
"capture_source": {
|
|
"platform": "telegram",
|
|
"transport": "authenticated_chrome_telegram_ui",
|
|
"chat_label": "Leo",
|
|
"chat_id": "-5146042086",
|
|
"captured_at_utc": "2026-07-15T00:03:40Z",
|
|
"captured_by": "fabricated",
|
|
"final_screenshot": str(fake_png),
|
|
"final_accessibility": str(fake_text),
|
|
"browser_provenance": "",
|
|
"final_capture_reuses_turn_3": False,
|
|
"final_capture_reuse_reason": "",
|
|
},
|
|
"messages": [
|
|
{
|
|
"sequence": item["sequence"],
|
|
"prompt_id": item["prompt_id"],
|
|
"sent_text": item["message"],
|
|
"reply": f"fabricated reply {item['sequence']}",
|
|
"sent_at_utc": f"2026-07-15T00:0{item['sequence']}:00Z",
|
|
"reply_at_utc": f"2026-07-15T00:0{item['sequence']}:20Z",
|
|
"telegram_message_id": f"sent-{item['sequence']}",
|
|
"reply_message_id": f"reply-{item['sequence']}",
|
|
"turn_screenshot": str(fake_png),
|
|
"turn_accessibility": str(fake_text),
|
|
}
|
|
for item in packet["exact_messages"]
|
|
],
|
|
}
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, None, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["current_tier"] != "T3_live_readonly"
|
|
assert receipt["telegram_visible_messages_sent"] is False
|
|
assert receipt["checks"]["capture_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"]["codex_rollout_log_is_trusted_external_file"] 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"]["per_turn_screenshot_hashes_distinct_nonempty"] is False
|
|
|
|
|
|
def test_state_token_tampering_is_rejected(tmp_path: Path) -> None:
|
|
packet, packet_path, before, after, capture, trusted = verified_fixture(tmp_path)
|
|
before = copy.deepcopy(before)
|
|
before["generated_at_utc"] = "2026-07-14T23:58:00Z"
|
|
|
|
receipt = run_verifier(packet, packet_path, before, after, capture, trusted, tmp_path)
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["checks"]["preflight_state_token_derivation_valid"] is False
|