mirror of
https://github.com/Sea-Haven-Industries/shoc-frontend-new.git
synced 2026-09-30 06:53:12 +00:00
628 lines
22 KiB
Python
628 lines
22 KiB
Python
#!/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 (
|
|
CONTROLLED_UPDATE_ADDRESSES,
|
|
ENVIRONMENT_CONFIG,
|
|
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"
|
|
DEPLOY_POLICY_ADDRESS = (
|
|
"module.environment_owned.aws_iam_role_policy.github_deploy"
|
|
)
|
|
DISTRIBUTION_ADDRESS = (
|
|
"module.environment_owned.aws_cloudfront_distribution.site"
|
|
)
|
|
ROLE_ADDRESS = "module.environment_owned.aws_iam_role.github_deploy"
|
|
TAG_UPDATE_ADDRESSES = CONTROLLED_UPDATE_ADDRESSES - {
|
|
BUCKET_POLICY_ADDRESS,
|
|
DEPLOY_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': <string>}}"]
|
|
|
|
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 _expected_pre_adoption_deploy_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}"
|
|
)
|
|
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 _canonical({"Version": "2012-10-17", "Statement": statements})
|
|
|
|
|
|
def _expected_deploy_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": [
|
|
{
|
|
"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 _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)
|
|
if address == BUCKET_POLICY_ADDRESS
|
|
else _expected_pre_adoption_deploy_policy(environment, distribution_id)
|
|
)
|
|
expected_after = (
|
|
_expected_bucket_policy(environment, distribution_id)
|
|
if address == BUCKET_POLICY_ADDRESS
|
|
else _expected_deploy_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 _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 in {BUCKET_POLICY_ADDRESS, DEPLOY_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("<unknown>: 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("<unknown>: 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")
|
|
|
|
if mode == "import":
|
|
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 != ["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())
|