generate_api_auth_spec.py

"""Generate the OpenAPI spec from the app, annotated with auth requirements.

This generator builds the FastAPI application, exports its OpenAPI document via
``app.openapi()``, and enriches every operation with authentication metadata:

  * a ``components.securitySchemes`` block,
  * a per-operation ``security`` requirement (so the docs render a lock badge),
  * an ``x-required-role`` extension for machine readers, and
  * a short bold ``Access:`` note prepended to each operation description.

The committed docs/static/frigate-api.yaml is the output of this script. It is
generated rather than hand-maintained so it stays complete and current; the docs
build (docusaurus-plugin-openapi-docs) consumes it as-is.

The access level for an endpoint is determined by BOTH its route-level
dependency (``require_role``/``allow_any_authenticated``/``allow_public``/
``require_camera_access``) AND the global "secure by default" admin dependency,
which is bypassed only for the paths listed in ``require_admin_by_default``.
Those exempt lists are read directly from the function's closure so this script
stays in lockstep with ``frigate/api/auth.py`` instead of duplicating them.

Many handlers enforce per-camera access by calling ``require_camera_access``
inside the handler body rather than as a route dependency, which dependency
introspection cannot see. We recover those from the handler's bytecode (see
``_handler_enforces_camera``) and promote an otherwise "any authenticated"
operation to camera-scoped.

Usage (from the repository root):

    python3 generate_api_auth_spec.py            # write the spec
    python3 generate_api_auth_spec.py --check    # CI guard: fail if stale

The process exits non-zero if the generated document fails structural
validation, or (in --check mode) if the committed spec is out of date.
"""

import argparse
import difflib
import inspect
import io
import logging
import sys
from pathlib import Path

from fastapi import FastAPI
from fastapi.routing import APIRoute
from ruamel.yaml import YAML
from ruamel.yaml.scalarstring import LiteralScalarString

from frigate.api import app as main_app
from frigate.api import (
    auth,
    camera,
    chat,
    classification,
    debug_replay,
    event,
    export,
    hardware,
    media,
    motion_search,
    notices,
    notification,
    preview,
    record,
    review,
)
from frigate.api.auth import require_admin_by_default

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("generate_api_auth_spec")

REPO_ROOT = Path(__file__).resolve().parent
OUTPUT_SPEC = REPO_ROOT / "docs" / "static" / "frigate-api.yaml"

HTTP_METHODS = {"get", "post", "put", "delete", "patch"}

# Banner written at the top of the generated spec.
HEADER = (
    "# Generated by generate_api_auth_spec.py — do not edit by hand.\n"
    "# Regenerate with:  python3 generate_api_auth_spec.py\n"
    "# The empty info.title is intentional: a docusaurus-openapi-docs convention\n"
    "# that suppresses the generated API introduction page.\n"
)

# Post-processing applied on top of the raw app.openapi() export. These live
# only in the published spec, not in the app, so they are reproduced here.
SPEC_TITLE = ""
SPEC_SERVERS = [
    {"url": "https://demo.frigate.video/api"},
    {"url": "http://localhost:5001/api"},
]

# Access levels, ordered from least to most privileged. The string values are
# also what we emit as ``x-required-role``.
PUBLIC = "public"
AUTHENTICATED = "any"
CAMERA = "camera"
ALL_CAMERAS = "all_cameras"
ADMIN = "admin"

ADMIN_SCHEME = "frigateAdminAuth"
USER_SCHEME = "frigateUserAuth"

SECURITY_SCHEMES = {
    ADMIN_SCHEME: {
        "type": "apiKey",
        "in": "cookie",
        "name": "frigate_token",
        "description": (
            "Authenticated session whose resolved role is 'admin'. The session "
            "is established via the JWT cookie issued by POST /login, or via "
            "proxy auth headers (remote-user / remote-role) when Frigate runs "
            "behind an authenticating reverse proxy."
        ),
    },
    USER_SCHEME: {
        "type": "apiKey",
        "in": "cookie",
        "name": "frigate_token",
        "description": (
            "Any authenticated session (role 'viewer' or higher), established "
            "via the JWT cookie issued by POST /login, or via proxy auth "
            "headers when Frigate runs behind an authenticating reverse proxy."
        ),
    },
}

