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()