148 lines
5.7 KiB
Python
148 lines
5.7 KiB
Python
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"}},
|
|
},
|
|
},
|
|
)
|
|
|
|
# Remove legacy cookieAuth; JWT bearer is the only auth now
|
|
|
|
# Bearer (JWT) auth for v2
|
|
security_schemes.setdefault(
|
|
"bearerAuth",
|
|
{
|
|
"type": "http",
|
|
"scheme": "bearer",
|
|
"bearerFormat": "JWT",
|
|
"description": "JWT access token in Authorization header",
|
|
},
|
|
)
|
|
|
|
# Normalize responses (RFC7807) but do not add cookie auth
|
|
paths = spec.get("paths", {})
|
|
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"}
|
|
|
|
# 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 protected routes: /api/v1/users/me/* and /api/v1/households/*
|
|
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]
|