Coverage for lambda / tls-certificate-manager / handler.py: 100.00%

1"""Manage GCO's deployment-local root CA and regional ACM leaf certificates. 

2 

3The function has two entry points: 

4 

5* a CDK ``cr.Provider`` custom resource bootstraps the root and one imported 

6 certificate per workload region; and 

7* an EventBridge schedule renews leaves before expiry and performs staged, 

8 overlap-safe root rollover. 

9 

10The root private key is stored only in the KMS-encrypted Secrets Manager secret 

11named by ``ROOT_SECRET_ARN``. Leaf private keys are generated in memory and sent 

12directly to the regional ACM ``ImportCertificate`` API; they are never written 

13to logs, SSM, or durable Lambda storage. SSM contains only public trust material 

14and ACM certificate ARNs. 

15""" 

16 

17from __future__ import annotations 

18 

19import json 

20import logging 

21import os 

22import re 

23from dataclasses import dataclass 

24from datetime import UTC, datetime, timedelta 

25from typing import Any 

26 

27import boto3 

28from botocore.exceptions import BotoCoreError, ClientError 

29from cryptography import x509 

30from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm 

31from cryptography.hazmat.primitives import hashes, serialization 

32from cryptography.hazmat.primitives.asymmetric import ec 

33from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID 

34 

35# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

36# Generated at (UTC): 2026-09-18T02:11:36Z 

37# Generated from Git commit: b8faa9689385cea16155a285a7f70cf6d488e512 

38# Flowchart(s) generated from this file: 

39# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/tls-certificate-manager/handler.lambda_handler.html`` 

40# (PNG: ``diagrams/code_diagrams/lambda/tls-certificate-manager/handler.lambda_handler.png``) 

41# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``. 

42# <pyflowchart-code-diagram> END 

43 

44 

45LOGGER = logging.getLogger(__name__) 

46LOGGER.setLevel(logging.INFO) 

47 

48_SCHEMA_VERSION = 1 

49_REGION_RE = re.compile(r"^[a-z]{2,4}(?:-[a-z0-9]+)+-[0-9]+$") 

50_MANAGED_BY_TAG_VALUE = "gco-backend-tls-manager" 

51_CERTIFICATE_STATUSES = ( 

52 "PENDING_VALIDATION", 

53 "ISSUED", 

54 "INACTIVE", 

55 "EXPIRED", 

56 "VALIDATION_TIMED_OUT", 

57 "REVOKED", 

58 "FAILED", 

59) 

60_DNS_RE = re.compile( 

61 r"(?=.{1,253}\Z)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+" 

62 r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", 

63 re.IGNORECASE, 

64) 

65 

66 

67@dataclass(frozen=True) 

68class ManagerConfig: 

69 """Validated certificate policy shared by custom-resource and scheduled calls.""" 

70 

71 regions: tuple[str, ...] 

72 server_name: str 

73 project_name: str 

74 registry_region: str 

75 root_ca_parameter_name: str 

76 certificate_parameter_prefix: str 

77 root_generation: int 

78 root_validity_days: int 

79 root_rotate_before_days: int 

80 root_activation_delay_hours: int 

81 root_overlap_days: int 

82 leaf_validity_days: int 

83 leaf_rotate_before_days: int 

84 

85 @classmethod 

86 def from_event(cls, event: dict[str, Any]) -> ManagerConfig: 

87 properties = event.get("ResourceProperties") or event 

88 raw_regions = properties.get("Regions") 

89 if raw_regions is None: 

90 raw_environment = os.environ.get("CERTIFICATE_REGIONS") 

91 if raw_environment is None: 

92 raise ValueError("Regions are not configured") 

93 raw_regions = json.loads(raw_environment) 

94 if not isinstance(raw_regions, list): 

95 raise ValueError("Regions must be a list") 

96 # An empty list is a control-plane-only deployment: the root CA is still 

97 # managed (so a later scale-up chains to the same root) and every leaf 

98 # certificate the previous configuration issued is retired. 

