|
1 | | -import hashlib |
2 | | -import hmac |
| 1 | +import os |
| 2 | +from wsgiref.simple_server import make_server |
3 | 3 |
|
4 | 4 | import mailtrap as mt |
5 | 5 |
|
6 | | -# --- Direct verification (e.g. for unit tests or custom routers) ---------- |
7 | | -payload = '{"event":"delivery","message_id":"abc-123"}' |
8 | | -signing_secret = "8d9a3c0e7f5b2d4a6c1e9f8b3a7d5c2e" |
9 | | -signature = hmac.new( |
10 | | - signing_secret.encode("utf-8"), |
11 | | - payload.encode("utf-8"), |
12 | | - hashlib.sha256, |
13 | | -).hexdigest() |
14 | | - |
15 | | -if not mt.verify_signature(payload, signature, signing_secret): |
16 | | - raise SystemExit("Signature verification failed!") |
| 6 | +SIGNING_SECRET = os.environ["MAILTRAP_WEBHOOK_SIGNING_SECRET"] |
| 7 | + |
| 8 | + |
| 9 | +def app(environ, start_response): |
| 10 | + # Use the raw request body — parsing and re-serializing the JSON may |
| 11 | + # reorder keys or alter whitespace and invalidate the signature. |
| 12 | + length = int(environ.get("CONTENT_LENGTH") or 0) |
| 13 | + payload = environ["wsgi.input"].read(length).decode("utf-8") |
| 14 | + signature = environ.get("HTTP_MAILTRAP_SIGNATURE", "") |
| 15 | + |
| 16 | + if not mt.verify_signature(payload, signature, SIGNING_SECRET): |
| 17 | + start_response("401 Unauthorized", [("Content-Type", "text/plain")]) |
| 18 | + return [b"Invalid signature"] |
| 19 | + |
| 20 | + start_response("200 OK", [("Content-Type", "text/plain")]) |
| 21 | + return [b""] |
| 22 | + |
| 23 | + |
| 24 | +if __name__ == "__main__": |
| 25 | + with make_server("", 9292, app) as server: |
| 26 | + server.serve_forever() |
0 commit comments