Webhook Receiver

DWALLET_WEBHOOK_SECRET=
WEBHOOK_HOST=127.0.0.1
WEBHOOK_PORT=8080

aiohttp>=3.9
python-dotenv>=1.0

import os, json, hmac, hashlib, sqlite3
from aiohttp import web
from dotenv import load_dotenv

load_dotenv()
SECRET = os.environ["DWALLET_WEBHOOK_SECRET"].encode()
DB = sqlite3.connect("webhook.db")
DB.execute("CREATE TABLE IF NOT EXISTS deliveries(id TEXT PRIMARY KEY, event_type TEXT NOT NULL)")
DB.commit()

def verify(raw: bytes, timestamp: str, signature: str) -> bool:
    if not signature.startswith("v1="):
        return False
    digest = hmac.new(SECRET, timestamp.encode("ascii") + b"." + raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(signature[3:], digest)

async def webhook(request: web.Request):
    raw = await request.read()
    timestamp = request.headers.get("X-D-Wallet-Timestamp", "")
    signature = request.headers.get("X-D-Wallet-Signature", "")
    delivery_id = request.headers.get("X-D-Wallet-Delivery-ID", "")
    if not delivery_id or not verify(raw, timestamp, signature):
        raise web.HTTPUnauthorized()

    payload = json.loads(raw)
    try:
        DB.execute(
            "INSERT INTO deliveries(id,event_type) VALUES(?,?)",
            (delivery_id, str(payload.get("type", ""))),
        )
        DB.commit()
    except sqlite3.IntegrityError:
        return web.json_response({"ok": True, "duplicate": True})

    if payload.get("type") == "payment.completed":
        payment = payload.get("data", {})
        print("payment.completed", payment.get("payment_id"))

    return web.json_response({"ok": True})

app = web.Application()
app.router.add_post("/dwallet/webhook", webhook)

web.run_app(
    app,
    host=os.getenv("WEBHOOK_HOST", "127.0.0.1"),
    port=int(os.getenv("WEBHOOK_PORT", "8080")),
)