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

#!/usr/bin/env python3
"""Deterministic unit tests for the frontend Terraform plan checker."""

from __future__ import annotations

import copy
import json
import re
import subprocess
import sys
import tempfile
import unittest
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,
)

SCRIPT = Path(__file__).with_name("check-terraform-import-plan.py")
REPOSITORY = SCRIPT.parent.parent
BUCKET_POLICY = "module.environment_owned.aws_s3_bucket_policy.site"
BUCKET = "module.environment_owned.aws_s3_bucket.site"
DEPLOY_POLICY = "module.environment_owned.aws_iam_role_policy.github_deploy"
ROLE = "module.environment_owned.aws_iam_role.github_deploy"
DISTRIBUTION = "module.environment_owned.aws_cloudfront_distribution.site"
TAG_ADDRESSES = CONTROLLED_UPDATE_ADDRESSES - {BUCKET_POLICY, DEPLOY_POLICY}


def import_id(environment: str, address: str) -> str:
    expected = REQUIRED_IMPORT_IDS[environment][address]
    assert expected is not None, f"{environment} must pin an import ID for {address}"
    return expected


def distribution_id(environment: str) -> str:
    configured = ENVIRONMENT_CONFIG[environment]["distribution_id"]
    assert isinstance(configured, str), f"{environment} must pin a distribution ID"
    return configured


def pre_adoption_bucket_policy(environment: str) -> dict[str, Any]:
    config = ENVIRONMENT_CONFIG[environment]
    bucket_arn = f"arn:aws:s3:::{config['bucket_name']}"
    source = (
        "arn:aws:cloudfront::396287094661:distribution/"
        f"{distribution_id(environment)}"
    )
    return {
        "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": source}},
            },
            {
                "Effect": "Deny",
                "Principal": {"AWS": "*"},
                "Action": "s3:*",
                "Resource": [bucket_arn, f"{bucket_arn}/*"],
                "Condition": {"Bool": {"aws:SecureTransport": "false"}},
            },
        ],
    }


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


def tag_change(environment: str, address: str) -> dict[str, Any]:
    manager = {
        "HcpTerraformWorkspace": ENVIRONMENT_CONFIG[environment]["workspace_name"]
    }
    before_tags = {
        "Environment": environment,
        "ManagedBy": "cdk",
        "Project": "shoc-frontend",
    }
    after_tags = {
        "Environment": environment,
        "ManagedBy": "terraform",
        "Ownership": "terraform",
        "Project": "shoc-frontend",
    }
    if address == ROLE:
        before_tags.update(manager)
        after_tags.update(manager)
    if address == BUCKET:
        before_tags["aws-cdk:auto-delete-objects"] = "true"
    before: dict[str, Any] = {
        "tags": before_tags,
        "tags_all": before_tags,
    }
    after: dict[str, Any] = {
        "tags": after_tags,
        "tags_all": after_tags,
    }
    if address == DISTRIBUTION:
        before["id"] = distribution_id(environment)
        after["id"] = distribution_id(environment)
    return {"actions": ["update"], "before": before, "after": after}


def policy_change(environment: str, address: str) -> dict[str, Any]:
    if address == DEPLOY_POLICY:
        return {
            "actions": ["update"],
            "before": {"policy": json.dumps({"Version": "2012-10-17", "Statement": []})},
            "after": {
                "policy": json.dumps(
                    {
                        "Version": "2012-10-17",
                        "Statement": [{"Sid": "ListWebBucket", "Effect": "Allow"}],
                    }
                )
            },
        }
    if address != BUCKET_POLICY:
        raise AssertionError(f"{address} is not a reviewed policy update")
    return {
        "actions": ["update"],
        "before": {"policy": json.dumps(pre_adoption_bucket_policy(environment))},
        "after": {"policy": json.dumps(bucket_policy(environment))},
    }


