import hashlib
import hmac
import json
import os
import time
from flask import Flask, jsonify, request
app = Flask(__name__)
SIGNING_SECRET = os.environ["SECAPI_WEBHOOK_SIGNING_SECRET"]
MAX_SKEW_SECONDS = 300
def parse_signature(header: str) -> tuple[str, str]:
parts = {}
for item in header.split(","):
if "=" in item:
key, value = item.split("=", 1)
parts[key] = value
return parts.get("t", ""), parts.get("v1", "")
def verify_signature(raw_body: bytes, header: str) -> bool:
timestamp, provided = parse_signature(header)
if not timestamp or not provided:
return False
try:
signed_at = int(timestamp)
except ValueError:
return False
if abs(int(time.time()) - signed_at) > MAX_SKEW_SECONDS:
return False
expected = hmac.new(
SIGNING_SECRET.encode("utf-8"),
f"{timestamp}.".encode("utf-8") + raw_body,
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(provided, expected)
@app.post("/webhooks/secapi")
def handle_secapi_webhook():
raw_body = request.get_data()
signature = request.headers.get("x-secapi-signature", "")
if not verify_signature(raw_body, signature):
return jsonify({"error": "invalid_signature"}), 401
event = json.loads(raw_body)
if event.get("type") != "monitor.match":
return jsonify({"status": "ignored", "type": event.get("type")}), 200
data = event.get("data", {})
monitor = data.get("monitor", {})
for match in data.get("matches", []):
print(
f"{monitor.get('name')} matched "
f"{match.get('form', 'filing')} for {match.get('ticker', 'UNKNOWN')}: "
f"{match.get('htmlUrl') or match.get('id')}"
)
return jsonify({"received": True}), 200
@app.get("/health")
def health():
return jsonify({"status": "ok"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