99 regions = tuple(dict.fromkeys(str(region).strip() for region in raw_regions)) 

100 if any(_REGION_RE.fullmatch(region) is None for region in regions): 

101 raise ValueError("Regions must contain only valid AWS workload region names") 

102 

103 config = cls( 

104 regions=regions, 

105 server_name=str( 

106 properties.get("ServerName") or os.environ.get("BACKEND_TLS_SERVER_NAME", "") 

107 ).strip(), 

108 project_name=str( 

109 properties.get("ProjectName") or os.environ.get("PROJECT_NAME", "") 

110 ).strip(), 

111 registry_region=str( 

112 properties.get("RegistryRegion") or os.environ.get("REGISTRY_REGION", "") 

113 ).strip(), 

114 root_ca_parameter_name=str( 

115 properties.get("RootCaParameterName") 

116 or os.environ.get("ROOT_CA_PARAMETER_NAME", "") 

117 ).strip(), 

118 certificate_parameter_prefix=str( 

119 properties.get("CertificateParameterPrefix") 

120 or os.environ.get("CERTIFICATE_PARAMETER_PREFIX", "") 

121 ).strip(), 

122 root_generation=_positive_int( 

123 properties.get("RootGeneration", os.environ.get("ROOT_GENERATION", "1")), 

124 "RootGeneration", 

125 ), 

126 root_validity_days=_positive_int( 

127 properties.get("RootValidityDays", os.environ.get("ROOT_VALIDITY_DAYS", "3650")), 

128 "RootValidityDays", 

129 ), 

130 root_rotate_before_days=_positive_int( 

131 properties.get( 

132 "RootRotateBeforeDays", 

133 os.environ.get("ROOT_ROTATE_BEFORE_DAYS", "180"), 

134 ), 

135 "RootRotateBeforeDays", 

136 ), 

137 root_activation_delay_hours=_positive_int( 

138 properties.get( 

139 "RootActivationDelayHours", 

140 os.environ.get("ROOT_ACTIVATION_DELAY_HOURS", "24"), 

141 ), 

142 "RootActivationDelayHours", 

143 ), 

144 root_overlap_days=_positive_int( 

145 properties.get("RootOverlapDays", os.environ.get("ROOT_OVERLAP_DAYS", "45")), 

146 "RootOverlapDays", 

147 ), 

148 leaf_validity_days=_positive_int( 

149 properties.get("LeafValidityDays", os.environ.get("LEAF_VALIDITY_DAYS", "30")), 

150 "LeafValidityDays", 

151 ), 

152 leaf_rotate_before_days=_positive_int( 

153 properties.get( 

154 "LeafRotateBeforeDays", 

155 os.environ.get("LEAF_ROTATE_BEFORE_DAYS", "10"), 

156 ), 

157 "LeafRotateBeforeDays", 

158 ), 

159 ) 

160 config.validate() 

161 return config 

162 

163 def validate(self) -> None: 

164 if _DNS_RE.fullmatch(self.server_name) is None: 

165 raise ValueError("ServerName must be a valid private DNS name") 

166 if _REGION_RE.fullmatch(self.registry_region) is None: 

167 raise ValueError("RegistryRegion must be a valid AWS region") 

168 if not self.project_name: 

169 raise ValueError("ProjectName is required") 

170 if not self.root_ca_parameter_name.startswith(f"/{self.project_name}/backend-tls/"): 

171 raise ValueError("RootCaParameterName must stay inside the project TLS namespace") 

172 if not self.certificate_parameter_prefix.startswith(f"/{self.project_name}/backend-tls/"): 

173 raise ValueError( 

174 "CertificateParameterPrefix must stay inside the project TLS namespace" 

175 ) 

176 if self.root_rotate_before_days >= self.root_validity_days: 

177 raise ValueError("RootRotateBeforeDays must be less than RootValidityDays") 

178 if self.leaf_rotate_before_days >= self.leaf_validity_days: 