def make_plan(
    environment: str,
    *,
    mode: str = "import",
    controlled_updates: set[str] | None = None,
) -> dict[str, Any]:
    resources: list[dict[str, Any]] = []
    updates = controlled_updates or set()
    for address, resource_type in REQUIRED_RESOURCES[environment].items():
        if mode == "import":
            import_id_value = REQUIRED_IMPORT_IDS[environment][address]
            if import_id_value is None:
                change = {"actions": ["create"]}
            else:
                change = {
                    "actions": ["no-op"],
                    "importing": {"id": import_id_value},
                }
        elif mode == "post-import":
            change = {"actions": ["no-op"]}
        elif address in updates:
            change = (
                tag_change(environment, address)
                if address in TAG_ADDRESSES
                else policy_change(environment, address)
            )
        elif address in ALLOWED_CREATE_ADDRESSES:
            change = {"actions": ["create"]}
        else:
            change = {"actions": ["no-op"]}
            if address == DISTRIBUTION:
                change["after"] = {"id": distribution_id(environment)}
        resources.append(
            {
                "address": address,
                "mode": "managed",
                "type": resource_type,
                "change": change,
            }
        )
    return {"resource_changes": resources}


def resource(plan: dict[str, Any], address: str) -> dict[str, Any]:
    return next(
        item for item in plan["resource_changes"] if item["address"] == address
    )


def run_checker(
    plan: dict[str, Any],
    environment: str,
    *allowed_updates: str,
    post_import: bool = False,
) -> subprocess.CompletedProcess[str]:
    with tempfile.TemporaryDirectory() as directory:
        path = Path(directory) / "plan.json"
        path.write_text(json.dumps(plan), encoding="utf-8")
        command = [
            sys.executable,
            str(SCRIPT),
            str(path),
            "--environment",
            environment,
        ]
        if post_import:
            command.append("--post-import-no-op")
        for address in allowed_updates:
            command.extend(["--allow-update-address", address])
        return subprocess.run(
            command,
            check=False,
            capture_output=True,
            text=True,
        )


