shoc-frontend-new/scripts/test-terraform-release-plan-check.py

#!/usr/bin/env python3
"""Deterministic tests for check-terraform-release-plan.py."""

from __future__ import annotations

import importlib.util
import io
import subprocess
import sys
import urllib.request
from email.message import EmailMessage
from pathlib import Path
from urllib.request import Request

SCRIPT = Path(__file__).with_name("check-terraform-release-plan.py")
FIXTURES = Path(__file__).with_name("testdata") / "terraform-release-plans"
EXPECTED_LABEL = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-2-1"
EXPECTED_PREVIOUS = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-1-1"
PLAN_ID = "plan-8F5JFydVYAmtTjET"
POINTER_ADDRESS = "module.environment_owned.aws_s3_object.release_pointer"


def run_case(
    fixture_name: str,
    *,
    expected_label: str = EXPECTED_LABEL,
    expected_previous: str = EXPECTED_PREVIOUS,
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [
            sys.executable,
            str(SCRIPT),
            str(FIXTURES / fixture_name),
            "--expected-version-label",
            expected_label,
            "--expected-previous-version-label",
            expected_previous,
        ],
        check=False,
        capture_output=True,
        text=True,
    )


class FakeResponse:
    def __init__(
        self,
        *,
        url: str,
        status: int,
        headers: dict[str, str] | None = None,
        body: bytes = b"",
    ) -> None:
        self.url = url
        self.status = status
        self.headers = headers or {}
        self._body = body

    def read(self) -> bytes:
        return self._body

    def close(self) -> None:
        return None


def load_check_module():
    spec = importlib.util.spec_from_file_location(
        "check_terraform_release_plan", SCRIPT
    )
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    spec.loader.exec_module(module)
    return module


def test_download_pinning() -> list[str]:
    module = load_check_module()
    fixture = (FIXTURES / "version-only.json").read_bytes()
    archive_url = "https://archivist.terraform.io/v1/object/example"
    calls: list[str] = []

    def fake_urlopen(request: Request, **_kwargs):
        url = request.full_url
        calls.append(url)
        host = request.host if hasattr(request, "host") else ""
        if url.startswith("https://app.terraform.io/api/v2/plans/"):
            if request.get_header("Authorization") != "Bearer test-token":
                raise AssertionError("API request is missing the bearer token")
            if "/runs" in url or "/apply" in url or "/discard" in url:
                raise AssertionError(f"download contacted a run-control path: {url}")
            return FakeResponse(
                url=url,
                status=307,
                headers={"Location": archive_url},
            )
        if url == archive_url:
            if request.get_header("Authorization"):
                raise AssertionError("archivist request must not send TF_API_TOKEN")
            return FakeResponse(url=url, status=200, body=fixture)
        raise AssertionError(f"unexpected URL {url} host={host}")

    plan = module.download_plan_json(PLAN_ID, "test-token", urlopen=fake_urlopen)
    failures: list[str] = []
    addresses = [item["address"] for item in plan["resource_changes"]]
    if POINTER_ADDRESS not in addresses:
        failures.append("download did not return the version-only fixture")
    if calls != [
        f"https://app.terraform.io/api/v2/plans/{PLAN_ID}/json-output",
        archive_url,
    ]:
        failures.append(f"download URLs were {calls}")

    try:
        module.download_plan_json("run-not-a-plan", "test-token", urlopen=fake_urlopen)
        failures.append("invalid plan id was accepted")
    except ValueError:
        pass

    def redirect_elsewhere(request: Request, **_kwargs):
        return FakeResponse(
            url=request.full_url,
            status=307,
            headers={"Location": "https://evil.example/plan.json"},
        )

    try:
        module.download_plan_json(PLAN_ID, "test-token", urlopen=redirect_elsewhere)
        failures.append("redirect to a non-archivist host was accepted")
    except ValueError:
        pass

    def double_redirect(request: Request, **_kwargs):
        if request.full_url.startswith("https://app.terraform.io/"):
            return FakeResponse(
                url=request.full_url,
                status=307,
                headers={"Location": archive_url},
            )
        return FakeResponse(
            url=request.full_url,
            status=307,
            headers={"Location": "https://archivist.terraform.io/v1/object/other"},
        )

    try:
        module.download_plan_json(PLAN_ID, "test-token", urlopen=double_redirect)
        failures.append("second archivist redirect was accepted")
    except ValueError:
        pass

    def not_ready(request: Request, **_kwargs):
        return FakeResponse(url=request.full_url, status=204)

    try:
        module.download_plan_json(PLAN_ID, "test-token", urlopen=not_ready)
        failures.append("HTTP 204 was polled or accepted")
    except ValueError as exc:
        if "poll" not in str(exc):
            failures.append(f"HTTP 204 error was {exc}")

    source = SCRIPT.read_text(encoding="utf-8")
    for banned in ("/apply", "/discard", "/runs"):
        if banned in source:
            failures.append(f"download client contains run-control path {banned}")

    return failures


