"""Local teaching protocol, not a provider adapter. No external requests or model calls.""" from contextlib import closing from hashlib import sha256 import hmac from http.server import BaseHTTPRequestHandler, HTTPServer import json from pathlib import Path import secrets import sqlite3 import tempfile import threading import time import urllib.error import urllib.request WINDOW_SECONDS = 300 MAX_BODY = 16384 def signature(secret: bytes, timestamp: str, body: bytes) -> str: return "sha256=" + hmac.new(secret, timestamp.encode("ascii") + b"." + body, sha256).hexdigest() def initialize(path): with closing(sqlite3.connect(path)) as db, db: db.execute("CREATE TABLE inbox (event_id TEXT PRIMARY KEY, body_hash TEXT NOT NULL, state TEXT NOT NULL)") db.execute("CREATE TABLE work (event_id TEXT PRIMARY KEY, result TEXT NOT NULL)") def accept(path, secret, headers, raw, now): timestamp = headers.get("X-Lab-Timestamp", "") supplied = headers.get("X-Lab-Signature", "") if not timestamp.isascii() or not timestamp.isdecimal() or len(timestamp) > 12: return 401, {"error": "invalid_timestamp"} if abs(now - int(timestamp)) > WINDOW_SECONDS: return 401, {"error": "stale_timestamp"} if not supplied.isascii() or not hmac.compare_digest(signature(secret, timestamp, raw), supplied): return 401, {"error": "invalid_signature"} try: event = json.loads(raw) if not isinstance(event, dict) or set(event) != {"event_id", "type", "ticket_id"}: raise ValueError() if event["type"] != "ticket.created" or event["ticket_id"] != "DEMO-42": raise ValueError() event_id = event["event_id"] if not isinstance(event_id, str) or not 1 <= len(event_id) <= 80: raise ValueError() except (ValueError, TypeError, UnicodeError): return 400, {"error": "invalid_event"} digest = sha256(raw).hexdigest() with closing(sqlite3.connect(path)) as db, db: db.execute("BEGIN IMMEDIATE") existing = db.execute("SELECT body_hash, state FROM inbox WHERE event_id=?", (event_id,)).fetchone() if existing: if existing[0] != digest: return 409, {"error": "event_id_conflict"} return 200, {"event_id": event_id, "state": existing[1], "duplicate": True} db.execute("INSERT INTO inbox VALUES (?, ?, 'pending')", (event_id, digest)) # The context manager committed before acknowledging acceptance. return 202, {"event_id": event_id, "state": "pending", "duplicate": False} def process_one(path, fail_before_commit=False): with closing(sqlite3.connect(path)) as db, db: db.execute("BEGIN IMMEDIATE") row = db.execute("SELECT event_id FROM inbox WHERE state='pending' ORDER BY event_id LIMIT 1").fetchone() if not row: return False db.execute("INSERT INTO work VALUES (?, 'ready_for_triage')", row) if fail_before_commit: raise RuntimeError("injected failure before commit") db.execute("UPDATE inbox SET state='completed' WHERE event_id=?", row) return True def fixtures(): secret = secrets.token_bytes(32) rows = [] with tempfile.TemporaryDirectory(prefix="webhook-lesson-") as directory: path = Path(directory) / "inbox.sqlite" initialize(path) class Handler(BaseHTTPRequestHandler): def log_message(self, *_): pass def do_POST(self): self.connection.settimeout(5) if self.path != "/webhook": self.send_error(404); return try: length = int(self.headers.get("Content-Length", "-1")) except ValueError: length = -1 if not 0 <= length <= MAX_BODY: self.send_error(413); return raw = self.rfile.read(length) if len(raw) != length: self.send_error(400); return status, result = accept(path, secret, self.headers, raw, int(time.time())) output = json.dumps(result).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(output))) self.end_headers() self.wfile.write(output) server = HTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() body = b'{"event_id":"evt-demo-1","type":"ticket.created","ticket_id":"DEMO-42"}' stamp = str(int(time.time())) def send(name, raw=body, timestamp=stamp, signed_body=None, secret_override=None, expected=202): signed = raw if signed_body is None else signed_body headers = {"Content-Type": "application/json", "X-Lab-Timestamp": timestamp, "X-Lab-Signature": signature(secret if secret_override is None else secret_override, timestamp, signed)} request = urllib.request.Request(f"http://127.0.0.1:{server.server_port}/webhook", data=raw, headers=headers) try: response = urllib.request.urlopen(request, timeout=10) except urllib.error.HTTPError as error: response = error with response: result = json.load(response) assert response.status == expected, (name, response.status, result) rows.append({"case": name, "status": response.status, "result": result}) try: send("accepted") send("duplicate-pending", expected=200) send("changed-bytes", raw=body+b" ", signed_body=body, expected=401) send("wrong-secret", secret_override=b"fixture-wrong-key", expected=401) send("stale-delivery", timestamp=str(int(stamp)-301), expected=401) send("signed-invalid-json", raw=b"not json", expected=400) send("same-id-different-body", raw=body+b" ", expected=409) try: process_one(path, fail_before_commit=True) except RuntimeError: pass else: raise AssertionError("Injected worker failure must occur") with closing(sqlite3.connect(path)) as reopened: assert reopened.execute("SELECT COUNT(*) FROM work").fetchone()[0] == 0 assert reopened.execute("SELECT state FROM inbox").fetchone()[0] == "pending" assert process_one(path) is True assert process_one(path) is False send("duplicate-completed", expected=200) with closing(sqlite3.connect(path)) as reopened: count = reopened.execute("SELECT COUNT(*) FROM work").fetchone()[0] assert count == 1 print(json.dumps({"protocol": "local timestamp + raw body HMAC; not GitHub or Stripe", "http_cases": rows, "http_passed": len(rows), "reopened_after_rollback": True, "work_rows": count, "limits": "One local SQLite transaction; no remote side effects or process-kill test"}, indent=2)) finally: server.shutdown(); server.server_close(); thread.join() if __name__ == "__main__": fixtures()