"""Consumer-facing API documentation.
`GET /openapi.json` returns a merged OpenAPI document scoped to the calling
API key: every operation the key holds a grant on, across all services,
rewritten to the gateway's own routes (`/{slug}/...`). `GET /docs` serves a
Swagger UI portal around it.
"""
import re
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, JSONResponse
from sqlalchemy.orm import Session
from app import config, discovery, security
from app.database import get_db
from app.models import ApiKey, Endpoint, Service
from app.proxy import _error, _pattern_to_regex
router = APIRouter()
_HTTP_METHODS = ("get", "post", "put", "patch", "delete", "head", "options")
_REF_RE = re.compile(r"^#/components/([A-Za-z]+)/(.+)$")
def _rewrite_refs(obj, slug: str):
"""Namespace `$ref`s into a service's renamed components (atlas_Error)."""
if isinstance(obj, dict):
out = {}
for key, value in obj.items():
if key == "$ref" and isinstance(value, str):
m = _REF_RE.match(value)
out[key] = (f"#/components/{m.group(1)}/{slug}_{m.group(2)}"
if m else value)
else:
out[key] = _rewrite_refs(value, slug)
return out
if isinstance(obj, list):
return [_rewrite_refs(item, slug) for item in obj]
return obj
def _grant_matches(grant: Endpoint, method: str, path: str) -> bool:
return (grant.method in ("*", method)
and _pattern_to_regex(grant.path).match(path) is not None)
def _service_paths(service: Service, grants: list[Endpoint], spec: dict | None) -> dict:
"""The slug-prefixed path entries of one service, filtered to the grants."""
slug = service.slug
if spec:
included = {}
for path, operations in spec.get("paths", {}).items():
if not isinstance(operations, dict):
continue
kept = {}
for method, op in operations.items():
if method.lower() not in _HTTP_METHODS or not isinstance(op, dict):
continue
if any(_grant_matches(g, method.upper(), path) for g in grants):
kept[method] = {**op, "tags": [service.name]}
if kept:
entry = {k: v for k, v in operations.items() if k.lower() not in _HTTP_METHODS}
entry.update(kept)
included[f"/{slug}{path}"] = entry
return _rewrite_refs(included, slug)
# No OpenAPI document upstream — synthesize minimal entries from the catalog.
included = {}
for g in grants:
methods = [g.method.lower()] if g.method != "*" else ["get", "post", "put", "patch", "delete"]
entry = included.setdefault(f"/{slug}{g.path}", {})
for method in methods:
entry[method] = {
"summary": g.description or ("(any method)" if g.method == "*" else ""),
"tags": [service.name],
"responses": {"default": {"description": "Proxied upstream response"}},
}
return included
def build_key_spec(db: Session, api_key: ApiKey, gateway_url: str) -> dict:
by_service: dict[int, list[Endpoint]] = {}
for grant in api_key.endpoints:
by_service.setdefault(grant.service_id, []).append(grant)
paths: dict = {}
components: dict = {}
tags: list[dict] = []
for service_id, grants in by_service.items():
service = grants[0].service
if not service.is_active:
continue
spec = discovery.get_spec(db, service)
paths.update(_service_paths(service, grants, spec))
if spec:
for section, items in spec.get("components", {}).items():
if section == "securitySchemes" or not isinstance(items, dict):
continue
renamed = {f"{service.slug}_{name}": schema for name, schema in items.items()}
components.setdefault(section, {}).update(_rewrite_refs(renamed, service.slug))
tags.append({"name": service.name, "description": service.description or f"/{service.slug}"})
components["securitySchemes"] = {
"ApiKeyAuth": {"type": "apiKey", "in": "header", "name": config.API_KEY_HEADER},
}
return {
"openapi": "3.0.3",
"info": {
"title": "API Gateway",
"version": "1.0.0",
"description": f"Operations available to API key **{api_key.name}**. "
f"Send the key in the `{config.API_KEY_HEADER}` header.",
},
"servers": [{"url": gateway_url}],
"tags": sorted(tags, key=lambda t: t["name"]),
"paths": dict(sorted(paths.items())),
"components": components,
"security": [{"ApiKeyAuth": []}],
}
@router.get("/openapi.json", include_in_schema=False)
def key_scoped_openapi(request: Request, db: Session = Depends(get_db)):
plain_key = request.headers.get(config.API_KEY_HEADER)
if not plain_key:
return _error(401, "missing_api_key",
f"Provide your API key in the {config.API_KEY_HEADER} header.")
api_key = (db.query(ApiKey)
.filter(ApiKey.key_hash == security.hash_api_key(plain_key))
.one_or_none())
if api_key is None or not api_key.is_active or not api_key.user.is_active:
return _error(403, "invalid_api_key", "API key is unknown or has been revoked.")
spec = build_key_spec(db, api_key, str(request.base_url).rstrip("/"))
return JSONResponse(spec, headers={"Cache-Control": "no-store"})
_DOCS_PAGE = """
API Gateway · Docs
API Gateway
Enter your API key above to see the endpoints available to you.