railiance-platform/tests/test_service_jwt_lane.py

"""CCR-2026-0038 service-jwt-login lane: validator, invariant (ii), fake-bao applier."""
from __future__ import annotations

import copy
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-0038-secrets-engine-service-jwt-login.yaml"
GUARDED = "platform/data/workloads/keycape/secrets-engine-openbao"


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("service_jwt_lane", "scripts/service_jwt_lane.py")
cc = load("credential_change_cli", "scripts/credential-change.py")


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


FAKE_BAO = r'''#!/usr/bin/env python3
import json, os, sys
state = json.load(open(os.environ["FAKE_BAO_STATE"]))
open(os.environ["FAKE_BAO_LOG"], "a").write(json.dumps(sys.argv[1:]) + "\n")
args = [a for a in sys.argv[1:] if a != "-format=json"]
if args[:2] == ["auth", "list"]:
    print(json.dumps({k: {"type": v} for k, v in state["mounts"].items()}))
elif args[:2] == ["policy", "read"]:
    print(state["policy"])
elif args[0] == "read" and args[1].endswith("/config"):
    print(json.dumps({"data": state["config"]}))
elif args[0] == "read":
    print(json.dumps({"data": state["role"]}))
'''


class ServiceJwtLaneTests(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(_ccr["openbao"]["auth"]["bound_claims_confirmed"])
        evidence = _ccr["verification"]["evidence"]
        self.assertTrue(
            any(e["kind"] == "delegated_metadata_apply" and e["result"] == "passed"
                for e in evidence),
            evidence,
        )

    def test_invariant_ii_policy_has_no_capability_on_the_client_secret_path(self):
        policy = (ROOT / "openbao/policies/secrets-engine-login-self.hcl").read_text()
        stanzas = re.findall(r'path "([^"]+)"\s*{\s*capabilities = \[([^\]]*)\]', policy)
        for path, caps in stanzas:
            if path.startswith("platform/"):
                self.assertEqual(caps.strip(), '"deny"', path)
        self.assertIn((GUARDED, '"deny"'), stanzas)
        granted = {p for p, c in stanzas if c.strip() != '"deny"'}
        self.assertEqual(granted, {p for p, _ in lane.SELF_GRANTS})
        self.assertNotIn("*", policy)

    def test_validator_rejects_widening(self):
        cases = []
        c = ccr(); c["openbao"]["guarded_paths"] = ["platform/data/workloads/*"]; cases.append(c)
        c = ccr(); c["openbao"]["auth"]["mount"] = "netkingdom"; cases.append(c)
        c = ccr(); c["openbao"]["policy_name"] = "platform-admin"; cases.append(c)
        c = ccr(); c["target"]["environment"] = "staging"; cases.append(c)
        for case in cases:
            self.assertTrue(lane.validate(case, ROOT), case["openbao"])

    def test_validator_rejects_bad_role_file(self):
        with tempfile.TemporaryDirectory() as tmp:
            repo = Path(tmp)
            (repo / "openbao/auth").mkdir(parents=True)
            (repo / "openbao/policies").mkdir(parents=True)
            c = ccr()
            (repo / c["openbao"]["policy_file"]).write_text(lane.generated_policy_hcl(c))
            (repo / c["openbao"]["auth"]["config_file"]).write_text(
                (ROOT / c["openbao"]["auth"]["config_file"]).read_text())
            base = json.loads((ROOT / c["openbao"]["auth"]["role_file"]).read_text())
            self.assertEqual(self._role_errors(repo, c, base), [])
            for mutate in (
                lambda r: r.update(token_ttl="1h"),
                lambda r: r.update(token_no_default_policy=False),
                lambda r: r.update(token_policies=["secrets-engine-login-self", "default"]),
                lambda r: r.update(token_period="5m"),
                lambda r: r["bound_claims"].update(roles="*"),
                lambda r: r.update(bound_audiences=["other"]),
                lambda r: r.update(token_num_uses=0),
            ):
                role = copy.deepcopy(base); mutate(role)
                self.assertTrue(self._role_errors(repo, c, role), role)

    def _role_errors(self, repo, c, role):
        (repo / c["openbao"]["auth"]["role_file"]).write_text(json.dumps(role))
        return lane.validate(c, repo)

    def _fake(self, tmp: Path, mounts: dict, policy: str | None = None, role_ttl=300):
        c = ccr()
        role = lane.role_payload(c, ROOT); role["token_ttl"] = role_ttl
        role["token_max_ttl"] = role["token_explicit_max_ttl"] = 300
        state = {"mounts": mounts, "policy": policy if policy is not None else lane.generated_policy_hcl(c),
                 "config": lane.config_payload(c, ROOT), "role": role}
        (tmp / "state.json").write_text(json.dumps(state))
        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")
        os.environ["FAKE_BAO_LOG"] = str(tmp / "log")
        return c, str(bao)

    def _log(self, tmp: Path):
        return [json.loads(l) for l in (tmp / "log").read_text().splitlines()]

    def test_apply_enables_absent_mount_writes_in_order_and_reads_back(self):
        with tempfile.TemporaryDirectory() as t:
            tmp = Path(t)
            c, bao = self._fake(tmp, {"keycape-services/": "jwt"})
            # first auth list sees no mount; later readback sees jwt
            state = json.loads((tmp / "state.json").read_text())
            state["mounts"] = {}
            (tmp / "state.json").write_text(json.dumps(state))
            with self.assertRaises(lane.BaoError):  # readback: mount absent in fake
                lane.run_apply(c, ROOT, bao, quiet=True)
            verbs = [a[:2] for a in self._log(tmp)]
            self.assertEqual(verbs[:5], [["auth", "list"], ["policy", "write"], ["auth", "enable"],
                                         ["write", "auth/keycape-services/config"],
                                         ["write", "auth/keycape-services/role/secrets-engine-openbao"]])
            self.assertIn("jwt", self._log(tmp)[2])

    def test_apply_on_existing_jwt_mount_skips_enable_and_passes_readback(self):
        with tempfile.TemporaryDirectory() as t:
            tmp = Path(t)
            c, bao = self._fake(tmp, {"keycape-services/": "jwt"})
            lane.run_apply(c, ROOT, bao, quiet=True)
            verbs = [a[:2] for a in self._log(tmp)]
            self.assertNotIn(["auth", "enable"], verbs)
            for argv in self._log(tmp):
                self.assertFalse(any("token" in a.lower() and "=" in a for a in argv), argv)

    def test_apply_refuses_foreign_mount_type_before_any_write(self):
        with tempfile.TemporaryDirectory() as t:
            tmp = Path(t)
            c, bao = self._fake(tmp, {"keycape-services/": "oidc"})
            with self.assertRaises(lane.BaoError):
                lane.run_apply(c, ROOT, bao, quiet=True)
            self.assertEqual([a[:2] for a in self._log(tmp)], [["auth", "list"]])

    def test_readback_detects_policy_drift(self):
        with tempfile.TemporaryDirectory() as t:
            tmp = Path(t)
            c, bao = self._fake(tmp, {"keycape-services/": "jwt"}, policy='path "platform/data/*" {}')
            self.assertIn("policy text differs from source", lane.readback_mismatches(c, ROOT, bao))


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