"""plan-survey reads metadata only, never prints a token, and fails loudly.""" import datetime as dt import importlib.util import json import sys from pathlib import Path import pytest spec = importlib.util.spec_from_file_location("plan_survey", Path(__file__).parents[1] / "scripts/plan-survey.py") survey = importlib.util.module_from_spec(spec) sys.modules["plan_survey"] = survey spec.loader.exec_module(survey) TOKEN = "hvs.SYNTHETICgrantTOKENvalue0123456789" NOW = dt.datetime(2026, 10, 5, 12, 0, tzinfo=dt.timezone.utc) ARGS = ["--kv-path", "platform/workloads/inbound-library/imap", "--policy", "workload-kv-read-inbound-library-imap", "--approle", "inbound-library", "--consumer", "inbound-library", "--keyword", "imap"] CATALOG = """ version: 1 entries: - id: inbound-library-imap title: inbound-library IMAP mailbox need_keywords: [inbound-library, imap] owner_repo: ops-mason path_template: platform/workloads/inbound-library/imap status: draft - id: npm-publish-lane title: something else need_keywords: [npm] owner_repo: railiance-platform status: active """ POLICY_GRANTING = '''path "platform/data/workloads/inbound-library/imap" { capabilities = ["read"] } path "platform/metadata/workloads/inbound-library/imap" { capabilities = ["read"] } ''' def lookup(policies=("default", "ops-mason-survey")): return {"data": {"id": TOKEN, "accessor": "acc-SECRET", "display_name": "token-claude-mason-wp-0008", "policies": list(policies), "ttl": 2000}} class FakeBao: """Records every argv; answers from a table keyed by positional words.""" def __init__(self, table=None, denied=(), missing=()): self.calls = [] self.table = { ("token", "lookup"): lookup(), ("secrets", "list"): {"platform/": {"type": "kv", "options": {"version": "2"}, "description": "workloads"}, "sys/": {"type": "system"}}, ("auth", "list"): {"approle/": {"type": "approle"}, "kubernetes/": {"type": "kubernetes"}, "token/": {"type": "token"}}, ("policy", "list"): ["default", "root", "ops-mason-survey", "workload-kv-read-inbound-library-imap", "workload-kv-read-other"], ("list", "auth/approle/role"): ["inbound-library", "rein-openweights"], ("list", "auth/kubernetes/role"): ["external-secrets-activity-core"], ("read", "auth/approle/role/inbound-library"): {"data": { "token_policies": ["workload-kv-read-inbound-library-imap"], "token_ttl": 900}}, ("policy", "read", "workload-kv-read-inbound-library-imap"): POLICY_GRANTING, ("kv", "metadata", "get", "platform/workloads/inbound-library/imap"): {"data": { "current_version": 1, "created_time": "2026-10-01T00:00:00Z", "custom_metadata": {"description": "IMAP login for inbound-library"}}}, ("kv", "list", "platform/workloads/inbound-library"): ["imap"], } self.table.update(table or {}) self.denied, self.missing = set(denied), set(missing) def __call__(self, argv, env): assert argv[0] == "bao" assert env["BAO_TOKEN"] == TOKEN self.calls.append(argv) key = tuple(a for a in argv[1:] if not a.startswith("-")) if key in self.denied: return 2, "", "Error: permission denied" if key in self.missing or key not in self.table: return 2, "", "No value found" value = self.table[key] return 0, value if isinstance(value, str) else json.dumps(value), "" @pytest.fixture def catalog(tmp_path): p = tmp_path / "catalog.yaml" p.write_text(CATALOG) return p def run(fake, catalog, capsys, args=ARGS): rc = survey.main([*args, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN, "BAO_ADDR": "http://127.0.0.1:18200"}, now=NOW) return rc, capsys.readouterr().out def assert_metadata_only(calls): for argv in calls: pos = [a for a in argv[1:] if not a.startswith("-")] assert not (pos[:1] == ["kv"] and "get" in pos and "metadata" not in pos), argv assert pos[:2] != ["kv", "get"], argv assert not any("/data/" in a or a.startswith("data/") for a in pos), argv assert not any(seg in ("role-id", "secret-id") for a in pos for seg in a.split("/")), argv def test_existing_lane_found_and_only_metadata_read(catalog, capsys): fake = FakeBao() rc, out = run(fake, catalog, capsys) assert rc == 0 assert_metadata_only(fake.calls) assert "## 2. Existing-structure survey" in out assert "**Draft answer: yes.**" in out assert "token-claude-mason-wp-0008" in out and "BAO_ADDR=http://127.0.0.1:18200" in out assert "2026-10-05T12:00:00Z" in out assert "(5 policies, 1 relevant)" in out assert "inbound-library-imap owner=ops-mason" in out and "npm-publish-lane" not in out assert "rein-openweights" not in out assert TOKEN not in out and "acc-SECRET" not in out def test_no_existing_lane(catalog, capsys, tmp_path): empty = tmp_path / "empty.yaml" empty.write_text("version: 1\nentries: []\n") fake = FakeBao(table={("policy", "list"): ["default"], ("list", "auth/approle/role"): ["other"]}, missing={("kv", "metadata", "get", "platform/workloads/inbound-library/imap"), ("kv", "list", "platform/workloads/inbound-library")}) rc, out = run(fake, empty, capsys) assert rc == 0 assert "**Draft answer: no.**" in out assert "no metadata (path unused)" in out assert_metadata_only(fake.calls) def test_policy_without_bound_role_is_partial(catalog, capsys): fake = FakeBao(table={("list", "auth/approle/role"): ["other"]}) rc, out = run(fake, catalog, capsys) assert "**Draft answer: partial.**" in out assert "no surveyed role binds it" in out def test_missing_capability_degrades_and_is_reported(catalog, capsys): fake = FakeBao(denied={("list", "auth/approle/role"), ("list", "auth/kubernetes/role"), ("policy", "read", "workload-kv-read-inbound-library-imap")}) rc, out = run(fake, catalog, capsys) assert rc == 0 assert "not permitted by ops-mason-survey" in out assert "incomplete:" in out def test_no_grant_fails_loudly(monkeypatch, tmp_path): monkeypatch.setattr(survey.os.path, "expanduser", lambda p: str(tmp_path / "absent")) with pytest.raises(survey.SurveyError, match="no ops-mason-survey grant"): survey.main(ARGS, runner=FakeBao(), environ={}) @pytest.mark.parametrize("policies", [("root",), ("default", "ops-mason-survey", "platform-admin"), ("default",)]) def test_non_grant_token_is_refused(catalog, policies): fake = FakeBao(table={("token", "lookup"): lookup(policies)}) with pytest.raises(survey.SurveyError, match="not an ops-mason-survey grant") as exc: survey.main([*ARGS, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN}) assert TOKEN not in str(exc.value) assert len(fake.calls) == 1 def test_expired_grant_fails_loudly(catalog): fake = FakeBao(denied={("token", "lookup")}) with pytest.raises(survey.SurveyError, match="lookup failed"): survey.main([*ARGS, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN}) def test_script_exits_nonzero_without_grant(tmp_path): import os import subprocess env = {k: v for k, v in os.environ.items() if k not in ("BAO_TOKEN", "VAULT_TOKEN")} env["HOME"] = str(tmp_path) p = subprocess.run([sys.executable, str(Path(__file__).parents[1] / "scripts/plan-survey.py"), *ARGS], capture_output=True, text=True, env=env) assert p.returncode == 2 assert "ERROR: no ops-mason-survey grant" in p.stderr @pytest.mark.parametrize("args", [ ["kv", "get", "platform/workloads/x"], ["kv", "get", "-field=password", "platform/workloads/x"], ["read", "platform/data/workloads/x"], ["list", "platform/data/workloads"], ["kv", "metadata", "get", "platform/data/x"], ["read", "auth/approle/role/x/role-id"], ["write", "auth/approle/role/x/secret-id"], ["read", "sys/raw/x"], ["policy", "write", "x", "-"], ["token", "create"], ]) def test_guard_refuses_value_reads_and_writes(args): assert not survey.allowed_bao_call(args) bao = survey.Bao(FakeBao(), {"BAO_TOKEN": TOKEN}) with pytest.raises(survey.SurveyError, match="refused"): bao.call(*args) def test_kv_path_must_not_be_an_api_data_path(catalog): with pytest.raises(survey.SurveyError, match="logical KV path"): survey.main(["--kv-path", "platform/data/workloads/x", "--policy", "p", "--consumer", "c"], runner=FakeBao(), environ={"BAO_TOKEN": TOKEN}) def test_policy_glob_matching(): assert survey.path_matches("platform/data/workloads/*", "platform/data/workloads/a/b") assert survey.path_matches("platform/data/+/a/b", "platform/data/workloads/a/b") assert not survey.path_matches("platform/data/workloads/a", "platform/data/workloads/a/b")