Samuel AmarandClaude Haiku 4.5
f13160ec29
Include SVG favicon showing hexagon network topology with nodes and connections, integrated into admin console and /docs portal for consistent branding. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""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.
|
|
Operations keep their upstream tags — they become the second grouping
|
|
level; untagged operations fall into 'General'."""
|
|
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": list(op.get("tags") or ["General"])}
|
|
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": ["General"],
|
|
"responses": {"default": {"description": "Proxied upstream response"}},
|
|
}
|
|
return included
|
|
|
|
|
|
def build_key_spec(db: Session, api_key: ApiKey, gateway_url: str) -> dict:
|
|
"""Two-level structure: each service is an `x-tagGroups` entry (first
|
|
level, with its description in the intro), containing the upstream's own
|
|
operation tags (second level). Tag names shared by several services are
|
|
disambiguated with the service name."""
|
|
by_service: dict[int, list[Endpoint]] = {}
|
|
for grant in api_key.endpoints:
|
|
by_service.setdefault(grant.service_id, []).append(grant)
|
|
|
|
merged: list[tuple[Service, dict]] = [] # (service, its filtered paths)
|
|
used_tags: dict[int, list[str]] = {} # service id -> upstream tags, in order
|
|
tag_descriptions: dict[tuple[int, str], str] = {}
|
|
upstream_infos: dict[int, dict] = {} # service id -> upstream spec's `info`
|
|
components: dict = {}
|
|
|
|
ordered = sorted(by_service.values(), key=lambda g: g[0].service.name)
|
|
for grants in ordered:
|
|
service = grants[0].service
|
|
if not service.is_active:
|
|
continue
|
|
spec = discovery.get_spec(db, service)
|
|
paths = _service_paths(service, grants, spec)
|
|
if not paths:
|
|
continue
|
|
order: list[str] = []
|
|
for entry in paths.values():
|
|
for method, op in entry.items():
|
|
if method.lower() in _HTTP_METHODS and isinstance(op, dict):
|
|
for tag in op.get("tags", []):
|
|
if tag not in order:
|
|
order.append(tag)
|
|
used_tags[service.id] = order
|
|
if spec:
|
|
info = spec.get("info")
|
|
upstream_infos[service.id] = info if isinstance(info, dict) else {}
|
|
for tag in spec.get("tags") or []:
|
|
if isinstance(tag, dict) and tag.get("name") in order:
|
|
tag_descriptions[(service.id, tag["name"])] = tag.get("description", "")
|
|
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))
|
|
merged.append((service, paths))
|
|
|
|
# A tag name used by several services must stay unique in the flat tag list.
|
|
tag_counts: dict[str, int] = {}
|
|
for names in used_tags.values():
|
|
for name in names:
|
|
tag_counts[name] = tag_counts.get(name, 0) + 1
|
|
|
|
def final_name(service: Service, tag: str) -> str:
|
|
return tag if tag_counts[tag] == 1 else f"{tag} ({service.name})"
|
|
|
|
paths_out: dict = {}
|
|
tags_out: list[dict] = []
|
|
tag_groups: list[dict] = []
|
|
intro = [f"Operations available to API key **{api_key.name}**. "
|
|
f"Send the key in the `{config.API_KEY_HEADER}` header."]
|
|
for service, paths in merged:
|
|
finals = []
|
|
for tag in used_tags[service.id]:
|
|
name = final_name(service, tag)
|
|
finals.append(name)
|
|
tags_out.append({"name": name,
|
|
"description": tag_descriptions.get((service.id, tag), "")})
|
|
for entry in paths.values():
|
|
for method, op in entry.items():
|
|
if method.lower() in _HTTP_METHODS and isinstance(op, dict):
|
|
op["tags"] = [final_name(service, t) for t in op.get("tags", ["General"])]
|
|
paths_out.update(paths)
|
|
tag_groups.append({"name": service.name, "tags": finals})
|
|
info = upstream_infos.get(service.id, {})
|
|
section = f"## {service.name}\n\n"
|
|
title = info.get("title", "")
|
|
if title and title.lower() != service.name.lower():
|
|
section += f"*{title}*\n\n"
|
|
# The admin-entered description overrides the upstream's own.
|
|
description = service.description or info.get("description") or ""
|
|
if description:
|
|
section += description.strip() + "\n\n"
|
|
section += f"Routes under `/{service.slug}/…`"
|
|
intro.append(section)
|
|
|
|
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": "\n\n".join(intro),
|
|
},
|
|
"servers": [{"url": gateway_url}],
|
|
"tags": tags_out,
|
|
"x-tagGroups": tag_groups,
|
|
"paths": dict(sorted(paths_out.items())),
|
|
"components": components,
|
|
"security": [{"ApiKeyAuth": []}],
|
|
}
|
|
|
|
|
|
def resolve_api_key(request: Request, db: Session) -> ApiKey | JSONResponse:
|
|
"""Authenticate a consumer-facing request by X-API-Key. Returns the key,
|
|
or the error response to send back."""
|
|
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.")
|
|
return api_key
|
|
|
|
|
|
@router.get("/openapi.json", include_in_schema=False)
|
|
def key_scoped_openapi(request: Request, db: Session = Depends(get_db)):
|
|
api_key = resolve_api_key(request, db)
|
|
if isinstance(api_key, JSONResponse):
|
|
return api_key
|
|
|
|
spec = build_key_spec(db, api_key, str(request.base_url).rstrip("/"))
|
|
return JSONResponse(spec, headers={"Cache-Control": "no-store"})
|
|
|
|
|
|
_DOCS_PAGE = """<!doctype html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
|
<title>API Gateway · Docs</title>
|
|
<link rel="icon" type="image/svg+xml" href="/static/favicon.svg">
|
|
<style>
|
|
body { margin: 0; font-family: system-ui, sans-serif; }
|
|
#keybar {
|
|
display: flex; gap: 10px; align-items: center; flex-wrap: wrap;
|
|
padding: 14px 20px; background: #1b1b19; color: #e8e6df;
|
|
}
|
|
#keybar strong { font-size: 15px; }
|
|
#keybar input {
|
|
flex: 1; min-width: 260px; max-width: 480px; padding: 8px 10px;
|
|
border-radius: 8px; border: 1px solid #444; background: #121210;
|
|
color: #e8e6df; font-family: monospace;
|
|
}
|
|
#keybar button {
|
|
padding: 8px 16px; border-radius: 8px; border: 0;
|
|
background: #3987e5; color: #fff; font-weight: 600; cursor: pointer;
|
|
}
|
|
#keybar .msg { font-size: 13px; color: #e66767; }
|
|
#placeholder { padding: 48px 20px; text-align: center; color: #666; }
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div id="keybar">
|
|
<strong>API Gateway</strong>
|
|
<input id="apikey" type="password" placeholder="Paste your API key (X-API-Key)"
|
|
autocomplete="off" spellcheck="false">
|
|
<button id="load">Load my API docs</button>
|
|
<span class="msg" id="msg"></span>
|
|
</div>
|
|
<div id="placeholder">Enter your API key above to see the endpoints available to you.</div>
|
|
<div id="app"></div>
|
|
<script src="https://cdn.jsdelivr.net/npm/@scalar/api-reference"></script>
|
|
<script>
|
|
const input = document.getElementById('apikey');
|
|
const msg = document.getElementById('msg');
|
|
input.value = sessionStorage.getItem('gw_api_key') || '';
|
|
let reference = null;
|
|
|
|
async function load() {
|
|
const key = input.value.trim();
|
|
msg.textContent = '';
|
|
if (!key) { msg.textContent = 'An API key is required.'; return; }
|
|
const resp = await fetch('/openapi.json', { headers: { 'X-API-Key': key } });
|
|
if (!resp.ok) {
|
|
const err = await resp.json().catch(() => ({}));
|
|
msg.textContent = err.message || ('Error ' + resp.status);
|
|
return;
|
|
}
|
|
sessionStorage.setItem('gw_api_key', key);
|
|
document.getElementById('placeholder')?.remove();
|
|
const config = {
|
|
content: await resp.json(),
|
|
hideClientButton: true,
|
|
authentication: {
|
|
preferredSecurityScheme: 'ApiKeyAuth',
|
|
securitySchemes: { ApiKeyAuth: { value: key, token: key } },
|
|
apiKey: { token: key },
|
|
},
|
|
};
|
|
if (reference) {
|
|
reference.destroy?.();
|
|
document.getElementById('app').innerHTML = '';
|
|
}
|
|
reference = Scalar.createApiReference('#app', config);
|
|
}
|
|
document.getElementById('load').addEventListener('click', load);
|
|
input.addEventListener('keydown', e => { if (e.key === 'Enter') load(); });
|
|
if (input.value) load();
|
|
</script>
|
|
</body>
|
|
</html>"""
|
|
|
|
|
|
@router.get("/docs", include_in_schema=False)
|
|
def docs_portal():
|
|
return HTMLResponse(_DOCS_PAGE)
|