diff --git a/README.md b/README.md index 676273b..6652dc3 100644 --- a/README.md +++ b/README.md @@ -470,6 +470,18 @@ Default execution mode Remote. Never use HCP's "Quick setup AWS dynamic credentials" button — it writes the single `TFC_AWS_RUN_ROLE_ARN`, which collapses the plan/apply role split. Never project-scoped variable sets. +**VCS file triggers (PLAT-183).** Keep `file-triggers-enabled`. If the plan +packages Lambda source or otherwise reads files outside the working +directory, set `trigger-prefixes` or `trigger-patterns` to those paths. +`trigger-prefixes` add to the working directory; `trigger-patterns` replace +it and must include the working-directory glob. Do not disable file triggers +to make source-only merges queue; docs-only commits must not apply prod. +Record the live prefixes or patterns in the app README next to the workspace +name. Checklist: engineering-handbook +[aws-infrastructure.md](https://github.com/Sea-Haven-Industries/engineering-handbook/blob/main/aws-infrastructure.md#hcp-terraform-vcs-file-triggers). +Audit: `python3 scripts/check_hcp_workspace_triggers.py` (org `seahaven`, +projects `seahaven-mgmt`, `seahaven-prod`, `seahaven-dev`). + **Two-principal model (prod/dev HCP, PLAT-149).** Replace enumerated `StringEquals` on `iam:PermissionsBoundary` with a factory vs scoped split. @@ -515,7 +527,13 @@ on the SAM guardrail. Do not add `ArnLike` to deploy-substrate. 2. HCP workspace `-` exists. Auto-apply off. Vars still empty. IAM lives in this workspace's state, so this workspace must assume `hcptf-bootstrap` for the first apply. Default bootstrap trust does not - include it. Never a project-scoped variable set. + include it. Never a project-scoped variable set. Enable VCS file triggers + before the first merge to the tracked branch. If `terraform/` (or the + working directory) packages `src/`, `functions/`, `lambda/`, or + `lambdas/`, list those paths in `trigger-prefixes` or `trigger-patterns` + and write them in the app README next to the workspace name. See + engineering-handbook + [HCP Terraform VCS file triggers](https://github.com/Sea-Haven-Industries/engineering-handbook/blob/main/aws-infrastructure.md#hcp-terraform-vcs-file-triggers). 3. `scripts/create-hcptf-bootstrap-roles.sh --account prod|dev --allow-workspace -` (OAA). Trust stays exact `StringEquals` on `iam-bootstrap-` plus this one workspace. Never `StringLike`. diff --git a/scripts/check_hcp_workspace_triggers.py b/scripts/check_hcp_workspace_triggers.py new file mode 100755 index 0000000..7bbd68c --- /dev/null +++ b/scripts/check_hcp_workspace_triggers.py @@ -0,0 +1,225 @@ +#!/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()) diff --git a/scripts/test_check_hcp_workspace_triggers.py b/scripts/test_check_hcp_workspace_triggers.py new file mode 100644 index 0000000..a52c91e --- /dev/null +++ b/scripts/test_check_hcp_workspace_triggers.py @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 +"""Tests for check_hcp_workspace_triggers.py (stdlib unittest, no live HCP).""" +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +SCRIPTS = Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPTS)) +import check_hcp_workspace_triggers as chk # noqa: E402 + + +def _ws( + name: str, + project_id: str, + *, + file_triggers: bool = True, + prefixes: list[str] | None = None, + patterns: list[str] | None = None, + wd: str = "terraform", + identifier: str | None = "Sea-Haven-Industries/sample", +) -> dict: + vcs = {"identifier": identifier} if identifier else None + return { + "attributes": { + "name": name, + "file-triggers-enabled": file_triggers, + "trigger-prefixes": prefixes or [], + "trigger-patterns": patterns or [], + "working-directory": wd, + "vcs-repo": vcs, + }, + "relationships": {"project": {"data": {"id": project_id}}}, + } + + +class WorkingDirScanTests(unittest.TestCase): + def test_detects_relative_src_in_tf(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + repo = Path(tmp) + tf = repo / "terraform" + tf.mkdir() + (tf / "lambda.tf").write_text( + 'source_dir = "${path.module}/../src/shared"\n' + ) + hits = chk.working_dir_reads_outside(repo, "terraform") + self.assertEqual(hits, ["terraform/lambda.tf"]) + + def test_detects_repo_src_in_build_script(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + repo = Path(tmp) + tf = repo / "terraform" + tf.mkdir() + (tf / "build_packages.sh").write_text( + 'SRC="$(cd "${ROOT}/../src" && pwd)"\n' + ) + hits = chk.working_dir_reads_outside(repo, "terraform") + self.assertEqual(hits, ["terraform/build_packages.sh"]) + + def test_ignores_terraform_only_tree(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + repo = Path(tmp) + tf = repo / "terraform" + tf.mkdir() + (tf / "s3.tf").write_text('resource "aws_s3_bucket" "x" {}\n') + self.assertEqual(chk.working_dir_reads_outside(repo, "terraform"), []) + + +class ClassifyTests(unittest.TestCase): + def test_empty_triggers_with_outside_refs_is_defect(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "sample" + (repo / "terraform").mkdir(parents=True) + (repo / "terraform" / "lambda.tf").write_text( + 'source_dir = "${path.module}/../src/app"\n' + ) + row = chk.classify_workspace( + _ws("sample-prod", "p1")["attributes"], + "seahaven-prod", + root, + ) + self.assertTrue(row["defect"]) + + def test_empty_triggers_without_outside_refs_is_ok(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "sample" + (repo / "terraform").mkdir(parents=True) + (repo / "terraform" / "s3.tf").write_text("resource aws_s3_bucket x {}\n") + row = chk.classify_workspace( + _ws("sample-prod", "p1")["attributes"], + "seahaven-prod", + root, + ) + self.assertFalse(row["defect"]) + + def test_prefixes_cover_outside_refs(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "sample" + (repo / "terraform").mkdir(parents=True) + (repo / "terraform" / "lambda.tf").write_text( + 'source_dir = "${path.module}/../src/app"\n' + ) + row = chk.classify_workspace( + _ws("sample-prod", "p1", prefixes=["terraform", "src"])["attributes"], + "seahaven-prod", + root, + ) + self.assertFalse(row["defect"]) + self.assertTrue(row["outside_refs"]) + + def test_missing_clone_does_not_fail(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + row = chk.classify_workspace( + _ws("sample-prod", "p1")["attributes"], + "seahaven-prod", + Path(tmp), + ) + self.assertEqual(row["inspect"], "missing-clone") + self.assertFalse(row["defect"]) + + +class CheckFilterTests(unittest.TestCase): + def test_skips_disabled_file_triggers_and_other_projects(self) -> None: + payload = { + "data": [ + _ws("cli-only", "prod", file_triggers=False), + _ws("shoc-dev", "ext"), + _ws("app-prod", "prod", patterns=["terraform/**/*", "src/**/*"]), + ], + "included": [ + {"id": "prod", "type": "projects", "attributes": {"name": "seahaven-prod"}}, + { + "id": "ext", + "type": "projects", + "attributes": {"name": "seahaven-external-dev"}, + }, + ], + "links": {"next": None}, + } + with mock.patch.object(chk, "api_get", return_value=payload): + with tempfile.TemporaryDirectory() as tmp: + rows = chk.check("token", Path(tmp)) + names = [row["name"] for row in rows] + self.assertEqual(names, ["app-prod"]) + + +class ReportTests(unittest.TestCase): + def test_format_includes_defect_count(self) -> None: + text = chk.format_report( + [ + { + "name": "bad-prod", + "project": "seahaven-prod", + "trigger_prefixes": [], + "trigger_patterns": [], + "outside_refs": ["terraform/lambda.tf"], + "inspect": "ok", + "defect": True, + } + ] + ) + self.assertIn("defects: 1", text) + self.assertIn("bad-prod", text) + + +if __name__ == "__main__": + unittest.main()