shoc-backend/scripts/check-terraform-import-plan.py

#!/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", "<unknown>")
        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())