Coverage for scripts / generate_api_gateway_openapi.py: 100.00%

1#!/usr/bin/env python3 

2"""Generate the committed OpenAPI documents for the two GCO API Gateway REST APIs. 

3 

4The four FastAPI services describe themselves (``scripts/generate_openapi.py`` 

5asks each application for ``app.openapi()``). The two AWS API Gateway front 

6doors have no runtime to ask: they are CDK constructs. This script synthesizes 

7those stacks in-process — no ``cdk`` CLI, no AWS call — reads the 

8``AWS::ApiGateway::*`` resources out of the CloudFormation templates and writes 

9one OpenAPI 3.0.3 document per API: 

10 

11* ``docs/openapi/api-gateway-global.json`` — ``GCOApiGatewayGlobalStack`` 

12 (``<project>-global-api``): the edge-optimized single entry point. 

13* ``docs/openapi/api-gateway-regional.json`` — ``GCORegionalApiGatewayStack`` 

14 (``<project>-regional-api-<region>``): the per-region IAM bridge. 

15 

16Every operation records API Gateway's authorization as an OpenAPI security 

17requirement, the integration (``x-amazon-apigateway-integration``: type, 

18timeout, response transfer mode) and, under ``x-gco-backend``, the Lambda that 

19receives the request and the hops it takes from there (Global Accelerator, the 

20in-cluster Gateway, a regional API Gateway) — the same route table the 

21interaction diagram in ``diagrams/api_specs/`` is drawn from. Routes that exist 

22only under a deployment condition carry ``x-gco-condition``: the global stack is 

23synthesized three times (full, without analytics, without Global Accelerator) 

24and a route's condition is whichever knob makes it appear. 

25 

26Usage:: 

27 

28 python scripts/generate_api_gateway_openapi.py # write the documents 

29 python scripts/generate_api_gateway_openapi.py --check # fail if anything is stale 

30 

31``--check`` regenerates in memory and compares, so a route added to a stack 

32without regenerating is caught in review. Prose for a route (its tag, summary 

33and description) lives in ``ROUTE_DOCS`` here; a synthesized route with no entry 

34fails generation rather than shipping an undocumented operation. 

35""" 

36 

37from __future__ import annotations 

38 

39import argparse 

40import json 

41import re 

42import sys 

43from dataclasses import dataclass 

44from pathlib import Path 

45from typing import Any 

46 

47REPO_ROOT = Path(__file__).resolve().parent.parent 

48OUTPUT_DIR = REPO_ROOT / "docs" / "openapi" 

49 

50#: Filename stems of the documents this generator owns. 

51DOCUMENT_NAMES: tuple[str, ...] = ("api-gateway-global", "api-gateway-regional") 

52 

53OPENAPI_VERSION = "3.0.3" 

54#: The version of the gateway route contract, independent of the release 

55#: number so a version bump alone never makes these documents stale. 

56API_VERSION = "1.0.0" 

57GENERATOR = "scripts/generate_api_gateway_openapi.py" 

58 

59#: CDK context key that stamps every resource with its construct path. 

60PATH_METADATA_CONTEXT = "aws:cdk:enable-path-metadata" 

61#: Construct id of the inline Lambda that stands in for the analytics stack's 

62#: presigned-URL function while the global stack is synthesized here. 

63ANALYTICS_STAND_IN_ID = "AnalyticsPresignedUrlStandIn" 

64#: Account of the placeholder ARNs the synthesis needs for cross-stack inputs. 

65#: Nothing deploys from here; ``plain`` renders it back as ``${AWS::AccountId}``. 

66PLACEHOLDER_ACCOUNT = "000000000000" 

67 

68_HASH_SUFFIX_RE = re.compile(r"[0-9A-F]{8}$") 

69 

70 

71class ApiGatewayDocumentError(RuntimeError): 

72 """A synthesized API cannot be described faithfully by this generator.""" 

73 

74 

75# ─── route documentation ───────────────────────────────────────────────────── 

76 

77 

78@dataclass(frozen=True) 

79class Hop: 

80 """One hop on a request's way from a proxy Lambda to the service that answers.""" 

81 

82 id: str 

83 label: str 

84 #: Stems of the ``docs/openapi/`` documents describing this hop, if any. 

85 documents: tuple[str, ...] = () 

86 

87 def as_json(self) -> dict[str, Any]: 

88 return {"id": self.id, "label": self.label, "documents": list(self.documents)} 

89 

90 

91@dataclass(frozen=True) 

92class Backend: 

93 """The Lambda behind an integration and the hops that follow it.""" 

94 

95 name: str 

96 source: str 

97 description: str 

98 hops: tuple[Hop, ...] 

