Webhook handling with signature verification and idempotency

import hashlib
import hmac
from fastapi import APIRouter, Request, HTTPException

router = APIRouter()

def verify_hmac_signature(
    payload: bytes,
    signature: str,
    secret: str,
    algorithm: str = 'sha256',
    prefix: str = 'sha256=',
) -> bool:
    """Verify HMAC signature (GitHub/Stripe style)."""
    expected = hmac.new(
        secret.encode(),
        payload,
        getattr(hashlib, algorithm),
    ).hexdigest()
    provided = signature.removeprefix(prefix)
    return hmac.compare_digest(expected, provided)  # Timing-safe comparison

@router.post('/webhooks/github')
async def github_webhook(
    request: Request,
    db: DbSession,
    x_hub_signature_256: str = Header(None),
):
    payload = await request.body()

    if not verify_hmac_signature(
        payload,
        x_hub_signature_256 or '',
        settings.github_webhook_secret,
        prefix='sha256=',
    ):
        raise HTTPException(401, 'Invalid signature')

    event = request.headers.get('X-GitHub-Event')
    delivery_id = request.headers.get('X-GitHub-Delivery')

    # Idempotency:
    if await is_processed(db, delivery_id):
        return {'status': 'duplicate'}

    body = await request.json()
    match event:
        case 'push': await handle_push(body)
        case 'pull_request': await handle_pr(body)

    await mark_processed(db, delivery_id)
    return {'status': 'ok'}