openapi-first/openapi_first/security.py

"""
OpenAPI security scheme parsing and auto-generated auth dependencies.

Reads `securitySchemes` and per-operation `security` from an OpenAPI spec,
resolves `{ENV_VAR}` placeholders in `x-` extension fields, and generates
FastAPI dependencies for token validation (e.g., Bearer JWT introspection).
"""

import os
import re
from typing import Any, Callable

import httpx
from fastapi import Depends, Request, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

_env_pattern = re.compile(r"\{(\w+)\}")


def _resolve_env(value: str) -> str:
    """Replace {ENV_VAR} placeholders with values from os.environ."""
    def _replace(m: re.Match) -> str:
        return os.environ.get(m.group(1), "")
    return _env_pattern.sub(_replace, value)


def _resolve_scheme(scheme: dict) -> dict:
    """Recursively resolve env vars in all string-valued fields of a scheme."""
    resolved = {}
    for key, value in scheme.items():
        if isinstance(value, str):
            resolved[key] = _resolve_env(value)
        elif isinstance(value, dict):
            resolved[key] = _resolve_scheme(value)
        else:
            resolved[key] = value
    return resolved


def parse_security_schemes(spec: dict) -> dict[str, dict]:
    """Extract and resolve environment variables in security schemes."""
    raw = spec.get("components", {}).get("securitySchemes", {})
    return {name: _resolve_scheme(scheme) for name, scheme in raw.items()}


def _make_bearer_dependency(introspect_url: str | None) -> Callable:
    """
    Create a FastAPI dependency that validates a Bearer JWT.

    If *introspect_url* is provided the dependency calls that endpoint
    with ``{"token": "<token>"}`` and expects ``{"active": true, "user": …}``
    back.  The resolved user dict is stored on ``request.state.user``.

    Without an introspection URL the dependency only extracts the raw token
    and stores it on ``request.state.token`` — no remote validation.
    """
    bearer = HTTPBearer(auto_error=False)

    async def _bearer_dep(
        request: Request,
        credentials: HTTPAuthorizationCredentials | None = Depends(bearer),
    ) -> None:
        if credentials is None:
            raise HTTPException(status_code=401, detail="Not authenticated")

        token = credentials.credentials

        if introspect_url:
            try:
                async with httpx.AsyncClient(timeout=5) as client:
                    resp = await client.post(
                        introspect_url,
                        json={"token": token},
                    )
            except httpx.RequestError:
                raise HTTPException(
                    status_code=503,
                    detail="Authentication service unavailable",
                )

            if resp.status_code != 200:
                raise HTTPException(status_code=401, detail="Invalid or expired token")

            body = resp.json()
            if not body.get("active", False):
                raise HTTPException(status_code=401, detail="Invalid or expired token")

            request.state.user = body.get("user")
        else:
            request.state.token = token

    return _bearer_dep


def _build_dependency(scheme_name: str, scheme: dict) -> Callable | None:
    """Return a FastAPI dependency callable for *scheme*, or ``None``."""
    scheme_type = scheme.get("type")

    if scheme_type == "http" and scheme.get("scheme") == "bearer":
        server_url = scheme.get("x-server-url", "")
        introspect_path = scheme.get("x-introspect-path", "/introspect")
        introspect_url = server_url + introspect_path if server_url else None
        return _make_bearer_dependency(introspect_url)

    return None


def make_security_dependencies(
    spec: dict,
    security_schemes: dict[str, dict],
) -> dict[str, list[Any]]:
    """
    Build a mapping of ``METHOD:/path`` → list of ``Depends(...)``.

    The effective security for each operation is resolved by:

    1. Using the operation-level ``security`` field if present.
    2. Falling back to the top-level ``security`` field.
    3. An empty list means *no auth required* for that operation.
    """
    global_security = spec.get("security", [])
    paths = spec.get("paths", {})

    result: dict[str, list[Any]] = {}

    for path, methods in paths.items():
        for http_method, operation in methods.items():
            if http_method.startswith("x-"):
                continue

            op_security = operation.get("security", global_security)

            depends_list: list[Any] = []
            for sec_req in op_security:
                for scheme_name in sec_req:
                    scheme = security_schemes.get(scheme_name)
                    if scheme:
                        dep = _build_dependency(scheme_name, scheme)
                        if dep:
                            depends_list.append(Depends(dep))

            key = f"{http_method.upper()}:{path}"
            result[key] = depends_list

    return result