# How each access level maps to a rendered note.
ACCESS_NOTES = {
    PUBLIC: "**Access:** Public — no authentication required.",
    AUTHENTICATED: "**Access:** Any authenticated user.",
    CAMERA: "**Access:** Authenticated user with access to the referenced camera.",
    ALL_CAMERAS: "**Access:** Authenticated user with access to all cameras.",
    ADMIN: "**Access:** Admin role required.",
}


def build_app() -> FastAPI:
    """Build a bare app with every router mounted.

    This mirrors the router set wired up in frigate.api.fastapi_app. It omits
    the global admin dependency and all runtime state; the OpenAPI route table
    and the per-route dependencies are all we need to export and classify.
    """
    app = FastAPI()
    routers = [
        auth.router,
        camera.router,
        chat.router,
        classification.router,
        review.router,
        main_app.router,
        preview.router,
        notification.router,
        export.router,
        hardware.router,
        notices.router,
        event.router,
        media.router,
        motion_search.router,
        record.router,
        debug_replay.router,
    ]
    for router in routers:
        app.include_router(router)
    return app


def read_exempt_rules() -> tuple[set[str], tuple[str, ...]]:
    """Read the admin-exemption lists straight from the auth dependency closure.

    Reading them here (rather than copying) keeps this generator in sync with
    frigate/api/auth.py automatically.
    """
    closure = inspect.getclosurevars(require_admin_by_default()).nonlocals
    exempt_paths = set(closure["EXEMPT_PATHS"])
    exempt_prefixes = tuple(closure["EXEMPT_PREFIXES"])
    return exempt_paths, exempt_prefixes


def _first_segment(path: str) -> str:
    return path.split("/", 2)[1] if path.startswith("/") and len(path) > 1 else ""


def _route_markers(route: APIRoute) -> tuple[set[str], list[str] | None]:
    """Return the set of recognized auth markers on a route's dependencies."""
    markers: set[str] = set()
    admin_roles: list[str] | None = None

    for dep in route.dependant.dependencies:
        call = dep.call
        qualname = getattr(call, "__qualname__", "") or ""
        name = getattr(call, "__name__", "") or ""

        if "role_checker" in qualname:
            markers.add(ADMIN)
            try:
                roles = inspect.getclosurevars(call).nonlocals.get("required_roles")
                if roles:
                    admin_roles = list(roles)
            except (TypeError, ValueError):
                pass
        elif name in ("require_camera_access", "require_go2rtc_stream_access"):
            markers.add(CAMERA)
        elif name == "require_full_camera_access":
            markers.add(ALL_CAMERAS)
        elif "auth_checker" in qualname:
            markers.add(AUTHENTICATED)
        elif "public_checker" in qualname:
            markers.add(PUBLIC)

    return markers, admin_roles


def _handler_enforces_camera(route: APIRoute) -> bool:
    """True if the route handler calls require_camera_access in its body.

    Such calls are invisible to dependency introspection. We detect them from
    the handler's compiled bytecode: a global name referenced anywhere in the
    function appears in ``__code__.co_names``. This catches direct calls (all of
    them, currently); a call hidden behind a helper function would be missed.
    """
    code = getattr(route.endpoint, "__code__", None)
    return bool(code and "require_camera_access" in code.co_names)


def classify_route(
    route: APIRoute,
    exempt_paths: set[str],
    exempt_prefixes: tuple[str, ...],
) -> tuple[str, list[str] | None, str | None]:
    """Resolve the effective access level for a route.

    Returns (access_level, roles, flag). ``flag`` is a human-readable note when
    the result needed inference or revealed a possible inconsistency.
    """
    level, roles, flag = _classify_base(route, exempt_paths, exempt_prefixes)

    # In-body require_camera_access enforcement is invisible to dependency
    # introspection. When the effective access would otherwise be "any
    # authenticated", the handler's per-camera check is the real constraint, so
    # promote it to camera-scoped. Admin/public are left alone: for admin the
    # role is the binding requirement and the camera check is only defensive.
    if level == AUTHENTICATED and _handler_enforces_camera(route):
        return CAMERA, None, None

    return level, roles, flag


