tegwick
4e46aa3b4f
Hub decision 9833715e is resolved, decided by Bernd Worsch at 2026-10-15T00:07:05.087085Z, and is distinct from the disclosure-consent digest. The plan stays reviewed. Assistant: grok Assistant-Session: 01a16071-5425-78d3-80b4-20fe2ee30f0a
218 lines
8.5 KiB
Python
218 lines
8.5 KiB
Python
"""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
|
|
)
|