179 raise ValueError("LeafRotateBeforeDays must be less than LeafValidityDays") 

180 if self.root_validity_days <= self.leaf_validity_days: 

181 raise ValueError("RootValidityDays must exceed LeafValidityDays") 

182 if self.root_overlap_days <= self.leaf_validity_days: 

183 raise ValueError("RootOverlapDays must exceed LeafValidityDays") 

184 

185 def certificate_parameter_name(self, region: str) -> str: 

186 return f"{self.certificate_parameter_prefix}{region}" 

187 

188 

189def _positive_int(value: Any, name: str) -> int: 

190 try: 

191 parsed = int(value) 

192 except (TypeError, ValueError) as exc: 

193 raise ValueError(f"{name} must be a positive integer") from exc 

194 if parsed <= 0: 

195 raise ValueError(f"{name} must be a positive integer") 

196 return parsed 

197 

198 

199def _now() -> datetime: 

200 return datetime.now(UTC) 

201 

202 

203def _iso(value: datetime) -> str: 

204 return value.astimezone(UTC).isoformat().replace("+00:00", "Z") 

205 

206 

207def _parse_iso(value: Any, field: str) -> datetime: 

208 if not isinstance(value, str): 

209 raise ValueError(f"Invalid root state: {field} is missing") 

210 try: 

211 parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) 

212 except ValueError as exc: 

213 raise ValueError(f"Invalid root state: {field} is not an ISO timestamp") from exc 

214 if parsed.tzinfo is None: 

215 parsed = parsed.replace(tzinfo=UTC) 

216 return parsed.astimezone(UTC) 

217 

218 

219def _certificate_not_after(certificate: x509.Certificate) -> datetime: 

220 value = getattr(certificate, "not_valid_after_utc", None) 

221 if isinstance(value, datetime): 

222 return value 

223 return certificate.not_valid_after.replace(tzinfo=UTC) 

224 

225 

226def _serialize_private_key(private_key: ec.EllipticCurvePrivateKey) -> str: 

227 return private_key.private_bytes( 

228 encoding=serialization.Encoding.PEM, 

229 format=serialization.PrivateFormat.PKCS8, 

230 encryption_algorithm=serialization.NoEncryption(), 

231 ).decode("ascii") 

232 

233 

234def _serialize_certificate(certificate: x509.Certificate) -> str: 

235 return certificate.public_bytes(serialization.Encoding.PEM).decode("ascii") 

236 

237 

238def _root_subject(project_name: str, generation: int) -> x509.Name: 

239 return x509.Name( 

240 [ 

241 x509.NameAttribute(NameOID.ORGANIZATION_NAME, "GCO deployment-local PKI"), 

242 x509.NameAttribute( 

243 NameOID.COMMON_NAME, 

244 f"{project_name} backend root generation {generation}", 

245 ), 

246 ] 

247 ) 

248 

249 

250def _generate_root(config: ManagerConfig, generation: int) -> dict[str, Any]: 

251 now = _now() 

252 private_key = ec.generate_private_key(ec.SECP256R1()) 

253 subject = _root_subject(config.project_name, generation) 

254 certificate = ( 

255 x509.CertificateBuilder() 

256 .subject_name(subject) 

257 .issuer_name(subject) 

258 .public_key(private_key.public_key()) 

259 .serial_number(x509.random_serial_number()) 

260 .not_valid_before(now - timedelta(minutes=5)) 

261 .not_valid_after(now + timedelta(days=config.root_validity_days)) 

262 .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) 

263 .add_extension( 

264 x509.KeyUsage( 

265 digital_signature=False, 

266 content_commitment=False, 

267 key_encipherment=False, 

268 data_encipherment=False, 

269 key_agreement=False, 

270 key_cert_sign=True, 

271 crl_sign=True, 

272 encipher_only=False, 

273 decipher_only=False, 

274 ), 

275 critical=True, 

276 ) 

277 .add_extension( 

278 x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()), 

279 critical=False, 

280 ) 