def _classify_base(
    route: APIRoute,
    exempt_paths: set[str],
    exempt_prefixes: tuple[str, ...],
) -> tuple[str, list[str] | None, str | None]:
    """Resolve the access level from route-level dependencies and exempt rules."""
    markers, admin_roles = _route_markers(route)
    path = route.path
    is_camera_path = _first_segment(path) == "{camera_name}"
    exempt = path in exempt_paths or path.startswith(exempt_prefixes) or is_camera_path

    # Explicit route-level markers win, in order of specificity.
    if ADMIN in markers:
        return ADMIN, admin_roles or ["admin"], None
    if ALL_CAMERAS in markers:
        return ALL_CAMERAS, None, None
    if CAMERA in markers:
        return CAMERA, None, None
    if AUTHENTICATED in markers:
        if exempt:
            return AUTHENTICATED, None, None
        # The route opts in to any-authenticated, but the global admin check is
        # not bypassed for this path, so admin is what actually gets enforced.
        return (
            ADMIN,
            ["admin"],
            (
                "route declares allow_any_authenticated but path is not exempt from "
                "the global admin check; admin is effectively enforced"
            ),
        )
    if PUBLIC in markers:
        if exempt:
            return PUBLIC, None, None
        return (
            ADMIN,
            ["admin"],
            (
                "route declares allow_public but path is not exempt from the global "
                "admin check; admin is effectively enforced"
            ),
        )

    # No explicit auth marker: governed purely by the global default.
    if not exempt:
        return ADMIN, ["admin"], None

    # Exempt with no route dependency: the global admin check is bypassed and
    # there is no route-level gate, so authorization (if any) happens inside the
    # handler. Infer from the path shape and flag for confirmation.
    if is_camera_path:
        return (
            CAMERA,
            None,
            (
                "no route-level dependency; camera-scoped path, authorization "
                "assumed to be enforced in the handler"
            ),
        )
    return (
        AUTHENTICATED,
        None,
        (
            "path is exempt from the global admin check but has no route-level "
            "dependency; confirm authorization is enforced in the handler"
        ),
    )


def build_access_map(
    app: FastAPI,
    exempt_paths: set[str],
    exempt_prefixes: tuple[str, ...],
) -> dict[tuple[str, str], dict]:
    """Map (path, lowercase method) -> classification details."""
    access_map: dict[tuple[str, str], dict] = {}
    for route in app.routes:
        if not isinstance(route, APIRoute):
            continue
        level, roles, flag = classify_route(route, exempt_paths, exempt_prefixes)
        for method in route.methods:
            if method in ("HEAD", "OPTIONS"):
                continue
            # the OpenAPI paths drop convertors such as :path, like path_format
            access_map[(route.path_format, method.lower())] = {
                "level": level,
                "roles": roles,
                "flag": flag,
                "path": route.path,
                "method": method,
            }
    return access_map


def security_for(level: str) -> list:
    """Build the OpenAPI ``security`` value for an access level."""
    if level == PUBLIC:
        return []
    if level == ADMIN:
        return [{ADMIN_SCHEME: []}]
    # AUTHENTICATED, CAMERA and ALL_CAMERAS all require any authenticated
    # session; the camera scoping is conveyed in the note and x-required-role.
    return [{USER_SCHEME: []}]


def required_role_value(level: str, roles: list[str] | None):
    if level == ADMIN and roles and roles != ["admin"]:
        return roles
    return level


def annotate_description(operation: dict, note: str) -> None:
    existing = operation.get("description")
    if not existing:
        operation["description"] = note
        return
    operation["description"] = LiteralScalarString(
        f"{note}\n\n{str(existing).rstrip()}"
    )