99 

100 

101GLOBAL_ACCELERATOR = Hop( 

102 "global-accelerator", 

103 "AWS Global Accelerator: TCP/443 pass-through to the regional internal ALB", 

104) 

105CLUSTER_GATEWAY = Hop( 

106 "cluster-gateway", 

107 "Internal ALB — the Kubernetes Gateway gco-system/gco-gateway; HTTPRoute " 

108 "gco-routes picks the Service by longest path prefix", 

109 ("cluster-gateway",), 

110) 

111CONTROL_PLANE_SERVICES = Hop( 

112 "control-plane-services", 

113 "manifest-processor (catch-all and /api/v1/manifests) or health-monitor " 

114 "(/api/v1/health, /api/v1/metrics, /healthz)", 

115 ("manifest-processor", "health-monitor"), 

116) 

117INFERENCE_PROXY = Hop( 

118 "inference-proxy", 

119 "inference-proxy Service: resolves the endpoint by name, enforces its state " 

120 "and the serving-path allowlist", 

121 ("inference-proxy",), 

122) 

123INFERENCE_ENDPOINT = Hop( 

124 "inference-endpoint", 

125 "The endpoint's model-server Service in gco-inference (<name>, <name>-canary " 

126 "or <name>-proxy), streamed back through every hop", 

127) 

128REGIONAL_API = Hop( 

129 "api-gateway-regional", 

130 "Each region's API Gateway bridge, called with SigV4 by the aggregator's own " 

131 "execution role (GET/DELETE api/v1/jobs, GET api/v1/health, api/v1/status, " 

132 "api/v1/policy)", 

133 ("api-gateway-regional",), 

134) 

135REGIONAL_PROXY = Hop( 

136 "regional-api-proxy", 

137 "regional-api-proxy VPC Lambda: resolves the ALB from SSM " 

138 "/<project>/alb-hostname-<region> and signs the HMAC envelope", 

139) 

140 

141#: ``(api kind, Lambda construct id) -> Backend``. A synthesized integration 

142#: whose Lambda is not listed here fails generation: every new front-door 

143#: Lambda must say where it sends requests. 

144BACKENDS: dict[tuple[str, str], Backend] = { 

145 ("global", "ApiGatewayProxyFunction"): Backend( 

146 name="api-gateway-proxy", 

147 source="lambda/api-gateway-proxy/handler.py", 

148 description=( 

149 "Buffered control-plane proxy. Fetches the HMAC signing key from Secrets " 

150 "Manager, rejects region-pinning headers and base64 bodies, signs the method, " 

151 "path, query, body digest, timestamp and nonce into the X-GCO-* envelope and " 

152 "forwards over private-root TLS within 28 s." 

153 ), 

154 hops=(GLOBAL_ACCELERATOR, CLUSTER_GATEWAY, CONTROL_PLANE_SERVICES), 

155 ), 

156 ("global", "InferenceStreamingProxyFunction"): Backend( 

157 name="inference-streaming-proxy", 

158 source="lambda/inference-streaming-proxy/index.mjs", 

159 description=( 

160 "Node.js response-streaming proxy (ROUTING_MODE=global). Accepts GET, HEAD " 

161 "and POST under /inference/<endpoint>/..., caps the body at 1 MiB, signs the " 

162 "HMAC envelope and streams the model server's response for up to 15 minutes." 

163 ), 

164 hops=(GLOBAL_ACCELERATOR, CLUSTER_GATEWAY, INFERENCE_PROXY, INFERENCE_ENDPOINT), 

165 ), 

166 ("global", "CrossRegionAggregatorFunction"): Backend( 

167 name="cross-region-aggregator", 

168 source="lambda/cross-region-aggregator/handler.py", 

169 description=( 

170 "Discovers every <project>-regional-api-<region> stack's RegionalApiEndpoint " 

171 "output, fans the request out to each regional API with SigV4 and merges the " 

172 "answers into one response." 

173 ), 

174 hops=(REGIONAL_API, REGIONAL_PROXY, CLUSTER_GATEWAY, CONTROL_PLANE_SERVICES), 

175 ), 

176 ("global", ANALYTICS_STAND_IN_ID): Backend( 

177 name="analytics-presigned-url", 

178 source="lambda/analytics-presigned-url/handler.py", 

179 description=( 

180 "The analytics stack's presigned-URL Lambda (GCOAnalyticsStack): exchanges the " 

181 "Cognito identity for a SageMaker Studio presigned login URL. Synthesized here " 

182 "with an inline stand-in because the analytics stack is not part of this API's " 

183 "template." 

184 ), 

185 hops=(), 

186 ), 

187 ("regional", "RegionalProxyFunction"): Backend( 

188 name="regional-api-proxy", 

189 source="lambda/regional-api-proxy/handler.py", 

190 description=( 

191 "Buffered VPC Lambda. Resolves the internal ALB hostname registered in SSM " 

192 "/<project>/alb-hostname-<region>, verifies it is an account-owned internal " 

193 "ALB, signs the HMAC envelope and forwards inside the VPC within 28 s." 

194 ), 

195 hops=(CLUSTER_GATEWAY, CONTROL_PLANE_SERVICES), 

196 ), 

197 ("regional", "InferenceStreamingProxyFunction"): Backend( 

198 name="inference-streaming-proxy", 

199 source="lambda/inference-streaming-proxy/index.mjs", 

200 description=( 

201 "Node.js response-streaming VPC Lambda (ROUTING_MODE=regional): the same " 

202 "handler as the global route, reaching the ALB directly instead of through " 

203 "Global Accelerator." 

204 ), 

205 hops=(CLUSTER_GATEWAY, INFERENCE_PROXY, INFERENCE_ENDPOINT), 

206 ), 

207} 

