#!/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': }}"] 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(": 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(": 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())