shoc-frontend-new/scripts/hcp-run-guard.py

#!/usr/bin/env python3
"""Guard HCP Terraform runs used by GitHub content CD.

Subcommands:
  check-and-discard  Refuse unsafe workspace settings. Discard a blocking
                     non-speculative VCS run so GitHub CD can create-run.
  reconcile-apply    Treat an HCP run whose status is already ``applied`` as
                     success when the GitHub apply-run step reported failure.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from typing import Any, Callable

API = "https://app.terraform.io/api/v2"
DEFAULT_WORKSPACE = "shoc-frontend-new-dev"
EXPECTED_TRIGGER_PATTERNS = [
    "terraform/live/dev/**",
    "terraform/live/modules/**",
]
DISCARDABLE = {
    "pending",
    "planned",
    "cost_estimated",
    "policy_checked",
    "policy_override",
}
APPLYING = {"applying", "apply_queued"}

HttpGet = Callable[[str], dict[str, Any]]
HttpPost = Callable[[str, dict[str, Any]], int]


class GuardError(Exception):
    """Refused to continue."""


def _headers(token: str) -> dict[str, str]:
    return {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/vnd.api+json",
    }


def default_get(token: str) -> HttpGet:
    def get(url: str) -> dict[str, Any]:
        request = urllib.request.Request(url, headers=_headers(token))
        with urllib.request.urlopen(request, timeout=30) as response:
            return json.load(response)

    return get


def default_post(token: str) -> HttpPost:
    def post(url: str, payload: dict[str, Any]) -> int:
        data = json.dumps(payload).encode()
        request = urllib.request.Request(
            url, data=data, method="POST", headers=_headers(token)
        )
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                return int(response.status)
        except urllib.error.HTTPError as exc:
            if exc.code in (409, 404):
                body = exc.read().decode("utf-8", "replace")
                print(f"discard returned HTTP {exc.code}: {body}")
                return exc.code
            raise

    return post


def require_token(token: str) -> str:
    if not token:
        raise GuardError("TF_API_TOKEN is required")
    return token


def check_invariants(attrs: dict[str, Any], workspace: str) -> None:
    if attrs.get("auto-apply") is True:
        raise GuardError(f"{workspace} auto-apply is on; refuse to continue")
    if not attrs.get("speculative-enabled"):
        raise GuardError("speculative plans are off; refuse to continue")
    if (attrs.get("vcs-repo") or {}).get("tags-regex"):
        raise GuardError("tag-based VCS triggering is set; refuse to continue")
    if attrs.get("trigger-patterns") != EXPECTED_TRIGGER_PATTERNS:
        raise GuardError(
            "trigger-patterns must be "
            f"{EXPECTED_TRIGGER_PATTERNS}; got {attrs.get('trigger-patterns')}"
        )


def check_and_discard(
    *,
    workspace: str,
    token: str,
    get: HttpGet | None = None,
    post: HttpPost | None = None,
) -> int:
    token = require_token(token)
    get = get or default_get(token)
    post = post or default_post(token)
    workspace_payload = get(
        f"{API}/organizations/seahaven/workspaces/{workspace}"
    )["data"]
    attrs = workspace_payload["attributes"]
    check_invariants(attrs, workspace)
    if not attrs.get("locked"):
        print("workspace is unlocked")
        return 0

    current = (
        workspace_payload.get("relationships", {})
        .get("current-run", {})
        .get("data")
    )
    if not current:
        raise GuardError("workspace is locked without a current run")
    run_id = current["id"]
    run = get(f"{API}/runs/{run_id}")["data"]
    run_attrs = run["attributes"]
    status = run_attrs.get("status")
    plan_only = run_attrs.get("plan-only")
    print(f"current run {run_id} status={status} plan-only={plan_only}")
    if plan_only:
        print("speculative run does not block GitHub CD")
        return 0
    if status in APPLYING:
        raise GuardError(f"{run_id} is {status}; wait, do not discard an apply")
    if status not in DISCARDABLE:
        raise GuardError(f"{run_id} status {status} is not discardable")
    code = post(
        f"{API}/runs/{run_id}/actions/discard",
        {
            "comment": (
                "Discarded so GitHub CD can create the content-release applyable run"
            )
        },
    )
    print(f"discarded {run_id} http={code}")
    return 0


def reconcile_apply(
    *,
    run_id: str,
    apply_outcome: str,
    token: str,
    get: HttpGet | None = None,
) -> int:
    token = require_token(token)
    if not run_id:
        raise GuardError("run id is required")
    if apply_outcome == "success":
        print("Apply succeeded.")
        return 0
    get = get or default_get(token)
    status = get(f"{API}/runs/{run_id}")["data"]["attributes"]["status"]
    print(f"HCP run {run_id} status={status}")
    if status == "applied":
        return 0
    raise GuardError(
        f"Apply failed: GitHub outcome={apply_outcome} HCP status={status}"
    )


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)

    check = sub.add_parser("check-and-discard")
    check.add_argument("--workspace", default=DEFAULT_WORKSPACE)
    check.add_argument("--token", default=os.environ.get("TF_API_TOKEN", ""))

    reconcile = sub.add_parser("reconcile-apply")
    reconcile.add_argument("--run-id", required=True)
    reconcile.add_argument(
        "--apply-outcome",
        default=os.environ.get("APPLY_OUTCOME", ""),
    )
    reconcile.add_argument("--token", default=os.environ.get("TF_API_TOKEN", ""))
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        if args.command == "check-and-discard":
            return check_and_discard(workspace=args.workspace, token=args.token)
        return reconcile_apply(
            run_id=args.run_id,
            apply_outcome=args.apply_outcome,
            token=args.token,
        )
    except GuardError as exc:
        print(str(exc), file=sys.stderr)
        return 1


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