mirror of
https://github.com/Sea-Haven-Industries/shoc-frontend-new.git
synced 2026-09-30 09:13:11 +00:00
505 lines
18 KiB
Python
505 lines
18 KiB
Python
#!/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]
|
|
if expected is not None:
|
|
return expected
|
|
suffix = address.rsplit(".", 1)[-1].replace("_", "-")
|
|
return f"tf-poc-generated-{suffix}"
|
|
|
|
|
|
def distribution_id(environment: str) -> str:
|
|
configured = ENVIRONMENT_CONFIG[environment]["distribution_id"]
|
|
return configured if isinstance(configured, str) else "ETFPOCGENERATED123"
|
|
|
|
|
|
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 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]:
|
|
after_policy = (
|
|
bucket_policy(environment)
|
|
if address == BUCKET_POLICY
|
|
else deploy_policy(environment)
|
|
)
|
|
before_policy = {"Version": "2012-10-17", "Statement": []}
|
|
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_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",
|
|
),
|
|
"staging": (
|
|
"local.hosted_zone_id",
|
|
"local.certificate_arn",
|
|
"local.github_oidc_arn",
|
|
"local.cache_policy_id",
|
|
),
|
|
"tf-poc": (
|
|
"var.hosted_zone_id",
|
|
"var.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 address in CONTROLLED_UPDATE_ADDRESSES:
|
|
with self.subTest(address=address):
|
|
self.assert_passes(
|
|
make_plan(
|
|
"tf-poc",
|
|
mode="controlled",
|
|
controlled_updates={address},
|
|
),
|
|
"tf-poc",
|
|
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_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()
|