fix(cutover): write Slack secrets into empty Terraform shells

DescribeSecret succeeds on HCP-created shells with no version, so skip-if-exists left roster and Slack tokens unset.

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())

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 == []