munch-ease-backend/api/openapi.py

from __future__ import annotations

from typing import Any

from fastapi import FastAPI


def extend_with_problem_and_cookie_auth(app: FastAPI) -> None:
    """Augment FastAPI's OpenAPI spec with RFC7807 responses and cookie auth.

    This mutates the app's OpenAPI generation in-place while delegating to the
    original generator for the base schema.
    """
    original_openapi = app.openapi

    def custom_openapi() -> dict[str, Any]:
        spec = original_openapi()
        components = spec.setdefault("components", {})
        responses = components.setdefault("responses", {})
        security_schemes = components.setdefault("securitySchemes", {})

        # Standard ProblemDetails responses
        responses.setdefault(
            "Problem400",
            {
                "description": "Bad Request",
                "content": {
                    "application/problem+json": {},
                    "application/json": {"schema": {"$ref": "#/components/schemas/ProblemDetails"}},
                },
            },
        )
        responses.setdefault(
            "Problem404",
            {
                "description": "Not Found",
                "content": {
                    "application/problem+json": {},
                    "application/json": {"schema": {"$ref": "#/components/schemas/ProblemDetails"}},
                },
            },
        )
        responses.setdefault(
            "Problem422",
            {
                "description": "Validation Error",
                "content": {
                    "application/problem+json": {
                        "schema": {"$ref": "#/components/schemas/ProblemDetails"}
                    },
                    "application/json": {"schema": {"$ref": "#/components/schemas/ProblemDetails"}},
                },
            },
        )
        responses.setdefault(
            "Problem403",
            {
                "description": "Forbidden",
                "content": {
                    "application/problem+json": {},
                    "application/json": {"schema": {"$ref": "#/components/schemas/ProblemDetails"}},
                },
            },
        )

        # Cookie-based auth for v1 documentation (does not enforce at runtime)
        security_schemes.setdefault(
            "cookieAuth",
            {
                "type": "apiKey",
                "in": "cookie",
                "name": "user_id",
                "description": "Authentication via user_id cookie (session-style).",
            },
        )

        # Bearer (JWT) auth for v2
        security_schemes.setdefault(
            "bearerAuth",
            {
                "type": "http",
                "scheme": "bearer",
                "bearerFormat": "JWT",
                "description": "JWT access token in Authorization header",
            },
        )

        # Normalize v1 responses and mark cookie security for known endpoints
        paths = spec.get("paths", {})
        protected_ops: set[str] = {
            "parseRecipe",
            "createRecipe",
            "deleteRecipe",
            "markMealConsumed",
            "deleteMeal",
            "purchaseIngredients",
            "getMyShoppingList",
            "syncMyShoppingList",
            "requestMeal",
            "unrequestMeal",
            "refresh",
        }
        for path, ops in paths.items():
            if not isinstance(path, str) or not path.startswith("/api/v1/"):
                continue
            if not isinstance(ops, dict):
                continue
            for _method, op in ops.items():
                if not isinstance(op, dict):
                    continue
                resp = op.get("responses")
                if not isinstance(resp, dict):
                    continue
                if "400" in resp:
                    resp["400"] = {"$ref": "#/components/responses/Problem400"}
                if "404" in resp:
                    resp["404"] = {"$ref": "#/components/responses/Problem404"}
                if "422" not in resp:
                    resp["422"] = {"$ref": "#/components/responses/Problem422"}

                op_id = op.get("operationId")
                if isinstance(op_id, str) and op_id in protected_ops:
                    security = op.setdefault("security", [])
                    if not any(isinstance(s, dict) and "cookieAuth" in s for s in security):
                        security.append({"cookieAuth": []})

                    # Ensure the cookie parameter is documented as required integer (non-null)
                    params = op.get("parameters")
                    if isinstance(params, list):
                        for p in params:
                            if not isinstance(p, dict):
                                continue
                            if p.get("in") == "cookie" and p.get("name") == "user_id":
                                p["required"] = True
                                schema = p.setdefault("schema", {})
                                if isinstance(schema, dict):
                                    schema.clear()
                                    schema.update({"type": "integer", "title": "User Id"})

        # Keep endpoint-specific schemas driven by route declarations only (no forced overrides)

        # Normalize outward-facing shopping list storeName enum to avoid empty-string value
        schemas = components.setdefault("schemas", {})
        # Define outward-only enum for store names
        schemas.setdefault(
            "StoreNameOut",
            {
                "type": "string",
                "enum": ["woolworths", "coles", "home"],
                "title": "StoreNameOut",
            },
        )
        # Replace any storeName prop that points to StoreEnum (which includes "") with outward StoreNameOut
        for schema in schemas.values():
            if not isinstance(schema, dict):
                continue
            props = schema.get("properties")
            if not isinstance(props, dict):
                continue
            store = props.get("storeName")
            if isinstance(store, dict) and store.get("$ref") == "#/components/schemas/StoreEnum":
                props["storeName"] = {"$ref": "#/components/schemas/StoreNameOut"}

    # Mark bearer security for v2 routes we know require auth
        # Simple heuristic: underline select paths under /api/v1/users/me and /api/v1/households/* that are protected
        for path, ops in paths.items():
            if not isinstance(path, str) or not path.startswith("/api/v1/"):
                continue
            if not isinstance(ops, dict):
                continue
            needs_bearer = (
                path.startswith("/api/v1/users/me/")
                or path.startswith("/api/v1/households/")
            )
            if not needs_bearer:
                continue
            for _method, op in ops.items():
                if not isinstance(op, dict):
                    continue
                security = op.setdefault("security", [])
                if not any(isinstance(s, dict) and "bearerAuth" in s for s in security):
                    security.append({"bearerAuth": []})
                # ensure 403 Problem is defined on these operations
                resp = op.setdefault("responses", {})
                if "403" not in resp:
                    resp["403"] = {"$ref": "#/components/responses/Problem403"}

        return spec

    # Rebind app.openapi to our generator (FastAPI supports this pattern). Mypy needs a narrow ignore here.
    app.openapi = custom_openapi  # type: ignore[method-assign]