281 .sign(private_key, hashes.SHA256()) 

282 ) 

283 return { 

284 "generation": generation, 

285 "private_key_pem": _serialize_private_key(private_key), 

286 "certificate_pem": _serialize_certificate(certificate), 

287 "created_at": _iso(now), 

288 "not_after": _iso(_certificate_not_after(certificate)), 

289 } 

290 

291 

292def _validate_root_record(record: Any, field: str) -> tuple[Any, x509.Certificate]: 

293 if not isinstance(record, dict): 

294 raise ValueError(f"Invalid root state: {field} is missing") 

295 generation = record.get("generation") 

296 if type(generation) is not int or generation <= 0: 

297 raise ValueError(f"Invalid root state: {field}.generation") 

298 private_pem = record.get("private_key_pem") 

299 certificate_pem = record.get("certificate_pem") 

300 if not isinstance(private_pem, str) or not isinstance(certificate_pem, str): 

301 raise ValueError(f"Invalid root state: {field} key or certificate is missing") 

302 try: 

303 private_key = serialization.load_pem_private_key(private_pem.encode("ascii"), None) 

304 certificate = x509.load_pem_x509_certificate(certificate_pem.encode("ascii")) 

305 except (TypeError, ValueError) as exc: 

306 raise ValueError(f"Invalid root state: {field} contains malformed PEM") from exc 

307 if not isinstance(private_key, ec.EllipticCurvePrivateKey): 

308 raise ValueError(f"Invalid root state: {field} key must be ECDSA") 

309 key_public = private_key.public_key().public_bytes( 

310 serialization.Encoding.DER, 

311 serialization.PublicFormat.SubjectPublicKeyInfo, 

312 ) 

313 cert_public = certificate.public_key().public_bytes( 

314 serialization.Encoding.DER, 

315 serialization.PublicFormat.SubjectPublicKeyInfo, 

316 ) 

317 if key_public != cert_public: 

318 raise ValueError(f"Invalid root state: {field} key does not match its certificate") 

319 try: 

320 constraints = certificate.extensions.get_extension_for_class(x509.BasicConstraints).value 

321 except x509.ExtensionNotFound as exc: 

322 raise ValueError(f"Invalid root state: {field} is not a CA") from exc 

323 if not constraints.ca: 

324 raise ValueError(f"Invalid root state: {field} is not a CA") 

325 _parse_iso(record.get("not_after"), f"{field}.not_after") 

326 return private_key, certificate 

327 

328 

329def _load_root_state() -> dict[str, Any] | None: 

330 client = boto3.client("secretsmanager") 

331 try: 

332 response = client.get_secret_value(SecretId=os.environ["ROOT_SECRET_ARN"]) 

333 except ClientError as exc: 

334 if exc.response.get("Error", {}).get("Code") == "ResourceNotFoundException": 

335 return None 

336 raise 

337 secret_string = response.get("SecretString") 

338 if not secret_string: 

339 return None 

340 try: 

341 state = json.loads(secret_string) 

342 except json.JSONDecodeError as exc: 

343 raise ValueError("Root CA secret contains invalid JSON") from exc 

344 if isinstance(state, dict) and state.get("state") == "UNINITIALIZED": 

345 return None 

346 if not isinstance(state, dict) or state.get("schema_version") != _SCHEMA_VERSION: 

347 raise ValueError("Root CA secret has an unsupported schema") 

348 _validate_root_record(state.get("current"), "current") 

349 pending = state.get("pending") 

350 if pending is not None: 

351 _validate_root_record(pending, "pending") 

352 _parse_iso(pending.get("activate_after"), "pending.activate_after") 

353 published_at = pending.get("trust_bundle_published_at") 

354 if published_at is not None: 

355 _parse_iso(published_at, "pending.trust_bundle_published_at") 

356 previous = state.get("previous", []) 

357 if not isinstance(previous, list): 

358 raise ValueError("Invalid root state: previous must be a list") 

359 for index, item in enumerate(previous): 

