"""Rotation Lambda for the SHOC webhook HMAC secret. Secrets Manager invokes this on the 30-day rotation schedule for ``workorder-ingest/shoc-webhook-hmac`` (docs/shoc-webhook-contract.md section 6.1). Secret value shape:: {"keys": [{"kid": "", "secret": "<64 hex chars>"}, ...]} newest first, truncated to 2 entries (one overlap cycle). The emitter always signs with ``keys[0]``; the SHOC receiver accepts any listed ``kid``. This is *single-user* rotation: no external system holds the value -- both sides re-fetch from Secrets Manager on their own <=5-minute cache TTLs -- so the standard 4-step protocol collapses to createSecret/finishSecret, with setSecret a logged no-op and testSecret validating the pending value's shape. The secret material is machine-generated HMAC key bytes and is NEVER logged; structured logs carry only step names, version ids, and ``kid`` values. """ import json import logging import re import secrets from datetime import datetime, timezone import boto3 logger = logging.getLogger() logger.setLevel(logging.INFO) # One overlap cycle: the new key plus the previous key (contract section 6.1). MAX_KEYS = 2 # 32 random bytes -> secrets.token_hex emits 64 lowercase hex chars. SECRET_NUM_BYTES = 32 SECRET_HEX_PATTERN = re.compile(r"[0-9a-f]{64}") KID_FORMAT = "%Y-%m-%dT%H" # Forced re-rotation within the same hour needs a distinct kid. KID_FORMAT_INTRA_HOUR = "%Y-%m-%dT%H%M" # Lazy cached Secrets Manager client. Keeps the public attribute name # ``secretsmanager_client`` so the test monkeypatch target is stable. secretsmanager_client = None def _get_client(): global secretsmanager_client if secretsmanager_client is None: secretsmanager_client = boto3.client("secretsmanager") return secretsmanager_client def _current_keys(client, secret_id: str) -> list: """Read the AWSCURRENT ``keys`` list, tolerating malformed prior values. The CDK-seeded bootstrap value is ``{"keys": [], "bootstrap_entropy": ...}``; any unparseable or wrong-shape prior value likewise falls back to ``[]`` (rotation then simply starts a fresh key list -- the material is fully regenerable, receivers refresh within their TTL). """ try: raw = client.get_secret_value(SecretId=secret_id, VersionStage="AWSCURRENT") parsed = json.loads(raw["SecretString"]) except (client.exceptions.ResourceNotFoundException, json.JSONDecodeError): # Genuinely-absent or corrupt current value -> start a fresh list. # NB: this does NOT catch transient failures (throttling, KMS/IAM # blips): those re-raise so Secrets Manager marks the rotation failed # and retries with the prior AWSCURRENT intact, rather than silently # dropping the overlap key and stranding in-flight deliveries. logger.warning( json.dumps( {"event": "hmac_rotation_current_unreadable", "fallback": "empty"} ) ) return [] keys = parsed.get("keys") if isinstance(parsed, dict) else None if not isinstance(keys, list) or not all( isinstance(key, dict) and isinstance(key.get("kid"), str) and isinstance(key.get("secret"), str) for key in keys ): logger.warning( json.dumps( {"event": "hmac_rotation_current_malformed", "fallback": "empty"} ) ) return [] return keys def _create_secret(client, secret_id: str, token: str) -> None: """Stage a new AWSPENDING value: fresh key prepended, list truncated to 2.""" versions = client.describe_secret(SecretId=secret_id).get("VersionIdsToStages", {}) if "AWSCURRENT" in versions.get(token, []): logger.info( json.dumps( {"event": "hmac_rotation_create_already_current", "version_id": token} ) ) return try: client.get_secret_value( SecretId=secret_id, VersionId=token, VersionStage="AWSPENDING" ) # This token's version already holds a value: retried createSecret # invocation, nothing to do. logger.info( json.dumps( {"event": "hmac_rotation_create_idempotent", "version_id": token} ) ) return except client.exceptions.ResourceNotFoundException: pass current_keys = _current_keys(client, secret_id) now = datetime.now(timezone.utc) existing_kids = {key["kid"] for key in current_keys} new_kid = now.strftime(KID_FORMAT) if new_kid in existing_kids: # Collision against ANY retained kid (not just keys[0]): repeated # same-hour forced rotations must never reissue a kid for a different # secret, or a receiver's cached kid->secret map desyncs. Append a # random suffix guaranteed distinct from the retained set. new_kid = f"{now.strftime(KID_FORMAT_INTRA_HOUR)}{secrets.token_hex(2)}" new_key = {"kid": new_kid, "secret": secrets.token_hex(SECRET_NUM_BYTES)} client.put_secret_value( SecretId=secret_id, ClientRequestToken=token, SecretString=json.dumps({"keys": [new_key] + current_keys[: MAX_KEYS - 1]}), VersionStages=["AWSPENDING"], ) logger.info( json.dumps( { "event": "hmac_rotation_pending_staged", "version_id": token, "kid": new_kid, "key_count": 1 + len(current_keys[: MAX_KEYS - 1]), } ) ) def _set_secret(client, secret_id: str, token: str) -> None: """No-op: no external system holds the value (single-user rotation).""" logger.info( json.dumps({"event": "hmac_rotation_set_secret_noop", "version_id": token}) ) def _test_secret(client, secret_id: str, token: str) -> None: """Validate the AWSPENDING value's shape; raise ValueError to fail rotation.""" pending = client.get_secret_value( SecretId=secret_id, VersionId=token, VersionStage="AWSPENDING" ) # json.JSONDecodeError is a ValueError subclass: bad JSON fails the rotation. parsed = json.loads(pending["SecretString"]) keys = parsed.get("keys") if isinstance(parsed, dict) else None if not isinstance(keys, list) or not keys: raise ValueError("AWSPENDING value has no keys list") head = keys[0] if not isinstance(head, dict) or not isinstance(head.get("kid"), str): raise ValueError("AWSPENDING keys[0] has no string kid") secret_value = head.get("secret") if not isinstance(secret_value, str) or not SECRET_HEX_PATTERN.fullmatch( secret_value ): raise ValueError("AWSPENDING keys[0].secret is not 64 lowercase hex chars") # Log only the rotation token, never a value pulled from the parsed # secret dict. kid is non-sensitive (it rides X-SH-Key-Id in the clear) # and is already logged at stage time in _create_secret, but subscripting # the secret-bearing dict here trips CodeQL's clear-text-logging taint # (py/clear-text-logging-sensitive-data) and is fragile if a later edit # swaps the field -- version_id already correlates this step to the stage. logger.info(json.dumps({"event": "hmac_rotation_test_ok", "version_id": token})) def _finish_secret(client, secret_id: str, token: str) -> None: """Move AWSCURRENT to the pending version (idempotent on retry).""" versions = client.describe_secret(SecretId=secret_id).get("VersionIdsToStages", {}) current_version_id = None for version_id, stages in versions.items(): if "AWSCURRENT" in stages: current_version_id = version_id break if current_version_id == token: logger.info( json.dumps( {"event": "hmac_rotation_finish_idempotent", "version_id": token} ) ) return stage_move = { "SecretId": secret_id, "VersionStage": "AWSCURRENT", "MoveToVersionId": token, } if current_version_id is not None: stage_move["RemoveFromVersionId"] = current_version_id client.update_secret_version_stage(**stage_move) logger.info( json.dumps( { "event": "hmac_rotation_finished", "moved_to": token, "removed_from": current_version_id, } ) ) def handler(event, context): """Secrets Manager rotation entry point. ``event`` = ``{"SecretId", "ClientRequestToken", "Step"}``; dispatches the standard rotation steps. Unknown steps raise (rotation marked failed). """ secret_id = event["SecretId"] token = event["ClientRequestToken"] step = event["Step"] logger.info( json.dumps({"event": "hmac_rotation_step", "step": step, "version_id": token}) ) client = _get_client() steps = { "createSecret": _create_secret, "setSecret": _set_secret, "testSecret": _test_secret, "finishSecret": _finish_secret, } if step not in steps: raise ValueError(f"Unknown rotation step: {step}") steps[step](client, secret_id, token)