#!/usr/bin/env python3 """Reject unsafe actions in a live Terraform import plan.""" from __future__ import annotations import argparse import json import sys from pathlib import Path from terraform_import_plan_resources import ( DEV_IMPORT_BASELINE, IMPORT_BASELINES, IMPORT_IDS, REQUIRED_RESOURCES, STAGING_IMPORT_BASELINE, ) ALLOWED_MANAGED_TYPES = { resource_type for resources in REQUIRED_RESOURCES.values() for resource_type in resources.values() } UNSAFE_ACTIONS = {"create", "delete"} def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("plan_json", type=Path) parser.add_argument( "--environment", required=True, choices=sorted(REQUIRED_RESOURCES), help="Exact environment ownership boundary expected in the plan.", ) parser.add_argument( "--allow-update-address", action="append", default=[], metavar="ADDRESS", help=( "Allow an in-place update to this exact address after the initial " "no-op import is proven. Repeat for each reviewed update." ), ) parser.add_argument( "--evidence-out", type=Path, help="Write machine-readable proof after every import assertion passes.", ) return parser.parse_args() def validate_dev_import_baseline( resources_by_address: dict[str, dict], violations: list[str] ) -> None: environment_address = ( "module.environment.aws_elastic_beanstalk_environment.this" ) route_address = "module.environment.aws_route53_record.api_alias[0]" expected_tags = DEV_IMPORT_BASELINE["environment_tags"] expected_alias = DEV_IMPORT_BASELINE["api_alias"] for side in ("before", "after"): environment = ( resources_by_address.get(environment_address, {}) .get("change", {}) .get(side) or {} ) if environment.get("tags") != expected_tags: violations.append( f"{environment_address}: {side} environment tags do not match " f"the exact dev import baseline" ) if environment.get("setting") != []: violations.append( f"{environment_address}: {side} contains managed EB settings " "during the import-only phase" ) route = ( resources_by_address.get(route_address, {}) .get("change", {}) .get(side) or {} ) aliases = route.get("alias") or [] if len(aliases) != 1 or aliases[0] != expected_alias: violations.append( f"{route_address}: {side} alias does not match the exact " "live ALB target and zone" ) def validate_staging_import_baseline( resources_by_address: dict[str, dict], violations: list[str] ) -> None: environment_address = ( "module.environment.aws_elastic_beanstalk_environment.this" ) route_address = "module.environment.aws_route53_record.api_cname[0]" expected_tags = STAGING_IMPORT_BASELINE["environment_tags"] expected_cname = STAGING_IMPORT_BASELINE["api_cname"] for side in ("before", "after"): environment = ( resources_by_address.get(environment_address, {}) .get("change", {}) .get(side) or {} ) if environment.get("tags") != expected_tags: violations.append( f"{environment_address}: {side} environment tags do not match " f"the exact staging import baseline" ) if environment.get("setting") != []: violations.append( f"{environment_address}: {side} contains managed EB settings " "during the import-only phase" ) route = ( resources_by_address.get(route_address, {}) .get("change", {}) .get(side) or {} ) actual_cname = { "records": route.get("records"), "ttl": route.get("ttl"), } if actual_cname != expected_cname: violations.append( f"{route_address}: {side} CNAME does not match the exact " "live ALB target and TTL" ) def main() -> int: args = parse_args() plan = json.loads(args.plan_json.read_text(encoding="utf-8")) violations: list[str] = [] managed = 0 updates = 0 allowed_update_addresses = set(args.allow_update_address) seen_update_addresses: set[str] = set() seen_addresses: set[str] = set() required_resources = REQUIRED_RESOURCES[args.environment] resources_by_address: dict[str, dict] = {} initial_import = not allowed_update_addresses for resource in plan.get("resource_changes", []): if resource.get("mode", "managed") != "managed": continue resource_type = resource.get("type", "") address = resource.get("address", "") actions = set(resource.get("change", {}).get("actions", [])) managed += 1 seen_addresses.add(address) resources_by_address[address] = resource if resource_type not in ALLOWED_MANAGED_TYPES: violations.append( f"{address}: managed type {resource_type!r} is outside the live ownership boundary" ) expected_type = required_resources.get(address) if expected_type is None: violations.append( f"{address}: managed address is outside the live ownership boundary" ) elif resource_type != expected_type: violations.append( f"{address}: expected managed type {expected_type!r}, got {resource_type!r}" ) unsafe = sorted(actions & UNSAFE_ACTIONS) if unsafe: violations.append(f"{address}: unsafe actions {unsafe}") if "update" in actions: updates += 1 seen_update_addresses.add(address) if address not in allowed_update_addresses: violations.append( f"{address}: update is not explicitly allowlisted" ) if initial_import and args.environment in IMPORT_IDS: if actions != {"no-op"}: violations.append( f"{address}: initial {args.environment} import actions " "must be ['no-op'], " f"got {sorted(actions)}" ) expected_import_id = IMPORT_IDS[args.environment].get(address) actual_import_id = ( resource.get("change", {}).get("importing") or {} ).get("id") if actual_import_id != expected_import_id: violations.append( f"{address}: expected import id {expected_import_id!r}, " f"got {actual_import_id!r}" ) for unused in sorted(allowed_update_addresses - seen_update_addresses): violations.append(f"{unused}: allowlisted update address is not updating") for missing in sorted(set(required_resources) - seen_addresses): violations.append(f"{missing}: required managed resource is absent") if initial_import and args.environment == "dev": validate_dev_import_baseline(resources_by_address, violations) elif initial_import and args.environment == "staging": validate_staging_import_baseline(resources_by_address, violations) if violations: print("FAIL: live Terraform plan is not import-safe", file=sys.stderr) for violation in violations: print(f" - {violation}", file=sys.stderr) return 1 mode = "controlled update" if allowed_update_addresses else "no-op import" if args.evidence_out: evidence = { "environment": args.environment, "mode": mode, "managed_resources": managed, "updates": updates, "creates": 0, "deletes": 0, "replacements": 0, "imports": { address: ( resource.get("change", {}).get("importing") or {} ).get("id") for address, resource in sorted(resources_by_address.items()) }, } if args.environment in IMPORT_BASELINES and initial_import: evidence["asserted_live_baseline"] = IMPORT_BASELINES[args.environment] args.evidence_out.write_text( json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) print( f"PASS: {mode} plan has {managed} managed resources, " f"{updates} updates, and no create/delete/replace actions" ) return 0 if __name__ == "__main__": raise SystemExit(main())