360 if not isinstance(item, dict) or not isinstance(item.get("certificate_pem"), str): 

361 raise ValueError(f"Invalid root state: previous[{index}]") 

362 x509.load_pem_x509_certificate(item["certificate_pem"].encode("ascii")) 

363 _parse_iso(item.get("retire_after"), f"previous[{index}].retire_after") 

364 retired_regions = state.get("retired_regions", []) 

365 if ( 

366 not isinstance(retired_regions, list) 

367 or any( 

368 not isinstance(region, str) or _REGION_RE.fullmatch(region) is None 

369 for region in retired_regions 

370 ) 

371 or len(retired_regions) != len(set(retired_regions)) 

372 ): 

373 raise ValueError("Invalid root state: retired_regions") 

374 return state 

375 

376 

377def _save_root_state(state: dict[str, Any]) -> None: 

378 boto3.client("secretsmanager").put_secret_value( 

379 SecretId=os.environ["ROOT_SECRET_ARN"], 

380 SecretString=json.dumps(state, separators=(",", ":")), 

381 ) 

382 

383 

384def _publish_trust_bundle(config: ManagerConfig, state: dict[str, Any]) -> None: 

385 certificates = [state["current"]["certificate_pem"]] 

386 pending = state.get("pending") 

387 if pending is not None: 

388 certificates.append(pending["certificate_pem"]) 

389 certificates.extend(item["certificate_pem"] for item in state.get("previous", [])) 

390 bundle = "".join(cert.rstrip() + "\n" for cert in certificates) 

391 boto3.client("ssm", region_name=config.registry_region).put_parameter( 

392 Name=config.root_ca_parameter_name, 

393 Value=bundle, 

394 Type="String", 

395 Overwrite=True, 

396 Description="Public GCO backend TLS root trust bundle; contains no private key", 

397 ) 

398 

399 

400def _ensure_root(config: ManagerConfig) -> tuple[dict[str, Any], bool]: 

401 """Ensure root state, stage/promote rollover, and publish the public bundle.""" 

402 now = _now() 

403 state = _load_root_state() 

404 changed = False 

405 if state is None: 

406 state = { 

407 "schema_version": _SCHEMA_VERSION, 

408 "current": _generate_root(config, config.root_generation), 

409 "pending": None, 

410 "previous": [], 

411 "retired_regions": [], 

412 } 

413 changed = True 

414 elif "retired_regions" not in state: 

415 # Additive migration for root state written before regional retirement 

416 # tracking existed. The schema remains compatible with old secrets. 

417 state["retired_regions"] = [] 

418 changed = True 

419 

420 _, current_certificate = _validate_root_record(state["current"], "current") 

421 current_generation = int(state["current"]["generation"]) 

422 pending = state.get("pending") 

423 pending_generation = int(pending["generation"]) if pending is not None else 0 

424 current_expiry = _certificate_not_after(current_certificate) 

425 should_rotate_for_expiry = current_expiry <= now + timedelta( 

426 days=config.root_rotate_before_days 

427 ) 

428 desired_generation = max( 

429 config.root_generation, 

430 current_generation + 1 if should_rotate_for_expiry else current_generation, 

431 ) 

432 

433 if desired_generation > max(current_generation, pending_generation): 

434 pending = _generate_root(config, desired_generation) 

435 pending["activate_after"] = _iso(now + timedelta(hours=config.root_activation_delay_hours)) 

436 state["pending"] = pending 

437 changed = True 

438 LOGGER.info( 

439 "Staged root generation %d; activation begins after the trust propagation delay", 

440 desired_generation, 

441 ) 

442 

443 pending = state.get("pending") 

444 if ( 

445 pending is not None 

446 and pending.get("trust_bundle_published_at") is not None 

447 and _parse_iso(pending["activate_after"], "pending.activate_after") <= now 

448 ): 

449 previous = list(state.get("previous", [])) 

