"""Python 3.9+ raw-body HMAC + durable SQLite inbox. No third-party dependencies. Use --self-test offline, or --serve behind an HTTPS reverse proxy. No outbound calls. """ import argparse import hashlib import hmac import json import os import re import sqlite3 import tempfile from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path def open_inbox(filename): db = sqlite3.connect(filename, timeout=5) db.execute('PRAGMA journal_mode=WAL') db.execute('PRAGMA synchronous=FULL') db.execute('CREATE TABLE IF NOT EXISTS inbox(event_id TEXT PRIMARY KEY, digest TEXT NOT NULL, payload TEXT NOT NULL, processed INTEGER NOT NULL DEFAULT 0, received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)') db.commit() return db def accept(db, raw, signature, secret): if not isinstance(raw, bytes) or len(raw) > 65536: return {'status': 413} if not secret or not isinstance(signature, str) or not re.fullmatch(r'[a-fA-F0-9]{64}', signature): return {'status': 401} expected = hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature.lower()): return {'status': 401} try: event = json.loads(raw) except (ValueError, UnicodeDecodeError): return {'status': 400} if (not isinstance(event, dict) or event.get('type') != 'transfer.confirmed' or event.get('network') != 'mainnet' or event.get('asset') not in ('USDT', 'TRX') or not isinstance(event.get('event_id'), str) or not 1 <= len(event['event_id']) <= 200 or not isinstance(event.get('watch_id'), str) or not event['watch_id'] or not isinstance(event.get('amount_raw'), str) or not re.fullmatch(r'[0-9]+', event['amount_raw']) or not isinstance(event.get('amount'), str) or not re.fullmatch(r'[0-9]+(\.[0-9]+)?', event['amount'])): return {'status': 400} digest = hashlib.sha256(raw).hexdigest() with db: inserted = db.execute('INSERT OR IGNORE INTO inbox(event_id,digest,payload) VALUES(?,?,?)', (event['event_id'], digest, raw.decode('utf-8'))).rowcount existing = db.execute('SELECT digest FROM inbox WHERE event_id=?', (event['event_id'],)).fetchone() if existing[0] != digest: return {'status': 409} # Return success only after commit. Processing/credits belong in a separate durable worker. return {'status': 200, 'event_id': event['event_id'], 'duplicate': inserted == 0} def self_test(): with tempfile.TemporaryDirectory(prefix='tronhooks-example-') as directory: filename = str(Path(directory) / 'inbox.sqlite') raw = json.dumps({'event_id': 'offline-event', 'type': 'transfer.confirmed', 'network': 'mainnet', 'asset': 'USDT', 'amount': '9007199254740993.000001', 'amount_raw': '9007199254740993000001', 'watch_id': 'offline-watch'}).encode() secret = 'offline-fixture-secret-not-a-production-key' sign = lambda body: hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() db = open_inbox(filename) assert accept(db, raw, sign(raw), secret)['duplicate'] is False for signature in (None, 'xx', '0' * 64): assert accept(db, raw, signature, secret)['status'] == 401 assert accept(db, raw + b' ', sign(raw), secret)['status'] == 401 assert accept(db, raw, sign(raw), 'wrong-secret')['status'] == 401 assert accept(db, raw, sign(raw), secret)['duplicate'] is True assert accept(db, b'{', sign(b'{'), secret)['status'] == 400 invalid = json.dumps({**json.loads(raw), 'amount': 2}).encode() assert accept(db, invalid, sign(invalid), secret)['status'] == 400 conflict = json.dumps({**json.loads(raw), 'amount': '2'}).encode() assert accept(db, conflict, sign(conflict), secret)['status'] == 409 db.close() db = open_inbox(filename) assert accept(db, raw, sign(raw), secret)['duplicate'] is True assert db.execute('SELECT count(*) FROM inbox').fetchone()[0] == 1 assert json.loads(db.execute('SELECT payload FROM inbox').fetchone()[0])['amount'] == '9007199254740993.000001' db.close() try: accept(db, raw, sign(raw), secret) raise AssertionError('Closed storage must fail') except sqlite3.ProgrammingError: pass print('PASS: raw signatures, invalid input, durable replay, conflict, decimal precision, storage failure') def serve(): secret = os.environ.get('TRONHOOKS_WEBHOOK_SECRET', '') if len(secret) < 16: raise ValueError('Set TRONHOOKS_WEBHOOK_SECRET locally (at least 16 characters)') db = open_inbox(os.environ.get('INBOX_DB', './inbox.sqlite')) class Handler(BaseHTTPRequestHandler): def do_POST(self): if self.path != '/webhook': self.send_error(404) return if not self.headers.get('Content-Type', '').lower().startswith('application/json'): self.send_error(415) return # This minimal receiver requires Content-Length. Reject chunked/ambiguous framing. length = self.headers.get('Content-Length', '') if self.headers.get('Transfer-Encoding') or not length.isdigit() or len(self.headers.get_all('Content-Length', [])) != 1: self.send_error(400) return size = int(length) if size > 65536: self.send_error(413) return self.connection.settimeout(10) try: raw = self.rfile.read(size) if len(raw) != size: self.send_error(400) return result = accept(db, raw, self.headers.get('X-Signature'), secret) self.send_response(result['status']) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(result).encode()) except Exception: self.send_error(503) # Do not acknowledge data that was not durably stored. def log_message(self, *_args): pass # Do not log request paths, headers or credentials. HTTPServer(('127.0.0.1', int(os.environ.get('PORT', '8787'))), Handler).serve_forever() if __name__ == '__main__': parser = argparse.ArgumentParser() group = parser.add_mutually_exclusive_group(required=True) group.add_argument('--self-test', action='store_true') group.add_argument('--serve', action='store_true') options = parser.parse_args() self_test() if options.self_test else serve()