mirror of
https://github.com/Sea-Haven-Industries/shoc-backend.git
synced 2026-09-30 07:13:12 +00:00
114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
|
|
#!/usr/bin/env python3
|
||
|
|
"""Reject unsafe actions in a live Terraform import plan."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from terraform_import_plan_resources import REQUIRED_RESOURCES
|
||
|
|
|
||
|
|
|
||
|
|
ALLOWED_MANAGED_TYPES = {
|
||
|
|
resource_type
|
||
|
|
for resources in REQUIRED_RESOURCES.values()
|
||
|
|
for resource_type in resources.values()
|
||
|
|
}
|
||
|
|
UNSAFE_ACTIONS = {"create", "delete"}
|
||
|
|
|
||
|
|
|
||
|
|
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.",
|
||
|
|
)
|
||
|
|
parser.add_argument(
|
||
|
|
"--allow-update-address",
|
||
|
|
action="append",
|
||
|
|
default=[],
|
||
|
|
metavar="ADDRESS",
|
||
|
|
help=(
|
||
|
|
"Allow an in-place update to this exact address after the initial "
|
||
|
|
"no-op import is proven. Repeat for each reviewed update."
|
||
|
|
),
|
||
|
|
)
|
||
|
|
return parser.parse_args()
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
args = parse_args()
|
||
|
|
plan = json.loads(args.plan_json.read_text(encoding="utf-8"))
|
||
|
|
violations: list[str] = []
|
||
|
|
managed = 0
|
||
|
|
updates = 0
|
||
|
|
allowed_update_addresses = set(args.allow_update_address)
|
||
|
|
seen_update_addresses: set[str] = set()
|
||
|
|
seen_addresses: set[str] = set()
|
||
|
|
required_resources = REQUIRED_RESOURCES[args.environment]
|
||
|
|
|
||
|
|
for resource in plan.get("resource_changes", []):
|
||
|
|
if resource.get("mode", "managed") != "managed":
|
||
|
|
continue
|
||
|
|
|
||
|
|
resource_type = resource.get("type", "")
|
||
|
|
address = resource.get("address", "<unknown>")
|
||
|
|
actions = set(resource.get("change", {}).get("actions", []))
|
||
|
|
managed += 1
|
||
|
|
seen_addresses.add(address)
|
||
|
|
|
||
|
|
if resource_type not in ALLOWED_MANAGED_TYPES:
|
||
|
|
violations.append(
|
||
|
|
f"{address}: managed type {resource_type!r} is outside the live ownership boundary"
|
||
|
|
)
|
||
|
|
|
||
|
|
expected_type = required_resources.get(address)
|
||
|
|
if expected_type is None:
|
||
|
|
violations.append(
|
||
|
|
f"{address}: managed address is outside the live ownership boundary"
|
||
|
|
)
|
||
|
|
elif resource_type != expected_type:
|
||
|
|
violations.append(
|
||
|
|
f"{address}: expected managed type {expected_type!r}, got {resource_type!r}"
|
||
|
|
)
|
||
|
|
|
||
|
|
unsafe = sorted(actions & UNSAFE_ACTIONS)
|
||
|
|
if unsafe:
|
||
|
|
violations.append(f"{address}: unsafe actions {unsafe}")
|
||
|
|
|
||
|
|
if "update" in actions:
|
||
|
|
updates += 1
|
||
|
|
seen_update_addresses.add(address)
|
||
|
|
if address not in allowed_update_addresses:
|
||
|
|
violations.append(
|
||
|
|
f"{address}: update is not explicitly allowlisted"
|
||
|
|
)
|
||
|
|
|
||
|
|
for unused in sorted(allowed_update_addresses - seen_update_addresses):
|
||
|
|
violations.append(f"{unused}: allowlisted update address is not updating")
|
||
|
|
|
||
|
|
for missing in sorted(set(required_resources) - seen_addresses):
|
||
|
|
violations.append(f"{missing}: required managed resource is absent")
|
||
|
|
|
||
|
|
if violations:
|
||
|
|
print("FAIL: live Terraform plan is not import-safe", file=sys.stderr)
|
||
|
|
for violation in violations:
|
||
|
|
print(f" - {violation}", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
|
||
|
|
mode = "controlled update" if allowed_update_addresses else "no-op import"
|
||
|
|
print(
|
||
|
|
f"PASS: {mode} plan has {managed} managed resources, "
|
||
|
|
f"{updates} updates, and no create/delete/replace actions"
|
||
|
|
)
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|