450 previous.insert( 

451 0, 

452 { 

453 "generation": state["current"]["generation"], 

454 "certificate_pem": state["current"]["certificate_pem"], 

455 "retire_after": _iso(now + timedelta(days=config.root_overlap_days)), 

456 }, 

457 ) 

458 promoted = dict(pending) 

459 promoted.pop("activate_after", None) 

460 promoted.pop("trust_bundle_published_at", None) 

461 state["current"] = promoted 

462 state["pending"] = None 

463 state["previous"] = previous 

464 changed = True 

465 LOGGER.info("Promoted root generation %d", promoted["generation"]) 

466 

467 active_previous = [ 

468 item 

469 for item in state.get("previous", []) 

470 if _parse_iso(item["retire_after"], "previous.retire_after") > now 

471 ] 

472 if len(active_previous) != len(state.get("previous", [])): 

473 state["previous"] = active_previous 

474 changed = True 

475 LOGGER.info("Removed expired previous root certificates from the trust bundle") 

476 

477 if changed: 

478 _save_root_state(state) 

479 _publish_trust_bundle(config, state) 

480 

481 # Activation delay starts only after SSM accepted a bundle containing the 

482 # pending root. A prolonged SSM outage therefore cannot consume the safety 

483 # window and promote a root that warm proxy caches never had a chance to 

484 # observe. Missing markers on legacy pending records are repaired safely. 

485 pending = state.get("pending") 

486 if pending is not None and pending.get("trust_bundle_published_at") is None: 

487 published_at = _now() 

488 pending["trust_bundle_published_at"] = _iso(published_at) 

489 pending["activate_after"] = _iso( 

490 published_at + timedelta(hours=config.root_activation_delay_hours) 

491 ) 

492 _save_root_state(state) 

493 changed = True 

494 LOGGER.info( 

495 "Confirmed trust publication for pending root generation %d; activation begins %s", 

496 pending["generation"], 

497 pending["activate_after"], 

498 ) 

499 return state, changed 

500 

501 

502def _generate_leaf( 

503 config: ManagerConfig, 

504 root_record: dict[str, Any], 

505) -> tuple[bytes, bytes, datetime]: 

506 root_key, root_certificate = _validate_root_record(root_record, "current") 

507 now = _now() 

508 root_expiry = _certificate_not_after(root_certificate) 

509 requested_expiry = now + timedelta(days=config.leaf_validity_days) 

510 leaf_expiry = min(requested_expiry, root_expiry - timedelta(days=1)) 

511 if leaf_expiry <= now + timedelta(days=config.leaf_rotate_before_days): 

512 raise RuntimeError("Current root expires too soon to issue a safe leaf certificate") 

513 

514 leaf_key = ec.generate_private_key(ec.SECP256R1()) 

515 subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, config.server_name)]) 

516 certificate = ( 

517 x509.CertificateBuilder() 

518 .subject_name(subject) 

519 .issuer_name(root_certificate.subject) 

520 .public_key(leaf_key.public_key()) 

521 .serial_number(x509.random_serial_number()) 

522 .not_valid_before(now - timedelta(minutes=5)) 

523 .not_valid_after(leaf_expiry) 

524 .add_extension( 

525 x509.SubjectAlternativeName([x509.DNSName(config.server_name)]), 

526 critical=False, 

527 ) 

528 .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) 

529 .add_extension( 

530 x509.KeyUsage( 

531 digital_signature=True, 

532 content_commitment=False, 

533 key_encipherment=False, 

534 data_encipherment=False, 

535 key_agreement=False, 

536 key_cert_sign=False, 

537 crl_sign=False, 

538 encipher_only=False, 

539 decipher_only=False, 

540 ), 

541 critical=True, 

542 ) 

543 .add_extension( 

544 x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), 

545 critical=False, 

546 ) 

547 .add_extension( 

548 x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()), 

549 critical=False, 

550 ) 

551 .add_extension( 

552 x509.AuthorityKeyIdentifier.from_issuer_public_key(root_key.public_key()), 

553 critical=False, 

554 ) 

