#!/usr/bin/env python3 """Migrate SSM parameter paths from /acdl/... → /nova/... (REQ-161, P3). The Nova rebrand (v1.15) moves the SSM parameter namespace prefix from `/acdl/{env}/{contractId}/{output}` to `/nova/{env}/{contractId}/{output}`. This script copies every existing `/acdl/...` parameter to its `/nova/...` twin (same value, same Type, SecureString preserved, same KMS key), verifies the copy round-trips, then deletes the old `/acdl/...` parameter. Design: - **Dry-run by default.** Prints the planned copy/delete operations without touching AWS. Pass ``--apply`` to execute. - **Idempotent.** If the `/nova/...` target already exists with the same value, the copy is skipped (and reported as a no-op); the old `/acdl/...` parameter is still deleted (the migration is re-runnable). If the target exists with a *different* value, the copy is skipped with a WARNING and the old parameter is NOT deleted (manual review required) unless ``--force`` is passed. - **Path-mapping logic is unit-tested** (see ``tests/test_migrate_ssm_paths.py``); the AWS I/O is thin boto3 glue around ``map_path()``. Usage: python3 scripts/migrate_ssm_paths.py # dry-run, /acdl → /nova python3 scripts/migrate_ssm_paths.py --apply # execute python3 scripts/migrate_ssm_paths.py --source /acdl --dest /nova --apply python3 scripts/migrate_ssm_paths.py --region us-east-1 --apply This script does NOT need live AWS to be importable; the boto3 client is constructed lazily inside ``run()`` so the module can be imported + the path-mapping logic unit-tested without credentials. """ from __future__ import annotations import argparse import sys from typing import Optional try: import boto3 except ImportError: # pragma: no cover - boto3 is a test dep boto3 = None # type: ignore # --------------------------------------------------------------------------- # Path-mapping logic (pure, unit-tested) # --------------------------------------------------------------------------- def map_path(source_path: str, source_prefix: str = "/acdl", dest_prefix: str = "/nova") -> str: """Map an SSM parameter path from the source prefix to the dest prefix. The match is on a *path-segment* boundary: ``/acdl`` matches ``/acdl/dev/...`` but a literal like ``/acdl-platform`` is left untouched (it does not start with the ``/acdl/`` segment). A path that does not start with the source prefix (as a leading segment) raises ``ValueError`` so callers can filter or surface stray parameters. Examples: >>> map_path("/acdl/dev/svc-x/output") '/nova/dev/svc-x/output' >>> map_path("/acdl/dev/c-1/vpc_id", "/acdl", "/nova") '/nova/dev/c-1/vpc_id' >>> map_path("/acdl/qa/c-2/db_endpoint") '/nova/qa/c-2/db_endpoint' """ if not source_path.startswith(source_prefix + "/"): raise ValueError( f"path {source_path!r} does not start with source prefix " f"{source_prefix!r} (as a path segment)" ) return dest_prefix + source_path[len(source_prefix):] def list_acdl_params(client, source_prefix: str = "/acdl"): """List all SSM parameters whose Name starts with ``source_prefix/``. Uses ``DescribeParameters`` with a ParameterFilters Path prefix (the documented, pagination-friendly way to scope by path). Returns a list of parameter-summary dicts (Name, Type, KeyId, ...). """ params: list[dict] = [] paginator = client.get_paginator("describe_parameters") iterator = paginator.paginate( ParameterFilters=[ {"Key": "Path", "Option": "Recursive", "Values": [source_prefix + "/"]} ] ) for page in iterator: for p in page.get("Parameters", []): params.append(p) return params def copy_one_param(client, source_name: str, dest_name: str, force: bool = False) -> str: """Copy a single SSM parameter from source to dest. Returns one of: ``"copied"``, ``"skipped-equal"`` (already migrated), ``"skipped-mismatch"`` (dest exists with a different value; needs --force to overwrite), ``"overwritten"`` (force=True overwrote a mismatching dest). """ src = client.get_parameter(Name=source_name, WithDecryption=True) value = src["Parameter"]["Value"] ptype = src["Parameter"]["Type"] key_id = src["Parameter"].get("KeyId") # Check if dest already exists try: dst = client.get_parameter(Name=dest_name, WithDecryption=True) if dst["Parameter"]["Value"] == value: return "skipped-equal" if not force: return "skipped-mismatch" except client.exceptions.ParameterNotFound: pass # target doesn't exist yet → proceed to put except Exception as e: # P4 (REQ-168): narrow the broad swallow — only ParameterNotFound # is an expected "proceed to put" condition. Any other AWS error # (auth, throttling, service) must surface, not be swallowed. import sys sys.stderr.write( f"migrate_ssm_paths: get_parameter({dest_name}) failed: " f"{type(e).__name__}: {e}\n" ) raise put_kwargs = { "Name": dest_name, "Value": value, "Type": ptype, "Overwrite": True, } if ptype == "SecureString" and key_id: put_kwargs["KeyId"] = key_id client.put_parameter(**put_kwargs) return "overwritten" if force else "copied" def verify_one_param(client, source_name: str, dest_name: str) -> bool: """Verify the dest parameter holds the same value as the source.""" src = client.get_parameter(Name=source_name, WithDecryption=True) dst = client.get_parameter(Name=dest_name, WithDecryption=True) return src["Parameter"]["Value"] == dst["Parameter"]["Value"] def delete_one_param(client, name: str) -> None: """Delete a single SSM parameter.""" client.delete_parameter(Name=name) def run( source_prefix: str = "/acdl", dest_prefix: str = "/nova", region: Optional[str] = None, apply: bool = False, force: bool = False, client=None, ) -> dict: """Run the migration. Returns a summary dict. When ``apply`` is False (default, dry-run), no AWS mutations happen — the function lists the source parameters and reports the planned copy/delete operations. When ``apply`` is True, it copies, verifies, and deletes. A pre-built boto3 SSM ``client`` may be injected for testing. """ if apply and client is None: if boto3 is None: raise RuntimeError("boto3 is required for --apply (live AWS)") client = boto3.client("ssm", region_name=region) if region else boto3.client("ssm") if client is None and apply: raise RuntimeError("boto3 SSM client required for --apply") summary = {"listed": 0, "copied": 0, "skipped_equal": 0, "skipped_mismatch": 0, "verified": 0, "deleted": 0, "errors": 0, "plan": []} params = list_acdl_params(client, source_prefix) if apply else _dry_run_list(source_prefix, client) summary["listed"] = len(params) for p in params: src_name = p["Name"] try: dest_name = map_path(src_name, source_prefix, dest_prefix) except ValueError: summary["errors"] += 1 summary["plan"].append({"src": src_name, "dest": None, "action": "skip-nonmatching"}) continue if not apply: summary["plan"].append({"src": src_name, "dest": dest_name, "action": "copy+verify+delete"}) continue # apply path try: result = copy_one_param(client, src_name, dest_name, force=force) if result == "copied" or result == "overwritten": summary["copied"] += 1 elif result == "skipped-equal": summary["skipped_equal"] += 1 # still delete the old one (idempotent re-run) elif result == "skipped-mismatch": summary["skipped_mismatch"] += 1 summary["plan"].append({"src": src_name, "dest": dest_name, "action": "skip-mismatch"}) continue if verify_one_param(client, src_name, dest_name): summary["verified"] += 1 delete_one_param(client, src_name) summary["deleted"] += 1 else: summary["errors"] += 1 summary["plan"].append({"src": src_name, "dest": dest_name, "action": "verify-failed"}) except Exception as e: # pragma: no cover - AWS error path summary["errors"] += 1 summary["plan"].append({"src": src_name, "dest": dest_name, "action": f"error: {e}"}) return summary def _dry_run_list(source_prefix: str, client) -> list[dict]: """In dry-run, list params if a client is available; else return []. Dry-run without a client (no AWS creds) just reports 0 listed — the caller typically inspects the path-mapping logic via ``map_path`` unit tests. """ if client is None: return [] return list_acdl_params(client, source_prefix) def main(argv: Optional[list[str]] = None) -> int: parser = argparse.ArgumentParser( description="Migrate SSM parameter paths /acdl/... → /nova/... (REQ-161, P3)." ) parser.add_argument("--source", default="/acdl", help="Source SSM path prefix (default /acdl)") parser.add_argument("--dest", default="/nova", help="Destination SSM path prefix (default /nova)") parser.add_argument("--region", default=None, help="AWS region (default: boto3 default)") parser.add_argument("--apply", action="store_true", help="Execute the migration (default: dry-run)") parser.add_argument("--force", action="store_true", help="Overwrite a dest parameter that exists with a different value (default: skip)") args = parser.parse_args(argv) mode = "APPLY" if args.apply else "DRY-RUN" print(f"[migrate_ssm_paths] {mode}: {args.source} → {args.dest} (region={args.region or 'default'})") summary = run( source_prefix=args.source, dest_prefix=args.dest, region=args.region, apply=args.apply, force=args.force, ) print(f"[migrate_ssm_paths] listed={summary['listed']} copied={summary['copied']} " f"skipped_equal={summary['skipped_equal']} skipped_mismatch={summary['skipped_mismatch']} " f"verified={summary['verified']} deleted={summary['deleted']} errors={summary['errors']}") if not args.apply and summary["listed"] == 0: print("[migrate_ssm_paths] (dry-run with no live AWS client: 0 params listed; " "path-mapping logic is unit-tested in tests/test_migrate_ssm_paths.py)") return 0 if summary["errors"] == 0 else 1 if __name__ == "__main__": sys.exit(main())