#!/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 ( CONTROLLED_UPDATE_ADDRESSES, ENVIRONMENT_CONFIG, 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} 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 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 != 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": change: dict[str, Any] = { "actions": ["no-op"], "importing": {"id": import_id(environment, address)}, } 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) ) 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_only_dev_has_a_live_root(self) -> None: live_roots = sorted( path.name for path in (REPOSITORY / "terraform/live").iterdir() if path.is_dir() and path.name != "modules" ) self.assertEqual(["dev"], live_roots) def test_dev_root_pins_adoption_complete_in_code(self) -> None: source = (REPOSITORY / "terraform/live/dev/main.tf").read_text(encoding="utf-8") self.assertRegex(source, r"\n\s+adoption_complete\s+= true\n") self.assertRegex(source, r"adoption_complete\s+= local\.adoption_complete") self.assertNotIn('variable "adoption_complete"', source) for root_file in ("main.tf", "imports.tf", "outputs.tf", "providers.tf", "versions.tf"): self.assertNotIn( "variable ", (REPOSITORY / f"terraform/live/dev/{root_file}").read_text(encoding="utf-8"), root_file, ) def test_managed_modules_use_direct_pinned_inputs(self) -> None: expected = { "dev": ( "local.hosted_zone_id", "local.certificate_arn", "local.github_oidc_arn", "local.cache_policy_id", ), } for environment, values in expected.items(): source = ( REPOSITORY / f"terraform/live/{environment}/main.tf" ).read_text(encoding="utf-8") for name, value in zip( ( "hosted_zone_id", "certificate_arn", "github_oidc_provider_arn", "cache_policy_id", ), values, strict=True, ): self.assertIn(f"{name}", source) 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"]["assume_role_policy"] = "{}" 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_rejected(self) -> None: plan = make_plan("dev", mode="controlled", controlled_updates={ROLE}) role = resource(plan, ROLE)["change"] role["before"]["assume_role_policy"] = '{"Statement":[]}' role["after"]["assume_role_policy"] = '{"Statement":[{"Effect":"Allow"}]}' 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_release_prefix_only(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("ListReleasePrefixes", document) self.assertIn("PublishReleasePrefix", document) self.assertIn("ReadReleasePointer", document) self.assertIn("ReadDistribution", document) self.assertIn("cloudfront:GetDistribution", document) self.assertIn("cloudfront:GetDistributionConfig", document) self.assertIn("releases/*", document) self.assertNotIn("AssumeCdkBootstrapRoles", document) self.assertNotIn("DescribeStack", document) self.assertNotIn("CreateInvalidation", document) self.assertNotIn("ReadDeploymentBucket", document) self.assertNotIn("PublishAndRollbackSiteObjects", document) self.assertNotIn( "module.environment_owned.aws_iam_role_policy.github_deploy", CONTROLLED_UPDATE_ADDRESSES, ) def test_deploy_policy_is_not_eligible_for_controlled_update(self) -> None: plan = make_plan("dev", mode="controlled", controlled_updates=set()) self.assert_fails(plan, "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", DEPLOY_POLICY) 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()