shoc-backend/scripts/check-terraform-release-plan.py

#!/usr/bin/env python3
"""Reject HCP Terraform plans that are not a version-only Elastic Beanstalk update.

This script may read a local plan JSON file or download plan JSON from the
documented HashiCorp endpoint:

    GET https://app.terraform.io/api/v2/plans/:id/json-output

The download follows exactly one redirect, and only to archivist.terraform.io.
It does not create, apply, discard, or poll runs.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import ssl
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Callable
from urllib.parse import urlparse


RELEASE_ADDRESS = "module.environment.aws_elastic_beanstalk_environment.this"
API_HOST = "app.terraform.io"
ARCHIVE_HOST = "archivist.terraform.io"
PLAN_ID_RE = re.compile(r"^plan-[A-Za-z0-9]+$")
VERSION_LABEL_RE = re.compile(r"^[0-9a-f]{40}-[0-9]+-[0-9]+$")
IGNORED_ACTIONS = {"no-op", "read"}
UNSAFE_ACTIONS = {"create", "delete"}
# Wholly unknown computed attributes may be ignored. Nested unknowns on any
# other attribute are treated as changes so the version-only guard fails closed.
COMPUTED_UNKNOWN_ATTRIBUTES = frozenset({"instances", "load_balancers"})
REDIRECT_STATUSES = {301, 302, 303, 307, 308}

UrlOpen = Callable[..., Any]


class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
    """Return the redirect response instead of following it."""

    def http_error_301(self, req, fp, code, msg, headers):
        return self._capture(req, fp, code, headers)

    http_error_302 = http_error_303 = http_error_307 = http_error_308 = http_error_301

    @staticmethod
    def _capture(req, fp, code, headers):
        response = urllib.response.addinfourl(fp, headers, req.full_url, code=code)
        response.msg = "Redirect"
        return response


def _urlopen_without_redirects(
    *handlers: urllib.request.BaseHandler,
) -> UrlOpen:
    context = ssl.create_default_context()
    opener = urllib.request.build_opener(
        urllib.request.HTTPSHandler(context=context),
        _NoRedirectHandler,
        *handlers,
    )
    return opener.open


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument(
        "plan_json",
        type=Path,
        nargs="?",
        help="Local Terraform plan JSON. Mutually exclusive with --plan-id.",
    )
    source.add_argument(
        "--plan-id",
        help="HCP Terraform plan ID. Downloads JSON from app.terraform.io.",
    )
    parser.add_argument(
        "--expected-version-label",
        required=True,
        help="Immutable application version the plan must apply.",
    )
    parser.add_argument(
        "--evidence-out",
        type=Path,
        help="Write machine-readable proof after every assertion passes.",
    )
    return parser.parse_args()


def download_plan_json(
    plan_id: str,
    token: str,
    *,
    urlopen: UrlOpen | None = None,
    handlers: tuple[urllib.request.BaseHandler, ...] = (),
) -> dict[str, Any]:
    if not PLAN_ID_RE.fullmatch(plan_id):
        raise ValueError(f"plan id {plan_id!r} is not a valid HCP plan id")
    if not token:
        raise ValueError("TF_API_TOKEN is required to download plan JSON")

    opener = urlopen or _urlopen_without_redirects(*handlers)
    api_url = f"https://{API_HOST}/api/v2/plans/{plan_id}/json-output"
    request = urllib.request.Request(
        api_url,
        method="GET",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/vnd.api+json",
            "Accept": "application/json",
        },
    )
    first = _open_pinned(opener, request, allowed_host=API_HOST)
    try:
        if first.status == 204:
            raise ValueError(
                "plan JSON is not ready; refusing to poll the plans endpoint"
            )
        if first.status not in REDIRECT_STATUSES:
            raise ValueError(
                f"expected a redirect from {API_HOST}, got HTTP {first.status}"
            )
        location = first.headers.get("Location")
        if not location:
            raise ValueError(f"{API_HOST} redirect is missing a Location header")
        archive = urlparse(location)
        if archive.scheme != "https" or archive.hostname != ARCHIVE_HOST:
            raise ValueError(
                "refusing redirect that is not https://"
                f"{ARCHIVE_HOST}/"
            )
        archive_request = urllib.request.Request(location, method="GET")
        second = _open_pinned(opener, archive_request, allowed_host=ARCHIVE_HOST)
        try:
            if second.status in REDIRECT_STATUSES:
                raise ValueError(
                    f"refusing a second redirect from {ARCHIVE_HOST}"
                )
            if second.status != 200:
                raise ValueError(
                    f"plan JSON download from {ARCHIVE_HOST} returned "
                    f"HTTP {second.status}"
                )
            payload = second.read()
        finally:
            second.close()
    finally:
        first.close()

    plan = json.loads(payload.decode("utf-8"))
    if not isinstance(plan, dict):
        raise ValueError("plan JSON must be an object")
    return plan


def _open_pinned(urlopen: UrlOpen, request: urllib.request.Request, *, allowed_host: str):
    parsed = urlparse(request.full_url)
    if parsed.scheme != "https" or parsed.hostname != allowed_host:
        raise ValueError(
            f"refusing to contact {parsed.scheme}://{parsed.hostname} "
            f"(pinned host is {allowed_host})"
        )
    context = ssl.create_default_context()
    try:
        return urlopen(request, context=context, timeout=30)
    except TypeError:
        return urlopen(request, timeout=30)


def _is_nested_unknown(value: Any) -> bool:
    if isinstance(value, dict):
        return any(item is True or _is_nested_unknown(item) for item in value.values())
    if isinstance(value, list):
        return any(item is True or _is_nested_unknown(item) for item in value)
    return False


def changed_attributes(change: dict[str, Any]) -> set[str]:
    before = change.get("before") or {}
    after = change.get("after") or {}
    unknown = change.get("after_unknown") or {}
    keys = set(before) | set(after) | set(unknown)
    changed: set[str] = set()
    for key in keys:
        unknown_value = unknown.get(key)
        if unknown_value is True:
            if key in COMPUTED_UNKNOWN_ATTRIBUTES:
                continue
            changed.add(key)
            continue
        if _is_nested_unknown(unknown_value):
            changed.add(key)
            continue
        if before.get(key) != after.get(key):
            changed.add(key)
    return changed


def validate_plan(plan: dict[str, Any], expected_label: str) -> list[str]:
    violations: list[str] = []
    if not VERSION_LABEL_RE.fullmatch(expected_label):
        violations.append(
            "expected version label must be <full-sha>-<run-id>-<attempt>"
        )
        return violations

    updates: list[dict[str, Any]] = []
    for resource in plan.get("resource_changes", []):
        if resource.get("mode", "managed") != "managed":
            continue
        address = resource.get("address", "<unknown>")
        change = resource.get("change") or {}
        actions = list(change.get("actions") or [])
        action_set = set(actions)
        if action_set <= IGNORED_ACTIONS:
            continue

        if change.get("importing"):
            violations.append(f"{address}: import actions are not allowed")

        unsafe = sorted(action_set & UNSAFE_ACTIONS)
        if unsafe:
            violations.append(f"{address}: unsafe actions {unsafe}")
        if "replace" in action_set or actions in (
            ["delete", "create"],
            ["create", "delete"],
        ):
            violations.append(f"{address}: replacement is not allowed")

        if "update" in action_set:
            updates.append(resource)
            if action_set != {"update"}:
                violations.append(
                    f"{address}: update must be the only action, got {actions}"
                )

        if address != RELEASE_ADDRESS and action_set - IGNORED_ACTIONS:
            violations.append(
                f"{address}: managed address is outside the version-only release"
            )

    if len(updates) != 1:
        violations.append(
            f"expected exactly one managed update, found {len(updates)}"
        )
        return violations

    resource = updates[0]
    address = resource.get("address", "<unknown>")
    if address != RELEASE_ADDRESS:
        violations.append(
            f"{address}: expected update address {RELEASE_ADDRESS}"
        )
        return violations

    change = resource.get("change") or {}
    changed = changed_attributes(change)
    if changed != {"version_label"}:
        violations.append(
            f"{address}: expected only version_label to change, found "
            f"{sorted(changed) if changed else 'no attribute changes'}"
        )

    after = change.get("after") or {}
    actual = after.get("version_label")
    if actual != expected_label:
        violations.append(
            f"{address}: after version_label {actual!r} does not match "
            f"{expected_label!r}"
        )

    unknown = change.get("after_unknown") or {}
    if unknown.get("version_label") is True:
        violations.append(f"{address}: version_label after value is unknown")

    return violations


def main() -> int:
    args = parse_args()
    if args.plan_id:
        try:
            plan = download_plan_json(args.plan_id, os.environ.get("TF_API_TOKEN", ""))
        except (OSError, ValueError, json.JSONDecodeError, urllib.error.URLError) as exc:
            print(f"FAIL: could not download plan JSON: {exc}", file=sys.stderr)
            return 1
    else:
        if args.plan_json is None:
            print("FAIL: plan JSON path or --plan-id is required", file=sys.stderr)
            return 1
        plan = json.loads(args.plan_json.read_text(encoding="utf-8"))

    violations = validate_plan(plan, args.expected_version_label)
    if violations:
        print("FAIL: Terraform plan is not a version-only release", file=sys.stderr)
        for violation in violations:
            print(f"  - {violation}", file=sys.stderr)
        return 1

    if args.evidence_out:
        evidence = {
            "address": RELEASE_ADDRESS,
            "expected_version_label": args.expected_version_label,
            "managed_updates": 1,
            "changed_attributes": ["version_label"],
            "creates": 0,
            "deletes": 0,
            "replacements": 0,
        }
        args.evidence_out.write_text(
            json.dumps(evidence, indent=2, sort_keys=True) + "\n",
            encoding="utf-8",
        )
    print(
        "PASS: version-only plan updates "
        f"{RELEASE_ADDRESS} version_label to {args.expected_version_label}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())