#!/usr/bin/env python3 """Controlled HTTP connector/action experiment. Standard library only; loopback only. No model, MCP, SMTP, external service or real human review is used. Bearer credentials and one-use approval capabilities exist in memory only. The driver exercises a separate reviewer credential; it does not impersonate a real reviewer. """ import argparse import copy import hashlib import json import platform import secrets import socket import sqlite3 import sys import threading import time import urllib.error import urllib.request import uuid from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent REPEATS = 3 CASES = [ {"id": "01-no-auth", "question": "Does an unauthenticated connector read fail?", "statuses": [401], "drafts": 0, "outbox": 0}, {"id": "02-read", "question": "Can read scope retrieve the synthetic ticket without a write?", "statuses": [200], "drafts": 0, "outbox": 0}, {"id": "03-read-cannot-send", "question": "Can read scope invoke send?", "statuses": [403], "drafts": 0, "outbox": 0}, {"id": "04-draft-cannot-send", "question": "Can draft scope save a draft but not send?", "statuses": [201, 403], "drafts": 1, "outbox": 0}, {"id": "05-send-without-approval", "question": "Does send scope alone bypass the backend approval boundary?", "statuses": [403], "drafts": 0, "outbox": 0}, {"id": "06-exact-approval-and-replay", "question": "Does a separately issued exact approval allow one send and reject replay?", "statuses": [201, 201, 409], "drafts": 0, "outbox": 1}, {"id": "07-changed-payload", "question": "Does changed content fail without consuming the exact approved capability?", "statuses": [201, 403, 201], "drafts": 0, "outbox": 1}, {"id": "08-direct-backend-bypass", "question": "Do connector client credentials fail against direct backend read/write?", "statuses": [401, 401], "drafts": 0, "outbox": 0}, {"id": "09-ticket-content-is-not-authority", "question": "Does constructed instruction text itself grant the read client send scope? No model interpretation is tested.", "statuses": [200, 403], "drafts": 0, "outbox": 0}, {"id": "10-client-cannot-approve", "question": "Can a send client issue its own backend approval or forge a capability?", "statuses": [401, 403], "drafts": 0, "outbox": 0} ] def now(): return datetime.now(timezone.utc).isoformat(timespec="microseconds") def canonical(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") def sha(value): return hashlib.sha256(value).hexdigest() def write_json(path, value): path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") class State: def __init__(self, db_path, fixtures): self.lock = threading.RLock() self.db = sqlite3.connect(db_path, check_same_thread=False) self.db.row_factory = sqlite3.Row self.db.executescript(""" CREATE TABLE active_drafts(id INTEGER PRIMARY KEY,case_id TEXT,ticket_id TEXT,recipient TEXT,body TEXT,payload_sha256 TEXT,created_at TEXT); CREATE TABLE active_outbox(id INTEGER PRIMARY KEY,case_id TEXT,ticket_id TEXT,recipient TEXT,body TEXT,payload_sha256 TEXT,created_at TEXT); CREATE TABLE draft_archive(id INTEGER PRIMARY KEY,case_id TEXT,ticket_id TEXT,recipient TEXT,body TEXT,payload_sha256 TEXT,created_at TEXT); CREATE TABLE outbox_archive(id INTEGER PRIMARY KEY,case_id TEXT,ticket_id TEXT,recipient TEXT,body TEXT,payload_sha256 TEXT,created_at TEXT); CREATE TABLE approval_events(id INTEGER PRIMARY KEY,case_id TEXT,event TEXT,payload_sha256 TEXT,reviewer TEXT,created_at TEXT); CREATE TABLE resets(id INTEGER PRIMARY KEY,case_id TEXT,drafts_before INTEGER,outbox_before INTEGER,capabilities_cleared INTEGER,created_at TEXT); """) self.fixtures = {t["id"]: t for t in fixtures["tickets"]} self.credentials = {key: secrets.token_urlsafe(32) for key in ("client_read", "client_draft", "client_send", "backend_read", "backend_draft", "backend_send", "reviewer")} self.secret_values = list(self.credentials.values()) self.approvals = {} self.case = "not-started" def role(self, authorization): if not authorization or not authorization.startswith("Bearer "): return None value = authorization[7:] return next((name for name, token in self.credentials.items() if secrets.compare_digest(value, token)), None) def reset(self, case): with self.lock: before = self.snapshot() cleared = len(self.approvals) self.case = case self.approvals.clear() self.db.execute("DELETE FROM active_drafts") self.db.execute("DELETE FROM active_outbox") self.db.execute("INSERT INTO resets(case_id,drafts_before,outbox_before,capabilities_cleared,created_at) VALUES(?,?,?,?,?)", (case, before["drafts"], before["outbox"], cleared, now())) self.db.commit() return {"at": now(), "before": before, "capabilities_cleared": cleared, "after": self.snapshot()} def snapshot(self): with self.lock: return {"drafts": self.db.execute("SELECT COUNT(*) FROM active_drafts").fetchone()[0], "outbox": self.db.execute("SELECT COUNT(*) FROM active_outbox").fetchone()[0], "outbox_rows": [dict(r) for r in self.db.execute("SELECT * FROM active_outbox ORDER BY id")], "draft_rows": [dict(r) for r in self.db.execute("SELECT * FROM active_drafts ORDER BY id")]} def validate_message(self, payload): if not isinstance(payload, dict) or set(payload) != {"ticket_id", "recipient", "body"}: return False ticket = self.fixtures.get(payload.get("ticket_id")) return bool(ticket and payload.get("recipient") == ticket["recipient"] and isinstance(payload.get("body"), str) and 0 < len(payload["body"]) <= 2000) def record_action(self, kind, payload): stamp = now() args = (self.case, payload["ticket_id"], payload["recipient"], payload["body"], sha(canonical(payload)), stamp) archive = "draft_archive" if kind == "drafts" else "outbox_archive" active = "active_drafts" if kind == "drafts" else "active_outbox" self.db.execute(f"INSERT INTO {archive}(case_id,ticket_id,recipient,body,payload_sha256,created_at) VALUES(?,?,?,?,?,?)", args) self.db.execute(f"INSERT INTO {active}(case_id,ticket_id,recipient,body,payload_sha256,created_at) VALUES(?,?,?,?,?,?)", args) self.db.commit() def approval(self, payload): capability = secrets.token_urlsafe(32) self.secret_values.append(capability) digest = sha(canonical(payload)) self.approvals[capability] = {"digest": digest, "used": False, "case": self.case} self.db.execute("INSERT INTO approval_events(case_id,event,payload_sha256,reviewer,created_at) VALUES(?,?,?,?,?)", (self.case, "issued", digest, "scripted-reviewer-separate-credential", now())) self.db.commit() return capability def request(base, path, method="GET", payload=None, bearer=None, approval=None): headers = {"Accept": "application/json"} if bearer: headers["Authorization"] = "Bearer " + bearer if approval: headers["X-Action-Approval"] = approval data = None if payload is None else canonical(payload) if data is not None: headers["Content-Type"] = "application/json" req = urllib.request.Request(base + path, data=data, headers=headers, method=method) # Do not use environment proxies; both destinations are controlled loopback. opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) try: response = opener.open(req, timeout=5) except urllib.error.HTTPError as error: response = error with response: return response.status, json.loads(response.read().decode("utf-8")) class Handler(BaseHTTPRequestHandler): server_version = "TokenwerkSyntheticLab/1" def log_message(self, *args): pass # Never log HTTP headers or capability values. def answer(self, status, payload): content = canonical(payload) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) def do_GET(self): self.handle_request() def do_POST(self): self.handle_request() def handle_request(self): try: length = int(self.headers.get("Content-Length", "0")) if length < 0 or length > 16384: return self.answer(413, {"error": "request_too_large"}) payload = json.loads(self.rfile.read(length).decode("utf-8")) if length else None status, result = self.route(payload) self.answer(status, result) except (ValueError, UnicodeError): self.answer(400, {"error": "invalid_request"}) except Exception: self.answer(500, {"error": "internal_lab_error"}) def backend_handler(state): class Backend(Handler): def route(self, payload): role = state.role(self.headers.get("Authorization")) with state.lock: if self.command == "GET" and self.path.startswith("/tickets/"): if role != "backend_read": return 401, {"error": "backend_auth_required"} ticket = state.fixtures.get(self.path.rsplit("/", 1)[-1]) return (200, {"ticket": ticket}) if ticket else (404, {"error": "not_found"}) if self.command == "POST" and self.path in ("/drafts", "/outbox", "/approvals"): required = {"/drafts": "backend_draft", "/outbox": "backend_send", "/approvals": "reviewer"}[self.path] if role != required: return 401, {"error": "backend_auth_required"} if not state.validate_message(payload): return 400, {"error": "invalid_synthetic_message"} if self.path == "/approvals": return 201, {"status": "approval_created", "approval": state.approval(payload)} if self.path == "/drafts": state.record_action("drafts", payload) return 201, {"status": "draft_saved", "payload_sha256": sha(canonical(payload))} capability = self.headers.get("X-Action-Approval") allowed = state.approvals.get(capability) if not allowed: return 403, {"error": "approval_required"} if allowed["used"]: return 409, {"error": "approval_already_used"} if allowed["case"] != state.case or allowed["digest"] != sha(canonical(payload)): return 403, {"error": "approval_payload_mismatch"} state.record_action("outbox", payload) allowed["used"] = True state.db.execute("INSERT INTO approval_events(case_id,event,payload_sha256,reviewer,created_at) VALUES(?,?,?,?,?)", (state.case, "consumed", allowed["digest"], "scripted-reviewer-separate-credential", now())) state.db.commit() return 201, {"status": "synthetic_outbox_row_created", "payload_sha256": allowed["digest"]} return 404, {"error": "not_found"} return Backend def connector_handler(state, backend_url): scopes = {"client_read": {"read"}, "client_draft": {"read", "draft"}, "client_send": {"read", "draft", "send"}} class Connector(Handler): def route(self, payload): role = state.role(self.headers.get("Authorization")) if role not in scopes: return 401, {"error": "connector_auth_required"} if self.command == "GET" and self.path.startswith("/tickets/"): return request(backend_url, self.path, bearer=state.credentials["backend_read"]) route = {"/drafts": ("draft", "backend_draft", "/drafts"), "/send": ("send", "backend_send", "/outbox")}.get(self.path) if self.command != "POST" or route is None: return 404, {"error": "not_found"} scope, credential, destination = route if scope not in scopes[role]: return 403, {"error": "connector_scope_missing", "required_scope": scope} return request(backend_url, destination, "POST", payload, state.credentials[credential], self.headers.get("X-Action-Approval")) return Connector def freeze(): path = ROOT / "frozen-design.json" if path.exists(): raise SystemExit("Frozen design already exists; refusing to overwrite.") design = {"schema_version": 1, "frozen_at": now(), "protocol": "HTTP/1.0 with JSON over two separate loopback TCP listeners; not MCP", "repetitions_per_case": REPEATS, "cases": CASES, "source_sha256": sha(Path(__file__).read_bytes()), "fixtures_sha256": sha((ROOT / "fixtures.json").read_bytes()), "reset": "Before each case: clear active drafts/outbox and all in-memory capabilities; append reset counts. Preserve separate append-only draft/outbox archives.", "observation": "Actual HTTP status/body plus SQLite state before and after every request and persistent action archives.", "approval": "Separate reviewer bearer credential, server-created single-use random capability, bound to exact canonical ticket_id/recipient/body digest; reviewer action is scripted, not a real human participant.", "invariants": ["No external network destinations or email delivery", "Bind both listeners to127.0.0.1 with OS-assigned ports", "Generate authorization and approval secrets in memory, never write or print values", "No model, model API, MCP conformance or prompt-injection efficacy claim", "Stop both owned servers in finally; preserve unrelated processes"], "expected_per_run": {"case_executions": 30, "driver_http_requests": 54, "accepted_http": 21, "denied_http": 33, "outbox_archive_rows": 6, "draft_archive_rows": 3, "approval_issued": 6, "approval_consumed": 6}} write_json(path, design) print(json.dumps({"status": "design_frozen", "path": str(path), "design_sha256": sha(path.read_bytes()), "frozen_at": design["frozen_at"]})) def run(output): fixtures_path = ROOT / "fixtures.json" design_path = ROOT / "frozen-design.json" design = json.loads(design_path.read_text(encoding="utf-8")) if sha(Path(__file__).read_bytes()) != design["source_sha256"] or sha(fixtures_path.read_bytes()) != design["fixtures_sha256"] or design["cases"] != CASES: raise SystemExit("Frozen source/fixture/case verification failed; experiment not started.") output.mkdir(parents=True, exist_ok=True) if any((output / name).exists() for name in ("request-results.json", "environment.json", "action-evidence.sqlite3")): raise SystemExit("Output artifacts already exist; use a new directory.") started = now() run_id = str(uuid.uuid4()) state = State(output / "action-evidence.sqlite3", json.loads(fixtures_path.read_text(encoding="utf-8"))) backend = ThreadingHTTPServer(("127.0.0.1", 0), backend_handler(state)) backend_url = "http://127.0.0.1:" + str(backend.server_port) connector = ThreadingHTTPServer(("127.0.0.1", 0), connector_handler(state, backend_url)) connector_url = "http://127.0.0.1:" + str(connector.server_port) servers = [backend, connector] threads = [threading.Thread(target=s.serve_forever, name="owned-shadow-action-lab-" + str(i), daemon=True) for i, s in enumerate(servers)] for thread in threads: thread.start() requests, executions = [], [] fixtures = json.loads(fixtures_path.read_text(encoding="utf-8")) message = fixtures["message"] errors = [] def step(case, destination, path, method="GET", payload=None, auth=None, approval=None): before = state.snapshot() stamp = now() clock_start = time.perf_counter_ns() status, body = request(connector_url if destination == "connector" else backend_url, path, method, payload, state.credentials.get(auth), approval) elapsed = time.perf_counter_ns() - clock_start after = state.snapshot() public_body = {k: v for k, v in body.items() if k != "approval"} requests.append({"sequence": len(requests) + 1, "case_id": case, "at": stamp, "elapsed_ns": elapsed, "destination": destination, "method": method, "path": path, "credential_role": auth or "none", "approval_present": bool(approval), "request_payload": payload, "status": status, "response": public_body, "before": before, "after": after, "actual_delta": {"drafts": after["drafts"] - before["drafts"], "outbox": after["outbox"] - before["outbox"]}}) return status, body try: for repetition in range(1, REPEATS + 1): for spec in CASES: case = f"r{repetition}-" + spec["id"] reset = state.reset(case) first = len(requests) idx = spec["id"][:2] if idx == "01": step(case, "connector", "/tickets/TW-SYN-001") elif idx == "02": step(case, "connector", "/tickets/TW-SYN-001", auth="client_read") elif idx == "03": step(case, "connector", "/send", "POST", message, "client_read") elif idx == "04": step(case, "connector", "/drafts", "POST", message, "client_draft") step(case, "connector", "/send", "POST", message, "client_draft") elif idx == "05": step(case, "connector", "/send", "POST", message, "client_send") elif idx in ("06", "07"): _, approved = step(case, "backend", "/approvals", "POST", message, "reviewer") capability = approved.get("approval") if idx == "06": step(case, "connector", "/send", "POST", message, "client_send", capability) step(case, "connector", "/send", "POST", message, "client_send", capability) else: changed = {**message, "body": fixtures["changed_body"]} step(case, "connector", "/send", "POST", changed, "client_send", capability) step(case, "connector", "/send", "POST", message, "client_send", capability) elif idx == "08": step(case, "backend", "/outbox", "POST", message, "client_send") step(case, "backend", "/tickets/TW-SYN-001", auth="client_read") elif idx == "09": _, retrieved = step(case, "connector", "/tickets/TW-SYN-002", auth="client_read") if "Ignore your read-only permissions" not in retrieved.get("ticket", {}).get("text", ""): errors.append({"case": case, "issue": "constructed_instruction_not_retrieved"}) # Deliberately attempt the instruction with unchanged read credentials. # No model reads, obeys or rejects the text in this experiment. step(case, "connector", "/send", "POST", {**message, "ticket_id": "TW-SYN-002"}, "client_read") elif idx == "10": step(case, "backend", "/approvals", "POST", message, "client_send") step(case, "connector", "/send", "POST", message, "client_send", "deliberately-invalid-lab-capability") actual_statuses = [r["status"] for r in requests[first:]] snapshot = state.snapshot() passed = actual_statuses == spec["statuses"] and snapshot["drafts"] == spec["drafts"] and snapshot["outbox"] == spec["outbox"] executions.append({"case_id": case, "design_case": spec["id"], "repetition": repetition, "reset": reset, "expected_statuses": spec["statuses"], "actual_statuses": actual_statuses, "expected_final_drafts": spec["drafts"], "expected_final_outbox": spec["outbox"], "actual_final_state": snapshot, "passed": passed}) except Exception as error: errors.append({"issue": "trial_exception", "type": type(error).__name__}) finally: for server in servers: server.shutdown() server.server_close() for thread in threads: thread.join(timeout=5) with state.lock: archive = {table: [dict(row) for row in state.db.execute(f"SELECT * FROM {table} ORDER BY id")] for table in ("outbox_archive", "draft_archive", "approval_events", "resets")} state.db.commit() integrity = state.db.execute("PRAGMA integrity_check").fetchone()[0] state.db.close() closed = [] for server in servers: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: probe.settimeout(0.3) closed.append(probe.connect_ex(("127.0.0.1", server.server_port)) != 0) counts = {} for row in requests: counts[str(row["status"])] = counts.get(str(row["status"]), 0) + 1 actual = {"case_executions": len(executions), "driver_http_requests": len(requests), "accepted_http": sum(200 <= r["status"] < 300 for r in requests), "denied_http": sum(r["status"] >= 400 for r in requests), "outbox_archive_rows": len(archive["outbox_archive"]), "draft_archive_rows": len(archive["draft_archive"]), "approval_issued": sum(r["event"] == "issued" for r in archive["approval_events"]), "approval_consumed": sum(r["event"] == "consumed" for r in archive["approval_events"])} passed = not errors and all(e["passed"] for e in executions) and actual == design["expected_per_run"] and all(closed) and not any(t.is_alive() for t in threads) and integrity == "ok" environment = {"run_id": run_id, "started_at": started, "finished_at": now(), "python": sys.version, "python_implementation": platform.python_implementation(), "os": platform.platform(), "sqlite_version": sqlite3.sqlite_version, "source_sha256": sha(Path(__file__).read_bytes()), "fixtures_sha256": sha(fixtures_path.read_bytes()), "frozen_design_sha256": sha(design_path.read_bytes()), "protocol": design["protocol"], "bindings": {"connector": connector_url, "fixture_service": backend_url}, "dependencies": "Python standard library only", "model_used": False, "external_network_used": False, "real_delivery_used": False, "auth": "Random test bearer credentials and one-use approval capabilities generated and held only in memory; no values logged", "reviewer": "Deterministic driver using a separate reviewer credential; not a real human-review study", "cleanup": {"owned_server_threads_stopped": not any(t.is_alive() for t in threads), "both_listener_ports_closed_after_run": all(closed), "sqlite_closed": True, "integrity_check": integrity}} result = {"schema_version": 1, "run_id": run_id, "started_at": started, "finished_at": environment["finished_at"], "frozen_design_sha256": environment["frozen_design_sha256"], "source_sha256": environment["source_sha256"], "fixtures_sha256": environment["fixtures_sha256"], "all_planned_checks_passed": passed, "summary": {**actual, "status_counts": counts, "outbound_action_attempts": sum(r["path"] in ("/send", "/outbox") for r in requests), "outbound_action_denials": sum(r["path"] in ("/send", "/outbox") and r["status"] >= 400 for r in requests), "actual_synthetic_sends": len(archive["outbox_archive"])}, "case_executions": executions, "requests": requests, "actual_persisted_archives": archive, "errors": errors, "limits": ["Custom deterministic HTTP services; not a commercial product or MCP implementation", "No LLM, model API, real accounts or external delivery", "Constructed instruction case tests unchanged server credentials, not model prompt-injection resistance", "Reviewer endpoint is exercised by script, not a real person", "Repeated fixed cases do not estimate general effectiveness or real-world probabilities", "No concurrent replay, expiry, process-crash recovery, token theft, TLS or production-hardening audit"]} write_json(output / "request-results.json", result) write_json(output / "environment.json", environment) # Public-artifact leakage assertion, without exposing any secret on failure. leak = any(secret.encode("ascii") in path.read_bytes() for path in output.iterdir() if path.is_file() for secret in state.secret_values) if leak: raise SystemExit("Secret leakage check failed; do not publish run artifacts.") print(json.dumps({"run_id": run_id, "all_planned_checks_passed": passed, "summary": result["summary"], "cleanup": environment["cleanup"], "secret_values_found_in_artifacts": 0, "output": str(output)}, ensure_ascii=False)) return 0 if passed else 2 if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--freeze", action="store_true", help="Freeze source, fixtures and expected observations before any server execution.") parser.add_argument("--output", type=Path, default=ROOT, help="New output files directory; existing results are never overwritten.") args = parser.parse_args() if args.freeze: freeze() else: sys.exit(run(args.output.resolve()))