shoc-frontend-new/scripts/check-terraform-import-plan.py

#!/usr/bin/env python3
"""Reject plans that violate the frontend Terraform adoption boundary."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

from terraform_import_plan_resources import (
    ALLOWED_CREATE_ADDRESSES,
    CONTROLLED_UPDATE_ADDRESSES,
    ENVIRONMENT_CONFIG,
    GITHUB_OIDC_PROVIDER_ARN,
    GITHUB_REPO,
    REQUIRED_IMPORT_IDS,
    REQUIRED_RESOURCES,
)

BUCKET_POLICY_ADDRESS = "module.environment_owned.aws_s3_bucket_policy.site"
BUCKET_ADDRESS = "module.environment_owned.aws_s3_bucket.site"
DISTRIBUTION_ADDRESS = (
    "module.environment_owned.aws_cloudfront_distribution.site"
)
ROLE_ADDRESS = "module.environment_owned.aws_iam_role.github_deploy"
ROLE_POLICY_ADDRESS = "module.environment_owned.aws_iam_role_policy.github_deploy"
TAG_UPDATE_ADDRESSES = CONTROLLED_UPDATE_ADDRESSES - {
    BUCKET_POLICY_ADDRESS,
    ROLE_ADDRESS,
    ROLE_POLICY_ADDRESS,
}
OWNERSHIP_TAGS = {
    "Environment": None,
    "ManagedBy": "terraform",
    "Ownership": "terraform",
    "Project": "shoc-frontend",
}


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.",
    )
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument(
        "--post-import-no-op",
        action="store_true",
        help=(
            "Require all managed resources to be no-op after import and forbid "
            "import metadata."
        ),
    )
    modes.add_argument(
        "--allow-update-address",
        action="append",
        default=[],
        metavar="ADDRESS",
        help=(
            "Enter controlled-update mode and allow one exact reviewed address. "
            "Repeat for every expected update."
        ),
    )
    return parser.parse_args()


def _load_plan(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError("plan JSON root must be an object")
    if not isinstance(value.get("resource_changes"), list):
        raise ValueError("plan JSON must contain a resource_changes array")
    return value


def _validate_import_metadata(
    *,
    address: str,
    change: dict[str, Any],
    environment: str,
) -> list[str]:
    importing = change.get("importing")
    if not isinstance(importing, dict) or set(importing) != {"id"}:
        return [f"{address}: import metadata must be exactly {{'id': <string>}}"]

    import_id = importing.get("id")
    if not isinstance(import_id, str) or not import_id.strip():
        return [f"{address}: import ID must be a non-empty string"]
    if import_id.startswith("REPLACE_WITH_"):
        return [f"{address}: import ID is still a placeholder"]

    expected = REQUIRED_IMPORT_IDS[environment][address]
    if expected is not None and import_id != expected:
        return [f"{address}: expected import ID {expected!r}, got {import_id!r}"]

    other_environment_ids = {
        imports[address]
        for name, imports in REQUIRED_IMPORT_IDS.items()
        if name != environment and imports[address] is not None
    }
    if import_id in other_environment_ids:
        return [f"{address}: import ID belongs to another environment"]
    return []


def _contains_unknown(value: Any) -> bool:
    if value is True:
        return True
    if isinstance(value, dict):
        return any(_contains_unknown(item) for item in value.values())
    if isinstance(value, list):
        return any(_contains_unknown(item) for item in value)
    return False


def _changed_leaf_paths(
    before: Any,
    after: Any,
    path: tuple[str, ...] = (),
) -> set[tuple[str, ...]]:
    if isinstance(before, dict) and isinstance(after, dict):
        result: set[tuple[str, ...]] = set()
        for key in set(before) | set(after):
            result.update(
                _changed_leaf_paths(
                    before.get(key),
                    after.get(key),
                    (*path, str(key)),
                )
            )
        return result
    if before != after:
        return {path}
    return set()


def _canonical(value: Any) -> Any:
    if isinstance(value, dict):
        return {key: _canonical(value[key]) for key in sorted(value)}
    if isinstance(value, list):
        items = [_canonical(item) for item in value]
        return sorted(items, key=lambda item: json.dumps(item, sort_keys=True))
    return value


def _parse_policy(value: Any, address: str, side: str) -> tuple[Any, list[str]]:
    if not isinstance(value, str):
        return None, [f"{address}: {side} policy must be a JSON string"]
    try:
        document = json.loads(value)
    except json.JSONDecodeError:
        return None, [f"{address}: {side} policy is not valid JSON"]
    if not isinstance(document, dict):
        return None, [f"{address}: {side} policy must be a JSON object"]
    return _canonical(document), []


def _distribution_id(
    plan: dict[str, Any],
    environment: str,
) -> str | None:
    configured = ENVIRONMENT_CONFIG[environment]["distribution_id"]
    if isinstance(configured, str):
        return configured
    for resource in plan["resource_changes"]:
        if not isinstance(resource, dict) or resource.get("address") != DISTRIBUTION_ADDRESS:
            continue
        after = resource.get("change", {}).get("after")
        if isinstance(after, dict):
            identifier = after.get("id")
            if isinstance(identifier, str) and identifier.strip():
                return identifier
    return None


def _expected_pre_adoption_bucket_policy(
    environment: str,
    distribution_id: str,
) -> dict[str, Any]:
    config = ENVIRONMENT_CONFIG[environment]
    bucket_arn = f"arn:aws:s3:::{config['bucket_name']}"
    distribution_arn = (
        f"arn:aws:cloudfront::396287094661:distribution/{distribution_id}"
    )
    return _canonical(
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {
                        "AWS": config["bucket_auto_delete_helper_role_arn"]
                    },
                    "Action": [
                        "s3:DeleteObject*",
                        "s3:GetBucket*",
                        "s3:List*",
                        "s3:PutBucketPolicy",
                    ],
                    "Resource": [bucket_arn, f"{bucket_arn}/*"],
                },
                {
                    "Effect": "Allow",
                    "Principal": {"Service": "cloudfront.amazonaws.com"},
                    "Action": "s3:GetObject",
                    "Resource": f"{bucket_arn}/*",
                    "Condition": {
                        "StringEquals": {"AWS:SourceArn": distribution_arn}
                    },
                },
                {
                    "Effect": "Deny",
                    "Principal": {"AWS": "*"},
                    "Action": "s3:*",
                    "Resource": [bucket_arn, f"{bucket_arn}/*"],
                    "Condition": {"Bool": {"aws:SecureTransport": "false"}},
                },
            ],
        }
    )


def _expected_bucket_policy(environment: str, distribution_id: str) -> dict[str, Any]:
    bucket = ENVIRONMENT_CONFIG[environment]["bucket_name"]
    bucket_arn = f"arn:aws:s3:::{bucket}"
    distribution_arn = (
        f"arn:aws:cloudfront::396287094661:distribution/{distribution_id}"
    )
    return _canonical(
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {"Service": "cloudfront.amazonaws.com"},
                    "Action": "s3:GetObject",
                    "Resource": f"{bucket_arn}/*",
                    "Condition": {
                        "StringEquals": {"AWS:SourceArn": distribution_arn}
                    },
                },
                {
                    "Effect": "Deny",
                    "Principal": {"AWS": "*"},
                    "Action": "s3:*",
                    "Resource": [bucket_arn, f"{bucket_arn}/*"],
                    "Condition": {"Bool": {"aws:SecureTransport": "false"}},
                },
            ],
        }
    )


def _validate_tag_update(
    address: str,
    before: dict[str, Any],
    after: dict[str, Any],
    environment: str,
) -> list[str]:
    changed = _changed_leaf_paths(before, after)
    invalid = {
        path
        for path in changed
        if len(path) != 2 or path[0] not in {"tags", "tags_all"}
    }
    violations = [
        f"{address}: controlled tag update changes forbidden path {'.'.join(path)}"
        for path in sorted(invalid)
    ]
    expected = {**OWNERSHIP_TAGS, "Environment": environment}
    if address == ROLE_ADDRESS:
        expected["HcpTerraformWorkspace"] = ENVIRONMENT_CONFIG[environment][
            "workspace_name"
        ]
    if address == BUCKET_ADDRESS:
        expected["aws-cdk:auto-delete-objects"] = None
    expected_after = {
        key: value for key, value in expected.items() if value is not None
    }
    for tag_attribute in ("tags", "tags_all"):
        if after.get(tag_attribute) != expected_after:
            violations.append(
                f"{address}: {tag_attribute} must exactly match adopted ownership tags"
            )
    for path in sorted(changed - invalid):
        key = path[1]
        if key not in expected:
            violations.append(f"{address}: tag {key!r} is not an ownership tag")
        elif key == "aws-cdk:auto-delete-objects" and key in after.get(path[0], {}):
            violations.append(
                f"{address}: legacy auto-delete ownership tag was not removed"
            )
        elif after.get(path[0], {}).get(key) != expected[key]:
            violations.append(
                f"{address}: tag {key!r} does not have its expected adopted value"
            )
    if not changed:
        violations.append(f"{address}: update has no changed leaf values")
    return violations


def _validate_policy_update(
    address: str,
    before: dict[str, Any],
    after: dict[str, Any],
    environment: str,
    distribution_id: str | None,
) -> list[str]:
    changed = _changed_leaf_paths(before, after)
    if changed != {("policy",)}:
        return [f"{address}: policy update changes forbidden attributes {sorted(changed)!r}"]
    before_policy, violations = _parse_policy(before.get("policy"), address, "before")
    after_policy, after_violations = _parse_policy(
        after.get("policy"), address, "after"
    )
    violations.extend(after_violations)
    if before_policy == after_policy:
        violations.append(f"{address}: policy semantics did not change")
    if distribution_id is None:
        violations.append(
            f"{address}: cannot verify policy without the pinned distribution ID"
        )
        return violations
    expected_before = _expected_pre_adoption_bucket_policy(
        environment, distribution_id
    )
    expected_after = _expected_bucket_policy(environment, distribution_id)
    if before_policy is not None and before_policy != expected_before:
        violations.append(f"{address}: pre-adoption policy semantics are not exact")
    if after_policy is not None and after_policy != expected_after:
        violations.append(f"{address}: post-adoption policy semantics are not exact")
    return violations


def _expected_github_deploy_assume_policy(environment: str) -> dict[str, Any]:
    return _canonical(
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "GithubDeployOidc",
                    "Effect": "Allow",
                    "Action": "sts:AssumeRoleWithWebIdentity",
                    "Principal": {"Federated": GITHUB_OIDC_PROVIDER_ARN},
                    "Condition": {
                        "StringEquals": {
                            "token.actions.githubusercontent.com:aud": (
                                "sts.amazonaws.com"
                            ),
                            "token.actions.githubusercontent.com:sub": (
                                f"repo:{GITHUB_REPO}:environment:{environment}"
                            ),
                        },
                        "StringLike": {
                            "token.actions.githubusercontent.com:job_workflow_ref": [
                                (
                                    f"{GITHUB_REPO}/.github/workflows/"
                                    "deploy-web.yaml@refs/heads/main"
                                ),
                                (
                                    f"{GITHUB_REPO}/.github/workflows/"
                                    "deploy-web.yaml@refs/tags/v*"
                                ),
                            ],
                        },
                    },
                }
            ],
        }
    )


def _validate_role_assume_policy(
    address: str,
    before: dict[str, Any],
    after: dict[str, Any],
    environment: str,
) -> list[str]:
    before_policy, violations = _parse_policy(
        before.get("assume_role_policy"), address, "before"
    )
    after_policy, after_violations = _parse_policy(
        after.get("assume_role_policy"), address, "after"
    )
    violations.extend(after_violations)
    if before_policy == after_policy:
        violations.append(f"{address}: assume_role_policy semantics did not change")
    expected_after = _expected_github_deploy_assume_policy(environment)
    if after_policy is not None and after_policy != expected_after:
        violations.append(
            f"{address}: post-adoption assume_role_policy semantics are not exact"
        )
    return violations


def _validate_role_update(
    address: str,
    before: dict[str, Any],
    after: dict[str, Any],
    environment: str,
) -> list[str]:
    changed = _changed_leaf_paths(before, after)
    allowed_roots = {"tags", "tags_all", "assume_role_policy", "description"}
    invalid = {path for path in changed if not path or path[0] not in allowed_roots}
    violations = [
        f"{address}: controlled role update changes forbidden path {'.'.join(path)}"
        for path in sorted(invalid)
    ]
    if not changed:
        violations.append(f"{address}: update has no changed leaf values")
    expected = {
        **OWNERSHIP_TAGS,
        "Environment": environment,
        "HcpTerraformWorkspace": ENVIRONMENT_CONFIG[environment]["workspace_name"],
    }
    expected_after = {key: value for key, value in expected.items() if value is not None}
    for tag_attribute in ("tags", "tags_all"):
        if after.get(tag_attribute) != expected_after:
            violations.append(
                f"{address}: {tag_attribute} must exactly match adopted ownership tags"
            )
    if any(path and path[0] == "assume_role_policy" for path in changed):
        violations.extend(
            _validate_role_assume_policy(address, before, after, environment)
        )
    return violations


def _validate_iam_policy_update(
    address: str,
    before: dict[str, Any],
    after: dict[str, Any],
) -> list[str]:
    changed = _changed_leaf_paths(before, after)
    if changed != {("policy",)}:
        return [f"{address}: policy update changes forbidden attributes {sorted(changed)!r}"]
    if before.get("policy") == after.get("policy"):
        return [f"{address}: policy semantics did not change"]
    return []


def _validate_controlled_update(
    address: str,
    change: dict[str, Any],
    environment: str,
    distribution_id: str | None,
) -> list[str]:
    violations: list[str] = []
    replace_paths = change.get("replace_paths", [])
    if replace_paths not in (None, []):
        violations.append(f"{address}: replace_paths must be empty")
    if _contains_unknown(change.get("after_unknown", {})):
        violations.append(f"{address}: controlled update contains unknown values")
    before = change.get("before")
    after = change.get("after")
    if not isinstance(before, dict) or not isinstance(after, dict):
        return [*violations, f"{address}: controlled update requires before/after objects"]
    if address in TAG_UPDATE_ADDRESSES:
        violations.extend(_validate_tag_update(address, before, after, environment))
    elif address == ROLE_ADDRESS:
        violations.extend(_validate_role_update(address, before, after, environment))
    elif address == ROLE_POLICY_ADDRESS:
        violations.extend(_validate_iam_policy_update(address, before, after))
    elif address == BUCKET_POLICY_ADDRESS:
        violations.extend(
            _validate_policy_update(
                address,
                before,
                after,
                environment,
                distribution_id,
            )
        )
    return violations


def check_plan(
    plan: dict[str, Any],
    *,
    environment: str,
    mode: str,
    allowed_updates: set[str],
) -> list[str]:
    violations: list[str] = []
    invalid_allowed = allowed_updates - CONTROLLED_UPDATE_ADDRESSES
    for address in sorted(invalid_allowed):
        violations.append(
            f"{address}: address is not eligible for the controlled adoption update"
        )

    distribution_id = _distribution_id(plan, environment)
    seen_addresses: set[str] = set()
    seen_updates: set[str] = set()
    required_resources = REQUIRED_RESOURCES[environment]
    for resource in plan["resource_changes"]:
        if not isinstance(resource, dict):
            violations.append("<unknown>: resource change must be an object")
            continue
        if resource.get("mode", "managed") != "managed":
            continue
        address = resource.get("address")
        if not isinstance(address, str):
            violations.append("<unknown>: managed resource has no valid address")
            continue
        if address in seen_addresses:
            violations.append(f"{address}: duplicate managed resource change")
        seen_addresses.add(address)

        expected_type = required_resources.get(address)
        if expected_type is None:
            violations.append(f"{address}: managed address is outside the ownership boundary")
        elif resource.get("type") != expected_type:
            violations.append(
                f"{address}: expected managed type {expected_type!r}, "
                f"got {resource.get('type')!r}"
            )

        change = resource.get("change")
        if not isinstance(change, dict):
            violations.append(f"{address}: missing change object")
            continue
        actions = change.get("actions")
        if not isinstance(actions, list) or not all(
            isinstance(action, str) for action in actions
        ):
            violations.append(f"{address}: actions must be a string array")
            continue

        if change.get("replace_paths") not in (None, []):
            violations.append(f"{address}: replace_paths must be empty")

        import_id = REQUIRED_IMPORT_IDS[environment].get(address)
        if mode == "import":
            if address in ALLOWED_CREATE_ADDRESSES and import_id is None:
                if actions != ["create"]:
                    violations.append(
                        f"{address}: import mode requires create for deploy parameters, got {actions!r}"
                    )
                if "importing" in change:
                    violations.append(
                        f"{address}: import metadata is forbidden for created deploy parameters"
                    )
            else:
                if actions != ["no-op"]:
                    violations.append(
                        f"{address}: import mode requires no-op, got {actions!r}"
                    )
                if expected_type is not None:
                    violations.extend(
                        _validate_import_metadata(
                            address=address,
                            change=change,
                            environment=environment,
                        )
                    )
        elif mode == "post-import":
            if actions != ["no-op"]:
                violations.append(
                    f"{address}: post-import mode requires no-op, got {actions!r}"
                )
            if "importing" in change:
                violations.append(
                    f"{address}: import metadata is forbidden in post-import mode"
                )
        else:
            if "importing" in change:
                violations.append(
                    f"{address}: import metadata is forbidden in controlled-update mode"
                )
            if actions == ["update"]:
                seen_updates.add(address)
                if address not in allowed_updates:
                    violations.append(f"{address}: update is not explicitly allowlisted")
                else:
                    violations.extend(
                        _validate_controlled_update(
                            address,
                            change,
                            environment,
                            distribution_id,
                        )
                    )
            elif actions == ["create"] and address in ALLOWED_CREATE_ADDRESSES:
                pass
            elif actions != ["no-op"]:
                violations.append(f"{address}: unsafe controlled actions {actions!r}")

    for missing in sorted(set(required_resources) - seen_addresses):
        violations.append(f"{missing}: required managed resource is absent")
    for unused in sorted(allowed_updates - seen_updates):
        violations.append(f"{unused}: allowlisted update address is not updating")
    return violations


def main() -> int:
    args = parse_args()
    try:
        plan = _load_plan(args.plan_json)
    except (OSError, ValueError, json.JSONDecodeError) as error:
        print(f"FAIL: unable to read Terraform plan JSON: {error}", file=sys.stderr)
        return 1

    allowed_updates = set(args.allow_update_address or [])
    if args.post_import_no_op:
        mode = "post-import"
    elif allowed_updates:
        mode = "controlled"
    else:
        mode = "import"
    violations = check_plan(
        plan,
        environment=args.environment,
        mode=mode,
        allowed_updates=allowed_updates,
    )
    if violations:
        print("FAIL: Terraform plan is not adoption-safe", file=sys.stderr)
        for violation in violations:
            print(f"  - {violation}", file=sys.stderr)
        return 1

    label = {
        "import": "zero-change import",
        "post-import": "post-import no-op",
        "controlled": "controlled update",
    }[mode]
    print(
        f"PASS: {label} plan has {len(REQUIRED_RESOURCES[args.environment])} "
        f"managed resources and {len(allowed_updates)} exact updates"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())