import os
import re
import jwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
WEBHOOK_PUBLIC_KEY_PEM = os.environ["ATONOM_WEBHOOK_PUBLIC_KEY"] # full PEM string
_bearer_re = re.compile(r"^Bearer\s+(.+)$", re.IGNORECASE)
def verify_atonom_webhook(authorization_header: str | None) -> dict:
if not authorization_header:
raise ValueError("Missing Authorization header")
m = _bearer_re.match(authorization_header.strip())
if not m:
raise ValueError("Authorization must be a Bearer token")
token = m.group(1).strip()
public_key = serialization.load_pem_public_key(
WEBHOOK_PUBLIC_KEY_PEM.encode("utf-8"),
backend=default_backend(),
)
payload = jwt.decode(
token,
public_key,
algorithms=["PS256"],
options={"require": ["exp", "iat", "tenant_id"]},
leeway=60, # seconds; optional clock skew tolerance
)
return payload