RPF-WP-0067: owner answers for NK-WP-0040-T02 and value-safe identity metadata exporter for NK-WP-0054-T04

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 308701@bnt-lap001
Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56

Makefile

@@ -666,3 +666,14 @@ generated-work-records:
 	$(MAKE) --no-print-directory -f .generated-work-records.mk generated-work-records
 generated-work-records-check:
 	$(MAKE) --no-print-directory -f .generated-work-records.mk generated-work-records-check
+
+##@ Service-identity metadata (NK-WP-0054-T04, RPF-WP-0067)
+.PHONY: service-identity-metadata-tests service-identity-metadata-openbao service-identity-metadata-postgresql-sql
+service-identity-metadata-tests: ## Offline fixture tests for the value-safe identity metadata exporter
+	PYTHONDONTWRITEBYTECODE=1 python3 -m unittest -v tests.test_service_identity_metadata_export
+
+service-identity-metadata-openbao: ## Read-only OpenBao auth-role/policy export via the operator's bao login (OUT=file)
+	python3 scripts/service_identity_metadata_export.py --output "$${OUT:-openbao-identity-metadata.json}" openbao
+
+service-identity-metadata-postgresql-sql: ## Print the read-only pg_roles/pg_auth_members SELECT
+	@python3 scripts/service_identity_metadata_export.py postgresql-sql

WORK-RECORDS.md

@@ -46,6 +46,7 @@
 | workplan | RPF-WP-0064 | active | — | workplans/RPF-WP-0064-isolated-original-openbao-restore.md |
 | workplan | RPF-WP-0065 | proposed | — | workplans/RPF-WP-0065-netkingdom-role-claim-bindings.md |
 | workplan | RPF-WP-0066 | proposed | — | workplans/RPF-WP-0066-openbao-durable-audit-snapshots-conformance.md |
+| workplan | RPF-WP-0067 | active | — | workplans/RPF-WP-0067-netkingdom-identity-receipt-and-metadata-answers.md |
 | task | RPF-WP-ADHOC-2026-10-10-T01 | done | — | workplans/ADHOC-2026-10-10.md |
 | task | RPF-WP-ADHOC-2026-10-11-T01 | done | — | workplans/ADHOC-2026-10-11.md |
 | task | RPF-WP-ADHOC-2026-10-11-T02 | done | — | workplans/ADHOC-2026-10-11.md |
@@ -216,6 +217,9 @@
 | task | RPF-WP-0066-T04 | wait | — | workplans/RPF-WP-0066-openbao-durable-audit-snapshots-conformance.md |
 | task | RPF-WP-0066-T05 | wait | — | workplans/RPF-WP-0066-openbao-durable-audit-snapshots-conformance.md |
 | task | RPF-WP-0066-T06 | todo | — | workplans/RPF-WP-0066-openbao-durable-audit-snapshots-conformance.md |
+| task | RPF-WP-0067-T01 | done | — | workplans/RPF-WP-0067-netkingdom-identity-receipt-and-metadata-answers.md |
+| task | RPF-WP-0067-T02 | done | — | workplans/RPF-WP-0067-netkingdom-identity-receipt-and-metadata-answers.md |
+| task | RPF-WP-0067-T03 | wait | — | workplans/RPF-WP-0067-netkingdom-identity-receipt-and-metadata-answers.md |
 | intake | RPF-IN-0001 | closed | — | intakes/intakes.md |
 | intake | RPF-IN-0002 | open | — | intakes/intakes.md |
 | intake | RPF-IN-0003 | open | — | intakes/intakes.md |

scripts/service_identity_metadata_export.py

