"""MASON-WP-0013 and MASON-WP-0014 evidence stays name-only and fail-closed.""" import importlib.util import json import re import sys from pathlib import Path import pytest import yaml ROOT = Path(__file__).parents[1] spec = importlib.util.spec_from_file_location( "plan_survey", ROOT / "scripts" / "plan-survey.py" ) survey = importlib.util.module_from_spec(spec) sys.modules["plan_survey"] = survey spec.loader.exec_module(survey) READER_ARGS = [ "--kv-path", "platform/workloads/reuse/reuse-surface/forgejo-source-read", "--policy", "workload-kv-read-reuse-surface-runtime-eso", "--k8s-role", "reuse-surface-runtime-eso", "--consumer", "reuse-surface", "--keyword", "forgejo", ] MAS_ARGS = [ "--kv-path", "platform/suppliers/scaleway/mas-tem-relay", "--policy", "workload-kv-read-mas-tem-relay", "--k8s-role", "external-secrets-mas-tem-relay", "--consumer", "mail-address-semantics", ] def _task_status(text: str, task_id: str) -> str: match = re.search(rf"id: {re.escape(task_id)}\nstatus: (\w+)", text) assert match, task_id return match.group(1) def _frontmatter(text: str) -> dict: return yaml.safe_load(text.split("---", 2)[1]) def _flat(text: str) -> str: return re.sub(r"\s+", " ", text) def _intake(intake_id: str) -> dict: text = (ROOT / "intakes" / "intakes.md").read_text() blocks = re.findall(r"```yaml\n(.*?)```", text, re.S) for block in blocks: data = yaml.safe_load(block) if isinstance(data, dict) and data.get("id") == intake_id: return data raise AssertionError(intake_id) @pytest.mark.parametrize("args", [READER_ARGS, MAS_ARGS]) def test_plan_survey_refuses_without_grant(args, monkeypatch, tmp_path, capsys): monkeypatch.delenv("BAO_TOKEN", raising=False) monkeypatch.setattr(survey, "GRANT_FILE", str(tmp_path / "absent-grant")) calls = [] def runner(argv, env): calls.append(argv) raise AssertionError("survey must not call bao without a grant") with pytest.raises(survey.SurveyError, match="no ops-mason-survey grant"): survey.main(args, runner=runner, environ={"BAO_ADDR": "http://127.0.0.1:18200"}) assert calls == [] assert capsys.readouterr().out == "" def test_mas_teardown_intake_is_residual_and_value_free(): data = _intake("MASON-IN-0007") description = data["description"] assert data["origin"] == "residual" assert data["origin_ref"] == "MASON-WP-0014" assert data["status"] == "open" assert "external-secrets-mas-tem-relay" in description assert "workload-kv-read-mas-tem-relay" in description assert "platform/suppliers/scaleway/mas-tem-relay" in description assert "metadata delete" in description assert "Scaleway API key" in description assert "MAS-WP-0003" in description lowered = description.lower() for banned in ("password", "username", "hvs.", "token=", "begin private"): assert banned not in lowered def test_mas_delivery_check_records_metadata_and_sync(): workplan = (ROOT / "workplans" / "MASON-WP-0014-mas-tem-relay-smtp.md").read_text() plan = (ROOT / "plans" / "mas-tem-relay-smtp.md").read_text() assert _frontmatter(workplan)["status"] == "finished" assert _task_status(workplan, "MASON-WP-0014-T04") == "done" for text in (workplan, plan): flat = _flat(text) assert "token-claude-MASON-WP-0014-T04" in flat assert "ops-mason-survey" in flat assert "current_version=2" in flat assert "SecretSynced" in flat assert "binding name `mas-relay`" in flat assert "MASON-IN-0007" in flat assert "hvs." not in flat def test_private_reader_survey_records_lane_without_approving_apply(): workplan = ( ROOT / "workplans" / "MASON-WP-0013-reuse-surface-private-source-reader.md" ).read_text() plan = (ROOT / "plans" / "reuse-surface-private-source-reader.md").read_text() proposal = ( ROOT / "builds" / "reuse-surface-private-source-reader" / "catalog-proposal.yaml" ).read_text() assert _frontmatter(workplan)["status"] == "finished" assert _frontmatter(plan)["status"] == "reviewed" assert _frontmatter(plan)["approved_by"] is None assert _task_status(workplan, "MASON-WP-0013-T02") == "done" assert _task_status(workplan, "MASON-WP-0013-T03") == "done" assert _task_status(workplan, "MASON-WP-0013-T04") == "done" assert "cust81-forge-reader-r1.jsonl" in workplan assert "seven_exact_no_extra_private_grants" in workplan assert "coulomb/rapp-core-hub" in workplan flat_plan = _flat(plan) flat_workplan = _flat(workplan) assert "token-claude-MASON-WP-0014-T04" in flat_plan assert "current_version=1" in flat_plan assert "reuse-surface-runtime-eso" in flat_plan assert "workload-kv-read-reuse-surface-runtime-eso" in flat_plan assert "Q1:" in flat_plan and "Q2:" in flat_plan and "Q3:" in flat_plan assert "9833715e-a22a-45c0-ab6c-881f0f4f1a0e" in flat_workplan assert "status `resolved`" in flat_workplan assert "2026-10-15T00:07:05.087085Z" in flat_workplan assert "CCR-2026-0046" in flat_workplan assert "4a4602ad926e3af855d6ddba4731430d33e4a7432c507ba0a5193b9f43056b52" in flat_workplan assert "RPF-WP-0058-T05" in flat_workplan assert "reviewed_present" in flat_workplan assert "status: draft" in proposal assert "not-activated" in proposal assert "warden_executes: false" in proposal def test_reader_survey_entry_point_is_metadata_only(tmp_path, capsys): catalog = tmp_path / "catalog.yaml" catalog.write_text("version: 1\nentries: []\n") token = "hvs.SYNTHETICgrantTOKENvalue0123456789" calls = [] eso = "workload-kv-read-reuse-surface-runtime-eso" writer = "workload-kv-read-reuse-surface-runtime" def runner(argv, env): calls.append(argv) assert env["BAO_TOKEN"] == token 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) assert pos[:2] != ["kv", "get"] assert not any("/data/" in a for a in pos) key = tuple(pos) if key == ("token", "lookup"): body = {"data": {"policies": ["default", "ops-mason-survey"], "display_name": "token-claude-MASON-WP-0014-T04", "ttl": 1000, "id": token}} elif key == ("secrets", "list"): body = {"platform/": {"type": "kv", "options": {"version": "2"}}} elif key == ("auth", "list"): body = {"kubernetes/": {"type": "kubernetes"}} elif key == ("policy", "list"): body = [eso, writer] elif key == ("list", "auth/kubernetes/role"): body = ["reuse-surface-runtime-eso"] elif key == ("read", "auth/kubernetes/role/reuse-surface-runtime-eso"): body = {"data": {"token_policies": [eso], "bound_service_account_names": ["reuse-surface-eso"], "bound_service_account_namespaces": ["reuse"], "audience": "openbao"}} elif key == ("policy", "read", eso): return 0, ( 'path "platform/data/workloads/reuse/reuse-surface/forgejo-source-read" ' '{ capabilities = ["read"] }\n' ), "" elif key == ("policy", "read", writer): return 0, 'path "platform/data/workloads/reuse/reuse-surface/runtime-secrets" { capabilities = ["read"] }\n', "" elif key == ("kv", "metadata", "get", "platform/workloads/reuse/reuse-surface/forgejo-source-read"): body = {"data": {"current_version": 1, "created_time": "2026-10-15T00:00:00Z", "updated_time": "2026-10-15T00:00:00Z", "custom_metadata": {}}} elif key == ("kv", "list", "platform/workloads/reuse/reuse-surface"): body = ["forgejo-source-read"] else: return 2, "", "No value found" return 0, json.dumps(body), "" rc = survey.main( [*READER_ARGS, "--catalog", str(catalog)], runner=runner, environ={"BAO_TOKEN": token, "BAO_ADDR": "http://127.0.0.1:18200"}, ) out = capsys.readouterr().out assert rc == 0 assert "token-claude-MASON-WP-0014-T04" in out assert "platform/workloads/reuse/reuse-surface/forgejo-source-read" in out assert "current_version=1" in out assert "no stanza covers" in out assert token not in out assert any( [a for a in argv[1:] if not a.startswith("-")][:3] == ["kv", "metadata", "get"] for argv in calls )