tegwick
9f67565ea3
scripts/plan-survey.py runs under the ops-mason-build grant and prints a pasteable section 2: relevant secrets/auth/policy/role listings with counts, policy stanzas covering the target for name-matched policies, KV metadata existence, matching ops-warden catalog entries, timestamp, BAO_ADDR, token display_name and a draft yes/no/partial answer. Metadata only: an allowlist refuses kv get, */data/* and role-id/secret-id before any process starts; a non-grant token is refused and the token is never printed. Denied queries degrade to 'not permitted by ops-mason-build'. Tests use a fake bao runner. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 1170398@bnt-lap001 Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583
216 lines
9 KiB
Python
216 lines
9 KiB
Python
"""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-build")):
|
|
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-build", "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-build" 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-build grant"):
|
|
survey.main(ARGS, runner=FakeBao(), environ={})
|
|
|
|
|
|
@pytest.mark.parametrize("policies", [("root",), ("default", "ops-mason-build", "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-build 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-build 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")
|