class ImportPlanCheckerTests(unittest.TestCase):
    def assert_passes(
        self,
        plan: dict[str, Any],
        environment: str,
        *allowed_updates: str,
        post_import: bool = False,
    ) -> None:
        result = run_checker(
            plan,
            environment,
            *allowed_updates,
            post_import=post_import,
        )
        self.assertEqual(0, result.returncode, result.stdout + result.stderr)

    def assert_fails(
        self,
        plan: dict[str, Any],
        environment: str,
        *allowed_updates: str,
        post_import: bool = False,
    ) -> None:
        result = run_checker(
            plan,
            environment,
            *allowed_updates,
            post_import=post_import,
        )
        self.assertNotEqual(0, result.returncode, result.stdout + result.stderr)

    def test_cloudfront_function_source_matches_exact_nine_line_join(self) -> None:
        source = (
            REPOSITORY / "terraform/live/modules/environment-owned/main.tf"
        ).read_text(encoding="utf-8")
        expected = """  spa_rewrite_code = join("\\n", [
    "function handler(event) {",
    "  var request = event.request;",
    "  var uri = request.uri;",
    "  // No file extension after the last slash -> a client-side route.",
    "  if (uri.lastIndexOf('.') <= uri.lastIndexOf('/')) {",
    "    request.uri = '/index.html';",
    "  }",
    "  return request;",
    "}",
  ])"""
        self.assertIn(expected, source)

    def test_live_roots_are_not_flattened(self) -> None:
        live = REPOSITORY / "terraform" / "live"
        self.assertTrue((live / "dev" / "versions.tf").is_file())
        self.assertTrue((live / "dev" / "main.tf").is_file())
        self.assertTrue((live / "staging" / "versions.tf").is_file())
        self.assertTrue((live / "staging" / "main.tf").is_file())
        self.assertTrue((live / "modules" / "environment-owned" / "main.tf").is_file())
        self.assertFalse((REPOSITORY / "terraform" / "versions.tf").exists())
        self.assertFalse((REPOSITORY / "terraform" / "main.tf").exists())

    def test_adoption_complete_is_pinned_in_locals(self) -> None:
        dev = (REPOSITORY / "terraform/live/dev/main.tf").read_text(encoding="utf-8")
        staging = (REPOSITORY / "terraform/live/staging/main.tf").read_text(
            encoding="utf-8"
        )
        self.assertRegex(dev, r"adoption_complete\s+= true")
        self.assertRegex(staging, r"adoption_complete\s+= true")
        self.assertNotIn('variable "adoption_complete"', dev)
        self.assertNotIn('variable "environment"', dev)
        self.assertNotIn('variable "release_version_label"', dev)

    def test_managed_modules_use_direct_pinned_inputs(self) -> None:
        source = (REPOSITORY / "terraform/live/dev/main.tf").read_text(encoding="utf-8")
        expected = {
            "hosted_zone_id": "local.hosted_zone_id",
            "certificate_arn": "local.certificate_arn",
            "github_oidc_provider_arn": "local.github_oidc_arn",
            "cache_policy_id": "local.cache_policy_id",
        }
        for name, value in expected.items():
            self.assertRegex(source, rf"{name}\s+= {re.escape(value)}")
        self.assertNotRegex(
            source,
            r"(hosted_zone_id|certificate_arn|github_oidc_provider_arn|cache_policy_id)\s+= module\.inventory",
        )

    def test_exact_import_plan_passes_for_every_environment(self) -> None:
        for environment in REQUIRED_RESOURCES:
            with self.subTest(environment=environment):
                self.assert_passes(make_plan(environment), environment)

    def test_import_missing_extra_wrong_type_and_cross_environment_fail(self) -> None:
        for mutation in ("missing", "extra", "wrong-type", "cross-environment"):
            plan = make_plan("dev")
            if mutation == "missing":
                plan["resource_changes"].pop()
            elif mutation == "extra":
                plan["resource_changes"].append(
                    {
                        "address": "module.inventory.aws_route53_zone.site",
                        "mode": "managed",
                        "type": "aws_route53_zone",
                        "change": {
                            "actions": ["no-op"],
                            "importing": {"id": "Z00000000000000000000"},
                        },
                    }
                )
            elif mutation == "wrong-type":
                plan["resource_changes"][0]["type"] = "aws_s3_object"
            else:
                resource(plan, DISTRIBUTION)["change"]["importing"]["id"] = (
                    REQUIRED_IMPORT_IDS["staging"][DISTRIBUTION]
                )
            with self.subTest(mutation=mutation):
                self.assert_fails(plan, "dev")

    def test_import_rejects_mutation_and_invalid_metadata(self) -> None:
        for actions in (["create"], ["update"], ["delete"], ["delete", "create"]):
            plan = make_plan("dev")
            plan["resource_changes"][0]["change"]["actions"] = actions
            with self.subTest(actions=actions):
                self.assert_fails(plan, "dev")
        plan = make_plan("dev")
        plan["resource_changes"][0]["change"]["importing"] = {"id": ""}
        self.assert_fails(plan, "dev")

    def test_post_import_no_op_passes(self) -> None:
        self.assert_passes(
            make_plan("staging", mode="post-import"),
            "staging",
            post_import=True,
        )

    def test_post_import_rejects_import_metadata_and_update(self) -> None:
        plan = make_plan("dev", mode="post-import")
        plan["resource_changes"][0]["change"]["importing"] = {"id": "unexpected"}
        self.assert_fails(plan, "dev", post_import=True)
        plan = make_plan("dev", mode="post-import")
        plan["resource_changes"][0]["change"]["actions"] = ["update"]
        self.assert_fails(plan, "dev", post_import=True)

    def test_every_allowed_controlled_diff_passes(self) -> None:
        for environment in REQUIRED_RESOURCES:
            for address in CONTROLLED_UPDATE_ADDRESSES:
                with self.subTest(environment=environment, address=address):
                    self.assert_passes(
                        make_plan(
                            environment,
                            mode="controlled",
                            controlled_updates={address},
                        ),
                        environment,
                        address,
                    )

    def test_full_exact_controlled_allowlist_passes(self) -> None:
        addresses = tuple(sorted(CONTROLLED_UPDATE_ADDRESSES))
        self.assert_passes(
            make_plan(
                "dev",
                mode="controlled",
                controlled_updates=set(addresses),
            ),
            "dev",
            *addresses,
        )

    def test_tag_update_rejects_extra_attribute_and_wrong_value(self) -> None:
        plan = make_plan("dev", mode="controlled", controlled_updates={ROLE})
        resource(plan, ROLE)["change"]["after"]["max_session_duration"] = 7200
        self.assert_fails(plan, "dev", ROLE)
        plan = make_plan("dev", mode="controlled", controlled_updates={ROLE})
        resource(plan, ROLE)["change"]["after"]["tags"]["ManagedBy"] = "attacker"
        self.assert_fails(plan, "dev", ROLE)

    def test_tag_update_requires_complete_adopted_tag_sets(self) -> None:
        plan = make_plan("dev", mode="controlled", controlled_updates={BUCKET})
        del resource(plan, BUCKET)["change"]["after"]["tags"]["Ownership"]
        self.assert_fails(plan, "dev", BUCKET)

    def test_role_trust_change_is_allowed(self) -> None:
        for environment in REQUIRED_RESOURCES:
            plan = make_plan(
                environment, mode="controlled", controlled_updates={ROLE}
            )
            role = resource(plan, ROLE)["change"]
            role["before"]["assume_role_policy"] = '{"Statement":[]}'
            role["after"]["assume_role_policy"] = json.dumps(
                github_deploy_assume_policy(environment)
            )
            with self.subTest(environment=environment):
                self.assert_passes(plan, environment, ROLE)

    def test_role_trust_rejects_mutated_document(self) -> None:
        for mutation in ("principal", "missing-sub"):
            plan = make_plan("dev", mode="controlled", controlled_updates={ROLE})
            role = resource(plan, ROLE)["change"]
            policy = github_deploy_assume_policy("dev")
            if mutation == "principal":
                policy["Statement"][0]["Principal"] = {"AWS": "*"}
            else:
                del policy["Statement"][0]["Condition"]["StringEquals"][
                    "token.actions.githubusercontent.com:sub"
                ]
            role["before"]["assume_role_policy"] = '{"Statement":[]}'
            role["after"]["assume_role_policy"] = json.dumps(policy)
            with self.subTest(mutation=mutation):
                self.assert_fails(plan, "dev", ROLE)

    def test_bucket_policy_rejects_malicious_principal_and_extra_statement(self) -> None:
        for mutation in ("principal", "extra"):
            plan = make_plan(
                "dev",
                mode="controlled",
                controlled_updates={BUCKET_POLICY},
            )
            policy = copy.deepcopy(bucket_policy("dev"))
            if mutation == "principal":
                policy["Statement"][0]["Principal"] = {"AWS": "*"}
            else:
                policy["Statement"].append(
                    {
                        "Effect": "Allow",
                        "Principal": {"AWS": "*"},
                        "Action": "s3:*",
                        "Resource": "*",
                    }
                )
            resource(plan, BUCKET_POLICY)["change"]["after"]["policy"] = json.dumps(
                policy
            )
            with self.subTest(mutation=mutation):
                self.assert_fails(plan, "dev", BUCKET_POLICY)

    def test_github_deploy_policy_is_bucket_root_sync(self) -> None:
        source = (
            REPOSITORY / "terraform/live/modules/environment-owned/main.tf"
        ).read_text(encoding="utf-8")
        document = source.split('data "aws_iam_policy_document" "github_deploy" {', 1)[1]
        document = document.split("resource ", 1)[0]
        self.assertNotIn("var.adoption_complete", document)
        self.assertIn("ListWebBucket", document)
        self.assertIn("SyncWebBucket", document)
        self.assertIn("InvalidateDistribution", document)
        self.assertIn("DeployParams", document)
        self.assertIn("s3:DeleteObject", document)
        self.assertIn("cloudfront:CreateInvalidation", document)
        self.assertIn("cloudfront:GetInvalidation", document)
        self.assertIn("cloudfront:GetDistribution", document)
        self.assertIn("ssm:GetParameter", document)
        self.assertNotIn("ListReleasePrefixes", document)
        self.assertNotIn("PublishReleasePrefix", document)
        self.assertNotIn("ReadReleasePointer", document)
        self.assertNotIn("releases/*", document)
        self.assertNotIn("AssumeCdkBootstrapRoles", document)
        self.assertNotIn("DescribeStack", document)
        self.assertIn(
            "module.environment_owned.aws_iam_role_policy.github_deploy",
            CONTROLLED_UPDATE_ADDRESSES,
        )

    def test_github_deploy_assume_document_matches_module(self) -> None:
        source = (
            REPOSITORY / "terraform/live/modules/environment-owned/main.tf"
        ).read_text(encoding="utf-8")
        document = source.split(
            'data "aws_iam_policy_document" "github_deploy_assume" {', 1
        )[1]
        document = document.split(
            'data "aws_iam_policy_document" "github_deploy" {', 1
        )[0]
        self.assertIn("GithubDeployOidc", document)
        self.assertIn("sts:AssumeRoleWithWebIdentity", document)
        self.assertIn("token.actions.githubusercontent.com:aud", document)
        self.assertIn("sts.amazonaws.com", document)
        self.assertIn("token.actions.githubusercontent.com:sub", document)
        self.assertIn("local.github_subject", document)
        self.assertIn("token.actions.githubusercontent.com:job_workflow_ref", document)
        self.assertIn("deploy-web.yaml@refs/heads/main", document)
        self.assertIn("deploy-web.yaml@refs/tags/v*", document)

    def test_deploy_policy_controlled_update_passes(self) -> None:
        self.assert_passes(
            make_plan(
                "dev",
                mode="controlled",
                controlled_updates={DEPLOY_POLICY},
            ),
            "dev",
            DEPLOY_POLICY,
        )
        plan = make_plan("dev", mode="controlled", controlled_updates=set())
        resource(plan, DEPLOY_POLICY)["change"] = {
            "actions": ["update"],
            "before": {"policy": "{}"},
            "after": {"policy": '{"Version":"2012-10-17"}'},
        }
        self.assert_fails(plan, "dev")

    def test_policy_updates_require_exact_pre_adoption_state(self) -> None:
        for environment in REQUIRED_RESOURCES:
            plan = make_plan(
                environment,
                mode="controlled",
                controlled_updates={BUCKET_POLICY},
            )
            change = resource(plan, BUCKET_POLICY)["change"]
            before = json.loads(change["before"]["policy"])
            before["Statement"].append(
                {
                    "Sid": "UnexpectedDrift",
                    "Effect": "Deny",
                    "Action": "*",
                    "Resource": "*",
                }
            )
            change["before"]["policy"] = json.dumps(before)
            with self.subTest(environment=environment):
                self.assert_fails(plan, environment, BUCKET_POLICY)

    def test_controlled_update_rejects_unknown_and_replace_paths(self) -> None:
        for field, value in (
            ("after_unknown", {"tags": {"ManagedBy": True}}),
            ("replace_paths", [["tags"]]),
        ):
            plan = make_plan(
                "dev",
                mode="controlled",
                controlled_updates={ROLE},
            )
            resource(plan, ROLE)["change"][field] = value
            with self.subTest(field=field):
                self.assert_fails(plan, "dev", ROLE)

    def test_nonallowlisted_update_and_unused_allowlist_fail(self) -> None:
        plan = make_plan("dev", mode="controlled", controlled_updates={ROLE})
        self.assert_fails(plan, "dev", BUCKET_POLICY)
        plan = make_plan("dev", mode="controlled", controlled_updates=set())
        self.assert_fails(plan, "dev", ROLE)


if __name__ == "__main__":
    unittest.main()