208 

209 

210@dataclass(frozen=True) 

211class RouteDoc: 

212 """Human prose for one ``(METHOD, path)`` a stack exposes.""" 

213 

214 tag: str 

215 summary: str 

216 

217 

218_CONTROL_PLANE_SUMMARIES: dict[str, str] = { 

219 "GET": "Read a control-plane resource", 

220 "HEAD": "Probe a control-plane resource", 

221 "POST": "Create a control-plane resource or submit a job", 

222 "PUT": "Replace a control-plane resource", 

223 "PATCH": "Update a control-plane resource", 

224 "DELETE": "Delete a control-plane resource", 

225 "OPTIONS": "Describe the methods a control-plane path accepts", 

226} 

227_INFERENCE_SUMMARIES: dict[str, str] = { 

228 "GET": "Read an inference endpoint's health, info or model list", 

229 "HEAD": "Probe an inference endpoint", 

230 "POST": "Generate: forward a completion request and stream the response", 

231} 

232 

233#: ``(METHOD, path) -> RouteDoc`` for every route either stack exposes. 

234ROUTE_DOCS: dict[tuple[str, str], RouteDoc] = { 

235 **{ 

236 (method, "/api/v1/{proxy+}"): RouteDoc("control-plane", summary) 

237 for method, summary in _CONTROL_PLANE_SUMMARIES.items() 

238 }, 

239 **{ 

240 (method, "/inference/{proxy+}"): RouteDoc("inference", summary) 

241 for method, summary in _INFERENCE_SUMMARIES.items() 

242 }, 

243 ("GET", "/api/v1/global/jobs"): RouteDoc("global", "List jobs across every region"), 

244 ("DELETE", "/api/v1/global/jobs"): RouteDoc("global", "Bulk-delete jobs across every region"), 

245 ("GET", "/api/v1/global/health"): RouteDoc("global", "Health of every region's cluster"), 

246 ("GET", "/api/v1/global/status"): RouteDoc("global", "Aggregated status of every region"), 

247 ("GET", "/studio/login"): RouteDoc("studio", "Presigned SageMaker Studio login URL"), 

248 ("GET", "/studio/callback"): RouteDoc("studio", "OAuth redirect landing page (stub)"), 

249} 

250 

251#: Longer description shared by every method on a path. 

252PATH_DESCRIPTIONS: dict[str, str] = { 

253 "/api/v1/{proxy+}": ( 

254 "Catch-all for the control-plane API served by the in-cluster FastAPI services. " 

255 "`{proxy+}` is the rest of the path — for example `jobs`, `jobs/{namespace}/{name}`, " 

256 "`manifests`, `health`, `status` — so the operations available here are the " 

257 "`/api/v1/*` paths of the manifest-processor and health-monitor documents, minus " 

258 "`/api/v1/global/*`, which the aggregator serves. The request is forwarded " 

259 "unchanged (the stage prefix is not part of the backend path); the response is " 

260 "returned unchanged." 

261 ), 

262 "/inference/{proxy+}": ( 

263 "Streaming access to deployed inference endpoints. `{proxy+}` is " 

264 "`<endpoint-name>` or `<endpoint-name>/<serving-path>`; the in-cluster " 

265 "inference-proxy resolves the endpoint by name and allows only its serving and " 

266 "health paths (see the inference-proxy document). Bodies are limited to 1 MiB and " 

267 "the connection may stay open for API Gateway's full 15-minute streaming window." 

268 ), 

269 "/api/v1/global/jobs": ( 

270 "Fans out to `/api/v1/jobs` on every regional API Gateway and merges the " 

271 "results. Requires the regional bridges to be deployed; each region's answer is " 

272 "attributed to it in the response." 

273 ), 

274 "/api/v1/global/health": ( 

275 "Fans out to `/api/v1/health` on every regional API Gateway and returns one " 

276 "health object per region." 

277 ), 

278 "/api/v1/global/status": ( 

279 "Fans out to `/api/v1/status` on every regional API Gateway (answered by the " 

280 "manifest-processor) and merges templates, webhooks, resource limits and allowed " 

281 "namespaces per region." 

282 ), 

283 "/studio/login": ( 

284 "Authenticated with a Cognito ID token from the Studio user pool rather than " 

285 "SigV4. Returns a presigned SageMaker Studio URL for the caller's user profile." 

286 ), 

287 "/studio/callback": ( 

288 "Unauthenticated MOCK integration returning an empty 200 body: the Cognito " 

289 "hosted-UI OAuth redirect target. The browser redirect carries the authorization " 

290 "code as a query-string parameter and nothing reads the body." 

291 ), 

292} 

