diff --git a/scripts/cutover/copy_secrets.py b/scripts/cutover/copy_secrets.py new file mode 100644 index 0000000..c7ea130 --- /dev/null +++ b/scripts/cutover/copy_secrets.py @@ -0,0 +1,130 @@ +#!/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()) diff --git a/tests/scripts/test_copy_secrets.py b/tests/scripts/test_copy_secrets.py new file mode 100644 index 0000000..ae96426 --- /dev/null +++ b/tests/scripts/test_copy_secrets.py @@ -0,0 +1,99 @@ +"""copy_secrets.py writes Slack tokens into empty Terraform shells.""" + +import importlib.util +import sys +from pathlib import Path + +from botocore.exceptions import ClientError + +ROOT = Path(__file__).resolve().parents[2] + + +def _load(): + spec = importlib.util.spec_from_file_location( + "copy_secrets", ROOT / "scripts" / "cutover" / "copy_secrets.py" + ) + mod = importlib.util.module_from_spec(spec) + sys.modules["copy_secrets"] = mod + spec.loader.exec_module(mod) + return mod + + +mod = _load() + + +def _client_error(code: str) -> ClientError: + return ClientError({"Error": {"Code": code, "Message": code}}, "GetSecretValue") + + +class FakeSecrets: + def __init__(self, described, strings=None, get_errors=None): + self.described = set(described) + self.strings = dict(strings or {}) + self.get_errors = dict(get_errors or {}) + self.puts = [] + + def describe_secret(self, SecretId): + if SecretId not in self.described: + raise _client_error("ResourceNotFoundException") + return {"Name": SecretId} + + def get_secret_value(self, SecretId): + if SecretId in self.get_errors: + raise _client_error(self.get_errors[SecretId]) + if SecretId not in self.strings: + raise _client_error("ResourceNotFoundException") + return {"SecretString": self.strings[SecretId]} + + def put_secret_value(self, SecretId, SecretString): + self.puts.append((SecretId, SecretString)) + self.strings[SecretId] = SecretString + return {} + + +def test_execute_puts_into_empty_terraform_shells(): + src = FakeSecrets( + described=mod.COPY, + strings={name: f"{name}-value\n" for name in mod.COPY}, + ) + dst = FakeSecrets( + described=mod.COPY + mod.VERIFY_ONLY, + strings={name: "already-copied" for name in mod.VERIFY_ONLY}, + get_errors={name: "InvalidRequestException" for name in mod.COPY}, + ) + rc = mod.copy_secrets(src, dst, execute=True) + assert rc == 0 + assert [name for name, _ in dst.puts] == list(mod.COPY) + assert all(value.endswith("-value") and not value.endswith("\n") for _, value in dst.puts) + + +def test_skip_populated_copy_targets_and_never_write_3cx(): + src = FakeSecrets( + described=mod.COPY, + strings={name: "from-mgmt" for name in mod.COPY}, + ) + dst = FakeSecrets( + described=mod.COPY + mod.VERIFY_ONLY, + strings={ + **{name: "prod-already" for name in mod.COPY}, + **{name: "3cx-prod" for name in mod.VERIFY_ONLY}, + }, + ) + rc = mod.copy_secrets(src, dst, execute=True) + assert rc == 0 + assert dst.puts == [] + + +def test_dry_run_does_not_put(): + src = FakeSecrets( + described=mod.COPY, + strings={name: "from-mgmt" for name in mod.COPY}, + ) + dst = FakeSecrets( + described=mod.COPY + mod.VERIFY_ONLY, + strings={name: "3cx-prod" for name in mod.VERIFY_ONLY}, + get_errors={name: "InvalidRequestException" for name in mod.COPY}, + ) + rc = mod.copy_secrets(src, dst, execute=False) + assert rc == 0 + assert dst.puts == []