"""Durable local webhook queue. Python 3.12+, SQLite, no third-party packages.""" from __future__ import annotations import contextlib, hashlib, hmac, http.client, ipaddress, json, re, socket, sqlite3, ssl, threading, time, uuid from pathlib import Path from urllib.parse import urlsplit class Problem(ValueError): def __init__(self, message, status=400): super().__init__(message); self.status = status def canonical(value): try: return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False) except (ValueError, TypeError) as e: raise Problem('Payload must be finite JSON') from e def identifier(value): if not isinstance(value, str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,95}', value): raise Problem('Identifiers must be 1-96 ASCII letters, digits, _, ., :, or -') return value def sign(secret, timestamp, body): return 'v1=' + hmac.new(secret.encode(), str(timestamp).encode() + b'.' + body, hashlib.sha256).hexdigest() class DestinationPolicy: """Exact host approvals; resolve once, validate every address, connect to that IP.""" def __init__(self, approved_hosts=(), test_origins=(), resolver=socket.getaddrinfo): self.approved_hosts = frozenset(h.lower() for h in approved_hosts) self.test_origins = frozenset(test_origins) self.resolver = resolver def validate(self, url, resolve=True): try: u = urlsplit(url); host = u.hostname; port = u.port or (443 if u.scheme == 'https' else 80) except (ValueError, TypeError) as e: raise Problem('Invalid destination URL') from e if not host or len(url) > 2000 or u.username or u.password or u.fragment or any(ord(c) < 33 for c in url): raise Problem('Destination must have a host and no credentials, fragments, or whitespace') origin = f'{u.scheme}://{u.netloc}' local_fixture = origin in self.test_origins if local_fixture: if u.scheme != 'http' or host != '127.0.0.1': raise Problem('Test receivers must use an exact http://127.0.0.1:port origin') return u, '127.0.0.1', port if u.scheme != 'https' or port != 443 or host.lower() not in self.approved_hosts: raise Problem('Destination must be an administrator-approved HTTPS host on port 443') if not resolve: return u, None, port try: records = self.resolver(host, port, type=socket.SOCK_STREAM) except OSError as e: raise Problem('Destination DNS lookup failed', 503) from e addresses = sorted({r[4][0] for r in records}) if not addresses: raise Problem('Destination has no addresses') for raw in addresses: address = ipaddress.ip_address(raw) # Python versions have differed on special-range classification; reject these explicitly too. if (not address.is_global or address.is_multicast or address.is_unspecified or address.is_reserved or address.is_loopback or address.is_link_local or getattr(address, 'is_site_local', False) or getattr(address, 'ipv4_mapped', None) or getattr(address, 'sixtofour', None) or getattr(address, 'teredo', None)): raise Problem('Destination resolves to a prohibited address') return u, addresses[0], port class PinnedHTTPS(http.client.HTTPSConnection): def __init__(self, host, address, port, timeout): super().__init__(host, port, timeout=timeout, context=ssl.create_default_context()) self.address = address def connect(self): raw = socket.create_connection((self.address, self.port), self.timeout) try: self.sock = self._context.wrap_socket(raw, server_hostname=self.host) except BaseException: raw.close(); raise def send_request(policy, url, secret, event_id, body, timeout): u, address, port = policy.validate(url) # Validation and address pinning happen on every attempt. conn = (http.client.HTTPConnection(address, port, timeout=timeout) if u.scheme == 'http' else PinnedHTTPS(u.hostname, address, port, timeout)) timestamp = int(time.time()) path = u.path or '/' if u.query: path += '?' + u.query headers = {'Content-Type': 'application/json', 'X-Event-ID': event_id, 'X-Webhook-Timestamp': str(timestamp), 'X-Webhook-Signature': sign(secret, timestamp, body), 'User-Agent': 'LocalDeliveryQueue/1.0', 'Connection': 'close'} def interrupt(): if conn.sock is not None: try: conn.sock.shutdown(socket.SHUT_RDWR) except OSError: pass conn.close() deadline = threading.Timer(timeout, interrupt) deadline.daemon = True deadline.start() try: conn.request('POST', path, body=body, headers=headers) response = conn.getresponse() status = response.status # The status is enough; do not download an unbounded or slow response body. return status finally: deadline.cancel(); conn.close() SCHEMA = ''' CREATE TABLE IF NOT EXISTS endpoints( id TEXT PRIMARY KEY, url TEXT NOT NULL, secret TEXT NOT NULL, paused INTEGER NOT NULL DEFAULT 0 CHECK(paused IN (0,1)), max_inflight INTEGER NOT NULL CHECK(max_inflight BETWEEN 1 AND 16), max_attempts INTEGER NOT NULL CHECK(max_attempts BETWEEN 1 AND 10), timeout REAL NOT NULL CHECK(timeout > 0 AND timeout <= 30), retry_base REAL NOT NULL CHECK(retry_base > 0), created REAL NOT NULL); CREATE TABLE IF NOT EXISTS events( id TEXT PRIMARY KEY, payload TEXT, fingerprint TEXT NOT NULL, created REAL NOT NULL, purged_at REAL); CREATE TABLE IF NOT EXISTS jobs( id INTEGER PRIMARY KEY, event_id TEXT NOT NULL REFERENCES events(id), endpoint_id TEXT NOT NULL REFERENCES endpoints(id), status TEXT NOT NULL CHECK(status IN ('ready','leased','retry','delivered','dead')), attempts INTEGER NOT NULL DEFAULT 0, generation INTEGER NOT NULL DEFAULT 0, available REAL NOT NULL, lease_token TEXT, lease_until REAL, last_http INTEGER, last_error TEXT, UNIQUE(event_id,endpoint_id)); CREATE INDEX IF NOT EXISTS jobs_ready ON jobs(status,available,id); CREATE TABLE IF NOT EXISTS attempts( id INTEGER PRIMARY KEY, job_id INTEGER NOT NULL REFERENCES jobs(id), token TEXT UNIQUE NOT NULL, number INTEGER NOT NULL, generation INTEGER NOT NULL, started REAL NOT NULL, finished REAL, outcome TEXT NOT NULL, http_status INTEGER, error TEXT); CREATE TABLE IF NOT EXISTS audit( id INTEGER PRIMARY KEY, at REAL NOT NULL, action TEXT NOT NULL, subject TEXT NOT NULL, detail TEXT NOT NULL); ''' class Queue: def __init__(self, path, policy=None, clock=time.time): self.path = str(path); self.policy = policy or DestinationPolicy(); self.clock = clock Path(self.path).parent.mkdir(parents=True, exist_ok=True) with contextlib.closing(self.connect()) as db: db.executescript(SCHEMA) def connect(self): db = sqlite3.connect(self.path, timeout=10, isolation_level=None) db.row_factory = sqlite3.Row db.execute('PRAGMA foreign_keys=ON'); db.execute('PRAGMA journal_mode=WAL'); db.execute('PRAGMA synchronous=FULL') return db @contextlib.contextmanager def transaction(self): db = self.connect() try: db.execute('BEGIN IMMEDIATE'); yield db; db.commit() except BaseException: db.rollback(); raise finally: db.close() def audit(self, db, action, subject, detail=''): db.execute('INSERT INTO audit(at,action,subject,detail) VALUES(?,?,?,?)', (self.clock(), action, str(subject), detail)) def add_endpoint(self, endpoint_id, url, secret, max_inflight=1, max_attempts=4, timeout=2, retry_base=1): identifier(endpoint_id); self.policy.validate(url, resolve=False) if not isinstance(secret,str) or not 16 <= len(secret) <= 256: raise Problem('Signing secret must be 16-256 characters') if type(max_inflight) is not int or not 1 <= max_inflight <= 16: raise Problem('max_inflight must be 1-16') if type(max_attempts) is not int or not 1 <= max_attempts <= 10: raise Problem('max_attempts must be 1-10') if isinstance(timeout,bool) or not isinstance(timeout,(float,int)) or not 0 < timeout <= 30: raise Problem('timeout must be in (0,30]') if isinstance(retry_base,bool) or not isinstance(retry_base,(float,int)) or not 0 < retry_base <= 300: raise Problem('retry_base must be in (0,300]') with self.transaction() as db: try: db.execute('INSERT INTO endpoints(id,url,secret,max_inflight,max_attempts,timeout,retry_base,created) VALUES(?,?,?,?,?,?,?,?)', (endpoint_id,url,secret,max_inflight,max_attempts,timeout,retry_base,self.clock())) except sqlite3.IntegrityError as e: raise Problem('Endpoint already exists',409) from e self.audit(db,'endpoint_added',endpoint_id) def accept(self, event_id, kind, data, endpoints): identifier(event_id); identifier(kind) if not isinstance(data,dict): raise Problem('data must be a JSON object') if not isinstance(endpoints,list) or not endpoints or len(endpoints)>32: raise Problem('Choose 1-32 endpoint IDs') targets = sorted(set(identifier(x) for x in endpoints)) payload = canonical({'id':event_id,'type':kind,'data':data}) if len(payload.encode()) > 65536: raise Problem('Event payload exceeds 64 KiB',413) fingerprint = hashlib.sha256((payload + '\n' + canonical(targets)).encode()).hexdigest() with self.transaction() as db: old = db.execute('SELECT * FROM events WHERE id=?',(event_id,)).fetchone() if old: if not hmac.compare_digest(old['fingerprint'],fingerprint): raise Problem('Event ID already has different content or destinations',409) return {'id':event_id,'duplicate':True,'payload_purged':old['payload'] is None} for endpoint in targets: if not db.execute('SELECT 1 FROM endpoints WHERE id=?',(endpoint,)).fetchone(): raise Problem('Unknown endpoint: '+endpoint) now = self.clock() db.execute('INSERT INTO events(id,payload,fingerprint,created) VALUES(?,?,?,?)',(event_id,payload,fingerprint,now)) db.executemany("INSERT INTO jobs(event_id,endpoint_id,status,available) VALUES(?,?,'ready',?)",[(event_id,x,now) for x in targets]) self.audit(db,'event_accepted',event_id,f'{len(targets)} destinations') return {'id':event_id,'duplicate':False,'payload_purged':False} def claim(self, lease_seconds=None): now = self.clock() with self.transaction() as db: expired = db.execute("SELECT j.*,e.max_attempts FROM jobs j JOIN endpoints e ON e.id=j.endpoint_id WHERE j.status='leased' AND lease_until<=?",(now,)).fetchall() for row in expired: status = 'dead' if row['attempts'] >= row['max_attempts'] else 'retry' db.execute("UPDATE attempts SET finished=?,outcome='lease_expired',error='lease_expired' WHERE token=? AND outcome='inflight'",(now,row['lease_token'])) db.execute('UPDATE jobs SET status=?,available=?,lease_token=NULL,lease_until=NULL,last_error=? WHERE id=?',(status,now,'lease_expired',row['id'])) row = db.execute("""SELECT j.*,e.url,e.secret,e.timeout,e.max_attempts,e.retry_base,v.payload FROM jobs j JOIN endpoints e ON e.id=j.endpoint_id JOIN events v ON v.id=j.event_id WHERE j.status IN ('ready','retry') AND j.available<=? AND e.paused=0 AND v.payload IS NOT NULL AND (SELECT COUNT(*) FROM jobs a WHERE a.endpoint_id=j.endpoint_id AND a.status='leased' AND a.lease_until>?) < e.max_inflight ORDER BY j.available,j.id LIMIT 1""",(now,now)).fetchone() if not row: return None token = uuid.uuid4().hex duration = lease_seconds if lease_seconds is not None else row['timeout'] + 5 if duration <= 0: raise Problem('Lease duration must be positive') number = row['attempts']+1 db.execute("UPDATE jobs SET status='leased',attempts=?,lease_token=?,lease_until=? WHERE id=?",(number,token,now+duration,row['id'])) db.execute("INSERT INTO attempts(job_id,token,number,generation,started,outcome) VALUES(?,?,?,?,?,'inflight')",(row['id'],token,number,row['generation'],now)) return {**dict(row),'attempts':number,'lease_token':token,'lease_until':now+duration} def complete(self, job_id, token, http_status=None, error=None, permanent=False): now = self.clock() with self.transaction() as db: row = db.execute('SELECT j.*,e.max_attempts,e.retry_base FROM jobs j JOIN endpoints e ON e.id=j.endpoint_id WHERE j.id=?',(job_id,)).fetchone() if not row or row['status']!='leased' or row['lease_token']!=token or row['lease_until']<=now: self.audit(db,'stale_completion',job_id); return False success = http_status is not None and 200 <= http_status < 300 retryable_status = http_status is None or http_status in (408,425,429) or http_status >= 500 state = 'delivered' if success else ('dead' if permanent or not retryable_status or row['attempts']>=row['max_attempts'] else 'retry') available = now + min(300,row['retry_base']*2**(row['attempts']-1)) if state=='retry' else now db.execute('UPDATE jobs SET status=?,available=?,lease_token=NULL,lease_until=NULL,last_http=?,last_error=? WHERE id=?',(state,available,http_status,error,job_id)) db.execute('UPDATE attempts SET finished=?,outcome=?,http_status=?,error=? WHERE token=?',(now,state,http_status,error,token)) return True def work_once(self): job = self.claim() if job is None: return False try: status = send_request(self.policy,job['url'],job['secret'],job['event_id'],job['payload'].encode(),job['timeout']) self.complete(job['id'],job['lease_token'],http_status=status) except Problem as e: self.complete(job['id'],job['lease_token'],error='destination_blocked' if e.status!=503 else 'dns_failure',permanent=e.status!=503) except (OSError, http.client.HTTPException, TimeoutError): self.complete(job['id'],job['lease_token'],error='transport_failure') return True def pause(self, endpoint_id, paused): if type(paused) is not bool: raise Problem('paused must be boolean') with self.transaction() as db: if not db.execute('UPDATE endpoints SET paused=? WHERE id=?',(int(paused),endpoint_id)).rowcount: raise Problem('Unknown endpoint',404) self.audit(db,'endpoint_paused' if paused else 'endpoint_resumed',endpoint_id) def replay(self, job_id): with self.transaction() as db: row=db.execute('SELECT j.status,v.payload FROM jobs j JOIN events v ON v.id=j.event_id WHERE j.id=?',(job_id,)).fetchone() if not row: raise Problem('Unknown job',404) if row['status']!='dead': raise Problem('Only exhausted deliveries can be replayed',409) if row['payload'] is None: raise Problem('Payload retention expired; this delivery cannot be replayed',410) db.execute("UPDATE jobs SET status='ready',attempts=0,generation=generation+1,available=?,last_error=NULL WHERE id=?",(self.clock(),job_id)) self.audit(db,'job_replayed',job_id) def purge(self, before): with self.transaction() as db: count=db.execute("""UPDATE events SET payload=NULL,purged_at=? WHERE created