#!/usr/bin/env python3 """Signed delivery acceptance with a durable SQLite inbox. Python 3.10+, stdlib only.""" import argparse import hashlib import hmac import json import os import sqlite3 import time from datetime import datetime, timezone from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer MAX_BODY = 5 * 1024 * 1024 @contextmanager def connect(path): db = sqlite3.connect(path, timeout=5) db.row_factory = sqlite3.Row db.execute('PRAGMA synchronous=FULL') try: with db: yield db finally: db.close() def initialize(path): with connect(path) as db: db.execute('PRAGMA journal_mode=WAL') db.executescript('''CREATE TABLE IF NOT EXISTS jobs ( delivery_key TEXT PRIMARY KEY, event_id TEXT NOT NULL, body TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', result TEXT); CREATE TABLE IF NOT EXISTS failures (delivery_key TEXT PRIMARY KEY);''') def valid_signature(raw, timestamp, signature, secret): try: date = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) if date.tzinfo is None or abs(datetime.now(timezone.utc).timestamp() - date.timestamp()) > 300: return False expected = 'v2=' + hmac.new(secret.encode(), timestamp.encode() + b'.' + raw, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature) except (ValueError, TypeError): return False def accept(path, key, event): with connect(path) as db: db.execute('BEGIN IMMEDIATE') existing = db.execute( 'SELECT event_id FROM jobs WHERE delivery_key=?', (key,) ).fetchone() if existing: return 204 if existing['event_id'] == event['event_id'] else 409 db.execute( 'INSERT INTO jobs(delivery_key,event_id,body) VALUES (?,?,?)', (key, event['event_id'], json.dumps(event)) ) # SQLite committed before the HTTP acknowledgement. return 204 def reject_once(path, key): """Local outage fixture; separate from the acceptance implementation.""" with connect(path) as db: db.execute('BEGIN IMMEDIATE') if db.execute('SELECT 1 FROM jobs WHERE delivery_key=?', (key,)).fetchone(): return False if db.execute('SELECT 1 FROM failures WHERE delivery_key=?', (key,)).fetchone(): return False db.execute('INSERT INTO failures VALUES (?)', (key,)) return True def handler(path, secret, fail_first=False, endpoint_id="ep_demo"): class Receiver(BaseHTTPRequestHandler): def log_message(self, *args): pass # Do not log headers, secrets, or message content. def respond(self, status): self.send_response(status) self.send_header('Content-Length', '0') self.end_headers() def do_POST(self): self.connection.settimeout(10) if self.path != '/events': return self.respond(404) try: size = int(self.headers.get('Content-Length', '0')) if not 0 < size <= MAX_BODY: return self.respond(413) raw = self.rfile.read(size) if len(raw) != size: return self.respond(400) if not valid_signature(raw, self.headers.get('X-AgentOwl-Timestamp', ''), self.headers.get('X-AgentOwl-Signature', ''), secret): return self.respond(401) event = json.loads(raw) key = self.headers.get('X-AgentOwl-Idempotency-Key', '') if (not isinstance(event, dict) or not isinstance(event.get('event_id'), str) or not event['event_id'] or key != event['event_id'] + ':' + endpoint_id or len(key) > 256 or event.get('channel') not in ('email', 'sms', 'voice')): return self.respond(400) if fail_first and reject_once(path, key): return self.respond(503) status = accept(path, key, event) except (ValueError, TypeError, UnicodeError): return self.respond(400) except (sqlite3.Error, OSError): return self.respond(503) return self.respond(status) return Receiver def process_one(path): """Only writes a fixture result. External side effects need their own idempotency.""" with connect(path) as db: db.execute('BEGIN IMMEDIATE') job = db.execute("SELECT * FROM jobs WHERE status='pending' ORDER BY rowid LIMIT 1").fetchone() if not job: return None event = json.loads(job['body']) inline = event.get('payload', {}).get('inline', {}) if event.get('payload', {}).get('payload_ref_id'): result = {'event_id': event['event_id'], 'status': 'requires_payload_fetch'} else: text = inline.get('body', {}).get('text') if event['channel'] == 'email' else inline.get('text') result = {'event_id': event['event_id'], 'channel': event['channel'], 'status': 'example_processed', 'text': text} db.execute("UPDATE jobs SET status='done',result=? WHERE delivery_key=?", (json.dumps(result), job['delivery_key'])) return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('mode', choices=['serve', 'worker', 'inspect']) parser.add_argument('--db', default='receiver.sqlite3') parser.add_argument('--port', type=int, default=8788) parser.add_argument('--once', action='store_true') parser.add_argument('--endpoint', default='ep_demo', help='Expected endpoint ID; set to your real endpoint for hosted deliveries') parser.add_argument('--fail-first', action='store_true', help='Local demo only: reject first valid delivery per key') args = parser.parse_args() initialize(args.db) if args.mode == 'serve': secret = os.environ.get('AGENTOWL_SIGNING_SECRET') if not secret: parser.error('Set AGENTOWL_SIGNING_SECRET; activate v2 signing for real deliveries') server = ThreadingHTTPServer(('127.0.0.1', args.port), handler(args.db, secret, args.fail_first, args.endpoint)) print(f'Receiver listening on 127.0.0.1:{server.server_port}', flush=True) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() elif args.mode == 'inspect': with connect(args.db) as db: for row in db.execute('SELECT delivery_key,event_id,status,result FROM jobs ORDER BY rowid'): print(json.dumps(dict(row))) else: while True: result = process_one(args.db) if result: print(json.dumps(result), flush=True) if args.once: break if not result: time.sleep(.5) if __name__ == '__main__': main()