#!/usr/bin/env python3 """Loopback-only Call URL receipt lab. Python 3 standard library; no CRM writes. init creates new private state. serve runs a local GET receiver. inspect reads its receipt ledger. This is a teaching server, not a production ingress. """ import argparse from contextlib import closing from datetime import datetime, timezone import hashlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import hmac import json import os from pathlib import Path import re import secrets import sqlite3 import stat import sys from urllib.parse import parse_qs, urlsplit FIELDS = {"token", "source", "event", "call_id", "lead_id", "dispo"} def state_path(directory): directory = directory.expanduser().resolve() info = directory.stat() if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077: raise ValueError("State directory must be yours and private (mode 0700).") return directory def token(directory): fd = os.open(directory / "token.txt", os.O_RDONLY | os.O_NOFOLLOW) with os.fdopen(fd, encoding="ascii") as handle: info = os.fstat(handle.fileno()) if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o600: raise ValueError("Token must be your regular mode-0600 file.") value = handle.read().strip() if not re.fullmatch(r"[0-9a-f]{64}", value): raise ValueError("Token is not a generated lab token.") return value def connect(directory): db = sqlite3.connect(str(directory / "events.sqlite3"), timeout=5) db.execute("PRAGMA synchronous=FULL") return db def initialize(directory): directory.mkdir(mode=0o700) # Deliberately refuses an existing directory. directory = state_path(directory) fd = os.open(directory / "token.txt", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w", encoding="ascii") as handle: handle.write(secrets.token_hex(32) + "\n") with closing(connect(directory)) as db, db: db.execute("""CREATE TABLE receipts ( source TEXT NOT NULL, call_id TEXT NOT NULL, event TEXT NOT NULL, payload TEXT NOT NULL, payload_sha256 TEXT NOT NULL, received_utc TEXT NOT NULL, PRIMARY KEY (source, call_id, event) )""") os.chmod(directory / "events.sqlite3", 0o600) return directory def accept(directory, values): event = {key: values[key] for key in sorted(FIELDS - {"token"})} payload = json.dumps(event, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(payload.encode()).hexdigest() with closing(connect(directory)) as db, db: db.execute("BEGIN IMMEDIATE") prior = db.execute("SELECT payload_sha256 FROM receipts WHERE source=? AND call_id=? AND event=?", (event["source"], event["call_id"], event["event"])).fetchone() if prior: return (200, "duplicate") if prior[0] == digest else (409, "conflict") db.execute("INSERT INTO receipts VALUES (?,?,?,?,?,?)", (event["source"], event["call_id"], event["event"], payload, digest, datetime.now(timezone.utc).isoformat())) # The context manager has committed before the HTTP acknowledgement. return 201, "stored" def inspect(directory): with closing(connect(directory)) as db, db: return [json.loads(row[0]) for row in db.execute("SELECT payload FROM receipts ORDER BY source,call_id,event")] def make_server(directory, port=8765): directory = state_path(directory) secret = token(directory) if not (directory / "events.sqlite3").is_file(): raise ValueError("Run init first; receipt database is missing.") class Handler(BaseHTTPRequestHandler): def log_message(self, *_args): pass # Never put query tokens or event fields into an access log. def answer(self, code, result): body = (json.dumps({"result": result}, sort_keys=True) + "\n").encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): if len(self.path) > 2048: return self.answer(414, "request-too-long") parsed = urlsplit(self.path) if parsed.path != "/call-event": return self.answer(404, "unknown-path") try: items = parse_qs(parsed.query, keep_blank_values=True, strict_parsing=True, max_num_fields=12) except ValueError: return self.answer(400, "invalid-query") if any(len(v) != 1 for v in items.values()): return self.answer(400, "repeated-field") values = {key: value[0] for key, value in items.items()} supplied = values.get("token", "") if not supplied.isascii() or not hmac.compare_digest(supplied, secret): return self.answer(403, "rejected-token") if set(values) != FIELDS: return self.answer(400, "missing-or-extra-field") valid = (not any("--A--" in value or "--B--" in value for value in values.values()) and values["source"] == "TRAINING_A" and values["event"] == "dispo" and re.fullmatch(r"[A-Za-z0-9_.:-]{1,80}", values["call_id"]) and re.fullmatch(r"[0-9]{1,20}", values["lead_id"]) and re.fullmatch(r"[A-Za-z0-9_]{1,12}", values["dispo"])) if not valid: return self.answer(400, "invalid-event") try: code, result = accept(directory, values) except sqlite3.Error: return self.answer(503, "ledger-unavailable") self.answer(code, result) return ThreadingHTTPServer(("127.0.0.1", port), Handler) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("action", choices=["init", "serve", "inspect"]) parser.add_argument("--state", type=Path, default=Path("call-url-state")) parser.add_argument("--port", type=int, default=8765) args = parser.parse_args() os.umask(0o077) try: if args.action == "init": initialize(args.state) print("Private lab state created. The token was not printed.") elif args.action == "inspect": print(json.dumps(inspect(state_path(args.state)), indent=2, sort_keys=True)) else: with make_server(args.state, args.port) as server: print(f"Loopback lab listening on 127.0.0.1:{server.server_port}; Ctrl+C stops it.", flush=True) server.serve_forever() except KeyboardInterrupt: pass except (OSError, ValueError, sqlite3.Error) as error: print(f"Lab setup failed: {type(error).__name__}; inspect paths, permissions and free port privately.", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())