293 

294#: Descriptions of the greedy path parameters. 

295PATH_PARAMETERS: dict[str, str] = { 

296 "proxy": "The remainder of the backend path (API Gateway greedy path variable).", 

297} 

298 

299RESPONSE_DESCRIPTIONS: dict[str, str] = { 

300 "200": "Success; the backend's response is returned unchanged.", 

301 "400": ( 

302 "Bad request: rejected by the proxy before forwarding (for example an " 

303 "`X-GCO-Target-Region` header or a base64-encoded body) or by the backend." 

304 ), 

305 "401": "Unauthorized: the Cognito ID token is missing, expired or not from the Studio pool.", 

306 "403": ( 

307 "Forbidden: the caller's IAM identity or the API's resource policy denied " 

308 "`execute-api:Invoke`, or the backend rejected the request's HMAC envelope." 

309 ), 

310 "404": ( 

311 "Not found: the endpoint does not exist in this region or the path is outside " 

312 "the inference serving-path allowlist." 

313 ), 

314 "500": "Internal error in the proxy Lambda or the backend.", 

315 "502": "Bad gateway: the in-cluster proxy or the model server did not answer.", 

316} 

317 

318#: Security schemes, keyed by API Gateway's ``AuthorizationType``. 

319SECURITY_SCHEMES: dict[str, tuple[str, dict[str, Any]]] = { 

320 "AWS_IAM": ( 

321 "sigv4", 

322 { 

323 "type": "apiKey", 

324 "name": "Authorization", 

325 "in": "header", 

326 "x-amazon-apigateway-authtype": "awsSigv4", 

327 "description": ( 

328 "AWS Signature Version 4 for service `execute-api`, signed with IAM " 

329 "credentials of the deploying account that carry `execute-api:Invoke` on " 

330 "this API (for example `awscurl --service execute-api --region <region>`)." 

331 ), 

332 }, 

333 ), 

334 "COGNITO_USER_POOLS": ( 

335 "cognito", 

336 { 

337 "type": "apiKey", 

338 "name": "Authorization", 

339 "in": "header", 

340 "x-amazon-apigateway-authtype": "cognito_user_pools", 

341 "description": ( 

342 "ID token issued by the Studio Cognito user pool, validated by the " 

343 "`<project>-studio-cognito-authorizer` authorizer." 

344 ), 

345 }, 

346 ), 

347} 

348 

349CONDITIONS: dict[str, dict[str, str]] = { 

350 "global-accelerator": { 

351 "id": "global-accelerator", 

352 "description": ( 

353 "Present only in the commercial `aws` partition, where Global Accelerator " 

354 "provides the data path to the regional ALBs. Elsewhere the global API keeps " 

355 "only its aggregate routes and workload traffic uses each region's bridge." 

356 ), 

357 }, 

358 "analytics": { 

359 "id": "analytics", 

360 "description": "Present only when `analytics_environment.enabled` is true in cdk.json.", 

361 }, 

362} 

363 

364TAGS: dict[str, str] = { 

365 "control-plane": "Manifests, jobs, templates, webhooks, cost and status — proxied to the cluster.", 

366 "global": "Cross-region aggregation, answered by the aggregator Lambda.", 

367 "inference": "Streaming access to deployed model endpoints.", 

368 "studio": "SageMaker Studio access for the analytics environment.", 

369} 

370 

371 

372# ─── CloudFormation template reading ───────────────────────────────────────── 

373 

374 

375def _resources_of(template: dict[str, Any], resource_type: str) -> dict[str, dict[str, Any]]: 

376 resources = template.get("Resources") 