def _scripted_https_handler(fixture: bytes, archive_url: str):
    calls: list[str] = []
    api_prefix = "https://app.terraform.io/api/v2/plans/"

    class ScriptedHTTPSHandler(urllib.request.BaseHandler):
        handler_order = 100

        def https_open(self, req: Request):
            url = req.full_url
            calls.append(url)
            headers = EmailMessage()
            if url.startswith(api_prefix):
                headers["Location"] = archive_url
                body = b""
                status = 307
                msg = "Temporary Redirect"
            elif url == archive_url:
                body = fixture
                status = 200
                msg = "OK"
            else:
                raise AssertionError(f"unexpected URL {url}")
            response = urllib.response.addinfourl(
                io.BytesIO(body),
                headers,
                url,
                code=status,
            )
            response.msg = msg
            return response

    return ScriptedHTTPSHandler(), calls


def test_download_standard_opener_redirect() -> list[str]:
    """urllib follows the HCP 307; the guard must still inspect that first hop."""
    module = load_check_module()
    fixture = (FIXTURES / "version-only.json").read_bytes()
    archive_url = "https://archivist.terraform.io/v1/object/example"
    api_url = f"https://app.terraform.io/api/v2/plans/{PLAN_ID}/json-output"
    failures: list[str] = []

    following_handler, following_calls = _scripted_https_handler(fixture, archive_url)
    followed = urllib.request.build_opener(following_handler).open(api_url)
    try:
        if followed.status != 200:
            failures.append(
                f"standard opener first status was {followed.status}, not 200"
            )
        if following_calls != [api_url, archive_url]:
            failures.append(f"standard opener URLs were {following_calls}")
    finally:
        followed.close()

    guard_handler, guard_calls = _scripted_https_handler(fixture, archive_url)
    try:
        plan = module.download_plan_json(
            PLAN_ID,
            "test-token",
            handlers=(guard_handler,),
        )
    except ValueError as exc:
        failures.append(f"no-redirect download failed: {exc}")
        return failures

    addresses = [item["address"] for item in plan["resource_changes"]]
    if POINTER_ADDRESS not in addresses:
        failures.append("no-redirect download did not return the version-only fixture")
    if guard_calls != [api_url, archive_url]:
        failures.append(f"no-redirect download URLs were {guard_calls}")

    following_urlopen_handler, _ = _scripted_https_handler(fixture, archive_url)
    following_urlopen = urllib.request.build_opener(following_urlopen_handler).open
    try:
        module.download_plan_json(
            PLAN_ID,
            "test-token",
            urlopen=following_urlopen,
        )
        failures.append("redirect-following urlopen was accepted as the first hop")
    except ValueError as exc:
        if "expected a redirect" not in str(exc):
            failures.append(f"following urlopen error was {exc}")

    return failures


