#!/usr/bin/env python3 """Original LockedIn Labs webhook boundary lab. Python 3.11+, standard library. Offline byte verification and scratch SQLite only. Ephemeral keys and MAC values are never reported. This original envelope is not a provider SDK or public API. """ import argparse import base64 import binascii import hashlib import hmac import json import platform import re import secrets import sqlite3 NOW = 1_800_000_000 # Fixed fictional receiver time, not the machine's wall clock. SOURCE = "SYNTHETIC-SOURCE-A" CASES = ("valid", "body-changed", "body-reserialized", "wrong-key", "changed-delivery-id", "stale", "future", "repeat", "event-conflict", "same-event-new-bytes") ORIGINAL = b'{"event_id": "EVT-1", "kind": "synthetic.delta", "units": 4}' def sign(key, body, delivery="DEL-1", sent_at=NOW): message = delivery.encode("ascii") + b"." + str(sent_at).encode("ascii") + b"." + body mac = hmac.digest(key, message, "sha256") return {"lab-delivery-id": delivery, "lab-sent-at": str(sent_at), "lab-mac": "v1," + base64.b64encode(mac).decode("ascii")} def database(): db = sqlite3.connect(":memory:", isolation_level=None) db.row_factory = sqlite3.Row db.execute("CREATE TABLE receipts(source TEXT,event_id TEXT,body_hash TEXT,PRIMARY KEY(source,event_id))") db.execute("CREATE TABLE effects(units INTEGER NOT NULL)") db.execute("INSERT INTO effects VALUES(0)") return db def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("duplicate_json_key") result[key] = value return result def receive(db, key, body, headers, now=NOW, *, fault_after_receipt=False): steps = {"mac_comparisons": 0, "json_parses": 0, "receipt_transactions": 0} mac_valid, fresh = None, None def outcome(state, reason): return {"state": state, "reason": reason, "mac_valid": mac_valid, "timestamp_admitted": fresh, "steps": dict(steps)} if type(body) is not bytes or not 1 <= len(body) <= 4096: return outcome("rejected", "body_shape_or_limit") delivery, timestamp, encoded = [headers.get(field) for field in ("lab-delivery-id", "lab-sent-at", "lab-mac")] if not isinstance(delivery, str) or not re.fullmatch(r"DEL-[1-9][0-9]{0,3}", delivery): return outcome("rejected", "delivery_shape") if not isinstance(timestamp, str) or not re.fullmatch(r"[0-9]{10}", timestamp): return outcome("rejected", "timestamp_shape") if not isinstance(encoded, str) or not encoded.startswith("v1,") or len(encoded) != 47: return outcome("rejected", "mac_shape") try: supplied = base64.b64decode(encoded[3:], validate=True) except (ValueError, binascii.Error): return outcome("rejected", "mac_shape") if len(supplied) != 32: return outcome("rejected", "mac_shape") message = delivery.encode("ascii") + b"." + timestamp.encode("ascii") + b"." + body steps["mac_comparisons"] += 1 mac_valid = hmac.compare_digest(hmac.digest(key, message, "sha256"), supplied) if not mac_valid: return outcome("rejected", "mac_mismatch") age = now - int(timestamp) fresh = -30 <= age <= 300 if not fresh: return outcome("rejected", "timestamp_outside_policy") steps["json_parses"] += 1 try: payload = json.loads(body.decode("utf-8"), object_pairs_hook=unique_object) if type(payload) is not dict or set(payload) != {"event_id", "kind", "units"}: raise ValueError("payload_shape") if not isinstance(payload["event_id"], str) or not re.fullmatch(r"EVT-[1-9][0-9]{0,3}", payload["event_id"]): raise ValueError("event_identity") if payload["kind"] != "synthetic.delta" or type(payload["units"]) is not int or not 1 <= payload["units"] <= 9: raise ValueError("payload_contract") except (ValueError, UnicodeError, TypeError, RecursionError): return outcome("rejected", "payload_contract") body_hash = hashlib.sha256(body).hexdigest() steps["receipt_transactions"] += 1 db.execute("BEGIN IMMEDIATE") try: prior = db.execute("SELECT body_hash FROM receipts WHERE source=? AND event_id=?", (SOURCE, payload["event_id"])).fetchone() if prior: db.execute("COMMIT") return outcome("replayed" if prior["body_hash"] == body_hash else "conflict", "same_event_same_bytes" if prior["body_hash"] == body_hash else "event_bytes_conflict") db.execute("INSERT INTO receipts VALUES(?,?,?)", (SOURCE, payload["event_id"], body_hash)) if fault_after_receipt: raise RuntimeError("injected_local_failure") db.execute("UPDATE effects SET units=units+?", (payload["units"],)) db.execute("COMMIT") return outcome("applied", "receipt_and_local_effect_committed") except Exception: db.execute("ROLLBACK") return outcome("rolled_back", "local_transaction_failed") def snapshot(db): return {"receipt_count": db.execute("SELECT count(*) FROM receipts").fetchone()[0], "effect_units": db.execute("SELECT units FROM effects").fetchone()[0]} def run_case(name, age_seconds=0): if name not in CASES or type(age_seconds) is not int or age_seconds not in (0, 10, 301): raise ValueError("unsupported_case_or_age") key, wrong_key, db = secrets.token_bytes(32), secrets.token_bytes(32), database() attempts = [] compact = json.dumps(json.loads(ORIGINAL), separators=(",", ":")).encode("utf-8") def attempt(body=ORIGINAL, delivery="DEL-1", sent_at=NOW, signer=None, wire_body=None, wire_delivery=None): headers = sign(key if signer is None else signer, body, delivery, sent_at) actual = body if wire_body is None else wire_body if wire_delivery is not None: headers["lab-delivery-id"] = wire_delivery before = snapshot(db) result = receive(db, key, actual, headers) attempts.append({"delivery_id": headers["lab-delivery-id"], "sent_at": headers["lab-sent-at"], "body_utf8": actual.decode("utf-8"), "body_sha256": hashlib.sha256(actual).hexdigest(), "before": before, "verification_and_receiving": result, "after": snapshot(db)}) try: if name == "body-changed": attempt(wire_body=ORIGINAL.replace(b'"units": 4', b'"units": 7')) elif name == "body-reserialized": attempt(wire_body=compact) elif name == "wrong-key": attempt(signer=wrong_key) elif name == "changed-delivery-id": attempt(wire_delivery="DEL-2") elif name == "stale": attempt(sent_at=NOW-301) elif name == "future": attempt(sent_at=NOW+31) elif name in ("repeat", "event-conflict", "same-event-new-bytes"): attempt() other = ORIGINAL if name == "repeat" else compact if name == "same-event-new-bytes" else ORIGINAL.replace(b'"units": 4', b'"units": 7') attempt(body=other, delivery="DEL-2", sent_at=NOW+1) else: attempt(sent_at=NOW-age_seconds) return {"case": name, "receiver_time": NOW, "attempts": attempts, "final": snapshot(db)} finally: db.close() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--case", choices=CASES) parser.add_argument("--age-seconds", type=int, choices=(0, 10, 301), default=0, help="Change only the valid case's signed send time; receiver policy stays fixed.") args = parser.parse_args() print(json.dumps({"schema": "lockedin.webhook-receiving/v1", "python": platform.python_version(), "sqlite": sqlite3.sqlite_version, "scope": "Original HMAC-SHA256 envelope over delivery ID, timestamp and raw bytes; fixed fictional clock, ephemeral keys and scratch SQLite. Not a provider implementation, HTTP endpoint, key rotation, durable/distributed receipt or business authorization.", "cases": [run_case(name, args.age_seconds) for name in ([args.case] if args.case else CASES)]}, indent=2)) if __name__ == "__main__": main()