railiance-platform/tests/test_service_identity_metadata_export.py

import contextlib
import importlib.util
import io
import json
import shutil
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location("si_export", ROOT / "scripts/service_identity_metadata_export.py")
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
NOW = lambda: datetime(2026, 10, 28, 12, 0, tzinfo=timezone.utc)  # noqa: E731

# Source-derived fixture: the committed role and policy files, laid out at their live mount paths.
SOURCE_ROLES = {
    ("netkingdom", "platform-admin"): "netkingdom-platform-admin-role.json",
    ("netkingdom", "credential-broker-keycape-client-secret-generate"):
        "credential-broker-keycape-client-secret-generate-oidc-role.json",
    ("netkingdom", "fluid-telegram-hall-of-helix-workload-kv"): "fluid-telegram-hall-of-helix-oidc-role.json",
    ("netkingdom", "coding-agent"): "coding-agent-jwt-role.json",
    ("keycape-services", "secrets-engine-openbao"): "secrets-engine-jwt-role.json",
    ("approle", "coding-agent"): "coding-agent-approle.json",
}


def source_fixture(tmp: Path) -> Path:
    (tmp / "auth-list.json").write_text(json.dumps({
        "netkingdom/": {"type": "oidc"}, "keycape-services/": {"type": "jwt"},
        "approle/": {"type": "approle"}, "token/": {"type": "token"}}))
    for (mount, name), filename in SOURCE_ROLES.items():
        folder = tmp / "roles" / mount
        folder.mkdir(parents=True, exist_ok=True)
        shutil.copy(ROOT / "openbao/auth" / filename, folder / f"{name}.json")
    (tmp / "policies").mkdir()
    for policy in (ROOT / "openbao/policies").glob("*.hcl"):
        shutil.copy(policy, tmp / "policies" / policy.name)
    return tmp


def run(argv):
    out, err = io.StringIO(), io.StringIO()
    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
        code = m.main(argv, now=NOW)
    return code, out.getvalue(), err.getvalue()


class OpenBaoExportTests(unittest.TestCase):
    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp())
        self.addCleanup(shutil.rmtree, self.tmp)

    def export(self):
        code, out, err = run(["openbao", "--fixture-dir", str(source_fixture(self.tmp))])
        self.assertEqual(0, code, err)
        return json.loads(out)

    def role(self, doc, mount, name):
        return next(r for r in doc["openbao"]["roles"] if r["mount"] == mount and r["name"] == name)

    def test_envelope_and_mounts(self):
        doc = self.export()
        self.assertEqual(m.SCHEMA, doc["schema"])
        self.assertEqual("fixture", doc["source_kind"])
        self.assertEqual("2026-10-28T12:00:00Z", doc["collected_at"])
        self.assertIn({"path": "token", "type": "token"}, doc["openbao"]["auth_mounts"])
        self.assertEqual(6, len(doc["openbao"]["roles"]))

    def test_fluid_telegram_policy_is_exact_kv_operator_matrix(self):
        role = self.role(self.export(), "netkingdom", "fluid-telegram-hall-of-helix-workload-kv")
        self.assertEqual({"groups": ["fluid-telegram-operators"], "/assurance/level": "aal2"}, role["bound_claims"])
        self.assertEqual(900, role["token_ttl_seconds"])
        self.assertEqual(1800, role["token_explicit_max_ttl_seconds"])
        rules = {r["path"]: r["capabilities"] for r in role["effective_rules"]}
        base = "platform/data/workloads/coulomb/fluid-telegram/hall-of-helix/"
        self.assertEqual(["read"], rules[base + "operator-app"])
        self.assertEqual(["create", "read", "update"], rules[base + "bot-token"])
        self.assertFalse(any("metadata" in p or "delete" in c for p, c in rules.items()))
        self.assertTrue(role["effective_complete"])

    def test_coding_agent_deny_overrides_issue_core_runtime_read(self):
        doc = self.export()
        for mount in ("netkingdom", "approle"):
            role = self.role(doc, mount, "coding-agent")
            rules = {r["path"]: r["capabilities"] for r in role["effective_rules"]}
            self.assertEqual(["deny"], rules["platform/data/workloads/issue-core/issue-core/issue-core-runtime"])
            readable_data = [p for p, c in rules.items() if "/data/" in p and c != ["deny"]]
            self.assertEqual([], readable_data)
            self.assertEqual(900, role["token_max_ttl_seconds"])
        jwt = self.role(doc, "netkingdom", "coding-agent")
        self.assertEqual(["service", "agent"], jwt["bound_claims"]["principal_type"])

    def test_live_shape_drops_unlisted_fields_and_normalizes_ttls(self):
        view = m.role_view("approle", "approle", "x", {
            "token_ttl": 900, "secret_id_ttl": 300, "role_id": "ignored", "secret_id_accessor": "ignored",
            "policies": ["p"], "local_secret_ids": False})
        self.assertNotIn("role_id", view)
        self.assertNotIn("secret_id_accessor", view)
        self.assertEqual(["p"], view["token_policies"])
        self.assertEqual(300, view["secret_id_ttl_seconds"])

    def test_secret_shaped_value_refuses_without_echo(self):
        fixture = source_fixture(self.tmp)
        token = "hvs." + "A1b2C3d4" * 4
        path = fixture / "roles/netkingdom/coding-agent.json"
        data = json.loads(path.read_text())
        data["bound_claims"]["sub"] = token
        path.write_text(json.dumps(data))
        code, out, err = run(["openbao", "--fixture-dir", str(fixture)])
        self.assertEqual(3, code)
        self.assertEqual("", out)
        self.assertNotIn(token, err)
        self.assertIn("1 secret-shaped", err)

    def test_unreadable_policy_marks_role_incomplete(self):
        fixture = source_fixture(self.tmp)
        (fixture / "policies/agent-high-risk-boundary.hcl").unlink()
        code, out, _ = run(["openbao", "--fixture-dir", str(fixture)])
        role = self.role(json.loads(out), "netkingdom", "coding-agent")
        self.assertFalse(role["effective_complete"])

    def test_names_and_paths_are_not_false_positives(self):
        for value in ("credential-broker-keycape-client-secret-generate",
                      "platform/data/workloads/coulomb/fluid-telegram/hall-of-helix/operator-session",
                      "http://127.0.0.1:18200/ui/vault/auth/netkingdom/oidc/callback"):
            self.assertFalse(m.looks_secret(value), value)
        self.assertTrue(m.looks_secret("SCRAM-SHA-256$4096:abc"))
        self.assertTrue(m.looks_secret("Zx9" * 12))


