railiance-platform/tests/test_broker_grant_lane.py

"""RPF-WP-0054-T04: custody-generate-grant lane (CCR-2026-0040)."""
from __future__ import annotations

import importlib.util
import json
import os
import re
import stat
import sys
import tempfile
import unittest
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]
CCR_PATH = ROOT / "credential-change-requests/CCR-2026-0040-keycape-client-secret-generate-grant.yaml"


def load(name: str, rel: str):
    spec = importlib.util.spec_from_file_location(name, ROOT / rel)
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


lane = load("broker_grant_lane", "scripts/broker_grant_lane.py")
cc = load("credential_change_cli_grant", "scripts/credential-change.py")


def ccr() -> dict:
    return yaml.safe_load(CCR_PATH.read_text())


# Stateful fake: policies stored as text; token role key=value args stored the way
# OpenBao returns them (lists, booleans, seconds).
FAKE_BAO = r'''#!/usr/bin/env python3
import json, os, sys
path = os.environ["FAKE_BAO_STATE"]
state = json.load(open(path))
args = [a for a in sys.argv[1:] if a != "-format=json"]
if args[:2] == ["policy", "write"]:
    state["policies"][args[2]] = open(args[3]).read()
elif args[:2] == ["policy", "read"]:
    print(state["policies"][args[2]])
elif args[0] == "write":
    role = {}
    for kv in args[2:]:
        k, v = kv.split("=", 1)
        if k.endswith("_policies"): role[k] = v.split(",")
        elif v in ("true", "false"): role[k] = v == "true"
        elif k.endswith("_ttl"): role[k] = int(v[:-1]) * 60
        else: role[k] = v
    if state.get("tamper"): role["renewable"] = True
    state["roles"][args[1]] = role
elif args[0] == "read":
    print(json.dumps({"data": state["roles"][args[1]]}))
json.dump(state, open(path, "w"))
'''


def stanzas(text: str) -> dict[str, str]:
    return dict(re.findall(r'path "([^"]+)"\s*{\s*capabilities = \[([^\]]*)\]', text))


class BrokerGrantLaneTests(unittest.TestCase):
    def test_ccr_validates_and_records_the_delegated_apply(self):
        _ccr, errors, _w = cc.validate_ccr(CCR_PATH)
        self.assertEqual(errors, [])
        self.assertEqual(_ccr["status"], "applied")
        self.assertTrue(any(e["kind"] == "delegated_metadata_apply" and e["result"] == "passed"
                            for e in _ccr["verification"]["evidence"]))

    def test_unapproved_ccr_blocks_the_applier(self):
        c = ccr()
        c["status"] = "proposed"
        self.assertTrue(any("got proposed" in b for b in cc.applier_readiness_blockers(c)))

    def test_grant_policy_is_create_only(self):
        policy = stanzas((ROOT / ccr()["openbao"]["policy_file"]).read_text())
        self.assertEqual(policy["platform/data/workloads/keycape/+"], '"create"')
        self.assertEqual(policy["platform/metadata/workloads/keycape/+"], '"read"')
        for path in ("platform/data/workloads/keycape/secrets-engine-openbao",
                     "platform/metadata/workloads/keycape/secrets-engine-openbao"):
            self.assertEqual(policy[path], '"deny"')
        granted = " ".join(c for p, c in policy.items() if p != "auth/token/revoke-self")
        for word in ("update", "delete", "list", "patch", "sudo"):
            self.assertNotIn(f'"{word}"', granted)
        self.assertFalse(any("*" in p for p in policy))

    def test_issuer_policy_mints_only_through_this_role(self):
        issuer = stanzas((ROOT / ccr()["openbao"]["issuer_policy_file"]).read_text())
        creates = [p for p in issuer if p.startswith("auth/token/create")]
        self.assertEqual(creates, ["auth/token/create/keycape-client-secret-generate"])
        # Probe 2026-10-10: the operator must be able to end its own session.
        self.assertEqual(issuer["auth/token/revoke-self"], '"update"')

    def test_validator_rejects_drift(self):
        cases = [
            ("long ttl", lambda c: c["openbao"].update(token_ttl="1h")),
            ("prefix wildcard", lambda c: c["openbao"].update(kv_prefix="platform/workloads/*")),
            ("unknown grant", lambda c: c["openbao"].update(grant_id="keycape/other")),
            ("reserved glob", lambda c: c["openbao"].update(reserved_entries=["*"])),
            ("policy drift", lambda c: c["openbao"].update(reserved_entries=[])),
        ]
        for label, mutate in cases:
            with self.subTest(label):
                c = ccr()
                mutate(c)
                self.assertNotEqual(lane.validate(c, ROOT), [], label)

    def test_state_hub_evidence_summary_handles_prefix_lanes(self):
        posted = {}
        original = cc.state_hub_post_json
        cc.state_hub_post_json = lambda base, route, body: posted.update(body) or {"id": "x"}
        try:
            cc.record_evidence_state_hub(ccr(), "http://hub", "founder", "delegated_metadata_apply", "passed", ["ok"])
        finally:
            cc.state_hub_post_json = original
        self.assertIn("path=platform/workloads/keycape", posted["summary"])

    def _apply(self, tamper=False):
        tmp = Path(tempfile.mkdtemp())
        (tmp / "state.json").write_text(json.dumps({"policies": {}, "roles": {}, "tamper": tamper}))
        bao = tmp / "bao"
        bao.write_text(FAKE_BAO)
        bao.chmod(bao.stat().st_mode | stat.S_IEXEC)
        os.environ["FAKE_BAO_STATE"] = str(tmp / "state.json")
        error = None
        try:
            lane.run_apply(ccr(), ROOT, str(bao), quiet=True)
        except lane.BaoError as exc:
            error = exc
        return error, json.loads((tmp / "state.json").read_text())

    def test_apply_writes_three_objects_and_reads_back(self):
        error, state = self._apply()
        self.assertIsNone(error)
        self.assertEqual(sorted(state["policies"]), [
            "credential-broker-keycape-client-secret-generate-issuer",
            "keycape-client-secret-generate",
        ])
        role = state["roles"]["auth/token/roles/keycape-client-secret-generate"]
        self.assertEqual(role["allowed_policies"], ["keycape-client-secret-generate"])
        self.assertEqual(role["token_explicit_max_ttl"], 300)

    def test_readback_detects_role_drift(self):
        error, _state = self._apply(tamper=True)
        self.assertIsNotNone(error)
        self.assertIn("renewable", str(error))


if __name__ == "__main__":
    unittest.main()