377 if not isinstance(resources, dict): 

378 raise ApiGatewayDocumentError("template has no Resources") 

379 return { 

380 logical_id: resource 

381 for logical_id, resource in resources.items() 

382 if isinstance(resource, dict) and resource.get("Type") == resource_type 

383 } 

384 

385 

386def _properties(resource: dict[str, Any]) -> dict[str, Any]: 

387 properties = resource.get("Properties") 

388 return properties if isinstance(properties, dict) else {} 

389 

390 

391def plain(value: Any) -> Any: 

392 """CloudFormation intrinsics rendered as ``${...}`` placeholders, recursively. 

393 

394 ``{"Ref": "AWS::AccountId"}`` becomes ``${AWS::AccountId}``, a ``Fn::GetAtt`` 

395 becomes ``${Logical.Attribute}`` and a ``Fn::Join`` is joined — so a 

396 resource policy reads as the statement it deploys as, with deploy-time 

397 values named rather than guessed. 

398 """ 

399 if isinstance(value, dict): 

400 if set(value) == {"Ref"}: 

401 return f"${{{value['Ref']}}}" 

402 if set(value) == {"Fn::GetAtt"}: 

403 target = value["Fn::GetAtt"] 

404 return "${" + ".".join(str(part) for part in target) + "}" 

405 if set(value) == {"Fn::Join"}: 

406 delimiter, parts = value["Fn::Join"] 

407 return str(delimiter).join(str(plain(part)) for part in parts) 

408 if set(value) == {"Fn::Sub"}: 

409 template = value["Fn::Sub"] 

410 return template if isinstance(template, str) else str(plain(template[0])) 

411 return {str(key): plain(item) for key, item in value.items()} 

412 if isinstance(value, list): 

413 return [plain(item) for item in value] 

414 if isinstance(value, str): 

415 return value.replace(PLACEHOLDER_ACCOUNT, "${AWS::AccountId}") 

416 return value 

417 

418 

419def rest_api(template: dict[str, Any]) -> tuple[str, dict[str, Any]]: 

420 """The single ``AWS::ApiGateway::RestApi`` as ``(logical id, properties)``.""" 

421 apis = _resources_of(template, "AWS::ApiGateway::RestApi") 

422 if len(apis) != 1: 

423 raise ApiGatewayDocumentError(f"expected exactly one RestApi, found {len(apis)}") 

424 ((logical_id, resource),) = apis.items() 

425 return logical_id, _properties(resource) 

426 

427 

428def resource_paths(template: dict[str, Any], api_id: str) -> dict[str, str]: 

429 """``{resource logical id: full path}`` by walking ``ParentId`` to the API root.""" 

430 resources = { 

431 logical_id: _properties(resource) 

432 for logical_id, resource in _resources_of(template, "AWS::ApiGateway::Resource").items() 

433 } 

434 root = {"Fn::GetAtt": [api_id, "RootResourceId"]} 

435 paths: dict[str, str] = {} 

436 pending = dict(resources) 

437 while pending: 

438 progressed = False 

439 for logical_id, properties in list(pending.items()): 

440 parent = properties.get("ParentId") 

441 part = str(properties.get("PathPart", "")) 

442 if parent == root: 

443 parent_path = "" 

444 elif isinstance(parent, dict) and parent.get("Ref") in paths: 

445 parent_path = paths[str(parent["Ref"])] 

446 else: 

447 continue 

448 paths[logical_id] = f"{parent_path}/{part}" 

449 del pending[logical_id] 

450 progressed = True 

451 if not progressed: 

452 raise ApiGatewayDocumentError( 

453 f"unresolvable ParentId chain for resources {sorted(pending)}" 

454 ) 

455 return paths 

456 

457 

458def lambda_construct_id(template: dict[str, Any], logical_id: str) -> str: 

459 """The construct id of a Lambda from its path metadata, else its logical id.""" 

460 functions = _resources_of(template, "AWS::Lambda::Function") 

461 if logical_id not in functions: 

462 raise ApiGatewayDocumentError(f"integration target {logical_id} is not a Lambda function") 

463 metadata = functions[logical_id].get("Metadata") 

464 path = metadata.get("aws:cdk:path") if isinstance(metadata, dict) else None 

465 if isinstance(path, str) and path.count("/") >= 2: 

466 return path.split("/")[1] 

467 return _HASH_SUFFIX_RE.sub("", logical_id) 

468 

469 

470def integration_lambda(integration: dict[str, Any]) -> str | None: 

471 """The logical id of the Lambda an ``AWS_PROXY`` integration invokes.""" 

472 uri = integration.get("Uri") 

473 if not isinstance(uri, dict) or "Fn::Join" not in uri: 

