"""HookGet — webhook signature verification for Python.

One file, standard library only, Python 3.8+. Copy it into your project;
there is deliberately nothing to keep up to date.

    from hookget import verify, VerificationError

    @app.post("/hooks")
    def hooks():
        try:
            verify(os.environ["HOOKGET_SECRET"], request.headers, request.get_data())
        except VerificationError as err:
            return str(err), 400
        return "", 200          # acknowledge first, work afterwards

── THE ONE MISTAKE EVERYONE MAKES ────────────────────────────────────────────
Verify the RAW BODY — the exact bytes that arrived. Not ``json.dumps(
request.json)``, which re-serialises a parsed object and will differ from what
was signed the first time a payload contains a non-ASCII character, a float
that round-trips differently, or keys your parser reordered. In Flask that is
``request.get_data()``; in Django ``request.body``; in FastAPI ``await
request.body()``. Every "the signature does not match and I cannot see why"
ends here.

See https://www.standardwebhooks.com
License: Apache-2.0
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import time
from typing import Iterable, Mapping, NamedTuple, Optional, Sequence, Union

SECRET_PREFIX = "whsec_"
DEFAULT_TOLERANCE_SECONDS = 300


class VerificationError(Exception):
    """Raised for every failure, carrying a ``reason`` you can branch on."""

    def __init__(self, reason: str, message: str) -> None:
        super().__init__(message)
        self.reason = reason


class Verified(NamedTuple):
    id: str
    timestamp: int


def _secret_to_key(secret: str) -> bytes:
    """The secret is base64 and the HMAC key is its BYTES, not its text.

    Hashing the base64 string itself is the second most common bug here, and it
    produces a signature that is stable, plausible, and wrong.
    """
    body = secret[len(SECRET_PREFIX):] if secret.startswith(SECRET_PREFIX) else secret
    return base64.b64decode(body)


def sign(secret: str, id: str, timestamp: int, body: Union[str, bytes]) -> str:
    """The signed content is always ``{id}.{timestamp}.{body}``."""
    raw = body.encode("utf-8") if isinstance(body, str) else body
    signed = f"{id}.{timestamp}.".encode("utf-8") + raw
    digest = hmac.new(_secret_to_key(secret), signed, hashlib.sha256).digest()
    return base64.b64encode(digest).decode("ascii")


def _header(headers: Mapping[str, object], name: str) -> Optional[str]:
    # Header names are case-insensitive, and WSGI/ASGI frameworks disagree about
    # which case they hand you. Try the mapping's own lookup first — Werkzeug
    # and Starlette are already case-insensitive — then fall back to a scan.
    value = headers.get(name)
    if value is None:
        for key, candidate in headers.items():
            if key.lower() == name:
                value = candidate
                break
    if value is None:
        return None
    if isinstance(value, (list, tuple)):
        value = value[0] if value else None
    return None if value is None else str(value)


def verify(
    secrets: Union[str, Sequence[str]],
    headers: Mapping[str, object],
    body: Union[str, bytes],
    tolerance_seconds: int = DEFAULT_TOLERANCE_SECONDS,
    now: Optional[int] = None,
) -> Verified:
    """Verifies a delivery, or raises :class:`VerificationError`."""
    secret_list: Iterable[str] = [secrets] if isinstance(secrets, str) else secrets

    id_ = _header(headers, "webhook-id")
    raw_timestamp = _header(headers, "webhook-timestamp")
    signature_header = _header(headers, "webhook-signature")
    if not id_ or not raw_timestamp or not signature_header:
        raise VerificationError(
            "missing_headers",
            "missing webhook-id, webhook-timestamp or webhook-signature",
        )

    try:
        timestamp = int(raw_timestamp)
    except ValueError:
        raise VerificationError("bad_timestamp", "webhook-timestamp is not an integer") from None

    # A replay window. Without it a captured delivery stays valid forever, and
    # an attacker who ever saw one valid request can send it again at will.
    current = int(time.time()) if now is None else now
    if abs(current - timestamp) > tolerance_seconds:
        raise VerificationError(
            "timestamp_out_of_tolerance",
            f"timestamp is more than {tolerance_seconds}s away from now",
        )

    # Space-separated during rotation: both the old and the new secret sign the
    # same body, so a rotation never drops a delivery.
    presented = [p for p in signature_header.split(" ") if p]

    for secret in secret_list:
        expected = sign(secret, id_, timestamp, body)
        for candidate in presented:
            version, _, value = candidate.partition(",")
            if version != "v1" or not value:
                continue
            # compare_digest and not ==: a comparison that returns early leaks
            # the signature one byte at a time to anyone willing to measure.
            if hmac.compare_digest(value, expected):
                return Verified(id=id_, timestamp=timestamp)

    raise VerificationError("no_matching_signature", "no presented signature matched")


def is_valid(
    secrets: Union[str, Sequence[str]],
    headers: Mapping[str, object],
    body: Union[str, bytes],
    tolerance_seconds: int = DEFAULT_TOLERANCE_SECONDS,
    now: Optional[int] = None,
) -> bool:
    """Non-throwing form, for callers that prefer a boolean."""
    try:
        verify(secrets, headers, body, tolerance_seconds, now)
        return True
    except VerificationError:
        return False