Python webhook signature verifier

python3 make-test.py events/test-ping.json private/request-1

python3 verify.py private/request-1/body.json private/request-1/headers.json

#!/usr/bin/env python3
"""Verify a notification, not license validity. Python 3.11+, standard library.
Run: KDP_WEBHOOK_SECRET=YOUR_WEBHOOK_SECRET KDP_WEBHOOK_ENVIRONMENT=SANDBOX python3 verify.py body.json headers.json
Expected: JSON verified=true for valid bytes; verified=false and exit 1 otherwise.
"""
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import sys
import time


def verify(body: bytes, headers: dict, secret: str, environment: str, now: int) -> dict:
    contract = json.loads(Path(__file__).with_name('contract.json').read_text())
    if not secret or environment not in contract['environments'] or len(body) > contract['body_limit']:
        raise ValueError('Invalid configuration or body size')
    signature = headers.get('x-lmp-signature')
    if not isinstance(signature, str):
        raise ValueError('Missing signature')
    match = re.fullmatch(r't=([1-9][0-9]{0,10}),v1=([0-9a-f]{64})', signature)
    if not match or abs(now - int(match[1])) > contract['tolerance_seconds']:
        raise ValueError('Invalid or stale signature')
    message = match[1].encode('ascii') + b'.' + body
    digest = hmac.new(secret.encode('utf-8'), message, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(digest, match[2]):
        raise ValueError('Signature mismatch')
    event = json.loads(body.decode('utf-8', errors='strict'))
    if not isinstance(event, dict) or not isinstance(event.get('id'), str):
        raise ValueError('Invalid event')
    if not re.fullmatch(r'evt_[A-Za-z0-9_-]{1,100}', event['id']):
        raise ValueError('Invalid event ID')
    if event.get('event') not in contract['events'] or event.get('environment') != environment:
        raise ValueError('Wrong event or environment')
    if event['id'] != headers.get('x-lmp-event-id') or event['event'] != headers.get('x-lmp-event'):
        raise ValueError('Header and body mismatch')
    if not isinstance(event.get('data'), dict) or not isinstance(event.get('created_at'), str) or not event['created_at']:
        raise ValueError('Invalid event envelope')
    return event


def main() -> int:
    try:
        if len(sys.argv) != 3:
            raise ValueError('Two file paths required')
        with open(sys.argv[1], 'rb') as stream:
            body = stream.read(1048577)
        with open(sys.argv[2], 'rb') as stream:
            raw_headers = stream.read(32769)
        if len(raw_headers) > 32768:
            raise ValueError('Headers too large')
        headers = json.loads(raw_headers.decode('utf-8', errors='strict'))
        if not isinstance(headers, dict):
            raise ValueError('Invalid headers')
        event = verify(body, headers, os.environ.get('KDP_WEBHOOK_SECRET', ''),
                       os.environ.get('KDP_WEBHOOK_ENVIRONMENT', '').lower(), int(time.time()))
        print(json.dumps({'verified': True, 'event_id': event['id'], 'event': event['event'],
                          'environment': event['environment']}))
        return 0
    except (OSError, ValueError, TypeError, KeyError, OverflowError):
        print('{"verified":false}')
        return 1


if __name__ == '__main__':
    raise SystemExit(main())