474 return None 

475 for part in uri["Fn::Join"][1]: 

476 if isinstance(part, dict) and "Fn::GetAtt" in part: 

477 target, attribute = part["Fn::GetAtt"] 

478 if attribute == "Arn": 

479 return str(target) 

480 return None 

481 

482 

483def stage_settings(template: dict[str, Any]) -> dict[str, Any]: 

484 """Stage name, throttling, logging, metrics and tracing of the single stage.""" 

485 stages = _resources_of(template, "AWS::ApiGateway::Stage") 

486 if len(stages) != 1: 

487 raise ApiGatewayDocumentError(f"expected exactly one Stage, found {len(stages)}") 

488 (properties,) = (_properties(resource) for resource in stages.values()) 

489 settings = [ 

490 item 

491 for item in properties.get("MethodSettings", []) 

492 if isinstance(item, dict) and item.get("HttpMethod") == "*" 

493 ] 

494 method_settings = settings[0] if settings else {} 

495 return { 

496 "name": str(properties.get("StageName", "")), 

497 "throttling": { 

498 "rateLimit": method_settings.get("ThrottlingRateLimit"), 

499 "burstLimit": method_settings.get("ThrottlingBurstLimit"), 

500 }, 

501 "loggingLevel": method_settings.get("LoggingLevel"), 

502 "dataTraceEnabled": method_settings.get("DataTraceEnabled"), 

503 "metricsEnabled": method_settings.get("MetricsEnabled"), 

504 "tracingEnabled": properties.get("TracingEnabled"), 

505 "accessLogging": "AccessLogSetting" in properties, 

506 } 

507 

508 

509def waf_rules(template: dict[str, Any]) -> list[dict[str, Any]]: 

510 """Name, priority and action of every rule of the WebACL, if the API has one.""" 

511 acls = _resources_of(template, "AWS::WAFv2::WebACL") 

512 rules: list[dict[str, Any]] = [] 

513 for resource in acls.values(): 

514 for rule in _properties(resource).get("Rules", []): 

515 if not isinstance(rule, dict): 

516 continue 

517 statement = rule.get("Statement") or {} 

518 managed = statement.get("ManagedRuleGroupStatement") 

519 if isinstance(rule.get("Action"), dict) and rule["Action"]: 

520 action = next(iter(rule["Action"])).lower() 

521 elif isinstance(rule.get("OverrideAction"), dict) and "Count" in rule["OverrideAction"]: 

522 action = "count (group actions overridden)" 

523 else: 

524 # ``OverrideAction: None`` — the managed group's own rule actions apply. 

525 action = "group default" 

526 rules.append( 

527 { 

528 "name": rule.get("Name"), 

529 "priority": rule.get("Priority"), 

530 "action": action, 

531 "managedRuleGroup": managed.get("Name") if isinstance(managed, dict) else None, 

532 } 

533 ) 

534 return sorted(rules, key=lambda rule: int(rule["priority"])) 

535 

536 

537def resource_policy(properties: dict[str, Any]) -> list[dict[str, Any]]: 

538 """The API's resource policy statements with intrinsics rendered as placeholders.""" 

539 policy = properties.get("Policy") 

540 if not isinstance(policy, dict): 

541 return [] 

542 statements = policy.get("Statement", []) 

543 return [plain(statement) for statement in statements if isinstance(statement, dict)] 

544 

545 

546# ─── OpenAPI assembly ──────────────────────────────────────────────────────── 

547 

548 

549@dataclass(frozen=True) 

550class ApiDescription: 

551 """What the template cannot say about itself.""" 

552 

553 kind: str 

554 name: str 

555 title: str 

556 description: str 

557 stack_module: str 

558 construct: str 

559 stack_name: str 

560 region_default: str 

561 region_note: str 

562 endpoint_output: str 

563 synthesized_with: dict[str, Any] 

564 

565 

566Route = tuple[str, str] 

567 

568 

569def routes_of(template: dict[str, Any]) -> dict[Route, dict[str, Any]]: 

570 """``{(METHOD, path): method properties}`` for every method of the API.""" 

571 api_id, _ = rest_api(template) 

572 paths = resource_paths(template, api_id) 

573 root = {"Fn::GetAtt": [api_id, "RootResourceId"]} 

574 routes: dict[Route, dict[str, Any]] = {} 

575 for logical_id, resource in _resources_of(template, "AWS::ApiGateway::Method").items(): 

576 properties = _properties(resource) 

577 resource_ref = properties.get("ResourceId") 

578 if resource_ref == root: 

579 path = "/" 

580 elif isinstance(resource_ref, dict) and resource_ref.get("Ref") in paths: 

