meal-order-manager/functions/admin_authorizer/handler.py

"""API Gateway (HTTP API) Lambda authorizer for the meal-order-manager admin API.

Gates every ``/api/admin/*`` route at the gateway. The authorizer accepts the
existing Google ID token or a portal Cognito ID token in the Authorization
header, then confirms the caller is in the ``admin_emails`` allow-list
(DynamoDB CONFIG/SETTINGS).

Returns the HTTP API v2 *simple response* shape (``{"isAuthorized": bool}``);
a False result causes API Gateway to return 403 before the integration runs.
The in-handler ``_verify_admin`` check stays in place as defense-in-depth.
"""

import json
import logging
import os
import sys
import urllib.error
import urllib.parse
import urllib.request

from shared.cognito import (
    CognitoVerificationUnavailable,
    looks_like_cognito_token,
    verify_cognito_id_token,
)
from shared.db import get_settings
from shared.secrets import get_parameter

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
if not logger.handlers:
    logger.addHandler(logging.StreamHandler(sys.stderr))

ALLOWED_DOMAINS = {"seahavenind.com", "seahaven.com"}

_DENY = {"isAuthorized": False}
_ALLOW = {"isAuthorized": True}


def _get_google_client_id() -> str:
    param = os.environ.get("GOOGLE_CLIENT_ID_PARAM", "")
    if not param:
        return ""
    try:
        return get_parameter(param, decrypt=False) or ""
    except Exception as exc:  # ParameterNotFound or transient — fail closed
        logger.error("Failed to read Google client ID param: %s", exc)
        return ""


def _verify_google_token(token: str, client_id: str) -> dict | None:
    """Verify a Google ID token via the tokeninfo endpoint.

    Returns the decoded {name, email} on success, or None on any failure
    (bad token, wrong audience/domain, or service unavailable). The authorizer
    fails closed: any verification problem denies access.
    """
    try:
        qs = urllib.parse.urlencode({"id_token": token})
        req = urllib.request.Request(f"https://oauth2.googleapis.com/tokeninfo?{qs}")
        with urllib.request.urlopen(req, timeout=5) as resp:
            data = json.loads(resp.read())
    except urllib.error.HTTPError as exc:
        logger.warning("Google token rejected (HTTP %s)", exc.code)
        return None
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        logger.error("Google token verification service unavailable: %s", exc)
        return None
    except Exception as exc:
        logger.error("Google token verification failed: %s", exc)
        return None

    if data.get("aud") != client_id:
        logger.warning("Google token audience mismatch")
        return None
    if data.get("hd") not in ALLOWED_DOMAINS:
        logger.warning("Google token domain mismatch: %s", data.get("hd"))
        return None
    return {"name": data.get("name", ""), "email": data.get("email", "")}


def _extract_token(event) -> str:
    """Pull the bearer token from the Authorization header.

    HTTP API lowercases header names; check both for safety.
    """
    headers = event.get("headers") or {}
    raw = headers.get("authorization") or headers.get("Authorization") or ""
    if raw.lower().startswith("bearer "):
        return raw[7:].strip()
    return ""


def _verify_portal_token(token: str) -> dict | None:
    return verify_cognito_id_token(
        token,
        os.environ.get("PORTAL_COGNITO_ISSUER_PARAM", ""),
        os.environ.get("PORTAL_COGNITO_AUDIENCE_PARAM", ""),
    )


def lambda_handler(event, context):
    token = _extract_token(event)
    if not token:
        return _DENY

    if looks_like_cognito_token(token):
        try:
            user_info = _verify_portal_token(token)
        except CognitoVerificationUnavailable:
            logger.error("Cognito verification service unavailable; denying")
            return _DENY
    else:
        if not os.environ.get("GOOGLE_CLIENT_ID_PARAM", ""):
            logger.error("Authorizer misconfigured: GOOGLE_CLIENT_ID_PARAM unset")
            return _DENY
        client_id = _get_google_client_id()
        if not client_id:
            logger.error("Google client ID unavailable; denying")
            return _DENY
        user_info = _verify_google_token(token, client_id)
    if user_info is None:
        return _DENY

    try:
        admin_emails = {e.lower() for e in get_settings().get("admin_emails", [])}
    except Exception as exc:
        # DynamoDB unavailable / missing item: fail closed with DENY rather than
        # letting the unhandled exception surface as a 500 from the gateway.
        logger.error("Failed to load admin_emails from DynamoDB: %s", exc)
        return _DENY

    if user_info["email"].lower() not in admin_emails:
        logger.warning("Non-admin %s denied at gateway", user_info["email"])
        return _DENY

    logger.info("Admin %s authorized at gateway", user_info["email"])
    return _ALLOW