auth

azure_bootstrap.auth

Webhook + API-key authentication helpers.

Modules:

Name	Description
api_key	API-key header verification helper.
hmac	HMAC signature verification for webhooks.
webhook	Microsoft-Graph-style webhook authentication helpers.

Classes:

Name	Description
WebhookDedup	In-process dedup keyed on caller-supplied tuples.

Functions:

Name	Description
verify_api_key_header	FastAPI dependency. Raises HTTPException(401) on mismatch.
verify_hmac_signature	Constant-time HMAC-SHA256 verify (GitHub/Sumo sha256=… style).
install_graph_webhook_route	Register a Graph-flavored webhook route on the FastAPI app.
validation_token_handshake	Graph subscription-validation handshake — echo the token, or None.
verify_webhook_client_state	Constant-time comparison of received clientState against the configured value.

WebhookDedup

WebhookDedup(*, ttl_seconds: float = _DEFAULT_DEDUP_TTL_SECONDS, max_entries: int = _DEFAULT_DEDUP_MAX_ENTRIES)

In-process dedup keyed on caller-supplied tuples.

Thread-safe (single threading.Lock). Entries older than ttl_seconds are GC'd on every check; total entries capped at max_entries.

Methods:

Name	Description
reset	Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1.

Source code in azure_bootstrap/auth/webhook.py

def __init__(
    self,
    *,
    ttl_seconds: float = _DEFAULT_DEDUP_TTL_SECONDS,
    max_entries: int = _DEFAULT_DEDUP_MAX_ENTRIES,
) -> None:
    self._ttl = float(ttl_seconds)
    self._max_entries = int(max_entries)
    self._seen: dict[tuple[str, ...], float] = {}
    self._lock = threading.Lock()

reset

reset() -> None

Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1.

Source code in azure_bootstrap/auth/webhook.py

def reset(self) -> None:
    """Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1."""
    if os.environ.get("AZURE_BOOTSTRAP_ALLOW_RESET") != "1":
        raise RuntimeError(
            "WebhookDedup.reset is test-only — set AZURE_BOOTSTRAP_ALLOW_RESET=1"
        )
    with self._lock:
        self._seen.clear()

verify_api_key_header async

verify_api_key_header(x_api_key: str | None, *, env_var: str = 'API_KEY', fail_open_when_unset: bool = True) -> None

FastAPI dependency. Raises HTTPException(401) on mismatch.

When fail_open_when_unset is True (default) and the env var is unset or empty, the check passes — matches the v1 reference behavior. Strict mode (env required) is opt-in via fail_open_when_unset=False.

Imports FastAPI lazily so this module is importable without the fastapi extra; only callers that actually invoke the function pay the dep.

Source code in azure_bootstrap/security/__init__.py

async def verify_api_key_header(
    x_api_key: str | None,
    *,
    env_var: str = "API_KEY",
    fail_open_when_unset: bool = True,
) -> None:
    """FastAPI dependency. Raises ``HTTPException(401)`` on mismatch.

    When ``fail_open_when_unset`` is True (default) and the env var is unset
    or empty, the check passes — matches the v1 reference behavior. Strict
    mode (env required) is opt-in via ``fail_open_when_unset=False``.

    Imports FastAPI lazily so this module is importable without the ``fastapi``
    extra; only callers that actually invoke the function pay the dep.
    """
    import os

    expected = os.environ.get(env_var, "").strip()
    if not expected:
        if fail_open_when_unset:
            return
        from fastapi import HTTPException  # type: ignore[import-not-found]

        raise HTTPException(status_code=401, detail="API key not configured")
    if not compare_secrets(x_api_key, expected):
        _logger.debug(
            "API key validation failed",
            extra={"operation": "verify_api_key_header"},
        )
        from fastapi import HTTPException  # type: ignore[import-not-found]

        raise HTTPException(status_code=401, detail="Unauthorized")

verify_hmac_signature

verify_hmac_signature(secret: str, raw_body: bytes, header_value: str, *, prefix: str = 'sha256=') -> bool

Constant-time HMAC-SHA256 verify (GitHub/Sumo sha256=… style).

Parameters

secret: Shared signing secret. raw_body: Raw request body bytes (must not be re-serialized JSON). header_value: Value of the signature header (may include sha256= prefix). prefix: Expected algorithm prefix in the header value.

Source code in azure_bootstrap/auth/hmac.py

def verify_hmac_signature(
    secret: str,
    raw_body: bytes,
    header_value: str,
    *,
    prefix: str = "sha256=",
) -> bool:
    """Constant-time HMAC-SHA256 verify (GitHub/Sumo ``sha256=…`` style).

    Parameters
    ----------
    secret:
        Shared signing secret.
    raw_body:
        Raw request body bytes (must not be re-serialized JSON).
    header_value:
        Value of the signature header (may include ``sha256=`` prefix).
    prefix:
        Expected algorithm prefix in the header value.
    """
    if not secret or not header_value:
        return False
    provided = header_value.strip()
    if provided.startswith(prefix):
        provided = provided[len(prefix) :]
    elif "=" in provided:
        provided = provided.split("=", 1)[1]
    expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, provided.lower())