581 path = paths[str(resource_ref["Ref"])] 

582 else: 

583 raise ApiGatewayDocumentError(f"method {logical_id} has an unresolvable ResourceId") 

584 method = str(properties.get("HttpMethod", "")).upper() 

585 routes[(method, path)] = properties 

586 return routes 

587 

588 

589def _path_parameters(path: str) -> list[dict[str, Any]]: 

590 parameters = [] 

591 for name in re.findall(r"\{([^}]+)\}", path): 

592 bare = name.rstrip("+") 

593 parameters.append( 

594 { 

595 "name": bare, 

596 "in": "path", 

597 "required": True, 

598 "schema": {"type": "string"}, 

599 "description": PATH_PARAMETERS.get(bare, f"Path parameter `{bare}`."), 

600 } 

601 ) 

602 return parameters 

603 

604 

605def _operation_id(method: str, path: str) -> str: 

606 words = [word for word in re.split(r"[^A-Za-z0-9]+", path) if word] 

607 camel = "".join(word[:1].upper() + word[1:] for word in words) 

608 return f"{method.lower()}{camel}" if camel else f"{method.lower()}Root" 

609 

610 

611def _integration( 

612 template: dict[str, Any], kind: str, properties: dict[str, Any] 

613) -> tuple[dict[str, Any], dict[str, Any]]: 

614 """``(x-amazon-apigateway-integration, x-gco-backend)`` for one method.""" 

615 integration = properties.get("Integration") 

616 if not isinstance(integration, dict): 

617 raise ApiGatewayDocumentError("method has no Integration") 

618 extension: dict[str, Any] = { 

619 "type": str(integration.get("Type", "")).lower(), 

620 } 

621 if "IntegrationHttpMethod" in integration: 

622 extension["httpMethod"] = integration["IntegrationHttpMethod"] 

623 if "TimeoutInMillis" in integration: 

624 extension["timeoutInMillis"] = integration["TimeoutInMillis"] 

625 if "ResponseTransferMode" in integration: 

626 extension["responseTransferMode"] = integration["ResponseTransferMode"] 

627 

628 if extension["type"] == "mock": 

629 extension["requestTemplates"] = plain(integration.get("RequestTemplates", {})) 

630 extension["responses"] = plain(integration.get("IntegrationResponses", [])) 

631 backend: dict[str, Any] = { 

632 "integration": "mock", 

633 "description": "API Gateway answers directly; no Lambda and no backend is involved.", 

634 "hops": [], 

635 } 

636 return extension, backend 

637 

638 target = integration_lambda(integration) 

639 if target is None: 

640 raise ApiGatewayDocumentError(f"{extension['type']} integration has no Lambda target") 

641 construct_id = lambda_construct_id(template, target) 

642 known = BACKENDS.get((kind, construct_id)) 

643 if known is None: 

644 raise ApiGatewayDocumentError( 

645 f"no BACKENDS entry for ({kind!r}, {construct_id!r}); document where this " 

646 "Lambda sends requests in scripts/generate_api_gateway_openapi.py" 

647 ) 

648 function = _properties(_resources_of(template, "AWS::Lambda::Function")[target]) 

649 backend = { 

650 "integration": "lambda", 

651 "lambda": { 

652 "name": known.name, 

653 "source": known.source, 

654 "runtime": function.get("Runtime"), 

655 "handler": function.get("Handler"), 

656 "timeoutSeconds": function.get("Timeout"), 

657 "memorySizeMb": function.get("MemorySize"), 

658 }, 

659 "description": known.description, 

660 "hops": [hop.as_json() for hop in known.hops], 

661 } 

662 return extension, backend 

663 

664 

665def _security(properties: dict[str, Any], schemes: dict[str, dict[str, Any]]) -> list[Any]: 

666 authorization = str(properties.get("AuthorizationType", "NONE")).upper() 

667 if authorization == "NONE": 

668 return [] 

669 if authorization not in SECURITY_SCHEMES: 

670 raise ApiGatewayDocumentError(f"unsupported AuthorizationType {authorization!r}") 

671 name, scheme = SECURITY_SCHEMES[authorization] 

672 schemes[name] = scheme 

673 return [{name: []}] 

674 

675 

676def _responses(properties: dict[str, Any], mock: bool) -> dict[str, Any]: 

677 responses: dict[str, Any] = {} 

678 for response in properties.get("MethodResponses", []): 

679 if not isinstance(response, dict): 

680 continue 

681 status = str(response.get("StatusCode", "")) 

682 description = RESPONSE_DESCRIPTIONS.get(status, f"{status} response") 

683 if mock and status == "200": 

684 description = "Success; empty JSON body." 

