feat(hcp): flag workspaces that skip source-path file triggers (PLAT-183)

This commit is contained in:
Adam Moussa 2026-09-10 16:49:51 -04:00
parent fa940e69c6
commit 10f7c60991
No known key found for this signature in database
3 changed files with 417 additions and 1 deletions

View file

@ -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 `<stack>-<env>` 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 <stack>-<env>` (OAA). Trust stays exact `StringEquals`
on `iam-bootstrap-<env>` plus this one workspace. Never `StringLike`.

View file

@ -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())

View file

@ -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()