def base_document(raw: dict) -> dict:
    """Apply the docs pipeline post-processing with a stable top-level order."""
    info = dict(raw.get("info", {}))
    info["title"] = SPEC_TITLE
    return {
        "openapi": raw["openapi"],
        "info": info,
        "servers": [dict(server) for server in SPEC_SERVERS],
        "paths": raw["paths"],
        "components": raw.get("components", {}),
    }


def enrich(spec: dict, access_map: dict) -> tuple[dict, list, list]:
    """Add security schemes and per-operation auth metadata in place."""
    components = spec.setdefault("components", {})
    components["securitySchemes"] = dict(SECURITY_SCHEMES)

    counts: dict[str, int] = {}
    flagged: list[dict] = []
    unmatched: list[tuple[str, str]] = []

    for path, path_item in spec["paths"].items():
        for method, operation in path_item.items():
            if method.lower() not in HTTP_METHODS:
                continue
            details = access_map.get((path, method.lower()))
            if details is None:
                unmatched.append((method.upper(), path))
                continue

            level = details["level"]
            counts[level] = counts.get(level, 0) + 1
            operation["security"] = security_for(level)
            operation["x-required-role"] = required_role_value(level, details["roles"])
            annotate_description(operation, ACCESS_NOTES[level])

            if details["flag"]:
                flagged.append(details)

    return counts, flagged, unmatched


# Numeric defaults at or above this magnitude are treated as live Unix
# timestamps baked into the schema at import time. They make the export
# non-deterministic and document a meaningless frozen epoch, so they are
# stripped. No route defaults this way today, and route params that need "now"
# resolve it inside the handler. Kept as a guard against the pattern returning.
VOLATILE_DEFAULT_THRESHOLD = 1_000_000_000


def strip_volatile_defaults(node, trail: str = "") -> list[tuple[str, float]]:
    """Remove epoch-like numeric ``default`` values so the export is stable.

    Returns the (location, value) pairs that were removed, for reporting.
    """
    removed: list[tuple[str, float]] = []
    if isinstance(node, dict):
        default = node.get("default")
        if (
            isinstance(default, (int, float))
            and not isinstance(default, bool)
            and default >= VOLATILE_DEFAULT_THRESHOLD
        ):
            removed.append((trail, default))
            del node["default"]
        for key, value in node.items():
            removed.extend(strip_volatile_defaults(value, f"{trail}/{key}"))
    elif isinstance(node, list):
        for index, value in enumerate(node):
            removed.extend(strip_volatile_defaults(value, f"{trail}[{index}]"))
    return removed


def to_block_scalars(node):
    """Recursively render multi-line strings as literal block scalars.

    Produces readable, deterministic YAML (``|-`` blocks) instead of long
    double-quoted lines with escaped newlines.
    """
    if isinstance(node, dict):
        return {key: to_block_scalars(value) for key, value in node.items()}
    if isinstance(node, list):
        return [to_block_scalars(value) for value in node]
    if isinstance(node, str) and "\n" in node:
        return LiteralScalarString(node)
    return node


def _iter_refs(node):
    if isinstance(node, dict):
        for key, value in node.items():
            if key == "$ref" and isinstance(value, str):
                yield value
            else:
                yield from _iter_refs(value)
    elif isinstance(node, list):
        for value in node:
            yield from _iter_refs(value)


def validate(spec: dict) -> list[str]:
    """Structural sanity checks on the generated document."""
    problems: list[str] = []
    schemas = set(spec.get("components", {}).get("schemas", {}))
    defined_schemes = set(spec.get("components", {}).get("securitySchemes", {}))

    for ref in _iter_refs(spec):
        if ref.startswith("#/components/schemas/"):
            name = ref.rsplit("/", 1)[-1]
            if name not in schemas:
                problems.append(f"dangling $ref: {ref}")

    for path, path_item in spec.get("paths", {}).items():
        for method, operation in path_item.items():
            if method.lower() not in HTTP_METHODS or not isinstance(operation, dict):
                continue
            location = f"{method.upper()} {path}"
            if "x-required-role" not in operation:
                problems.append(f"missing x-required-role: {location}")
            if "security" not in operation:
                problems.append(f"missing security: {location}")
                continue
            for requirement in operation["security"]:
                for scheme in requirement:
                    if scheme not in defined_schemes:
                        problems.append(
                            f"undefined security scheme {scheme}: {location}"
                        )

    return sorted(set(problems))