install_graph_webhook_route

install_graph_webhook_route(app: Any, path: str, *, background_handler: Callable[[str], None], rate_limit_bucket: Any | None = None, dedup: WebhookDedup | None = None, counter_namespace: str = 'webhook') -> None

Register a Graph-flavored webhook route on the FastAPI app.

Pipeline order: validation token → rate limit → JSON parse → per-entry clientState → dedup → background dispatch → 202 Accepted.

Source code in azure_bootstrap/auth/webhook.py

def install_graph_webhook_route(
    app: Any,
    path: str,
    *,
    background_handler: Callable[[str], None],
    rate_limit_bucket: Any | None = None,
    dedup: WebhookDedup | None = None,
    counter_namespace: str = "webhook",
) -> None:
    """Register a Graph-flavored webhook route on the FastAPI app.

    Pipeline order: validation token → rate limit → JSON parse → per-entry
    clientState → dedup → background dispatch → 202 Accepted.
    """
    try:
        from fastapi import (  # type: ignore[import-not-found]
            BackgroundTasks,
            Request,
            Response,
        )
        from fastapi.responses import PlainTextResponse  # type: ignore[import-not-found]
    except ImportError as exc:  # pragma: no cover
        raise ImportError(
            "install_graph_webhook_route requires the `fastapi` extra: "
            "pip install azure-bootstrap[fastapi]"
        ) from exc

    @app.post(path, include_in_schema=False)
    async def _webhook(  # type: ignore[no-untyped-def]
        request: Request,
        background_tasks: BackgroundTasks,
    ) -> Response:
        # 1. Validation-token handshake (subscription creation)
        token = validation_token_handshake(request.query_params.get("validationToken"))
        if token is not None:
            bump_counter(f"{counter_namespace}.validation_token_handshake")
            return PlainTextResponse(token, status_code=200)

        # 2. Rate limit
        if rate_limit_bucket is not None and not rate_limit_bucket.consume(1.0):
            bump_counter(f"{counter_namespace}.rate_limited")
            return Response(status_code=429)

        # 3. Parse JSON
        try:
            payload = await request.json()
        except Exception:
            return Response(status_code=400)

        # 4. Iterate `value` entries
        entries = payload.get("value", []) if isinstance(payload, dict) else []
        for entry in entries:
            if not isinstance(entry, dict):
                continue
            received_state = entry.get("clientState")
            try:
                ok = verify_webhook_client_state(received_state)
            except ConfigurationError:
                # Endpoint unconfigured — refuse all entries; respond 401 no body.
                bump_counter(f"{counter_namespace}.client_state_mismatch")
                return Response(status_code=401)
            if not ok:
                bump_counter(f"{counter_namespace}.client_state_mismatch")
                return Response(status_code=401)

            resource_data = entry.get("resourceData") or {}
            subscription_id = entry.get("subscriptionId") or ""
            message_id = resource_data.get("id", "")
            if not message_id:
                continue
            key = (str(subscription_id), str(message_id))
            if dedup is not None and dedup.already_seen(key):
                _logger.info(
                    "webhook duplicate suppressed",
                    extra={
                        "operation": "webhook.dedup_skipped",
                        "subscription_id": subscription_id,
                        "message_id": message_id,
                    },
                )
                bump_counter(f"{counter_namespace}.dedup_skipped")
                continue
            bump_counter(f"{counter_namespace}.received")
            background_tasks.add_task(background_handler, message_id)

        return Response(status_code=202)

validation_token_handshake

validation_token_handshake(validation_token: str | None) -> str | None

Graph subscription-validation handshake — echo the token, or None.

Source code in azure_bootstrap/auth/webhook.py

def validation_token_handshake(validation_token: str | None) -> str | None:
    """Graph subscription-validation handshake — echo the token, or None."""
    if validation_token is None or validation_token == "":
        return None
    return validation_token

verify_webhook_client_state

verify_webhook_client_state(received_client_state: str | None, *, env_var: str = 'GRAPH_WEBHOOK_CLIENT_STATE') -> bool

Constant-time comparison of received clientState against the configured value.

Raises :class:ConfigurationError when env_var is unset — the webhook endpoint MUST be configured before accepting any requests.

Source code in azure_bootstrap/auth/webhook.py

def verify_webhook_client_state(
    received_client_state: str | None,
    *,
    env_var: str = "GRAPH_WEBHOOK_CLIENT_STATE",
) -> bool:
    """Constant-time comparison of received clientState against the configured value.

    Raises :class:`ConfigurationError` when ``env_var`` is unset — the webhook
    endpoint MUST be configured before accepting any requests.
    """
    expected = os.environ.get(env_var, "").strip()
    if not expected:
        raise ConfigurationError(f"webhook clientState env var {env_var!r} is not configured")
    return compare_secrets(received_client_state, expected)