def test_deploy_workflow_uses_script_flags() -> list[str]:
    workflow = (
        Path(__file__).resolve().parents[1] / ".github/workflows/deploy.yml"
    ).read_text(encoding="utf-8")
    failures: list[str] = []
    if workflow.count("--expected-version-label") < 2:
        failures.append(
            "deploy.yml must pass --expected-version-label on release and rollback"
        )
    if workflow.count("--expected-previous-version-label") < 2:
        failures.append(
            "deploy.yml must pass --expected-previous-version-label on release and rollback"
        )
    for forbidden in (
        "--expected-current-label",
        "--expected-previous-label",
        "--before-current-label",
        "--before-previous-label",
        "--current-origin-id",
        "--previous-origin-id",
        "CURRENT_ORIGIN_ID",
    ):
        if forbidden in workflow:
            failures.append(f"deploy.yml still passes unknown flag {forbidden}")
    return failures


def test_deploy_workflow_confirms_prefix_with_head_object() -> list[str]:
    workflow = (
        Path(__file__).resolve().parents[1] / ".github/workflows/deploy.yml"
    ).read_text(encoding="utf-8")
    failures: list[str] = []
    if "aws s3api head-object" not in workflow:
        failures.append(
            "deploy.yml must confirm the uploaded index.html with s3api head-object"
        )
    if "aws s3 ls" in workflow:
        failures.append("deploy.yml must not list the prefix with aws s3 ls")
    if any(
        line.lstrip().startswith("run:") and "npm run verify" in line
        for line in workflow.splitlines()
    ):
        failures.append(
            "deploy.yml must not re-run npm run verify; Frontend checks owns that gate"
        )
    if any(line.lstrip().startswith("pull_request:") for line in workflow.splitlines()):
        failures.append(
            "deploy.yml must not run on pull_request; Frontend checks owns PR verify"
        )
    return failures


def main() -> int:
    cases = [
        ("version-only", run_case("version-only.json"), 0),
        (
            "origin-timeout-normalization",
            run_case("origin-timeout-normalization.json"),
            0,
        ),
        ("origin-timeout-change", run_case("origin-timeout-change.json"), 1),
        ("wrong-label", run_case("wrong-label.json"), 1),
        ("wrong-before", run_case("wrong-before.json"), 1),
        ("extra-origin-change", run_case("extra-origin-change.json"), 1),
        ("iam-update", run_case("iam-update.json"), 1),
        ("dns-update", run_case("dns-update.json"), 1),
        ("create", run_case("create.json"), 1),
        ("delete", run_case("delete.json"), 1),
        ("replace", run_case("replace.json"), 1),
        ("multiple-updates", run_case("multiple-updates.json"), 1),
        ("nested-unknown", run_case("nested-unknown.json"), 1),
        ("unknown-only", run_case("unknown-only.json"), 1),
        ("empty", run_case("empty.json"), 1),
        ("missing-action", run_case("missing-action.json"), 1),
        ("extra-action", run_case("extra-action.json"), 1),
    ]
    failures = [
        (name, result, expected)
        for name, result, expected in cases
        if result.returncode != expected
    ]
    download_failures = test_download_pinning()
    redirect_failures = test_download_standard_opener_redirect()
    download_failures.extend(redirect_failures)
    download_failures.extend(test_deploy_workflow_uses_script_flags())
    download_failures.extend(test_deploy_workflow_confirms_prefix_with_head_object())
    if failures or download_failures:
        if failures:
            print(
                "FAIL: release plan-check cases failed: "
                + ", ".join(name for name, _, _ in failures),
                file=sys.stderr,
            )
            for name, result, expected in failures:
                print(
                    f"{name}: expected {expected}, got {result.returncode}\n"
                    f"{result.stdout}{result.stderr}",
                    file=sys.stderr,
                )
        for item in download_failures:
            print(f"FAIL: {item}", file=sys.stderr)
        return 1
    print("PASS: Terraform release plan safety checks")
    return 0


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