555 .sign(root_key, hashes.SHA256()) 

556 ) 

557 private_key_pem = leaf_key.private_bytes( 

558 encoding=serialization.Encoding.PEM, 

559 format=serialization.PrivateFormat.PKCS8, 

560 encryption_algorithm=serialization.NoEncryption(), 

561 ) 

562 certificate_pem = certificate.public_bytes(serialization.Encoding.PEM) 

563 return certificate_pem, private_key_pem, leaf_expiry 

564 

565 

566def _validated_certificate_arn(region: str, value: Any) -> str: 

567 """Validate one project-registry value before it authorizes mutation.""" 

568 if not isinstance(value, str): 

569 raise ValueError(f"Invalid ACM certificate ARN stored for {region}") 

570 certificate_arn = value.strip() 

571 expected_prefix = ( 

572 f"arn:{os.environ.get('AWS_PARTITION', 'aws')}:acm:{region}:" 

573 f"{os.environ.get('AWS_ACCOUNT_ID', '')}:certificate/" 

574 ) 

575 certificate_id = certificate_arn.removeprefix(expected_prefix) 

576 if ( 

577 not certificate_arn.startswith(expected_prefix) 

578 or re.fullmatch(r"[A-Za-z0-9-]+", certificate_id) is None 

579 ): 

580 raise ValueError(f"Invalid ACM certificate ARN stored for {region}") 

581 return certificate_arn 

582 

583 

584def _write_certificate_registry( 

585 config: ManagerConfig, 

586 region: str, 

587 certificate_arn: str, 

588) -> None: 

589 """Persist the canonical regional certificate association in SSM.""" 

590 boto3.client("ssm", region_name=config.registry_region).put_parameter( 

591 Name=config.certificate_parameter_name(region), 

592 Value=certificate_arn, 

593 Type="String", 

594 Overwrite=True, 

595 Description=f"Regional ACM certificate ARN for GCO backend TLS in {region}", 

596 ) 

597 

598 

599def _certificate_tags(acm_client: Any, certificate_arn: str) -> dict[str, str]: 

600 """Return a strict tag map for one ACM certificate.""" 

601 response = acm_client.list_tags_for_certificate(CertificateArn=certificate_arn) 

602 raw_tags = response.get("Tags", []) 

603 if not isinstance(raw_tags, list): 

604 raise ValueError("ACM returned malformed certificate tags") 

605 

606 tags: dict[str, str] = {} 

607 for raw_tag in raw_tags: 

608 if not isinstance(raw_tag, dict): 

609 raise ValueError("ACM returned a malformed certificate tag") 

610 key = raw_tag.get("Key") 

611 value = raw_tag.get("Value") 

612 if not isinstance(key, str) or not isinstance(value, str) or key in tags: 

613 raise ValueError("ACM returned a malformed certificate tag") 

614 tags[key] = value 

615 return tags 

616 

617 

618def _certificate_ownership_status(config: ManagerConfig, tags: dict[str, str]) -> str: 

619 """Classify ownership tags as ``owned``, ``legacy``, or ``conflicting``.""" 

620 expected = { 

621 "Project": config.project_name, 

622 "ManagedBy": _MANAGED_BY_TAG_VALUE, 

623 } 

624 if any(key in tags and tags[key] != value for key, value in expected.items()): 

625 return "conflicting" 

626 if all(tags.get(key) == value for key, value in expected.items()): 

627 return "owned" 

628 return "legacy" 

629 

630 

631def _require_certificate_ownership( 

632 config: ManagerConfig, 

633 region: str, 

634 certificate_arn: str, 

635 acm_client: Any, 

636) -> None: 

637 """Fail closed unless an ACM certificate has both GCO ownership tags.""" 

638 status = _certificate_ownership_status( 

639 config, 

640 _certificate_tags(acm_client, certificate_arn), 

641 ) 

642 if status != "owned": 

643 raise PermissionError( 

644 f"Registered ACM certificate for {region} has {status} ownership tags" 

645 ) 

