Coverage for lambda / tls-certificate-manager / handler.py: 100.00%
566 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Manage GCO's deployment-local root CA and regional ACM leaf certificates.
3The function has two entry points:
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.
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"""
17from __future__ import annotations
19import json
20import logging
21import os
22import re
23from dataclasses import dataclass
24from datetime import UTC, datetime, timedelta
25from typing import Any
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
35# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
36# Generated at (UTC): 2026-09-01T14:42:56Z
37# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
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
45LOGGER = logging.getLogger(__name__)
46LOGGER.setLevel(logging.INFO)
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)
67@dataclass(frozen=True)
68class ManagerConfig:
69 """Validated certificate policy shared by custom-resource and scheduled calls."""
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
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_regions = json.loads(os.environ.get("CERTIFICATE_REGIONS", "[]"))
91 if not isinstance(raw_regions, list):
92 raise ValueError("Regions must be a list")
93 regions = tuple(dict.fromkeys(str(region).strip() for region in raw_regions))
94 if not regions or any(_REGION_RE.fullmatch(region) is None for region in regions):
95 raise ValueError("At least one valid AWS workload region is required")
97 config = cls(
98 regions=regions,
99 server_name=str(
100 properties.get("ServerName") or os.environ.get("BACKEND_TLS_SERVER_NAME", "")
101 ).strip(),
102 project_name=str(
103 properties.get("ProjectName") or os.environ.get("PROJECT_NAME", "")
104 ).strip(),
105 registry_region=str(
106 properties.get("RegistryRegion") or os.environ.get("REGISTRY_REGION", "")
107 ).strip(),
108 root_ca_parameter_name=str(
109 properties.get("RootCaParameterName")
110 or os.environ.get("ROOT_CA_PARAMETER_NAME", "")
111 ).strip(),
112 certificate_parameter_prefix=str(
113 properties.get("CertificateParameterPrefix")
114 or os.environ.get("CERTIFICATE_PARAMETER_PREFIX", "")
115 ).strip(),
116 root_generation=_positive_int(
117 properties.get("RootGeneration", os.environ.get("ROOT_GENERATION", "1")),
118 "RootGeneration",
119 ),
120 root_validity_days=_positive_int(
121 properties.get("RootValidityDays", os.environ.get("ROOT_VALIDITY_DAYS", "3650")),
122 "RootValidityDays",
123 ),
124 root_rotate_before_days=_positive_int(
125 properties.get(
126 "RootRotateBeforeDays",
127 os.environ.get("ROOT_ROTATE_BEFORE_DAYS", "180"),
128 ),
129 "RootRotateBeforeDays",
130 ),
131 root_activation_delay_hours=_positive_int(
132 properties.get(
133 "RootActivationDelayHours",
134 os.environ.get("ROOT_ACTIVATION_DELAY_HOURS", "24"),
135 ),
136 "RootActivationDelayHours",
137 ),
138 root_overlap_days=_positive_int(
139 properties.get("RootOverlapDays", os.environ.get("ROOT_OVERLAP_DAYS", "45")),
140 "RootOverlapDays",
141 ),
142 leaf_validity_days=_positive_int(
143 properties.get("LeafValidityDays", os.environ.get("LEAF_VALIDITY_DAYS", "30")),
144 "LeafValidityDays",
145 ),
146 leaf_rotate_before_days=_positive_int(
147 properties.get(
148 "LeafRotateBeforeDays",
149 os.environ.get("LEAF_ROTATE_BEFORE_DAYS", "10"),
150 ),
151 "LeafRotateBeforeDays",
152 ),
153 )
154 config.validate()
155 return config
157 def validate(self) -> None:
158 if _DNS_RE.fullmatch(self.server_name) is None:
159 raise ValueError("ServerName must be a valid private DNS name")
160 if _REGION_RE.fullmatch(self.registry_region) is None:
161 raise ValueError("RegistryRegion must be a valid AWS region")
162 if not self.project_name:
163 raise ValueError("ProjectName is required")
164 if not self.root_ca_parameter_name.startswith(f"/{self.project_name}/backend-tls/"):
165 raise ValueError("RootCaParameterName must stay inside the project TLS namespace")
166 if not self.certificate_parameter_prefix.startswith(f"/{self.project_name}/backend-tls/"):
167 raise ValueError(
168 "CertificateParameterPrefix must stay inside the project TLS namespace"
169 )
170 if self.root_rotate_before_days >= self.root_validity_days:
171 raise ValueError("RootRotateBeforeDays must be less than RootValidityDays")
172 if self.leaf_rotate_before_days >= self.leaf_validity_days:
173 raise ValueError("LeafRotateBeforeDays must be less than LeafValidityDays")
174 if self.root_validity_days <= self.leaf_validity_days:
175 raise ValueError("RootValidityDays must exceed LeafValidityDays")
176 if self.root_overlap_days <= self.leaf_validity_days:
177 raise ValueError("RootOverlapDays must exceed LeafValidityDays")
179 def certificate_parameter_name(self, region: str) -> str:
180 return f"{self.certificate_parameter_prefix}{region}"
183def _positive_int(value: Any, name: str) -> int:
184 try:
185 parsed = int(value)
186 except (TypeError, ValueError) as exc:
187 raise ValueError(f"{name} must be a positive integer") from exc
188 if parsed <= 0:
189 raise ValueError(f"{name} must be a positive integer")
190 return parsed
193def _now() -> datetime:
194 return datetime.now(UTC)
197def _iso(value: datetime) -> str:
198 return value.astimezone(UTC).isoformat().replace("+00:00", "Z")
201def _parse_iso(value: Any, field: str) -> datetime:
202 if not isinstance(value, str):
203 raise ValueError(f"Invalid root state: {field} is missing")
204 try:
205 parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
206 except ValueError as exc:
207 raise ValueError(f"Invalid root state: {field} is not an ISO timestamp") from exc
208 if parsed.tzinfo is None:
209 parsed = parsed.replace(tzinfo=UTC)
210 return parsed.astimezone(UTC)
213def _certificate_not_after(certificate: x509.Certificate) -> datetime:
214 value = getattr(certificate, "not_valid_after_utc", None)
215 if isinstance(value, datetime):
216 return value
217 return certificate.not_valid_after.replace(tzinfo=UTC)
220def _serialize_private_key(private_key: ec.EllipticCurvePrivateKey) -> str:
221 return private_key.private_bytes(
222 encoding=serialization.Encoding.PEM,
223 format=serialization.PrivateFormat.PKCS8,
224 encryption_algorithm=serialization.NoEncryption(),
225 ).decode("ascii")
228def _serialize_certificate(certificate: x509.Certificate) -> str:
229 return certificate.public_bytes(serialization.Encoding.PEM).decode("ascii")
232def _root_subject(project_name: str, generation: int) -> x509.Name:
233 return x509.Name(
234 [
235 x509.NameAttribute(NameOID.ORGANIZATION_NAME, "GCO deployment-local PKI"),
236 x509.NameAttribute(
237 NameOID.COMMON_NAME,
238 f"{project_name} backend root generation {generation}",
239 ),
240 ]
241 )
244def _generate_root(config: ManagerConfig, generation: int) -> dict[str, Any]:
245 now = _now()
246 private_key = ec.generate_private_key(ec.SECP256R1())
247 subject = _root_subject(config.project_name, generation)
248 certificate = (
249 x509.CertificateBuilder()
250 .subject_name(subject)
251 .issuer_name(subject)
252 .public_key(private_key.public_key())
253 .serial_number(x509.random_serial_number())
254 .not_valid_before(now - timedelta(minutes=5))
255 .not_valid_after(now + timedelta(days=config.root_validity_days))
256 .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
257 .add_extension(
258 x509.KeyUsage(
259 digital_signature=False,
260 content_commitment=False,
261 key_encipherment=False,
262 data_encipherment=False,
263 key_agreement=False,
264 key_cert_sign=True,
265 crl_sign=True,
266 encipher_only=False,
267 decipher_only=False,
268 ),
269 critical=True,
270 )
271 .add_extension(
272 x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()),
273 critical=False,
274 )
275 .sign(private_key, hashes.SHA256())
276 )
277 return {
278 "generation": generation,
279 "private_key_pem": _serialize_private_key(private_key),
280 "certificate_pem": _serialize_certificate(certificate),
281 "created_at": _iso(now),
282 "not_after": _iso(_certificate_not_after(certificate)),
283 }
286def _validate_root_record(record: Any, field: str) -> tuple[Any, x509.Certificate]:
287 if not isinstance(record, dict):
288 raise ValueError(f"Invalid root state: {field} is missing")
289 generation = record.get("generation")
290 if type(generation) is not int or generation <= 0:
291 raise ValueError(f"Invalid root state: {field}.generation")
292 private_pem = record.get("private_key_pem")
293 certificate_pem = record.get("certificate_pem")
294 if not isinstance(private_pem, str) or not isinstance(certificate_pem, str):
295 raise ValueError(f"Invalid root state: {field} key or certificate is missing")
296 try:
297 private_key = serialization.load_pem_private_key(private_pem.encode("ascii"), None)
298 certificate = x509.load_pem_x509_certificate(certificate_pem.encode("ascii"))
299 except (TypeError, ValueError) as exc:
300 raise ValueError(f"Invalid root state: {field} contains malformed PEM") from exc
301 if not isinstance(private_key, ec.EllipticCurvePrivateKey):
302 raise ValueError(f"Invalid root state: {field} key must be ECDSA")
303 key_public = private_key.public_key().public_bytes(
304 serialization.Encoding.DER,
305 serialization.PublicFormat.SubjectPublicKeyInfo,
306 )
307 cert_public = certificate.public_key().public_bytes(
308 serialization.Encoding.DER,
309 serialization.PublicFormat.SubjectPublicKeyInfo,
310 )
311 if key_public != cert_public:
312 raise ValueError(f"Invalid root state: {field} key does not match its certificate")
313 try:
314 constraints = certificate.extensions.get_extension_for_class(x509.BasicConstraints).value
315 except x509.ExtensionNotFound as exc:
316 raise ValueError(f"Invalid root state: {field} is not a CA") from exc
317 if not constraints.ca:
318 raise ValueError(f"Invalid root state: {field} is not a CA")
319 _parse_iso(record.get("not_after"), f"{field}.not_after")
320 return private_key, certificate
323def _load_root_state() -> dict[str, Any] | None:
324 client = boto3.client("secretsmanager")
325 try:
326 response = client.get_secret_value(SecretId=os.environ["ROOT_SECRET_ARN"])
327 except ClientError as exc:
328 if exc.response.get("Error", {}).get("Code") == "ResourceNotFoundException":
329 return None
330 raise
331 secret_string = response.get("SecretString")
332 if not secret_string:
333 return None
334 try:
335 state = json.loads(secret_string)
336 except json.JSONDecodeError as exc:
337 raise ValueError("Root CA secret contains invalid JSON") from exc
338 if isinstance(state, dict) and state.get("state") == "UNINITIALIZED":
339 return None
340 if not isinstance(state, dict) or state.get("schema_version") != _SCHEMA_VERSION:
341 raise ValueError("Root CA secret has an unsupported schema")
342 _validate_root_record(state.get("current"), "current")
343 pending = state.get("pending")
344 if pending is not None:
345 _validate_root_record(pending, "pending")
346 _parse_iso(pending.get("activate_after"), "pending.activate_after")
347 published_at = pending.get("trust_bundle_published_at")
348 if published_at is not None:
349 _parse_iso(published_at, "pending.trust_bundle_published_at")
350 previous = state.get("previous", [])
351 if not isinstance(previous, list):
352 raise ValueError("Invalid root state: previous must be a list")
353 for index, item in enumerate(previous):
354 if not isinstance(item, dict) or not isinstance(item.get("certificate_pem"), str):
355 raise ValueError(f"Invalid root state: previous[{index}]")
356 x509.load_pem_x509_certificate(item["certificate_pem"].encode("ascii"))
357 _parse_iso(item.get("retire_after"), f"previous[{index}].retire_after")
358 retired_regions = state.get("retired_regions", [])
359 if (
360 not isinstance(retired_regions, list)
361 or any(
362 not isinstance(region, str) or _REGION_RE.fullmatch(region) is None
363 for region in retired_regions
364 )
365 or len(retired_regions) != len(set(retired_regions))
366 ):
367 raise ValueError("Invalid root state: retired_regions")
368 return state
371def _save_root_state(state: dict[str, Any]) -> None:
372 boto3.client("secretsmanager").put_secret_value(
373 SecretId=os.environ["ROOT_SECRET_ARN"],
374 SecretString=json.dumps(state, separators=(",", ":")),
375 )
378def _publish_trust_bundle(config: ManagerConfig, state: dict[str, Any]) -> None:
379 certificates = [state["current"]["certificate_pem"]]
380 pending = state.get("pending")
381 if pending is not None:
382 certificates.append(pending["certificate_pem"])
383 certificates.extend(item["certificate_pem"] for item in state.get("previous", []))
384 bundle = "".join(cert.rstrip() + "\n" for cert in certificates)
385 boto3.client("ssm", region_name=config.registry_region).put_parameter(
386 Name=config.root_ca_parameter_name,
387 Value=bundle,
388 Type="String",
389 Overwrite=True,
390 Description="Public GCO backend TLS root trust bundle; contains no private key",
391 )
394def _ensure_root(config: ManagerConfig) -> tuple[dict[str, Any], bool]:
395 """Ensure root state, stage/promote rollover, and publish the public bundle."""
396 now = _now()
397 state = _load_root_state()
398 changed = False
399 if state is None:
400 state = {
401 "schema_version": _SCHEMA_VERSION,
402 "current": _generate_root(config, config.root_generation),
403 "pending": None,
404 "previous": [],
405 "retired_regions": [],
406 }
407 changed = True
408 elif "retired_regions" not in state:
409 # Additive migration for root state written before regional retirement
410 # tracking existed. The schema remains compatible with old secrets.
411 state["retired_regions"] = []
412 changed = True
414 _, current_certificate = _validate_root_record(state["current"], "current")
415 current_generation = int(state["current"]["generation"])
416 pending = state.get("pending")
417 pending_generation = int(pending["generation"]) if pending is not None else 0
418 current_expiry = _certificate_not_after(current_certificate)
419 should_rotate_for_expiry = current_expiry <= now + timedelta(
420 days=config.root_rotate_before_days
421 )
422 desired_generation = max(
423 config.root_generation,
424 current_generation + 1 if should_rotate_for_expiry else current_generation,
425 )
427 if desired_generation > max(current_generation, pending_generation):
428 pending = _generate_root(config, desired_generation)
429 pending["activate_after"] = _iso(now + timedelta(hours=config.root_activation_delay_hours))
430 state["pending"] = pending
431 changed = True
432 LOGGER.info(
433 "Staged root generation %d; activation begins after the trust propagation delay",
434 desired_generation,
435 )
437 pending = state.get("pending")
438 if (
439 pending is not None
440 and pending.get("trust_bundle_published_at") is not None
441 and _parse_iso(pending["activate_after"], "pending.activate_after") <= now
442 ):
443 previous = list(state.get("previous", []))
444 previous.insert(
445 0,
446 {
447 "generation": state["current"]["generation"],
448 "certificate_pem": state["current"]["certificate_pem"],
449 "retire_after": _iso(now + timedelta(days=config.root_overlap_days)),
450 },
451 )
452 promoted = dict(pending)
453 promoted.pop("activate_after", None)
454 promoted.pop("trust_bundle_published_at", None)
455 state["current"] = promoted
456 state["pending"] = None
457 state["previous"] = previous
458 changed = True
459 LOGGER.info("Promoted root generation %d", promoted["generation"])
461 active_previous = [
462 item
463 for item in state.get("previous", [])
464 if _parse_iso(item["retire_after"], "previous.retire_after") > now
465 ]
466 if len(active_previous) != len(state.get("previous", [])):
467 state["previous"] = active_previous
468 changed = True
469 LOGGER.info("Removed expired previous root certificates from the trust bundle")
471 if changed:
472 _save_root_state(state)
473 _publish_trust_bundle(config, state)
475 # Activation delay starts only after SSM accepted a bundle containing the
476 # pending root. A prolonged SSM outage therefore cannot consume the safety
477 # window and promote a root that warm proxy caches never had a chance to
478 # observe. Missing markers on legacy pending records are repaired safely.
479 pending = state.get("pending")
480 if pending is not None and pending.get("trust_bundle_published_at") is None:
481 published_at = _now()
482 pending["trust_bundle_published_at"] = _iso(published_at)
483 pending["activate_after"] = _iso(
484 published_at + timedelta(hours=config.root_activation_delay_hours)
485 )
486 _save_root_state(state)
487 changed = True
488 LOGGER.info(
489 "Confirmed trust publication for pending root generation %d; activation begins %s",
490 pending["generation"],
491 pending["activate_after"],
492 )
493 return state, changed
496def _generate_leaf(
497 config: ManagerConfig,
498 root_record: dict[str, Any],
499) -> tuple[bytes, bytes, datetime]:
500 root_key, root_certificate = _validate_root_record(root_record, "current")
501 now = _now()
502 root_expiry = _certificate_not_after(root_certificate)
503 requested_expiry = now + timedelta(days=config.leaf_validity_days)
504 leaf_expiry = min(requested_expiry, root_expiry - timedelta(days=1))
505 if leaf_expiry <= now + timedelta(days=config.leaf_rotate_before_days):
506 raise RuntimeError("Current root expires too soon to issue a safe leaf certificate")
508 leaf_key = ec.generate_private_key(ec.SECP256R1())
509 subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, config.server_name)])
510 certificate = (
511 x509.CertificateBuilder()
512 .subject_name(subject)
513 .issuer_name(root_certificate.subject)
514 .public_key(leaf_key.public_key())
515 .serial_number(x509.random_serial_number())
516 .not_valid_before(now - timedelta(minutes=5))
517 .not_valid_after(leaf_expiry)
518 .add_extension(
519 x509.SubjectAlternativeName([x509.DNSName(config.server_name)]),
520 critical=False,
521 )
522 .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
523 .add_extension(
524 x509.KeyUsage(
525 digital_signature=True,
526 content_commitment=False,
527 key_encipherment=False,
528 data_encipherment=False,
529 key_agreement=False,
530 key_cert_sign=False,
531 crl_sign=False,
532 encipher_only=False,
533 decipher_only=False,
534 ),
535 critical=True,
536 )
537 .add_extension(
538 x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]),
539 critical=False,
540 )
541 .add_extension(
542 x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()),
543 critical=False,
544 )
545 .add_extension(
546 x509.AuthorityKeyIdentifier.from_issuer_public_key(root_key.public_key()),
547 critical=False,
548 )
549 .sign(root_key, hashes.SHA256())
550 )
551 private_key_pem = leaf_key.private_bytes(
552 encoding=serialization.Encoding.PEM,
553 format=serialization.PrivateFormat.PKCS8,
554 encryption_algorithm=serialization.NoEncryption(),
555 )
556 certificate_pem = certificate.public_bytes(serialization.Encoding.PEM)
557 return certificate_pem, private_key_pem, leaf_expiry
560def _validated_certificate_arn(region: str, value: Any) -> str:
561 """Validate one project-registry value before it authorizes mutation."""
562 if not isinstance(value, str):
563 raise ValueError(f"Invalid ACM certificate ARN stored for {region}")
564 certificate_arn = value.strip()
565 expected_prefix = (
566 f"arn:{os.environ.get('AWS_PARTITION', 'aws')}:acm:{region}:"
567 f"{os.environ.get('AWS_ACCOUNT_ID', '')}:certificate/"
568 )
569 certificate_id = certificate_arn.removeprefix(expected_prefix)
570 if (
571 not certificate_arn.startswith(expected_prefix)
572 or re.fullmatch(r"[A-Za-z0-9-]+", certificate_id) is None
573 ):
574 raise ValueError(f"Invalid ACM certificate ARN stored for {region}")
575 return certificate_arn
578def _write_certificate_registry(
579 config: ManagerConfig,
580 region: str,
581 certificate_arn: str,
582) -> None:
583 """Persist the canonical regional certificate association in SSM."""
584 boto3.client("ssm", region_name=config.registry_region).put_parameter(
585 Name=config.certificate_parameter_name(region),
586 Value=certificate_arn,
587 Type="String",
588 Overwrite=True,
589 Description=f"Regional ACM certificate ARN for GCO backend TLS in {region}",
590 )
593def _certificate_tags(acm_client: Any, certificate_arn: str) -> dict[str, str]:
594 """Return a strict tag map for one ACM certificate."""
595 response = acm_client.list_tags_for_certificate(CertificateArn=certificate_arn)
596 raw_tags = response.get("Tags", [])
597 if not isinstance(raw_tags, list):
598 raise ValueError("ACM returned malformed certificate tags")
600 tags: dict[str, str] = {}
601 for raw_tag in raw_tags:
602 if not isinstance(raw_tag, dict):
603 raise ValueError("ACM returned a malformed certificate tag")
604 key = raw_tag.get("Key")
605 value = raw_tag.get("Value")
606 if not isinstance(key, str) or not isinstance(value, str) or key in tags:
607 raise ValueError("ACM returned a malformed certificate tag")
608 tags[key] = value
609 return tags
612def _certificate_ownership_status(config: ManagerConfig, tags: dict[str, str]) -> str:
613 """Classify ownership tags as ``owned``, ``legacy``, or ``conflicting``."""
614 expected = {
615 "Project": config.project_name,
616 "ManagedBy": _MANAGED_BY_TAG_VALUE,
617 }
618 if any(key in tags and tags[key] != value for key, value in expected.items()):
619 return "conflicting"
620 if all(tags.get(key) == value for key, value in expected.items()):
621 return "owned"
622 return "legacy"
625def _require_certificate_ownership(
626 config: ManagerConfig,
627 region: str,
628 certificate_arn: str,
629 acm_client: Any,
630) -> None:
631 """Fail closed unless an ACM certificate has both GCO ownership tags."""
632 status = _certificate_ownership_status(
633 config,
634 _certificate_tags(acm_client, certificate_arn),
635 )
636 if status != "owned":
637 raise PermissionError(
638 f"Registered ACM certificate for {region} has {status} ownership tags"
639 )
642def _managed_root_certificates(state: dict[str, Any]) -> tuple[x509.Certificate, ...]:
643 """Load every current, pending, and previous root authorized by state."""
644 roots: list[x509.Certificate] = []
645 _, current = _validate_root_record(state.get("current"), "current")
646 roots.append(current)
648 pending = state.get("pending")
649 if pending is not None:
650 _, pending_certificate = _validate_root_record(pending, "pending")
651 roots.append(pending_certificate)
653 for index, previous in enumerate(state.get("previous", [])):
654 if not isinstance(previous, dict) or not isinstance(previous.get("certificate_pem"), str):
655 raise ValueError(f"Invalid root state: previous[{index}]")
656 try:
657 roots.append(
658 x509.load_pem_x509_certificate(previous["certificate_pem"].encode("ascii"))
659 )
660 except (TypeError, ValueError) as exc:
661 raise ValueError(f"Invalid root state: previous[{index}] certificate") from exc
662 return tuple(roots)
665def _certificate_has_server_name(
666 certificate: x509.Certificate,
667 server_name: str,
668) -> bool:
669 try:
670 names = certificate.extensions.get_extension_for_class(
671 x509.SubjectAlternativeName
672 ).value.get_values_for_type(x509.DNSName)
673 except x509.ExtensionNotFound:
674 return False
675 expected = server_name.casefold()
676 return any(name.casefold() == expected for name in names)
679def _certificate_signed_by_root(
680 certificate: x509.Certificate,
681 root_certificate: x509.Certificate,
682) -> bool:
683 if certificate.issuer != root_certificate.subject:
684 return False
685 root_public_key = root_certificate.public_key()
686 signature_algorithm = certificate.signature_hash_algorithm
687 if not isinstance(root_public_key, ec.EllipticCurvePublicKey) or signature_algorithm is None:
688 return False
689 try:
690 root_public_key.verify(
691 certificate.signature,
692 certificate.tbs_certificate_bytes,
693 ec.ECDSA(signature_algorithm),
694 )
695 except InvalidSignature, UnsupportedAlgorithm, TypeError, ValueError:
696 return False
697 return True
700def _recover_unregistered_certificate(
701 config: ManagerConfig,
702 region: str,
703) -> tuple[str | None, x509.Certificate | None]:
704 """Adopt the unique tagged ACM leaf left by a failed SSM registration.
706 SSM remains the canonical registry. Tagged ACM inventory is the durable
707 recovery channel for the narrow case where both that write and the
708 compensating delete fail. Ambiguous inventory fails closed rather than
709 importing another certificate.
710 """
711 acm_client = boto3.client("acm", region_name=region)
712 managed: list[tuple[str, x509.Certificate]] = []
713 paginator = acm_client.get_paginator("list_certificates")
714 for page in paginator.paginate(
715 CertificateStatuses=list(_CERTIFICATE_STATUSES),
716 Includes={"keyTypes": ["EC_prime256v1"]},
717 ):
718 for summary in page.get("CertificateSummaryList", []):
719 if summary.get("Type") not in (None, "IMPORTED"):
720 continue
721 certificate_arn = _validated_certificate_arn(
722 region,
723 summary.get("CertificateArn"),
724 )
725 tags = _certificate_tags(acm_client, certificate_arn)
726 if _certificate_ownership_status(config, tags) != "owned":
727 continue
728 detail = acm_client.describe_certificate(CertificateArn=certificate_arn)
729 if detail.get("Certificate", {}).get("Type") != "IMPORTED":
730 raise RuntimeError(
731 f"Managed ACM certificate inventory for {region} contains a non-imported leaf"
732 )
733 certificate_response = acm_client.get_certificate(CertificateArn=certificate_arn)
734 certificate_pem = certificate_response.get("Certificate")
735 if not isinstance(certificate_pem, str):
736 raise ValueError(f"ACM did not return a managed imported certificate for {region}")
737 managed.append(
738 (
739 certificate_arn,
740 x509.load_pem_x509_certificate(certificate_pem.encode("ascii")),
741 )
742 )
744 if len(managed) > 1:
745 raise RuntimeError(
746 f"Multiple unregistered managed ACM certificates were found for {region}"
747 )
748 if not managed:
749 return None, None
751 certificate_arn, certificate = managed[0]
752 _write_certificate_registry(config, region, certificate_arn)
753 LOGGER.warning(
754 "Recovered the tagged backend leaf certificate registry association in %s",
755 region,
756 )
757 return certificate_arn, certificate
760def _registered_certificate(
761 config: ManagerConfig,
762 region: str,
763 *,
764 migration_roots: tuple[x509.Certificate, ...] | None = None,
765) -> tuple[str | None, x509.Certificate | None]:
766 """Load a registered leaf, optionally adopting a proven legacy leaf.
768 Strict callers, including every cleanup path, omit ``migration_roots`` and
769 therefore reject missing or conflicting ownership tags. Reconciliation may
770 pass the roots from its authenticated secret state; only then can an
771 untagged legacy imported leaf be tagged after its SAN and signature prove
772 that it belongs to this deployment.
773 """
774 parameter_name = config.certificate_parameter_name(region)
775 ssm_client = boto3.client("ssm", region_name=config.registry_region)
776 try:
777 response = ssm_client.get_parameter(Name=parameter_name)
778 except ClientError as exc:
779 if exc.response.get("Error", {}).get("Code") == "ParameterNotFound":
780 return None, None
781 raise
782 certificate_arn = _validated_certificate_arn(
783 region,
784 response.get("Parameter", {}).get("Value"),
785 )
787 acm_client = boto3.client("acm", region_name=region)
788 ownership = _certificate_ownership_status(
789 config,
790 _certificate_tags(acm_client, certificate_arn),
791 )
792 if ownership == "conflicting":
793 raise PermissionError(
794 f"Registered ACM certificate for {region} has conflicting ownership tags"
795 )
796 if ownership == "legacy" and migration_roots is None:
797 raise PermissionError(f"Registered ACM certificate for {region} is missing ownership tags")
799 if ownership == "legacy":
800 detail = acm_client.describe_certificate(CertificateArn=certificate_arn)
801 certificate_detail = detail.get("Certificate")
802 if not isinstance(certificate_detail, dict) or certificate_detail.get("Type") != "IMPORTED":
803 raise PermissionError(
804 f"Legacy ACM certificate for {region} is not an imported certificate"
805 )
807 response = acm_client.get_certificate(CertificateArn=certificate_arn)
808 certificate_pem = response.get("Certificate")
809 if not isinstance(certificate_pem, str):
810 raise ValueError(f"ACM did not return the imported certificate for {region}")
811 certificate = x509.load_pem_x509_certificate(certificate_pem.encode("ascii"))
813 if ownership == "legacy":
814 assert migration_roots is not None
815 if not _certificate_has_server_name(certificate, config.server_name) or not any(
816 _certificate_signed_by_root(certificate, root) for root in migration_roots
817 ):
818 raise PermissionError(
819 f"Legacy ACM certificate for {region} is not a managed backend leaf"
820 )
821 acm_client.add_tags_to_certificate(
822 CertificateArn=certificate_arn,
823 Tags=[
824 {"Key": "Project", "Value": config.project_name},
825 {"Key": "ManagedBy", "Value": _MANAGED_BY_TAG_VALUE},
826 ],
827 )
828 LOGGER.warning(
829 "Migrated cryptographically verified legacy backend leaf ownership in %s",
830 region,
831 )
833 return certificate_arn, certificate
836def _leaf_needs_rotation(
837 config: ManagerConfig,
838 certificate: x509.Certificate | None,
839 root_certificate: x509.Certificate,
840) -> bool:
841 if certificate is None:
842 return True
843 if _certificate_not_after(certificate) <= _now() + timedelta(
844 days=config.leaf_rotate_before_days
845 ):
846 return True
847 return not _certificate_signed_by_root(
848 certificate,
849 root_certificate,
850 ) or not _certificate_has_server_name(certificate, config.server_name)
853def _ensure_certificate(
854 config: ManagerConfig,
855 state: dict[str, Any],
856 region: str,
857) -> tuple[str, datetime, bool]:
858 _, root_certificate = _validate_root_record(state["current"], "current")
859 certificate_arn, existing_certificate = _registered_certificate(
860 config,
861 region,
862 migration_roots=_managed_root_certificates(state),
863 )
864 if certificate_arn is None:
865 certificate_arn, existing_certificate = _recover_unregistered_certificate(
866 config,
867 region,
868 )
869 if not _leaf_needs_rotation(config, existing_certificate, root_certificate):
870 assert certificate_arn is not None
871 assert existing_certificate is not None
872 return certificate_arn, _certificate_not_after(existing_certificate), False
874 certificate_pem, private_key_pem, leaf_expiry = _generate_leaf(config, state["current"])
875 acm_client = boto3.client("acm", region_name=region)
876 import_args: dict[str, Any] = {
877 "Certificate": certificate_pem,
878 "PrivateKey": private_key_pem,
879 }
880 if certificate_arn is not None:
881 import_args["CertificateArn"] = certificate_arn
882 else:
883 import_args["Tags"] = [
884 {"Key": "Project", "Value": config.project_name},
885 {"Key": "ManagedBy", "Value": _MANAGED_BY_TAG_VALUE},
886 ]
887 response = acm_client.import_certificate(**import_args)
888 imported_arn = str(response["CertificateArn"])
889 try:
890 _write_certificate_registry(config, region, imported_arn)
891 except Exception: # noqa: BLE001 - compensate every failed registry write
892 # A first import has no stable ARN until the registry write succeeds.
893 # Delete only that newly-created certificate so a retry cannot leak an
894 # undiscoverable managed certificate or create another orphan. A
895 # reimport of an existing ARN must never be deleted on registry failure.
896 if certificate_arn is None:
897 try:
898 acm_client.delete_certificate(CertificateArn=imported_arn)
899 except Exception: # noqa: BLE001 - retain the original SSM failure
900 LOGGER.exception(
901 "Could not remove unregistered backend leaf certificate in %s",
902 region,
903 )
904 raise
905 LOGGER.info(
906 "%s backend leaf certificate in %s; ACM ARN association remains stable",
907 "Reimported" if certificate_arn else "Imported",
908 region,
909 )
910 return imported_arn, leaf_expiry, True
913def _publish_expiry_metrics(
914 config: ManagerConfig,
915 certificate_expiries: dict[str, datetime],
916 root_expiry: datetime,
917) -> None:
918 now = _now()
919 metric_data = [
920 {
921 "MetricName": "ReconciliationSuccess",
922 "Dimensions": [{"Name": "Project", "Value": config.project_name}],
923 "Value": 1.0,
924 "Unit": "Count",
925 },
926 {
927 "MetricName": "RootCertificateDaysToExpiry",
928 "Dimensions": [{"Name": "Project", "Value": config.project_name}],
929 "Value": max(0.0, (root_expiry - now).total_seconds() / 86400),
930 "Unit": "Count",
931 },
932 ]
933 metric_data.extend(
934 {
935 "MetricName": "LeafCertificateDaysToExpiry",
936 "Dimensions": [
937 {"Name": "Project", "Value": config.project_name},
938 {"Name": "Region", "Value": region},
939 ],
940 "Value": max(0.0, (expiry - now).total_seconds() / 86400),
941 "Unit": "Count",
942 }
943 for region, expiry in certificate_expiries.items()
944 )
945 try:
946 boto3.client("cloudwatch").put_metric_data(
947 Namespace="GCO/BackendTLS",
948 MetricData=metric_data,
949 )
950 except (BotoCoreError, ClientError) as exc:
951 LOGGER.warning("Could not publish backend TLS expiry metrics: %s", exc)
954def _reconcile(
955 config: ManagerConfig,
956 newly_retired: tuple[str, ...] = (),
957) -> dict[str, Any]:
958 state, root_changed = _ensure_root(config)
959 expiries: dict[str, datetime] = {}
960 rotated_regions: list[str] = []
961 for region in config.regions:
962 _, expiry, rotated = _ensure_certificate(config, state, region)
963 expiries[region] = expiry
964 if rotated:
965 rotated_regions.append(region)
967 cleaned_retired, pending_retired = _retry_retired_region_cleanup(
968 config,
969 state,
970 newly_retired,
971 )
972 _, root_certificate = _validate_root_record(state["current"], "current")
973 _publish_expiry_metrics(config, expiries, _certificate_not_after(root_certificate))
974 return {
975 "RootChanged": root_changed,
976 "RotatedRegions": rotated_regions,
977 "ManagedRegionCount": len(config.regions),
978 "CleanedRetiredRegions": cleaned_retired,
979 "PendingRetiredRegions": pending_retired,
980 }
983def _event_regions(properties: Any, field: str) -> tuple[str, ...]:
984 """Return a strict custom-resource region list without silent coercion."""
985 if not isinstance(properties, dict):
986 raise ValueError(f"{field} must be an object")
987 raw_regions = properties.get("Regions")
988 if not isinstance(raw_regions, list) or not raw_regions:
989 raise ValueError(f"{field}.Regions must be a non-empty list")
990 regions: list[str] = []
991 for value in raw_regions:
992 if not isinstance(value, str):
993 raise ValueError(f"{field}.Regions contains an invalid region")
994 region = value.strip()
995 if _REGION_RE.fullmatch(region) is None:
996 raise ValueError(f"{field}.Regions contains an invalid region")
997 if region in regions:
998 raise ValueError(f"{field}.Regions contains a duplicate region")
999 regions.append(region)
1000 return tuple(regions)
1003def _retired_regions_from_update(
1004 event: dict[str, Any],
1005 config: ManagerConfig,
1006) -> tuple[str, ...]:
1007 old_regions = _event_regions(event.get("OldResourceProperties"), "OldResourceProperties")
1008 current_regions = set(config.regions)
1009 return tuple(region for region in old_regions if region not in current_regions)
1012def _delete_parameter(client: Any, name: str) -> None:
1013 try:
1014 client.delete_parameter(Name=name)
1015 except ClientError as exc:
1016 if exc.response.get("Error", {}).get("Code") != "ParameterNotFound":
1017 raise
1020def _certificate_registry_regions(config: ManagerConfig) -> frozenset[str]:
1021 """Discover and validate every managed regional certificate parameter.
1023 Inventory and ownership validation finish before cleanup mutates anything.
1024 Names must be exact direct children of the project prefix, values must be
1025 account/partition/Region-scoped ACM ARNs, and every referenced certificate
1026 must already carry both expected ownership tags. Cleanup never performs
1027 legacy tag migration.
1028 """
1029 client = boto3.client("ssm", region_name=config.registry_region)
1030 certificates: dict[str, str] = {}
1031 seen_tokens: set[str] = set()
1032 next_token: str | None = None
1034 while True:
1035 request: dict[str, Any] = {
1036 "Path": config.certificate_parameter_prefix,
1037 "Recursive": True,
1038 "WithDecryption": False,
1039 }
1040 if next_token is not None:
1041 request["NextToken"] = next_token
1042 response = client.get_parameters_by_path(**request)
1043 parameters = response.get("Parameters")
1044 if not isinstance(parameters, list):
1045 raise ValueError("Certificate registry inventory returned malformed parameters")
1047 for parameter in parameters:
1048 if not isinstance(parameter, dict):
1049 raise ValueError("Certificate registry inventory contains a malformed entry")
1050 name = parameter.get("Name")
1051 if not isinstance(name, str) or not name.startswith(
1052 config.certificate_parameter_prefix
1053 ):
1054 raise ValueError("Certificate registry parameter is outside the project prefix")
1055 region = name.removeprefix(config.certificate_parameter_prefix)
1056 if _REGION_RE.fullmatch(region) is None or name != config.certificate_parameter_name(
1057 region
1058 ):
1059 raise ValueError(f"Malformed certificate registry parameter name: {name!r}")
1060 if region in certificates:
1061 raise ValueError(f"Duplicate certificate registry parameter for {region}")
1062 certificates[region] = _validated_certificate_arn(
1063 region,
1064 parameter.get("Value"),
1065 )
1067 token = response.get("NextToken")
1068 if token is None:
1069 break
1070 if not isinstance(token, str) or not token or token in seen_tokens:
1071 raise ValueError("Certificate registry inventory returned an invalid pagination token")
1072 seen_tokens.add(token)
1073 next_token = token
1075 for region, certificate_arn in certificates.items():
1076 _require_certificate_ownership(
1077 config,
1078 region,
1079 certificate_arn,
1080 boto3.client("acm", region_name=region),
1081 )
1082 return frozenset(certificates)
1085def _delete_regional_certificate(
1086 config: ManagerConfig,
1087 region: str,
1088 *,
1089 defer_in_use: bool,
1090) -> bool:
1091 """Delete one managed certificate and its ARN parameter when safe.
1093 ACM refuses deletion while an ALB listener still uses the certificate. An
1094 Update or scheduled reconciliation preserves the parameter and returns
1095 ``False`` so durable retired-region state can retry later. Delete events
1096 fail instead: once the custom resource disappears no scheduler remains to
1097 finish cleanup.
1098 """
1099 certificate_arn, _ = _registered_certificate(config, region)
1100 if certificate_arn is None:
1101 # A failed first import can leave a tagged ACM certificate without its
1102 # canonical SSM association when both registration and compensation
1103 # fail. Recover that association before deletion so CloudFormation
1104 # cleanup cannot strand the durable orphan that reconciliation knows
1105 # how to adopt.
1106 certificate_arn, _ = _recover_unregistered_certificate(config, region)
1107 if certificate_arn is not None:
1108 try:
1109 boto3.client("acm", region_name=region).delete_certificate(
1110 CertificateArn=certificate_arn
1111 )
1112 except ClientError as exc:
1113 code = exc.response.get("Error", {}).get("Code")
1114 if code == "ResourceInUseException" and defer_in_use:
1115 LOGGER.info(
1116 "Backend certificate in retired region %s is still attached; "
1117 "retaining its ARN for scheduled cleanup",
1118 region,
1119 )
1120 return False
1121 if code != "ResourceNotFoundException":
1122 raise
1123 _delete_parameter(
1124 boto3.client("ssm", region_name=config.registry_region),
1125 config.certificate_parameter_name(region),
1126 )
1127 return True
1130def _retry_retired_region_cleanup(
1131 config: ManagerConfig,
1132 state: dict[str, Any],
1133 newly_retired: tuple[str, ...],
1134) -> tuple[list[str], list[str]]:
1135 """Persist, retry, and remove retired regions only after complete cleanup."""
1136 active_regions = set(config.regions)
1137 pending = sorted((set(state.get("retired_regions", [])) | set(newly_retired)) - active_regions)
1138 if pending != state.get("retired_regions", []):
1139 state["retired_regions"] = pending
1140 _save_root_state(state)
1142 remaining: list[str] = []
1143 cleaned: list[str] = []
1144 for region in pending:
1145 if _delete_regional_certificate(config, region, defer_in_use=True):
1146 cleaned.append(region)
1147 else:
1148 remaining.append(region)
1149 if remaining != pending:
1150 state["retired_regions"] = remaining
1151 _save_root_state(state)
1152 return cleaned, remaining
1155def _cleanup(config: ManagerConfig) -> None:
1156 state = _load_root_state()
1157 registry_regions = _certificate_registry_regions(config)
1158 regions = set(config.regions) | set(registry_regions)
1159 if state is not None:
1160 regions.update(state.get("retired_regions", []))
1162 for region in sorted(regions):
1163 _delete_regional_certificate(config, region, defer_in_use=False)
1165 ssm_client = boto3.client("ssm", region_name=config.registry_region)
1166 _delete_parameter(ssm_client, config.root_ca_parameter_name)
1167 if state is not None and state.get("retired_regions"):
1168 state["retired_regions"] = []
1169 _save_root_state(state)
1170 LOGGER.info("Removed current and retired ACM certificates and public backend TLS parameters")
1173def lambda_handler(event: dict[str, Any], _context: Any) -> dict[str, Any]:
1174 """Handle scheduled reconciliation and CDK provider lifecycle events."""
1175 config = ManagerConfig.from_event(event)
1176 if event.get("Action") == "Rotate":
1177 LOGGER.info("Running scheduled backend TLS reconciliation")
1178 return _reconcile(config)
1180 request_type = event.get("RequestType")
1181 physical_id = event.get("PhysicalResourceId") or (
1182 f"{config.project_name}-backend-tls-certificates"
1183 )
1184 if request_type == "Delete":
1185 _cleanup(config)
1186 return {"PhysicalResourceId": physical_id}
1187 if request_type not in {"Create", "Update"}:
1188 raise ValueError(f"Unsupported certificate manager event: {request_type!r}")
1190 newly_retired = _retired_regions_from_update(event, config) if request_type == "Update" else ()
1191 result = _reconcile(config, newly_retired)
1192 return {
1193 "PhysicalResourceId": physical_id,
1194 "Data": result,
1195 }