def render(spec: dict) -> str:
    """Serialize the spec to the canonical YAML string (with the header)."""
    yaml = YAML()
    yaml.width = 80
    yaml.indent(mapping=2, sequence=4, offset=2)
    stream = io.StringIO()
    yaml.dump(spec, stream)
    return HEADER + stream.getvalue()


def build_spec() -> tuple[dict, dict, list, list, list]:
    app = build_app()
    exempt_paths, exempt_prefixes = read_exempt_rules()
    access_map = build_access_map(app, exempt_paths, exempt_prefixes)

    spec = base_document(app.openapi())
    normalized = strip_volatile_defaults(spec)
    counts, flagged, unmatched = enrich(spec, access_map)
    spec = to_block_scalars(spec)
    return spec, counts, flagged, unmatched, normalized


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Generate the annotated OpenAPI spec.")
    parser.add_argument(
        "--check",
        action="store_true",
        help="verify the committed spec is up to date without writing; "
        "exit non-zero if it would change",
    )
    args = parser.parse_args(argv)

    spec, counts, flagged, unmatched, normalized = build_spec()
    problems = validate(spec)
    rendered = render(spec)

    if args.check:
        return _check(rendered, problems)

    if problems:
        logger.error("Refusing to write — generated spec failed validation:")
        for problem in problems:
            logger.error("  %s", problem)
        return 1

    OUTPUT_SPEC.write_text(rendered)
    _report(counts, flagged, unmatched, normalized)
    logger.info("\nWrote %s", OUTPUT_SPEC.relative_to(REPO_ROOT))
    return 0


def _check(rendered: str, problems: list[str]) -> int:
    name = OUTPUT_SPEC.relative_to(REPO_ROOT)
    if problems:
        logger.error("Generated spec failed validation:")
        for problem in problems:
            logger.error("  %s", problem)
        return 1

    current = OUTPUT_SPEC.read_text() if OUTPUT_SPEC.exists() else ""
    if current == rendered:
        logger.info("%s is up to date", name)
        return 0

    logger.error(
        "%s is out of date. Regenerate with: python3 %s",
        name,
        Path(__file__).name,
    )
    diff = difflib.unified_diff(
        current.splitlines(),
        rendered.splitlines(),
        fromfile=f"{name} (committed)",
        tofile=f"{name} (generated)",
        lineterm="",
        n=2,
    )
    for shown, line in enumerate(diff):
        if shown >= 60:
            logger.error("  ... (diff truncated)")
            break
        logger.error("  %s", line)
    return 1


def _report(counts, flagged, unmatched, normalized) -> None:
    logger.info("Access levels applied:")
    for level in (PUBLIC, AUTHENTICATED, CAMERA, ADMIN):
        logger.info("  %-14s %d", level, counts.get(level, 0))
    logger.info("  %-14s %d", "total", sum(counts.values()))

    if normalized:
        logger.info("\nStripped volatile timestamp defaults (%d):", len(normalized))
        for location, value in normalized:
            logger.info("  %s = %s", location.lstrip("/"), value)

    if flagged:
        logger.info("\nFlagged for manual confirmation (%d):", len(flagged))
        for item in flagged:
            logger.info("  %-6s %s", item["method"], item["path"])
            logger.info("         -> %s (%s)", item["level"], item["flag"])

    if unmatched:
        logger.info(
            "\nOperations with no classification (%d) [unexpected]:", len(unmatched)
        )
        for method, path in unmatched:
            logger.info("  %-6s %s", method, path)


if __name__ == "__main__":
    sys.exit(main())