685 responses[status] = {"description": description} 

686 if not responses: 

687 raise ApiGatewayDocumentError("method declares no MethodResponses") 

688 return responses 

689 

690 

691def build_document( 

692 template: dict[str, Any], 

693 api: ApiDescription, 

694 *, 

695 conditions: dict[Route, str] | None = None, 

696 direct_access_template: dict[str, Any] | None = None, 

697) -> dict[str, Any]: 

698 """One OpenAPI document for the API a template deploys. 

699 

700 ``direct_access_template`` is the same API synthesized with 

701 ``api_gateway.regional_api_enabled``; the resource-policy statements it adds 

702 are reported separately so the opt-in is visible without a second document. 

703 """ 

704 _api_id, api_properties = rest_api(template) 

705 # Deploy-time references (the API id, the user pool ARN) say nothing about 

706 # the contract, so only the settings that describe behaviour are kept. 

707 validators = { 

708 logical_id: { 

709 key: value 

710 for key, value in plain(_properties(resource)).items() 

711 if key in {"Name", "ValidateRequestBody", "ValidateRequestParameters"} 

712 } 

713 for logical_id, resource in _resources_of( 

714 template, "AWS::ApiGateway::RequestValidator" 

715 ).items() 

716 } 

717 authorizers = { 

718 logical_id: { 

719 key: value 

720 for key, value in plain(_properties(resource)).items() 

721 if key in {"Name", "Type", "IdentitySource", "AuthorizerResultTtlInSeconds"} 

722 } 

723 for logical_id, resource in _resources_of(template, "AWS::ApiGateway::Authorizer").items() 

724 } 

725 schemes: dict[str, dict[str, Any]] = {} 

726 paths: dict[str, dict[str, Any]] = {} 

727 used_tags: set[str] = set() 

728 documented = set(ROUTE_DOCS) 

729 for (method, path), properties in sorted(routes_of(template).items()): 

730 doc = ROUTE_DOCS.get((method, path)) 

731 if doc is None: 

732 raise ApiGatewayDocumentError( 

733 f"no ROUTE_DOCS entry for {method} {path}; describe the route in " 

734 "scripts/generate_api_gateway_openapi.py" 

735 ) 

736 documented.discard((method, path)) 

737 integration, backend = _integration(template, api.kind, properties) 

738 operation: dict[str, Any] = { 

739 "operationId": _operation_id(method, path), 

740 "summary": doc.summary, 

741 "description": PATH_DESCRIPTIONS.get(path, doc.summary), 

742 "tags": [doc.tag], 

743 "security": _security(properties, schemes), 

744 "responses": _responses(properties, integration["type"] == "mock"), 

745 "x-amazon-apigateway-integration": integration, 

746 "x-gco-backend": backend, 

747 } 

748 used_tags.add(doc.tag) 

749 authorizer = properties.get("AuthorizerId") 

750 if isinstance(authorizer, dict) and authorizer.get("Ref") in authorizers: 

751 operation["x-amazon-apigateway-authorizer"] = authorizers[str(authorizer["Ref"])] 

752 validator = properties.get("RequestValidatorId") 

753 if isinstance(validator, dict) and validator.get("Ref") in validators: 

754 operation["x-amazon-apigateway-request-validator"] = validators[str(validator["Ref"])] 

755 if conditions and (method, path) in conditions: 

756 condition = CONDITIONS[conditions[(method, path)]] 

757 operation["x-gco-condition"] = condition 

758 operation["description"] += f"\n\n**Only when:** {condition['description']}" 

759 item = paths.setdefault(path, {}) 

760 parameters = _path_parameters(path) 

761 if parameters: 

762 item["parameters"] = parameters 

763 item[method.lower()] = operation 

764 

765 paths = dict(sorted(paths.items())) 

766 endpoint_types = api_properties.get("EndpointConfiguration", {}).get("Types", []) 

767 stage = stage_settings(template) 

768 server_url = "https://{api_id}.execute-api.{region}.{url_suffix}/{stage}" 

769 document: dict[str, Any] = { 

770 "openapi": OPENAPI_VERSION, 

771 "info": { 

772 "title": api.title, 

773 "version": API_VERSION, 

774 "description": ( 

775 f"{api_properties.get('Description', '')}\n\n{api.description}".strip() 

776 ), 

777 }, 

778 "servers": [ 

779 { 

780 "url": server_url, 

781 "description": ( 

782 f"Stage `{stage['name']}` of the deployed REST API. The full URL is the " 

783 f"`{api.endpoint_output}` output of the `{api.stack_name}` stack." 

784 ), 

785 "variables": { 

786 "api_id": { 

787 "default": "<api-id>",