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