#!/usr/bin/env python3 """Flag HCP workspaces whose VCS file triggers omit packaged source paths. Lists workspaces in seahaven-mgmt, seahaven-prod, and seahaven-dev with file-triggers-enabled=true. Fails when trigger-prefixes and trigger-patterns are both empty and the repo working directory references source trees outside itself (Lambda src, functions, lambda, lambdas). PLAT-183. Token: TFE_TOKEN, TF_TOKEN_app_terraform_io, or ~/.terraform.d/credentials.tfrc.json (app.terraform.io). """ from __future__ import annotations import argparse import json import os import re import sys import urllib.request from pathlib import Path from typing import Any ORG = "seahaven" PROJECTS = ("seahaven-mgmt", "seahaven-prod", "seahaven-dev") API = "https://app.terraform.io/api/v2" OUTSIDE_SOURCE = re.compile( r"(?:\.\./(?:src|functions|lambda|lambdas)\b)" r"|(?:\$\{(?:ROOT|REPO|FUNCS)\}/(?:src|functions|lambda|lambdas)\b)" ) SCAN_SUFFIXES = {".tf", ".sh"} def load_token() -> str: for key in ("TFE_TOKEN", "TF_TOKEN_app_terraform_io"): value = os.environ.get(key) if value: return value creds = Path.home() / ".terraform.d" / "credentials.tfrc.json" if creds.is_file(): data = json.loads(creds.read_text()) token = data.get("credentials", {}).get("app.terraform.io", {}).get("token") if token: return token raise SystemExit( "no HCP token: set TFE_TOKEN or configure ~/.terraform.d/credentials.tfrc.json" ) def api_get(token: str, path: str) -> dict[str, Any]: req = urllib.request.Request( API + path, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/vnd.api+json", }, ) with urllib.request.urlopen(req) as resp: return json.load(resp) def paginate(token: str, path: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: items: list[dict[str, Any]] = [] included: list[dict[str, Any]] = [] while path: payload = api_get(token, path) items.extend(payload.get("data") or []) included.extend(payload.get("included") or []) nxt = (payload.get("links") or {}).get("next") if not nxt: break if nxt.startswith(API): path = nxt[len(API) :] elif "/api/v2" in nxt: path = nxt.split("/api/v2", 1)[1] else: path = nxt return items, included def working_dir_reads_outside(repo_path: Path, working_directory: str) -> list[str]: """Return relative files under the working directory that reference source trees.""" wd = working_directory.strip().strip("/") root = repo_path / wd if wd else repo_path if not root.is_dir(): return [] hits: list[str] = [] for path in root.rglob("*"): if not path.is_file() or path.suffix not in SCAN_SUFFIXES: continue if "build" in path.parts: continue text = path.read_text(errors="replace") if OUTSIDE_SOURCE.search(text): hits.append(str(path.relative_to(repo_path))) return hits def local_repo_path(repo_root: Path, identifier: str | None) -> Path | None: if not identifier or "/" not in identifier: return None name = identifier.split("/", 1)[1] candidate = repo_root / name return candidate if candidate.is_dir() else None def classify_workspace( attrs: dict[str, Any], project: str, repo_root: Path, ) -> dict[str, Any]: vcs = attrs.get("vcs-repo") or {} identifier = vcs.get("identifier") if isinstance(vcs, dict) else None prefixes = attrs.get("trigger-prefixes") or [] patterns = attrs.get("trigger-patterns") or [] wd = attrs.get("working-directory") or "" file_triggers = bool(attrs.get("file-triggers-enabled")) local = local_repo_path(repo_root, identifier) outside: list[str] = [] inspect = "skipped" if local is not None: outside = working_dir_reads_outside(local, wd) inspect = "ok" elif identifier: inspect = "missing-clone" empty_triggers = not prefixes and not patterns defect = bool(file_triggers and empty_triggers and outside) return { "name": attrs.get("name"), "project": project, "file_triggers": file_triggers, "working_directory": wd, "trigger_prefixes": prefixes, "trigger_patterns": patterns, "vcs": identifier, "inspect": inspect, "outside_refs": outside, "defect": defect, } def check( token: str, repo_root: Path, org: str = ORG, projects: tuple[str, ...] = PROJECTS, ) -> list[dict[str, Any]]: workspaces, included = paginate( token, f"/organizations/{org}/workspaces?page%5Bsize%5D=100&include=project" ) project_names = { item["id"]: item["attributes"]["name"] for item in included if item.get("type") == "projects" } rows: list[dict[str, Any]] = [] for workspace in workspaces: attrs = workspace["attributes"] if not attrs.get("file-triggers-enabled"): continue project_id = ( (((workspace.get("relationships") or {}).get("project") or {}).get("data") or {}).get( "id" ) ) project = project_names.get(project_id, "") if project not in projects: continue rows.append(classify_workspace(attrs, project, repo_root)) return rows def format_report(rows: list[dict[str, Any]]) -> str: lines = [ "workspace project prefixes/patterns outside-refs", "-" * 96, ] for row in sorted(rows, key=lambda item: (item["project"], item["name"] or "")): prefixes = row["trigger_prefixes"] patterns = row["trigger_patterns"] trigger = ",".join(prefixes or patterns) or "(empty)" mark = "FAIL" if row["defect"] else "ok" refs = ",".join(row["outside_refs"][:3]) or row["inspect"] lines.append( f"{mark:4} {row['name']:36} {row['project']:18} {trigger:18} {refs}" ) defects = [row for row in rows if row["defect"]] lines.append("") lines.append(f"file-triggered workspaces: {len(rows)} defects: {len(defects)}") if defects: lines.append( "empty trigger-prefixes and trigger-patterns while the working " "directory reads source trees outside itself:" ) for row in defects: lines.append(f" {row['name']}: {', '.join(row['outside_refs'])}") return "\n".join(lines) def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--repo-root", type=Path, default=None, help="Directory of GitHub clones named after the repo (default: parent of this repo)", ) parser.add_argument("--org", default=ORG) return parser.parse_args(argv) def default_repo_root() -> Path: return Path(__file__).resolve().parents[1].parent def main(argv: list[str] | None = None) -> int: args = parse_args(argv) repo_root = (args.repo_root or default_repo_root()).resolve() rows = check(load_token(), repo_root, org=args.org) print(format_report(rows)) return 1 if any(row["defect"] for row in rows) else 0 if __name__ == "__main__": sys.exit(main())