Coverage for scripts / generate_cluster_gateway_openapi.py: 100.00%
197 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1#!/usr/bin/env python3
2"""Generate the committed OpenAPI document for the in-cluster Gateway (the internal ALB).
4Inside a region every request from the API Gateway proxies lands on one
5internal ALB: the Kubernetes Gateway ``gco-system/gco-gateway``, whose single
6``HTTPRoute`` (``gco-routes`` in
7``lambda/kubectl-applier-simple/manifests/post-helm-gateway.yaml``) picks a
8Service by longest path prefix. The Gateway has no schema of its own — it is a
9routing table over the FastAPI services — so this script composes one: for
10every path in the committed service documents (``docs/openapi/<service>.json``)
11it resolves the HTTPRoute rule that wins and keeps the operation if, and only
12if, the winning Service is the one that serves it. The result,
13``docs/openapi/cluster-gateway.json``, is the HTTP surface the ALB actually
14exposes: each operation is tagged with its backend Service and carries the rule
15that delivers it (``x-gco-route``); paths a Service implements but never
16receives are listed under ``x-gco-unreachable``; Services that are not ALB
17backends at all (the cost monitor) under ``x-gco-not-exposed``.
19Usage::
21 python scripts/generate_cluster_gateway_openapi.py # write the document
22 python scripts/generate_cluster_gateway_openapi.py --check # fail if it is stale
24The same resolution is enforced as a test by
25``tests/test_gateway_route_coverage.py``; this document is its reviewable form,
26and the interaction diagram in ``diagrams/api_specs/`` is drawn from it.
27"""
29from __future__ import annotations
31import argparse
32import json
33import sys
34from dataclasses import dataclass, field
35from pathlib import Path
36from typing import Any
38import yaml
40REPO_ROOT = Path(__file__).resolve().parent.parent
41if str(REPO_ROOT) not in sys.path:
42 sys.path.insert(0, str(REPO_ROOT))
44from scripts.generate_openapi import SERVICE_NAMES # noqa: E402
46OUTPUT_DIR = REPO_ROOT / "docs" / "openapi"
47DOCUMENT_NAME = "cluster-gateway"
48GATEWAY_MANIFEST = Path("lambda/kubectl-applier-simple/manifests/post-helm-gateway.yaml")
49GENERATOR = "scripts/generate_cluster_gateway_openapi.py"
50API_VERSION = "1.0.0"
52#: Header carrying the request-bound HMAC the proxies mint; the companion
53#: headers complete the envelope the auth middleware verifies.
54HMAC_SIGNATURE_HEADER = "x-gco-signature"
55HMAC_ENVELOPE_HEADERS: tuple[str, ...] = (
56 "x-gco-signature-version",
57 "x-gco-timestamp",
58 "x-gco-nonce",
59 HMAC_SIGNATURE_HEADER,
60)
62#: Services reachable only from another Service, never from the ALB. The value
63#: names the front door and the paths that reach it, and generation fails if
64#: the Service turns up as an HTTPRoute backend or the front door lacks the
65#: paths, so this table cannot drift from the manifests.
66INTERNAL_CONSUMERS: dict[str, dict[str, str]] = {
67 "cost-monitor": {
68 "via": "manifest-processor",
69 "pathPrefix": "/api/v1/cost",
70 "description": (
71 "Confined to manifest-processor traffic by a NetworkPolicy and runs no "
72 "authentication middleware of its own; the manifest processor's "
73 "`/api/v1/cost/*` routes are its authenticated front, reaching it at "
74 "`https://cost-monitor.gco-system.svc.cluster.local:8443` and verifying "
75 "its certificate against the cluster's internal CA."
76 ),
77 },
78}
80#: Where a backend Service sends requests next, when it is itself a proxy.
81#: ``hop`` is the id the API Gateway documents use for the same destination in
82#: their ``x-gco-backend`` chains, so the diagram draws one box for both.
83DOWNSTREAM: dict[str, dict[str, str]] = {
84 "inference-proxy": {
85 "hop": "inference-endpoint",
86 "target": "gco-inference/<endpoint>, <endpoint>-canary or <endpoint>-proxy Services",
87 "description": (
88 "Resolves `/inference/<endpoint>/...` by endpoint name from the inference "
89 "endpoint store, requires the endpoint to be running in this region, allows "
90 "only its serving and health paths, and streams "
91 "`https://<service>.gco-inference.svc.cluster.local:8443/<path>` back, "
92 "verifying the model pod's TLS sidecar against the cluster's internal CA."
93 ),
94 },
95}
98class ClusterGatewayDocumentError(RuntimeError):
99 """The manifest or a service document cannot be composed faithfully."""
102# ─── the Gateway manifest ────────────────────────────────────────────────────
105@dataclass(frozen=True)
106class Rule:
107 prefix: str
108 service: str
109 port: int
112@dataclass(frozen=True)
113class GatewaySpec:
114 name: str
115 namespace: str
116 gateway_class: str
117 controller: str
118 listener: dict[str, Any]
119 load_balancer: dict[str, Any]
120 target_group: dict[str, Any]
121 route_name: str
122 rules: tuple[Rule, ...] = field(default_factory=tuple)
124 def as_json(self) -> dict[str, Any]:
125 return {
126 "name": self.name,
127 "namespace": self.namespace,
128 "gatewayClassName": self.gateway_class,
129 "controllerName": self.controller,
130 "listener": self.listener,
131 "loadBalancer": self.load_balancer,
132 "targetGroup": self.target_group,
133 "httpRoute": self.route_name,
134 }
137def _one(documents: list[dict[str, Any]], kind: str, name: str | None = None) -> dict[str, Any]:
138 matches = [
139 document
140 for document in documents
141 if document.get("kind") == kind
142 and (name is None or document.get("metadata", {}).get("name") == name)
143 ]
144 if len(matches) != 1:
145 raise ClusterGatewayDocumentError(
146 f"expected exactly one {kind}{f' named {name}' if name else ''} in the gateway "
147 f"manifest, found {len(matches)}"
148 )
149 return matches[0]
152def load_gateway(project_root: Path) -> GatewaySpec:
153 """The Gateway, its ALB configuration and the HTTPRoute rules, as data."""
154 text = (project_root / GATEWAY_MANIFEST).read_text(encoding="utf-8")
155 documents = [document for document in yaml.safe_load_all(text) if isinstance(document, dict)]
156 gateway = _one(documents, "Gateway")
157 gateway_class = _one(documents, "GatewayClass")
158 route = _one(documents, "HTTPRoute")
159 listeners = gateway["spec"].get("listeners", [])
160 if len(listeners) != 1:
161 raise ClusterGatewayDocumentError(f"expected one Gateway listener, found {len(listeners)}")
162 parameters = gateway["spec"].get("infrastructure", {}).get("parametersRef", {})
163 load_balancer = _one(documents, "LoadBalancerConfiguration", parameters.get("name"))
164 lb_spec = load_balancer["spec"]
165 target_group = _one(
166 documents,
167 "TargetGroupConfiguration",
168 lb_spec.get("defaultTargetGroupConfiguration", {}).get("name"),
169 )["spec"]["defaultConfiguration"]
171 namespace = str(route["metadata"]["namespace"])
172 rules: list[Rule] = []
173 for rule in route["spec"]["rules"]:
174 backends = rule.get("backendRefs", [])
175 if len(backends) != 1:
176 raise ClusterGatewayDocumentError(
177 f"expected a single backendRef per rule, got {len(backends)}"
178 )
179 backend = backends[0]
180 if (
181 backend.get("kind", "Service") != "Service"
182 or backend.get("namespace", namespace) != namespace
183 ):
184 raise ClusterGatewayDocumentError(
185 f"unsupported backendRef {backend!r}: only same-namespace Services are modelled"
186 )
187 for match in rule.get("matches", []):
188 path = match.get("path", {})
189 if path.get("type") != "PathPrefix":
190 raise ClusterGatewayDocumentError(
191 f"unsupported path match type {path.get('type')!r} for "
192 f"{path.get('value')!r}: only PathPrefix precedence is modelled"
193 )
194 if set(match) - {"path"}:
195 raise ClusterGatewayDocumentError(
196 f"rule for {path.get('value')!r} matches on more than the path"
197 )
198 rules.append(Rule(str(path["value"]), str(backend["name"]), int(backend["port"])))
199 if rule.get("filters"):
200 raise ClusterGatewayDocumentError("HTTPRoute filters are not modelled")
202 listener = listeners[0]
203 return GatewaySpec(
204 name=str(gateway["metadata"]["name"]),
205 namespace=namespace,
206 gateway_class=str(gateway["spec"]["gatewayClassName"]),
207 controller=str(gateway_class["spec"]["controllerName"]),
208 listener={
209 "name": listener.get("name"),
210 "protocol": listener.get("protocol"),
211 "port": listener.get("port"),
212 "allowedRoutes": listener.get("allowedRoutes"),
213 },
214 load_balancer={
215 "scheme": lb_spec.get("scheme"),
216 "listenerConfigurations": lb_spec.get("listenerConfigurations"),
217 "loadBalancerAttributes": lb_spec.get("loadBalancerAttributes"),
218 "tags": lb_spec.get("tags"),
219 },
220 target_group={
221 "targetType": target_group.get("targetType"),
222 "protocol": target_group.get("protocol"),
223 "healthCheck": target_group.get("healthCheckConfig"),
224 "targetGroupAttributes": target_group.get("targetGroupAttributes"),
225 },
226 route_name=str(route["metadata"]["name"]),
227 rules=tuple(rules),
228 )
231def _segments(path: str) -> list[str]:
232 return [segment for segment in path.split("/") if segment]
235def resolve(path: str, rules: tuple[Rule, ...]) -> Rule | None:
236 """The winning rule for ``path``: the longest ``PathPrefix`` match on whole segments."""
237 best: Rule | None = None
238 target = _segments(path)
239 for rule in rules:
240 candidate = _segments(rule.prefix)
241 if target[: len(candidate)] != candidate:
242 continue
243 if best is None or len(candidate) > len(_segments(best.prefix)):
244 best = rule
245 return best
248# ─── composition ─────────────────────────────────────────────────────────────
251def load_services(openapi_dir: Path) -> dict[str, dict[str, Any]]:
252 """``{service: document}`` for every FastAPI service document."""
253 services: dict[str, dict[str, Any]] = {}
254 for service in SERVICE_NAMES:
255 path = openapi_dir / f"{service}.json"
256 if not path.is_file():
257 raise ClusterGatewayDocumentError(
258 f"missing {path.relative_to(openapi_dir.parent.parent)}; run "
259 "python scripts/generate_openapi.py first"
260 )
261 document = json.loads(path.read_text(encoding="utf-8"))
262 if not isinstance(document, dict) or not isinstance(document.get("paths"), dict):
263 raise ClusterGatewayDocumentError(f"{service}: document has no paths object")
264 services[service] = document
265 return services
268def _rewrite_refs(value: Any, renames: dict[str, str]) -> Any:
269 """Apply ``renames`` to every ``#/components/schemas/<name>`` reference."""
270 if isinstance(value, dict):
271 rewritten = {key: _rewrite_refs(item, renames) for key, item in value.items()}
272 ref = rewritten.get("$ref")
273 if isinstance(ref, str) and ref.startswith("#/components/schemas/"):
274 name = ref[len("#/components/schemas/") :]
275 rewritten["$ref"] = "#/components/schemas/" + renames.get(name, name)
276 return rewritten
277 if isinstance(value, list):
278 return [_rewrite_refs(item, renames) for item in value]
279 return value
282def merge_schemas(
283 services: dict[str, dict[str, Any]],
284) -> tuple[dict[str, Any], dict[str, dict[str, str]]]:
285 """One schema namespace for every service document.
287 Identical definitions (FastAPI's ``HTTPValidationError`` in every app) keep
288 their name; a name two services define differently is prefixed with the
289 service (``manifest-processor.Status``) in that service's operations only.
290 Returns the merged schemas and, per service, the renames to apply.
291 """
292 merged: dict[str, Any] = {}
293 owners: dict[str, str] = {}
294 renames: dict[str, dict[str, str]] = {}
295 for service in sorted(services):
296 schemas = services[service].get("components", {}).get("schemas", {})
297 renames[service] = {}
298 for name, schema in sorted(schemas.items()):
299 if name not in merged:
300 merged[name] = schema
301 owners[name] = service
302 elif merged[name] != schema:
303 renames[service][name] = f"{service}.{name}"
304 # A rename must also be visible inside the renamed schema's own references.
305 for service, mapping in renames.items():
306 schemas = services[service].get("components", {}).get("schemas", {})
307 for old, new in mapping.items():
308 merged[new] = _rewrite_refs(schemas[old], mapping)
309 return dict(sorted(merged.items())), renames
312def compose(project_root: Path) -> dict[str, Any]:
313 """The Gateway's OpenAPI document from the manifest and the service documents."""
314 gateway = load_gateway(project_root)
315 services = load_services(project_root / "docs" / "openapi")
316 backends = {rule.service for rule in gateway.rules}
317 unknown = backends - set(services)
318 if unknown:
319 raise ClusterGatewayDocumentError(
320 f"HTTPRoute backends without an OpenAPI document: {sorted(unknown)}"
321 )
322 for service, consumer in INTERNAL_CONSUMERS.items():
323 if service in backends:
324 raise ClusterGatewayDocumentError(
325 f"{service} is listed in INTERNAL_CONSUMERS but is an HTTPRoute backend"
326 )
327 front = services.get(consumer["via"])
328 prefix = consumer["pathPrefix"]
329 if front is None or not any(path.startswith(prefix) for path in front["paths"]):
330 raise ClusterGatewayDocumentError(
331 f"{consumer['via']} serves no {prefix}* path to front {service}"
332 )
334 from gco.services.auth_middleware import UNAUTHENTICATED_PATHS
336 schemas, renames = merge_schemas(services)
337 paths: dict[str, dict[str, Any]] = {}
338 unreachable: list[dict[str, Any]] = []
339 for service in sorted(services):
340 document = services[service]
341 for path in sorted(document["paths"]):
342 item = document["paths"][path]
343 winner = resolve(path, gateway.rules)
344 if service not in backends:
345 continue
346 if winner is None or winner.service != service:
347 unreachable.append(
348 {
349 "path": path,
350 "service": service,
351 "rule": winner.prefix if winner else None,
352 "deliveredTo": winner.service if winner else None,
353 }
354 )
355 continue
356 # Exactly one Service wins a path, so no two services can both land here.
357 composed = _rewrite_refs(item, renames[service])
358 route = {
359 "rule": winner.prefix,
360 "service": service,
361 "namespace": gateway.namespace,
362 "port": winner.port,
363 }
364 for method, operation in composed.items():
365 if method == "parameters" or not isinstance(operation, dict):
366 continue
367 operation["tags"] = [
368 service,
369 *[t for t in operation.get("tags", []) if t != service],
370 ]
371 operation["security"] = [] if path in UNAUTHENTICATED_PATHS else [{"gcoHmac": []}]
372 operation["x-gco-route"] = route
373 paths[path] = composed
375 tags = [
376 {
377 "name": service,
378 "description": (
379 f"{services[service].get('info', {}).get('title', service)} — operations the "
380 f"HTTPRoute delivers to the `{service}` Service."
381 ),
382 }
383 for service in sorted(backends)
384 ]
385 versions = {str(document.get("openapi", "3.1.0")) for document in services.values()}
386 description = (
387 f"The HTTP surface of the internal ALB that fronts a GCO region: the Kubernetes "
388 f"Gateway `{gateway.namespace}/{gateway.name}` (GatewayClass `{gateway.gateway_class}`, "
389 f"implemented by `{gateway.controller}`, the AWS Load Balancer Controller) with "
390 f"one {gateway.listener.get('protocol')}:{gateway.listener.get('port')} listener. "
391 f"Its HTTPRoute `{gateway.route_name}` sends each request to a Service by longest "
392 f"path prefix; every operation below is one a backend Service serves *and* the "
393 f"route delivers to it, tagged with that Service and annotated with the winning "
394 f"rule (`x-gco-route`). The ALB is `{gateway.load_balancer.get('scheme')}` and "
395 f"answers only the API Gateway proxy Lambdas (over Global Accelerator or from the "
396 f"regional bridge's VPC), which sign every request with the HMAC envelope "
397 f"(`{HMAC_SIGNATURE_HEADER}` and companions) that the services verify on every "
398 f"path except {', '.join(f'`{p}`' for p in sorted(UNAUTHENTICATED_PATHS))}. "
399 f"The ALB terminates TLS with a deployment-local certificate and re-encrypts to a "
400 f"TLS proxy sidecar on each pod; the service documents describe the same operations "
401 f"as their applications see them."
402 )
403 return {
404 "openapi": max(versions),
405 "info": {
406 "title": "GCO Cluster Gateway (internal ALB)",
407 "version": API_VERSION,
408 "description": description,
409 },
410 "servers": [
411 {
412 "url": "https://{alb_hostname}",
413 "description": (
414 "The region's internal ALB, registered in SSM as "
415 "`/<project>/alb-hostname-<region>` by ga-registration and kept current by "
416 "the health monitor; reachable from inside the VPC and from Global "
417 "Accelerator only. Clients present SNI `backend.<project>.gco.internal`."
418 ),
419 "variables": {
420 "alb_hostname": {
421 "default": "<internal-alb-hostname>",
422 "description": "Assigned by the AWS Load Balancer Controller at deploy time.",
423 }
424 },
425 }
426 ],
427 "tags": tags,
428 "paths": paths,
429 "components": {
430 "schemas": schemas,
431 "securitySchemes": {
432 "gcoHmac": {
433 "type": "apiKey",
434 "in": "header",
435 "name": HMAC_SIGNATURE_HEADER,
436 "description": (
437 "Request-bound HMAC envelope minted by the API Gateway proxy Lambdas "
438 f"({', '.join(f'`{h}`' for h in HMAC_ENVELOPE_HEADERS)}): the "
439 "signature covers the method, exact path and query, body digest, "
440 "timestamp and nonce under the shared signing key from Secrets Manager. "
441 "Verified by `gco/services/auth_middleware.py`; a failure is a 403. "
442 "Original caller identity is authenticated upstream by API Gateway IAM."
443 ),
444 }
445 },
446 },
447 "x-gco-source": {
448 "kind": "cluster-gateway",
449 "generator": GENERATOR,
450 "manifest": GATEWAY_MANIFEST.as_posix(),
451 "composedFrom": [f"docs/openapi/{service}.json" for service in sorted(backends)],
452 },
453 "x-gco-gateway": gateway.as_json(),
454 "x-gco-routes": [
455 {"prefix": rule.prefix, "service": rule.service, "port": rule.port}
456 for rule in gateway.rules
457 ],
458 "x-gco-unreachable": unreachable,
459 "x-gco-not-exposed": {
460 service: consumer
461 for service, consumer in sorted(INTERNAL_CONSUMERS.items())
462 if service in services
463 },
464 "x-gco-downstream": {
465 service: downstream
466 for service, downstream in sorted(DOWNSTREAM.items())
467 if service in backends
468 },
469 "x-gco-unauthenticated-paths": sorted(UNAUTHENTICATED_PATHS),
470 }
473def render(document: dict[str, Any]) -> str:
474 """Serialize deterministically so regeneration produces a stable diff."""
475 return json.dumps(document, indent=2, sort_keys=True) + "\n"
478def main(argv: list[str] | None = None) -> int:
479 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
480 parser.add_argument(
481 "--check",
482 action="store_true",
483 help="Do not write; exit non-zero if the committed document is stale.",
484 )
485 args = parser.parse_args(argv)
486 try:
487 rendered = render(compose(REPO_ROOT))
488 except ClusterGatewayDocumentError as exc:
489 print(f"ERROR: {exc}", file=sys.stderr)
490 return 2
492 OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
493 target = OUTPUT_DIR / f"{DOCUMENT_NAME}.json"
494 current = target.read_text(encoding="utf-8") if target.is_file() else None
495 relative = target.relative_to(REPO_ROOT)
496 if args.check:
497 if current == rendered:
498 return 0
499 print(f"{relative}: {'missing' if current is None else 'stale'}")
500 print(f"\nRegenerate with: python {GENERATOR}", file=sys.stderr)
501 return 1
502 if current == rendered:
503 print(f"{relative}: unchanged")
504 else:
505 target.write_text(rendered, encoding="utf-8")
506 print(f"{relative}: written")
507 return 0
510if __name__ == "__main__":
511 raise SystemExit(main())