afterhours-shift-manager/scripts/cutover/copy_secrets.py

#!/usr/bin/env python3
"""Copy afterhours secrets mgmt → prod. Dry-run unless --execute.

Terraform creates empty secret shells. Slack, signing, and roster tokens are
written into those shells when the dest has no current string value. Populated
dest values are left alone. 3CX secrets are verified only and never written.
Strips trailing newlines. Never prints secret values.
"""

from __future__ import annotations

import argparse
import sys

import boto3
from botocore.exceptions import ClientError

SRC_ACCOUNT = "328440206208"
DST_ACCOUNT = "011934824531"

COPY = [
    "afterhours-shift-manager/slack-bot-token",
    "afterhours-shift-manager/slack-signing-secret",
    "afterhours-shift-manager/roster-api-token",
]

VERIFY_ONLY = [
    "afterhours-shift-manager/3cx-domain",
    "afterhours-shift-manager/3cx-client-id",
    "afterhours-shift-manager/3cx-client-secret",
]

# Describe succeeds on a Terraform shell; GetSecretValue fails until a version exists.
_NO_VALUE_CODES = frozenset({"ResourceNotFoundException", "InvalidRequestException"})


def _client(profile: str, region: str):
    return boto3.Session(profile_name=profile, region_name=region).client("secretsmanager")


def _account(profile: str) -> str:
    return boto3.Session(profile_name=profile).client("sts").get_caller_identity()["Account"]


def secret_string(client, name: str) -> str | None:
    """Return the current SecretString, or None if the secret does not exist.

    An empty string means the secret exists (Terraform shell) but has no usable
    current version.
    """
    try:
        client.describe_secret(SecretId=name)
    except ClientError as exc:
        if exc.response["Error"]["Code"] == "ResourceNotFoundException":
            return None
        raise
    try:
        payload = client.get_secret_value(SecretId=name)
    except ClientError as exc:
        if exc.response["Error"]["Code"] in _NO_VALUE_CODES:
            return ""
        raise
    value = payload.get("SecretString")
    if value is None:
        return ""
    return value


def copy_secrets(src, dst, *, execute: bool) -> int:
    rc = 0

    for name in VERIFY_ONLY:
        value = secret_string(dst, name)
        if value is None:
            print(f"missing prod secret {name} (expected from PLAT-76)", file=sys.stderr)
            rc = 1
        elif not value.strip():
            print(f"empty prod 3cx secret {name} (do not overwrite from mgmt)", file=sys.stderr)
            rc = 1
        else:
            print(f"keep existing prod secret {name}")

    for name in COPY:
        src_value = secret_string(src, name)
        if src_value is None or not src_value.strip():
            print(f"missing mgmt secret {name}", file=sys.stderr)
            rc = 1
            continue
        dest_value = secret_string(dst, name)
        if dest_value is None:
            print(f"missing prod secret shell {name}", file=sys.stderr)
            rc = 1
            continue
        if dest_value.strip():
            print(f"skip populated prod secret {name}")
            continue
        print(f"would copy {name}")
        if not execute:
            continue
        value = src_value.rstrip("\n")
        dst.put_secret_value(SecretId=name, SecretString=value)
        print(f"wrote {name} ({len(value)} chars)")

    if not execute:
        print("dry-run; pass --execute to PutSecretValue")
    return rc


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--src-profile", required=True)
    parser.add_argument("--dst-profile", required=True)
    parser.add_argument("--region", default="us-east-1")
    parser.add_argument("--execute", action="store_true")
    args = parser.parse_args()

    if _account(args.src_profile) != SRC_ACCOUNT:
        print("src profile is not mgmt", file=sys.stderr)
        return 2
    if _account(args.dst_profile) != DST_ACCOUNT:
        print("dst profile is not prod", file=sys.stderr)
        return 2

    src = _client(args.src_profile, args.region)
    dst = _client(args.dst_profile, args.region)
    return copy_secrets(src, dst, execute=args.execute)


if __name__ == "__main__":
    raise SystemExit(main())