@@ -0,0 +1,292 @@
+#!/usr/bin/env python3
+"""Value-safe service-identity metadata export for OpenBao and PostgreSQL.
+
+NK-WP-0054-T04 compares NetKingdom's service-identity declarations with
+observed state. This exporter produces the observed side as non-secret JSON.
+
+OpenBao (read-only): ``bao auth list``, ``bao list auth/<mount>/role``,
+``bao read auth/<mount>/role/<name>`` and ``bao policy read <name>``. It never
+reads a role-id, secret-id, KV data, token or Secret. Only allowlisted role
+fields are emitted; every other field returned by OpenBao is dropped.
+
+PostgreSQL (read-only): ``postgresql-sql`` prints one SELECT over ``pg_roles``,
+``pg_auth_members`` and ``pg_database`` (never ``pg_authid`` or
+``rolpassword``); ``postgresql`` turns its single JSON output row into the
+export. The operator runs the SELECT with their own access; this script holds
+no database connection.
+
+Every output string is screened for secret shapes before printing. On a hit
+the export is refused and only a count is reported, never the value.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable
+
+SCHEMA = "railiance-platform.service-identity-metadata.v1"
+ROLE_MOUNT_TYPES = {"jwt", "oidc", "approle", "kubernetes"}
+ROLE_FIELDS = (
+    "role_type", "bound_audiences", "bound_subject", "bound_claims", "bound_claims_type",
+    "user_claim", "groups_claim", "claim_mappings", "oidc_scopes", "allowed_redirect_uris",
+    "bound_service_account_names", "bound_service_account_namespaces", "audience",
+    "bind_secret_id", "secret_id_num_uses", "secret_id_bound_cidrs", "token_bound_cidrs",
+    "token_policies", "policies", "token_no_default_policy", "token_type", "token_num_uses",
+)
+DURATION_FIELDS = ("token_ttl", "token_max_ttl", "token_explicit_max_ttl", "secret_id_ttl", "ttl", "max_ttl")
+SECRET_SHAPES = (
+    re.compile(r"^(hvs|hvb|hvr|s|b)\.[A-Za-z0-9_-]{20,}"),
+    re.compile(r"^eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}"),
+    re.compile(r"-----BEGIN [A-Z ]*(PRIVATE KEY|CERTIFICATE)"),
+    re.compile(r"^(SCRAM-SHA-256\$|md5[0-9a-f]{32}$)"),
+    re.compile(r"^[A-Fa-f0-9]{40,}$"),
+)
+OPAQUE = re.compile(r"^[A-Za-z0-9+/_=-]{32,}$")
+DIGEST_KEYS = {"sha256"}  # our own policy-text digests, not credentials
+POLICY_RULE = re.compile(r'path\s+"([^"]+)"\s*\{\s*capabilities\s*=\s*\[([^\]]*)\]', re.S)
+DURATION = re.compile(r"(\d+)([smhd])")
+UNIT = {"s": 1, "m": 60, "h": 3600, "d": 86400}
+
+PG_SQL = (
+    "SELECT json_build_object("
+    "'server_version', current_setting('server_version'),"
+    "'roles', COALESCE((SELECT json_agg(json_build_object("
+    "'name', r.rolname, 'superuser', r.rolsuper, 'create_role', r.rolcreaterole,"
+    "'create_db', r.rolcreatedb, 'can_login', r.rolcanlogin, 'replication', r.rolreplication,"
+    "'bypass_rls', r.rolbypassrls, 'inherit', r.rolinherit, 'connection_limit', r.rolconnlimit,"
+    "'valid_until', r.rolvaliduntil,"
+    "'member_of', COALESCE((SELECT json_agg(json_build_object('role', g.rolname, 'admin_option', m.admin_option)"
+    " ORDER BY g.rolname) FROM pg_auth_members m JOIN pg_roles g ON g.oid = m.roleid"
+    " WHERE m.member = r.oid), '[]'::json)) ORDER BY r.rolname)"
+    " FROM pg_roles r WHERE r.rolname !~ '^pg_'), '[]'::json),"
+    "'databases', COALESCE((SELECT json_agg(json_build_object('name', d.datname,"
+    " 'owner', pg_get_userbyid(d.datdba)) ORDER BY d.datname)"
+    " FROM pg_database d WHERE NOT d.datistemplate), '[]'::json))"
+)
+PG_ROLE_FIELDS = ("name", "superuser", "create_role", "create_db", "can_login", "replication",
+                  "bypass_rls", "inherit", "connection_limit", "valid_until", "member_of")
+
+
+class Refused(Exception):
+    """Raised when output would carry a secret-shaped value; never holds the value."""
+
+
+def seconds(value: Any) -> int | None:
+    if value is None or value == "":
+        return None
+    if isinstance(value, bool):
+        raise ValueError("duration")
+    if isinstance(value, (int, float)):
+        return int(value)
+    text = str(value).strip()
+    if text.isdigit():
+        return int(text)
+    parts = DURATION.findall(text)
+    if not parts or "".join(n + u for n, u in parts) != text:
+        raise ValueError("duration")
+    return sum(int(n) * UNIT[u] for n, u in parts)
+
+
+def policy_rules(text: str) -> list[dict[str, Any]]:
+    rules = [{"path": p, "capabilities": sorted(c.strip().strip('"') for c in caps.split(",") if c.strip())}
+             for p, caps in POLICY_RULE.findall(text)]
+    return sorted(rules, key=lambda r: r["path"])
+
+
+def effective_rules(policies: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Merge exact-path rules across a role's policies; deny wins (OpenBao semantics).
+
+    Glob and segment-wildcard paths are kept as written, not expanded."""
+    merged: dict[str, set[str]] = {}
+    for policy in policies:
+        for rule in policy["rules"]:
+            merged.setdefault(rule["path"], set()).update(rule["capabilities"])
+    out = []
+    for path, caps in sorted(merged.items()):
+        out.append({"path": path, "capabilities": ["deny"] if "deny" in caps else sorted(caps),
+                    "wildcard": "*" in path or "+" in path})
+    return out
+
+
+def role_view(mount: str, mount_type: str, name: str, data: dict[str, Any]) -> dict[str, Any]:
+    view: dict[str, Any] = {"mount": mount, "mount_type": mount_type, "name": name}
+    for key in ROLE_FIELDS:
+        if key in data and data[key] not in (None, "", [], {}):
+            view[key] = data[key]
+    for key in DURATION_FIELDS:
+        if key in data:
+            value = seconds(data[key])
+            if value is not None:
+                view[key + "_seconds"] = value
+    if "policies" in view and "token_policies" not in view:
+        view["token_policies"] = view["policies"]
+    view.pop("policies", None)
+    view.setdefault("role_type", mount_type)
+    return view
+
+
+class BaoCLI:
+    """Read-only calls through the operator's own ``bao`` login; no token is handled here."""
+
+    def __init__(self, bao: str = "bao") -> None:
+        self.bao = bao
+
+    def _json(self, *args: str) -> Any:
+        proc = subprocess.run([self.bao, *args, "-format=json"], capture_output=True, text=True, check=False)
+        if proc.returncode != 0:
+            raise RuntimeError(f"bao {args[0]} {args[1] if len(args) > 1 else ''} failed (exit {proc.returncode})")
+        return json.loads(proc.stdout or "null")
+
+    def auth_mounts(self) -> dict[str, Any]:
+        return self._json("auth", "list")
+
+    def roles(self, mount: str) -> list[str]:
+        proc = subprocess.run([self.bao, "list", "-format=json", f"auth/{mount}/role"],
+                              capture_output=True, text=True, check=False)
+        if proc.returncode != 0:
+            return []  # no roles, or list not permitted: reported as an empty mount
+        return list(json.loads(proc.stdout or "[]") or [])
+
+    def role(self, mount: str, name: str) -> dict[str, Any]:
+        reply = self._json("read", f"auth/{mount}/role/{name}")
+        return (reply or {}).get("data", {}) or {}
+
+    def policy(self, name: str) -> str | None:
+        proc = subprocess.run([self.bao, "policy", "read", name], capture_output=True, text=True, check=False)
+        return proc.stdout if proc.returncode == 0 else None
+
+
+class FixtureDir:
+    """Offline source: auth-list.json, roles/<mount>/<role>.json, policies/<name>.hcl."""
+
+    def __init__(self, root: Path) -> None:
+        self.root = root
+
+    def auth_mounts(self) -> dict[str, Any]:
+        return json.loads((self.root / "auth-list.json").read_text())
+
+    def roles(self, mount: str) -> list[str]:
+        folder = self.root / "roles" / mount
+        return sorted(p.stem for p in folder.glob("*.json")) if folder.is_dir() else []
+
+    def role(self, mount: str, name: str) -> dict[str, Any]:
+        return json.loads((self.root / "roles" / mount / f"{name}.json").read_text())
+
+    def policy(self, name: str) -> str | None:
+        path = self.root / "policies" / f"{name}.hcl"
+        return path.read_text() if path.is_file() else None
+
+
+def export_openbao(source: Any, mounts_filter: set[str] | None = None) -> dict[str, Any]:
+    mounts = []
+    roles = []
+    for path, meta in sorted((source.auth_mounts() or {}).items()):
+        mount = path.rstrip("/")
+        mtype = (meta or {}).get("type", "unknown")
+        mounts.append({"path": mount, "type": mtype})
+        if mtype not in ROLE_MOUNT_TYPES or (mounts_filter and mount not in mounts_filter):
+            continue
+        for name in source.roles(mount):
+            roles.append(role_view(mount, mtype, name, source.role(mount, name)))
+    policies: dict[str, dict[str, Any]] = {}
+    for name in sorted({p for r in roles for p in r.get("token_policies", [])}):
+        text = source.policy(name)
+        policies[name] = ({"name": name, "readable": False, "rules": []} if text is None else
+                          {"name": name, "readable": True, "sha256": hashlib.sha256(text.encode()).hexdigest(),
+                           "rules": policy_rules(text)})
+    for role in roles:
+        attached = [policies[p] for p in role.get("token_policies", [])]
+        role["effective_rules"] = effective_rules(attached)
+        role["effective_complete"] = (all(p["readable"] for p in attached)
+                                      and role.get("token_no_default_policy") is True)
+    return {"auth_mounts": mounts, "roles": roles, "policies": list(policies.values())}
+
+
+def export_postgresql(raw: str, cluster: str, namespace: str) -> dict[str, Any]:
+    lines = [line for line in raw.splitlines() if line.strip()]
+    if len(lines) != 1:
+        raise ValueError("expected exactly one JSON row from the postgresql-sql query")
+    data = json.loads(lines[0])
+    roles = [{k: r.get(k) for k in PG_ROLE_FIELDS} for r in data.get("roles", [])]
+    databases = [{"name": d["name"], "owner": d["owner"]} for d in data.get("databases", [])]
+    return {"cluster": cluster, "namespace": namespace, "server_version": data.get("server_version"),
+            "roles": roles, "databases": databases}
+
+
+def looks_secret(text: str) -> bool:
+    if any(p.search(text) for p in SECRET_SHAPES):
+        return True
+    # opaque high-entropy strings: mixed case plus digits, no spaces (names and paths are lower case)
+    return bool(OPAQUE.match(text) and re.search(r"[A-Z]", text) and re.search(r"[a-z]", text)
+                and re.search(r"[0-9]", text))
+
+
+def secret_hits(value: Any) -> int:
+    if isinstance(value, dict):
+        return sum(0 if k in DIGEST_KEYS else secret_hits(v) for k, v in value.items())
+    if isinstance(value, list):
+        return sum(secret_hits(v) for v in value)
+    if isinstance(value, str):
+        return int(looks_secret(value))
+    return 0
+
+
+def envelope(service: str, kind: str, body: dict[str, Any], now: Callable[[], datetime]) -> dict[str, Any]:
+    doc = {"schema": SCHEMA, "service": service,
+           "collected_at": now().astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+           "source_kind": kind, "value_safe": True, service: body}
+    hits = secret_hits(body)
+    if hits:
+        raise Refused(f"refused: {hits} secret-shaped value(s) in export; nothing printed")
+    return doc
+
+
+def main(argv: list[str] | None = None, now: Callable[[], datetime] = lambda: datetime.now(timezone.utc)) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    sub = parser.add_subparsers(dest="cmd", required=True)
+    bao = sub.add_parser("openbao", help="export auth roles and their policies")
+    bao.add_argument("--fixture-dir", type=Path, help="offline source instead of the bao CLI")
+    bao.add_argument("--bao", default="bao", help="bao CLI binary (uses the operator's own login)")
+    bao.add_argument("--mount", action="append", help="limit to these auth mounts (repeatable)")
+    sub.add_parser("postgresql-sql", help="print the read-only SELECT to run with psql -AtX")
+    pg = sub.add_parser("postgresql", help="export from the SELECT's single output row")
+    pg.add_argument("--input", type=Path, help="file with the psql output (default stdin)")
+    pg.add_argument("--cluster", required=True)
+    pg.add_argument("--namespace", default="databases")
+    parser.add_argument("--output", type=Path, help="write here instead of stdout")
+    args = parser.parse_args(argv)
+    try:
+        if args.cmd == "postgresql-sql":
+            print(PG_SQL)
+            return 0
+        if args.cmd == "openbao":
+            source = FixtureDir(args.fixture_dir) if args.fixture_dir else BaoCLI(args.bao)
+            doc = envelope("openbao", "fixture" if args.fixture_dir else "live",
+                           export_openbao(source, set(args.mount) if args.mount else None), now)
+        else:
+            raw = args.input.read_text() if args.input else sys.stdin.read()
+            doc = envelope("postgresql", "live" if not args.input else "file",
+                           export_postgresql(raw, args.cluster, args.namespace), now)
+    except Refused as exc:
+        print(str(exc), file=sys.stderr)
+        return 3
+    except (RuntimeError, ValueError, KeyError, json.JSONDecodeError) as exc:
+        print(f"export failed: {type(exc).__name__}", file=sys.stderr)
+        return 2
+    text = json.dumps(doc, indent=2, sort_keys=True) + "\n"
+    if args.output:
+        args.output.write_text(text)
+    else:
+        sys.stdout.write(text)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_service_identity_metadata_export.py

