#!/usr/bin/env python3 """Dependency-free Signal Trade webhook receiver with a durable SQLite inbox. Expose this loopback server only through a public HTTPS reverse proxy. It stores a verified delivery before acknowledging it; a dealer application must read the inbox and apply each event to its own records. """ from __future__ import annotations import hashlib import hmac import json import os import re import sqlite3 import time from contextlib import closing from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any MAX_BODY_BYTES = 1024 * 1024 REPLAY_WINDOW_SECONDS = 300 secret = os.environ.get("SIGNAL_TRADE_WEBHOOK_SECRET", "") host = os.environ.get("HOST", "127.0.0.1") port = int(os.environ.get("PORT", "8788")) database_path = Path(os.environ.get("SIGNAL_TRADE_WEBHOOK_DB", "signal-trade-webhooks.db")) if len(secret) < 32: raise RuntimeError("Set SIGNAL_TRADE_WEBHOOK_SECRET to the signing secret from Signal Trade settings.") if not 1 <= port <= 65535: raise RuntimeError("PORT must be a TCP port from 1 to 65535.") def connect() -> sqlite3.Connection: connection = sqlite3.connect(database_path, timeout=10) connection.execute("PRAGMA busy_timeout = 10000") connection.execute("PRAGMA synchronous = FULL") return connection database_path.parent.mkdir(parents=True, exist_ok=True) with closing(connect()) as database: with database: database.execute("PRAGMA journal_mode = WAL") database.execute("PRAGMA synchronous = FULL") database.execute( """CREATE TABLE IF NOT EXISTS webhook_inbox ( delivery_id TEXT PRIMARY KEY, event_id TEXT NOT NULL, event_name TEXT NOT NULL, received_at TEXT NOT NULL, body_sha256 TEXT NOT NULL, payload_json TEXT NOT NULL, processed_at TEXT )""" ) database.execute("CREATE INDEX IF NOT EXISTS idx_webhook_inbox_pending ON webhook_inbox(processed_at, received_at)") def verify_signature(raw_body: bytes, timestamp: str, signature: str, now_seconds: int | None = None) -> None: if not re.fullmatch(r"\d{1,12}", timestamp): raise ValueError("Invalid timestamp") now = int(time.time()) if now_seconds is None else now_seconds if abs(now - int(timestamp)) > REPLAY_WINDOW_SECONDS: raise ValueError("Expired timestamp") match = re.fullmatch(r"v1=([0-9a-fA-F]{64})", signature) if not match: raise ValueError("Invalid signature format") expected = hmac.new(secret.encode("utf-8"), timestamp.encode("ascii") + b"." + raw_body, hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, match.group(1).lower()): raise ValueError("Signature mismatch") def response(handler: BaseHTTPRequestHandler, status: int, message: str = "") -> None: body = message.encode("utf-8") handler.send_response(status) handler.send_header("Content-Type", "text/plain; charset=utf-8") handler.send_header("Cache-Control", "no-store") handler.send_header("Content-Length", str(len(body))) handler.end_headers() if body: handler.wfile.write(body) class WebhookHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def do_POST(self) -> None: if self.path != "/webhooks/signal": response(self, 404, "Not found") return length_text = self.headers.get("Content-Length", "") if not re.fullmatch(r"\d{1,10}", length_text): self.close_connection = True response(self, 411, "Content-Length required") return length = int(length_text) if length > MAX_BODY_BYTES: self.close_connection = True response(self, 413, "Payload too large") return raw_body = self.rfile.read(length) if len(raw_body) != length: self.close_connection = True response(self, 400, "Incomplete request body") return timestamp = self.headers.get("X-Signal-Webhook-Timestamp", "") signature = self.headers.get("X-Signal-Webhook-Signature", "") delivery_id = self.headers.get("X-Signal-Webhook-Delivery", "") try: verify_signature(raw_body, timestamp, signature) payload: Any = json.loads(raw_body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): response(self, 400, "Invalid JSON") return except ValueError: response(self, 401, "Webhook rejected") return if ( not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", delivery_id) or not isinstance(payload, dict) or not isinstance(payload.get("id"), str) or not payload["id"] or len(payload["id"]) > 128 or not isinstance(payload.get("event"), str) or not payload["event"] or len(payload["event"]) > 120 ): response(self, 400, "Invalid event envelope") return received_at = datetime.now(timezone.utc).isoformat() body_sha256 = hashlib.sha256(raw_body).hexdigest() payload_json = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) try: with closing(connect()) as database, database: database.execute( "INSERT INTO webhook_inbox(delivery_id,event_id,event_name,received_at,body_sha256,payload_json) VALUES(?,?,?,?,?,?)", (delivery_id, payload["id"], payload["event"], received_at, body_sha256, payload_json), ) except sqlite3.IntegrityError: try: with closing(connect()) as database: existing = database.execute( "SELECT event_id,body_sha256 FROM webhook_inbox WHERE delivery_id=?", (delivery_id,) ).fetchone() except sqlite3.Error: response(self, 503, "Inbox unavailable") return if existing and existing[0] == payload["id"] and existing[1] == body_sha256: response(self, 204) else: response(self, 409, "Delivery ID conflicts with a stored event") return except sqlite3.Error: response(self, 503, "Inbox unavailable") return # A 2xx is returned only after SQLite has durably committed the event. # Downstream processing should mark it processed after applying it. response(self, 204) def log_message(self, format_string: str, *args: object) -> None: # Avoid writing request details or delivery metadata to standard logs. return if __name__ == "__main__": server = ThreadingHTTPServer((host, port), WebhookHandler) server.daemon_threads = True print(f"Signal Trade webhook inbox listening on {host}:{port}/webhooks/signal") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close()