Jon Chery
257 lines
11 KiB
Python
257 lines
11 KiB
Python
#!/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()) |