@@ -0,0 +1,173 @@
+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()

workplans/RPF-WP-0067-netkingdom-identity-receipt-and-metadata-answers.md

@@ -0,0 +1,146 @@
+---
+id: RPF-WP-0067
+type: workplan
+title: "NetKingdom owner answers: execution-attribution receipt emitter and service-identity metadata export"
+domain: financials
+repo: railiance-platform
+status: active
+flavor: implementation
+owner: claude
+created: "2026-10-28"
+updated: "2026-10-28"
+related:
+  - NK-WP-0040
+  - NK-WP-0054
+  - AUDIT-WP-0014
+---
+
+# RPF-WP-0067 — NetKingdom owner answers (NK-WP-0040-T02, NK-WP-0054-T04)
+
+Receiving workplan for two NetKingdom briefs (State Hub messages
+`521e3dba-59db-46a8-ae7a-fa8ae2a4c9a7` and `f5d366f5-853f-4d3c-9570-a3275d27df25`).
+Answers are from railiance-platform source only. No live, cluster or OpenBao
+action was taken; no Secret value was read.
+
+## Answer the execution-attribution receipt emitter questions
+
+```task
+id: RPF-WP-0067-T01
+status: done
+priority: high
+```
+
+Question source: net-kingdom `canon/standards/playbook-capability-contract_v0.2.md`
+§"Execution Attribution Receipt (proposed)"; audit-core admission requirements in
+audit-core `docs/cross-repo-evidence-admission.md` §"Execution-attribution receipts —
+AUDIT-WP-0014". Emission is not implemented here.
+
+1. **Emitting component — not confirmable from this repository; none exists.**
+   railiance-platform contains no Railiance playbook runner or executor and no
+   `PlaybookCapability` declaration (search of `scripts/`, `tools/`, `lib/`,
+   `helm/`, `argocd/`, `Makefile`: the only `playbook` hits are unrelated). The
+   contract's reference declaration is
+   `railiance-infra/capabilities/playbooks/railiance-infra.bootstrap-host.yaml`
+   (S1 Ansible; contract v0.2 lines 375-385), and the contract itself says
+   "a separate playbook runner" executes (line 261). railiance-platform is the
+   S3 platform-services layer; the emitter must be named by the repository that
+   owns the runner (railiance-infra for S1; S2/S4 owners for theirs). This repo
+   should not be recorded as the emitter.
+2. **Wire binding — what this repo can state.** The only audit-core sender
+   binding in this repo is the existing event-sender pattern, which a future
+   emitter should reuse rather than invent:
+   - transport: HTTP `POST /v1/events` to in-cluster
+     `http://audit-core.audit-core.svc.cluster.local:8080`
+     (`scripts/native_factory_acceptance.py` line 48,
+     `scripts/native_factory_probe/producer.py` lines 30-36, 127);
+   - idempotency: `Idempotency-Key` header equal to the envelope `id`, with
+     `correlation_id` carrying the run id (`producer.py` lines 33-34, 130, 174-176);
+   - sender identity: a per-sender bearer from the audit-core sender registry at
+     OpenBao `platform/workloads/audit-core/senders`, projected by ESO to a
+     mounted credential file (`producer.py` line 31;
+     `scripts/audit_core_synthetic_sender_lane.py` docstring);
+     the envelope `source` must equal the sender name, tenant is declared;
+   - signing: none at the message level. Authenticity is the bearer plus
+     audit-core's own chaining; no detached or per-receipt signature exists in
+     this repo.
+   A playbook-runner sender identity does not exist in the registry; adding one
+   is a CCR through this repo once the runner owner names the sender.
+3. **When emitted — not confirmable.** No emitter exists, so no success/failure/
+   partial guarantee exists. Recommendation for the runner owner: emit on every
+   terminal outcome (success, failure, partial/aborted) with a stable `run_id`
+   as the idempotency key, and record absence of a receipt as an explicit gap.
+4. **Fields a platform-side emitter could not populate truthfully today:**
+   `initiator.iss`/`initiator.sub` and `delegation` (runs are started by an
+   operator shell or ssh; no IAM Profile token is presented to a runner);
+   `executing_workload` (Tenancy Posture `workload_identity` is still proposed);
+   `artifact_digest` for Ansible runs (no packaged artifact; would need a
+   computed tree digest); `decision_refs` (no machine link from CCR/approval
+   records to a run); `time_interval.bound` (no established skew bound, so it
+   must stay absent). `runtime_principal`, `tenant`, `environment`, `run_id`,
+   `source_commit`, `target`, `action`, `time_interval.start/end/clock_source`
+   are populatable by a runner that records them.
+
+Open: runner owner (railiance-infra and any S2/S4 runner) names the emitter;
+NetKingdom ships the receipt schema; audit-core implements AUDIT-WP-0014-T02.
+
+## Provide value-safe OpenBao and PostgreSQL identity metadata
+
+```task
+id: RPF-WP-0067-T02
+status: done
+priority: high
+```
+
+`scripts/service_identity_metadata_export.py` (tests:
+`tests/test_service_identity_metadata_export.py`, `make service-identity-metadata-tests`).
+Output schema `railiance-platform.service-identity-metadata.v1`. OpenBao: auth
+mounts, allowlisted role fields (bound claims/audiences/subject, service-account
+bindings, token policies, TTLs in seconds, token type/uses), policy rules and
+their SHA-256, and per-role `effective_rules` (exact paths merged, deny wins).
+PostgreSQL: `pg_roles` attributes, memberships and database owners. Role-ids,
+secret-ids, KV data, `pg_authid` and passwords are never read; secret-shaped
+output refuses the export and reports only a count.
+
+Proven from source for net-kingdom `tools/service-identity/declarations/openbao.yaml`:
+- `fluid-telegram-operators` grants: `openbao/auth/fluid-telegram-hall-of-helix-oidc-role.json`
+  lines 12-20 and `openbao/policies/workload-kv-fluid-telegram-hall-of-helix-operator.hcl`
+  lines 1-16 — read/create/update on four exact KV data entries under
+  `platform/workloads/coulomb/fluid-telegram/hall-of-helix/`, no metadata, list or
+  delete, 15-minute token, 30-minute explicit max.
+- `coding-agent` grants: `openbao/auth/coding-agent-jwt-role.json` lines 14-19 and
+  `openbao/auth/coding-agent-approle.json` lines 5-11 attach
+  `agent-high-risk-boundary` and `workload-kv-read-issue-core-runtime`; the boundary
+  denies the same data path the second policy grants
+  (`agent-high-risk-boundary.hcl` line 25 vs `workload-kv-read-issue-core-runtime.hcl`
+  line 1), and deny wins. Effective: no KV data read; metadata read only where the boundary allows it; token self-lookup/revoke and `sys/capabilities-self`.
+- `coding-agent` `agent.mode`: not provable. The JWT role binds no agent-mode
+  claim (`coding-agent-jwt-role.json` lines 8-13) and accepts `principal_type`
+  `service` or `agent` (line 10). The mode is issued by KeyCape (KEY-WP-0043).
+
+## Run the live export (founder)
+
+```task
+id: RPF-WP-0067-T03
+status: wait
+priority: medium
+blocking_reason: "Live read-only export needs the founder's OpenBao platform-admin login and cluster access."
+```
+
+```bash
+cd ~/railiance-platform
+bridge up openbao-ui-railiance01
+BAO_ADDR=http://127.0.0.1:18200 VAULT_ADDR=http://127.0.0.1:18200 \
+  warden access openbao-platform-admin-login --exec -- \
+  python3 scripts/service_identity_metadata_export.py \
+    --output ~/nk-wp-0054-t04-openbao-identity-metadata.json openbao
+
+SQL="$(python3 scripts/service_identity_metadata_export.py postgresql-sql)"
+P=$(ssh railiance01 kubectl -n databases get cluster net-kingdom-pg -o jsonpath='{.status.currentPrimary}')
+ssh railiance01 "kubectl -n databases exec $P -c postgres -- psql -U postgres -d postgres -AtX -c $(printf %q "$SQL")" \
+  | python3 scripts/service_identity_metadata_export.py \
+      --output ~/nk-wp-0054-t04-postgresql-identity-metadata.json postgresql --cluster net-kingdom-pg
+```
+
+Exit 0 writes the file; exit 3 means a secret-shaped value was found and nothing
+was written; exit 2 is a read failure. Return both JSON files to NetKingdom; they
+hold no secret values.