#!/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, 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 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 pre_adoption_deploy_policy(environment: str) -> dict[str, Any]: config = ENVIRONMENT_CONFIG[environment] bucket_arn = f"arn:aws:s3:::{config['bucket_name']}" distribution_arn = ( "arn:aws:cloudfront::396287094661:distribution/" f"{distribution_id(environment)}" ) statements: list[dict[str, Any]] = [] if environment == "dev": statements.append( { "Sid": "AssumeCdkBootstrapRoles", "Effect": "Allow", "Action": "sts:AssumeRole", "Resource": "arn:aws:iam::396287094661:role/cdk-hnb659fds-*", } ) statements.extend( [ { "Sid": "DescribeStack", "Effect": "Allow", "Action": "cloudformation:DescribeStacks", "Resource": ( "arn:aws:cloudformation:us-east-1:396287094661:stack/" f"{config['cloudformation_stack_name']}/*" ), }, { "Effect": "Allow", "Action": [ "s3:Abort*", "s3:DeleteObject*", "s3:GetBucket*", "s3:GetObject*", "s3:List*", "s3:PutObject", "s3:PutObjectLegalHold", "s3:PutObjectRetention", "s3:PutObjectTagging", "s3:PutObjectVersionTagging", ], "Resource": [bucket_arn, f"{bucket_arn}/*"], }, { "Sid": "InvalidateDistribution", "Effect": "Allow", "Action": [ "cloudfront:CreateInvalidation", "cloudfront:GetInvalidation", ], "Resource": distribution_arn, }, ] ) return {"Version": "2012-10-17", "Statement": statements} def deploy_policy(environment: str) -> dict[str, Any]: bucket = ENVIRONMENT_CONFIG[environment]["bucket_name"] bucket_arn = f"arn:aws:s3:::{bucket}" distribution_arn = ( "arn:aws:cloudfront::396287094661:distribution/" f"{distribution_id(environment)}" ) return { "Version": "2012-10-17", "Statement": [ { "Sid": "ReadDeploymentBucket", "Effect": "Allow", "Action": [ "s3:GetBucketLocation", "s3:GetBucketVersioning", "s3:ListBucket", "s3:ListBucketVersions", ], "Resource": bucket_arn, }, { "Sid": "PublishAndRollbackSiteObjects", "Effect": "Allow", "Action": [ "s3:DeleteObject", "s3:DeleteObjectVersion", "s3:GetObject", "s3:GetObjectVersion", "s3:PutObject", ], "Resource": f"{bucket_arn}/*", }, { "Sid": "InvalidateDistribution", "Effect": "Allow", "Action": [ "cloudfront:CreateInvalidation", "cloudfront:GetInvalidation", ], "Resource": distribution_arn, }, ], } 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]: before_policy = ( pre_adoption_bucket_policy(environment) if address == BUCKET_POLICY else pre_adoption_deploy_policy(environment) ) after_policy = ( bucket_policy(environment) if address == BUCKET_POLICY else deploy_policy(environment) ) return { "actions": ["update"], "before": {"policy": json.dumps(before_policy)}, "after": {"policy": json.dumps(after_policy)}, } 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_import_phase_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+= false\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_deploy_policy_rejects_resource_action_and_extra_statement(self) -> None: for mutation in ("resource", "action", "extra"): plan = make_plan( "staging", mode="controlled", controlled_updates={DEPLOY_POLICY}, ) policy = copy.deepcopy(deploy_policy("staging")) if mutation == "resource": policy["Statement"][0]["Resource"] = "*" elif mutation == "action": policy["Statement"][0]["Action"].append("iam:PassRole") else: policy["Statement"].append( { "Sid": "Extra", "Effect": "Allow", "Action": "s3:*", "Resource": "*", } ) resource(plan, DEPLOY_POLICY)["change"]["after"]["policy"] = json.dumps( policy ) with self.subTest(mutation=mutation): self.assert_fails(plan, "staging", DEPLOY_POLICY) def test_policy_updates_require_exact_pre_adoption_state(self) -> None: for environment in REQUIRED_RESOURCES: for address in (BUCKET_POLICY, DEPLOY_POLICY): plan = make_plan( environment, mode="controlled", controlled_updates={address}, ) change = resource(plan, address)["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, address=address): self.assert_fails(plan, environment, address) 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()