Verifying OpenFence webhook signatures

Copyimport hmac
import hashlib
import re
import time

# OpenFence MANDATES a 5-minute receiver freshness window.  The kwarg
# below exists for testability; production receivers must not pass a
# value above this floor.
_OPENFENCE_MAX_TOLERANCE_SECONDS = 300

# v1 signatures are 64 lowercase-hex chars (SHA-256 digest).
_V1_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")


def verify_openfence_signature(
    *,
    signing_secret: str,
    body: bytes,
    signature_header: str,
    timestamp_header: str,
    tolerance_seconds: int = _OPENFENCE_MAX_TOLERANCE_SECONDS,
    now_unix: int | None = None,
) -> bool:
    """Verify an OpenFence webhook signature.

    Returns True iff the signature matches AND the timestamp is within
    ``tolerance_seconds`` of now.  Returns False on any malformed input
    or mismatch.  This function NEVER raises to the caller — the caller
    is responsible for translating a False return into a 401 / drop.

    Arguments:
        signing_secret: the subscription's signing secret (the same
            value you registered with OpenFence).
        body: the RAW request body bytes — captured BEFORE any HTTP
            framework parses, decodes, or otherwise transforms the
            body.  Do not JSON-parse-and-re-serialize; whitespace and
            key-ordering differences will break verification.
        signature_header: the value of the X-OpenFence-Signature header,
            e.g. "t=1714780000,v1=abc123...".
        timestamp_header: the value of the X-OpenFence-Timestamp header.
        tolerance_seconds: how far the timestamp may be from now()
            before the delivery is treated as stale.  Defaults to 300s
            (5 minutes) — OpenFence's MUST.  Values ABOVE 300 are a
            spec violation and are rejected at call time; values below
            300 are permitted (a tighter window is always safe).
        now_unix: optional fixed wall-clock timestamp (seconds) to use
            for the freshness check.  Production callers leave this
            None; tests pass a fixed value to exercise the freshness
            window deterministically without monkey-patching
            ``time.time()``.
    """
    if tolerance_seconds > _OPENFENCE_MAX_TOLERANCE_SECONDS:
        raise ValueError(
            "tolerance_seconds exceeds OpenFence's 5-minute MUST; "
            "tighter windows are permitted, looser are not."
        )

    try:
        # Parse all `key=value` segments; reject duplicates and
        # malformed segments.  We tolerate unknown segments (e.g. a
        # future v2=...) and pick v1 explicitly — when OpenFence
        # ships v2, this sample needs updating to prefer the highest
        # version present.
        parts: dict[str, str] = {}
        for raw in signature_header.split(","):
            seg = raw.strip()
            if "=" not in seg:
                return False
            k, v = seg.split("=", 1)
            if k in parts:
                return False  # duplicate key — header tampering
            parts[k] = v

        if "t" not in parts or "v1" not in parts:
            return False

        t = int(parts["t"])  # ValueError caught below
        v1 = parts["v1"]

        # Pin the v1 hex shape — defense-in-depth around compare_digest.
        if not _V1_DIGEST_RE.fullmatch(v1):
            return False

        # Sibling header sanity check (the t= segment above is the
        # authoritative timestamp; this is a tamper-detection cross-check).
        if int(timestamp_header) != t:
            return False

        # Freshness window — symmetric: reject too-future as well as
        # too-past (clock-skew defense).
        now = now_unix if now_unix is not None else int(time.time())
        if abs(now - t) > tolerance_seconds:
            return False

        # Recompute over BYTES, not over a re-decoded string — the
        # producer signs `f"{t}.".encode("ascii") + body_bytes`.
        signed_input = f"{t}.".encode("ascii") + body
        expected = hmac.new(
            signing_secret.encode("utf-8"),
            signed_input,
            hashlib.sha256,
        ).hexdigest()
        return hmac.compare_digest(expected, v1)
    except (ValueError, TypeError, UnicodeDecodeError, AttributeError):
        # Any malformed input (non-integer t, missing/None timestamp
        # header, undecodable bytes anywhere) becomes a False return.
        # Per the docstring, this function does not raise to the caller.
        return False