646 

647 

648def _managed_root_certificates(state: dict[str, Any]) -> tuple[x509.Certificate, ...]: 

649 """Load every current, pending, and previous root authorized by state.""" 

650 roots: list[x509.Certificate] = [] 

651 _, current = _validate_root_record(state.get("current"), "current") 

652 roots.append(current) 

653 

654 pending = state.get("pending") 

655 if pending is not None: 

656 _, pending_certificate = _validate_root_record(pending, "pending") 

657 roots.append(pending_certificate) 

658 

659 for index, previous in enumerate(state.get("previous", [])): 

660 if not isinstance(previous, dict) or not isinstance(previous.get("certificate_pem"), str): 

661 raise ValueError(f"Invalid root state: previous[{index}]") 

662 try: 

663 roots.append( 

664 x509.load_pem_x509_certificate(previous["certificate_pem"].encode("ascii")) 

665 ) 

666 except (TypeError, ValueError) as exc: 

667 raise ValueError(f"Invalid root state: previous[{index}] certificate") from exc 

668 return tuple(roots) 

669 

670 

671def _certificate_has_server_name( 

672 certificate: x509.Certificate, 

673 server_name: str, 

674) -> bool: 

675 try: 

676 names = certificate.extensions.get_extension_for_class( 

677 x509.SubjectAlternativeName 

678 ).value.get_values_for_type(x509.DNSName) 

679 except x509.ExtensionNotFound: 

680 return False 

681 expected = server_name.casefold() 

682 return any(name.casefold() == expected for name in names) 

683 

684 

685def _certificate_signed_by_root( 

686 certificate: x509.Certificate, 

687 root_certificate: x509.Certificate, 

688) -> bool: 

689 if certificate.issuer != root_certificate.subject: 

690 return False 

691 root_public_key = root_certificate.public_key() 

692 signature_algorithm = certificate.signature_hash_algorithm 

693 if not isinstance(root_public_key, ec.EllipticCurvePublicKey) or signature_algorithm is None: 

694 return False 

695 try: 

696 root_public_key.verify( 

697 certificate.signature, 

698 certificate.tbs_certificate_bytes, 

699 ec.ECDSA(signature_algorithm), 

700 ) 

701 except InvalidSignature, UnsupportedAlgorithm, TypeError, ValueError: 

702 return False 

703 return True 

704 

705 

706def _recover_unregistered_certificate( 

707 config: ManagerConfig, 

708 region: str, 

709) -> tuple[str | None, x509.Certificate | None]: 

710 """Adopt the unique tagged ACM leaf left by a failed SSM registration. 

711 

712 SSM remains the canonical registry. Tagged ACM inventory is the durable 

713 recovery channel for the narrow case where both that write and the 

714 compensating delete fail. Ambiguous inventory fails closed rather than 

715 importing another certificate. 

716 """ 

717 acm_client = boto3.client("acm", region_name=region) 

718 managed: list[tuple[str, x509.Certificate]] = [] 

719 paginator = acm_client.get_paginator("list_certificates") 

720 for page in paginator.paginate( 

721 CertificateStatuses=list(_CERTIFICATE_STATUSES), 

722 Includes={"keyTypes": ["EC_prime256v1"]}, 

723 ): 

724 for summary in page.get("CertificateSummaryList", []): 

725 if summary.get("Type") not in (None, "IMPORTED"): 

726 continue 

727 certificate_arn = _validated_certificate_arn( 

728 region, 

729 summary.get("CertificateArn"), 

730 ) 

731 tags = _certificate_tags(acm_client, certificate_arn) 

732 if _certificate_ownership_status(config, tags) != "owned": 

733 continue 

734 detail = acm_client.describe_certificate(CertificateArn=certificate_arn) 

735 if detail.get("Certificate", {}).get("Type") != "IMPORTED": 

736 raise RuntimeError( 

737 f"Managed ACM certificate inventory for {region} contains a non-imported leaf"