procurement-ingest/lambdas/shared/web_ui_auth.py

import hmac
import logging
import os
import time

import boto3

logger = logging.getLogger()


_WEB_UI_AUTH_TOKEN_SECRET_ARN = os.environ.get("WEB_UI_AUTH_TOKEN_SECRET_ARN")
# Refresh the cached token this often so a rotated secret propagates without
# waiting for the execution environment to recycle (emergency-rotation path).
_AUTH_TOKEN_CACHE_TTL_SECONDS = 300
_auth_token_cache = None
_auth_token_cached_at = 0.0


def _get_auth_token() -> str | None:
    """Fetch the shared web UI auth token from Secrets Manager.

    Cached in the warm container for a short TTL so we don't hit Secrets Manager
    on every request, while still picking up a rotated secret within the TTL
    rather than only when the execution environment recycles. Returns None when
    not configured or unreadable (the caller then fails closed).
    """
    global _auth_token_cache, _auth_token_cached_at
    now = time.monotonic()
    if (
        _auth_token_cache is not None
        and now - _auth_token_cached_at < _AUTH_TOKEN_CACHE_TTL_SECONDS
    ):
        return _auth_token_cache
    if not _WEB_UI_AUTH_TOKEN_SECRET_ARN:
        return None
    secrets = boto3.client("secretsmanager")
    try:
        secret = secrets.get_secret_value(SecretId=_WEB_UI_AUTH_TOKEN_SECRET_ARN)
        _auth_token_cache = secret["SecretString"]
        _auth_token_cached_at = now
        return _auth_token_cache
    except Exception:
        # Fail closed (return None -> caller 401s) but surface the failure: a
        # Secrets Manager permission/config error would otherwise make every
        # request 401 with no operational signal. The secret value is never
        # logged.
        logger.exception(
            "Failed to fetch web UI auth token from Secrets Manager; "
            "denying request (failing closed)"
        )
        return None


def _header(event: dict, name: str) -> str:
    """Case-insensitive header lookup from a Lambda Function URL / APIGW event."""
    headers = event.get("headers") or {}
    name_lower = name.lower()
    for key, value in headers.items():
        if key.lower() == name_lower:
            return value or ""
    return ""


def is_authenticated(event: dict) -> bool:
    """Constant-time check of the request's shared secret against the configured
    token. Fails closed when no token is configured."""
    token = _get_auth_token()
    if not token:
        return False
    presented = _header(event, "x-auth-token")
    if not presented:
        auth = _header(event, "authorization")
        if auth.lower().startswith("bearer "):
            presented = auth[7:].strip()
    if not presented:
        return False
    # Compare as bytes: hmac.compare_digest raises TypeError on non-ASCII str
    # operands, which a crafted token (?token=%C3%A9 or a non-ASCII header)
    # would otherwise turn into an uncaught 500. Bytes always compare in
    # constant time, so a non-matching token fails closed (401) instead.
    return hmac.compare_digest(presented.encode("utf-8"), token.encode("utf-8"))