class PostgresExportTests(unittest.TestCase):
    ROW = json.dumps({
        "server_version": "16.4",
        "roles": [
            {"name": "postgres", "superuser": True, "create_role": True, "create_db": True, "can_login": True,
             "replication": True, "bypass_rls": True, "inherit": True, "connection_limit": -1,
             "valid_until": None, "member_of": []},
            {"name": "privacyidea", "superuser": False, "create_role": False, "create_db": False,
             "can_login": True, "replication": False, "bypass_rls": False, "inherit": True,
             "connection_limit": -1, "valid_until": None, "member_of": [], "rolpassword": "********"}],
        "databases": [{"name": "privacyidea_db", "owner": "privacyidea"}]})

    def test_export_projects_only_role_attributes(self):
        tmp = Path(tempfile.mkdtemp())
        self.addCleanup(shutil.rmtree, tmp)
        (tmp / "row.txt").write_text(self.ROW + "\n")
        code, out, err = run(["postgresql", "--cluster", "net-kingdom-pg", "--input", str(tmp / "row.txt")])
        self.assertEqual(0, code, err)
        doc = json.loads(out)["postgresql"]
        self.assertEqual("net-kingdom-pg", doc["cluster"])
        pi = next(r for r in doc["roles"] if r["name"] == "privacyidea")
        self.assertFalse(pi["superuser"])
        self.assertNotIn("rolpassword", pi)
        self.assertEqual([{"name": "privacyidea_db", "owner": "privacyidea"}], doc["databases"])

    def test_sql_is_read_only_and_never_touches_password_catalog(self):
        code, out, _ = run(["postgresql-sql"])
        self.assertEqual(0, code)
        sql = out.lower()
        self.assertTrue(sql.startswith("select "))
        for word in ("pg_authid", "rolpassword", "insert", "update ", "delete", "alter", "grant ", ";"):
            self.assertNotIn(word, sql)

    def test_rejects_multi_row_input(self):
        code, _, err = run(["postgresql", "--cluster", "c", "--input", "/dev/null"])
        self.assertEqual(2, code)
        self.assertIn("ValueError", err)


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