shoc-frontend-new/scripts/test-terraform-import-plan-check.py
Adam Moussa c96a259365
Some checks failed
Deploy dev content / Deploy shoc-frontend-new-dev through Terraform (push) Has been cancelled
refactor(cd): ship SPA content from GitHub on main (#220)
* ci(cd): convert SPA hosting to handbook HCP and GitHub content CD

Give HCP the bucket and CloudFront with an empty origin path. GitHub owns
bucket-root sync and invalidation so merge-to-main and a human staging tag
can deploy without creating HCP runs. G13 fails PRs that mix terraform/
with deployable application files.

* ci: run Frontend checks and Terraform CI on PRs to main and dev

Match backend 148 so a PR targeting origin/dev still gets the required
checks. Push remains main only.

* refactor(terraform): keep live/dev and live/staging as HCP roots

Leave the adopted working directories in place so this CD PR does not
retarget two live HCP workspaces. Flattening stays a later change.

* style: prettier terraform-validate.mjs

* fix(terraform): pin githubdeploy assume-role policy in import checker

Reject controlled role updates whose trust document is not the rendered
GitHub OIDC policy, matching the bucket-policy pin.
2026-09-18 14:30:20 -04:00

626 lines
24 KiB
Python
Executable file

#!/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_dev_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+= false")
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()