procurement-ingest/lambdas/wo/shoc_hmac_rotator/handler.py

"""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": "<YYYY-MM-DDTHH>", "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)