coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

NK-WP-0054-T04: first live conformance run (OpenBao, PostgreSQL)
All checks were successful
Bounded-execution contracts / contracts (push) Successful in 2m31s
Details
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 2s
Details

Browse source 
observe.py compares the service-identity declarations against
railiance-platform's value-safe live metadata (b0b6c0c) with a 7-day
freshness bound; read-only, offline, 9 tests. Result: 5 conformant,
2 unknown; undeclared keycape/platform-admin alias grants platform-admin
without aal2 (non-conformant), 14 undeclared roles unknown.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 149903@bnt-lap001
Assistant-Session: 4c20a536-1452-4a85-bc09-e1987d9817d6
This commit is contained in:
tegwick 2026-11-07 14:32:19 +01:00
parent e1cad11a80
commit 4a684baabf
6 changed files with 702 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

267
tools/service-identity/observe.py Normal file
View file

@ -0,0 +1,267 @@
#!/usr/bin/env python3
# /// script
# dependencies = ["PyYAML>=6,<7"]
# ///
"""Live conformance check: declared service identities against owner metadata (NK-WP-0054-T04).
Compares the checked-in declarations (`tools/service-identity/declarations/`) with value-safe, owner-provided
metadata (railiance-platform `service-identity-metadata.v1`, exported by `scripts/service_identity_metadata_export.py`).
Each mapped identity ends `conformant`, `non-conformant` or `unknown`; every finding names its owner. Observed roles
that no declaration covers are reported too, because an undeclared login path is itself a finding.
Offline and read-only: it reads two JSON files and the declarations, never contacts a service, never reads a Secret
and changes nothing. The result is a proposal input for posture feedback, not authority.
python3 tools/service-identity/observe.py --openbao FILE --postgresql FILE [--now ISO] [--max-age-days 7] [--json]
"""
from __future__ import annotations
import argparse
import datetime as dt
import json
import pathlib
import sys
from typing import Any
import yaml
ROOT = pathlib.Path(__file__).resolve().parents[2]
DECLARATIONS = ROOT / "tools/service-identity/declarations"
SCHEMA = "railiance-platform.service-identity-metadata.v1"
AAL2 = "/assurance/level"
# Declared identity -> the OpenBao roles that realize it, with the facts the declaration states about them.
# Kept here, not in the declaration schema, until the owner confirms the declarations (owner_confirmed: false).
OPENBAO_MAP: dict[tuple[str, str], list[dict[str, Any]]] = {
("openbao", "netkingdom-platform-admin"): [
{"mount": "netkingdom", "role": "platform-admin", "policies": ["platform-admin", "operator-custody"],
"groups": ["net-kingdom-admins"], "ttl": 3600}],
("openbao", "credential-broker-keycape-client-secret-generate"): [
{"mount": "netkingdom", "role": "credential-broker-keycape-client-secret-generate",
"policies": ["credential-broker-keycape-client-secret-generate-issuer"], "groups": ["net-kingdom-admins"],
"ttl": 900}],
("openbao", "fluid-telegram-operators"): [
{"mount": "netkingdom", "role": "fluid-telegram-hall-of-helix-workload-kv",
"policies": ["workload-kv-fluid-telegram-hall-of-helix-operator"], "groups": ["fluid-telegram-operators"],
"ttl": 900, "max_ttl": 1800, "kv_data_paths": 4, "no_list_delete": True}],
("openbao", "secrets-engine-openbao"): [
{"mount": "keycape-services", "role": "secrets-engine-openbao", "policies": ["secrets-engine-login-self"],
"ttl": 300, "bound_subject": "service:secrets-engine"}],
("openbao", "coding-agent"): [
{"mount": "netkingdom", "role": "coding-agent",
"policies": ["agent-high-risk-boundary", "workload-kv-read-issue-core-runtime"], "ttl": 900, "max_ttl": 900,
"observe_only": True},
{"mount": "approle", "role": "coding-agent-railiance-platform",
"policies": ["agent-high-risk-boundary", "workload-kv-read-issue-core-runtime"], "ttl": 900, "max_ttl": 900,
"observe_only": True}],
("bounded-executor", "net-kingdom-bounded-executor"): [
{"mount": "kubernetes", "role": "net-kingdom-bind-executor-svc-authelia-write",
"service_account": ("sso", "nk-bind-executor")}],
}
# Policies that make a role an administrator of OpenBao itself.
ADMIN_POLICIES = {"platform-admin", "operator-custody"}
def load_metadata(path: pathlib.Path, service: str, now: dt.datetime, max_age: dt.timedelta) -> tuple[dict, dict]:
doc = json.loads(path.read_text())
if doc.get("schema") != SCHEMA or doc.get("service") != service:
raise SystemExit(f"{path}: not {SCHEMA} for {service}")
if doc.get("value_safe") is not True:
raise SystemExit(f"{path}: not marked value_safe; refusing to read it")
collected = dt.datetime.fromisoformat(doc["collected_at"].replace("Z", "+00:00"))
age = now - collected
freshness = {"service": service, "collected_at": doc["collected_at"], "source_kind": doc.get("source_kind"),
"age_hours": round(age.total_seconds() / 3600, 1), "max_age_days": max_age.days,
"fresh": dt.timedelta(0) <= age <= max_age}
return doc[service], freshness
def finding(identity: str, owner: str, verdict: str, rule: str, detail: str) -> dict[str, str]:
return {"identity": identity, "owner": owner, "verdict": verdict, "rule": rule, "detail": detail}
def check_role(spec: dict, role: dict | None, declared: dict, ident: str, owner: str) -> list[dict]:
where = f"{spec['mount']}/{spec['role']}"
if role is None:
return [finding(ident, owner, "unknown", "observed",
f"role {where} is not in the export (missing, or created after collected_at)")]
out = []
if "policies" in spec and sorted(role.get("token_policies", [])) != sorted(spec["policies"]):
out.append(finding(ident, owner, "non-conformant", "grants",
f"{where} policies {role.get('token_policies')} != declared {spec['policies']}"))
claims = role.get("bound_claims") or {}
if declared.get("second_factor") == "required" and claims.get(AAL2) != "aal2":
out.append(finding(ident, owner, "non-conformant", "second_factor",
f"{where} does not bind {AAL2}=aal2"))
if "groups" in spec and sorted(claims.get("groups") or []) != sorted(spec["groups"]):
out.append(finding(ident, owner, "non-conformant", "consumers",
f"{where} binds groups {claims.get('groups')} != declared {spec['groups']}"))
if "ttl" in spec and role.get("token_ttl_seconds", 0) > spec["ttl"]:
out.append(finding(ident, owner, "non-conformant", "lifetime",
f"{where} token ttl {role.get('token_ttl_seconds')}s > declared {spec['ttl']}s"))
if "max_ttl" in spec and role.get("token_max_ttl_seconds") != spec["max_ttl"]:
out.append(finding(ident, owner, "non-conformant", "lifetime",
f"{where} max ttl {role.get('token_max_ttl_seconds')}s != declared {spec['max_ttl']}s"))
if "ttl" in spec and "max_ttl" not in spec and not role.get("token_max_ttl_seconds") \
and not role.get("token_explicit_max_ttl_seconds"):
out.append(finding(ident, owner, "unknown", "lifetime",
f"{where} has no role max ttl; renewal is bounded by the mount/system default, "
"which the export does not include"))
if "bound_subject" in spec and role.get("bound_subject") != spec["bound_subject"]:
out.append(finding(ident, owner, "non-conformant", "subject",
f"{where} bound_subject {role.get('bound_subject')} != {spec['bound_subject']}"))
if "service_account" in spec:
ns, sa = spec["service_account"]
if role.get("bound_service_account_names") != [sa] or role.get("bound_service_account_namespaces") != [ns]:
out.append(finding(ident, owner, "non-conformant", "consumers", f"{where} is not bound to exactly {ns}/{sa}"))
rules = role.get("effective_rules") or []
if spec.get("observe_only"):
data_reads = [r["path"] for r in rules if "/data/" in r["path"]
and set(r["capabilities"]) & {"read", "create", "update", "delete", "list"}
and "deny" not in r["capabilities"]]
if data_reads:
out.append(finding(ident, owner, "non-conformant", "grants", f"{where} can read KV data: {data_reads}"))
elif not role.get("effective_complete"):
out.append(finding(ident, owner, "unknown", "grants", f"{where} effective rules are incomplete"))
if "kv_data_paths" in spec:
data = [r for r in rules if "/data/" in r["path"]]
if len(data) != spec["kv_data_paths"] or any(r["wildcard"] for r in data):
out.append(finding(ident, owner, "non-conformant", "grants",
f"{where} has {len(data)} KV data paths (declared {spec['kv_data_paths']} exact)"))
if spec.get("no_list_delete") and any({"list", "delete"} & set(r["capabilities"]) for r in data):
out.append(finding(ident, owner, "non-conformant", "grants", f"{where} allows list or delete"))
return out
def check_openbao(meta: dict, decls: dict[str, dict]) -> tuple[list[dict], list[dict]]:
roles = {(r["mount"], r["name"]): r for r in meta["roles"]}
results, mapped = [], set()
for (service, ident), specs in OPENBAO_MAP.items():
doc = decls.get(service)
declared = next((i for i in (doc or {}).get("identities", []) if i["name"] == ident), None)
if declared is None:
continue
owner = declared.get("owner", doc.get("owner"))
findings = []
for spec in specs:
mapped.add((spec["mount"], spec["role"]))
findings += check_role(spec, roles.get((spec["mount"], spec["role"])), declared, ident, owner)
verdict = ("non-conformant" if any(f["verdict"] == "non-conformant" for f in findings)
else "unknown" if findings else "conformant")
results.append({"service": service, "evidence": "openbao", "identity": ident, "owner": owner, "verdict": verdict,
"observed_as": [f"{s['mount']}/{s['role']}" for s in specs], "findings": findings})
# Undeclared roles: alternative paths to a declared identity's privileges, and human logins without aal2.
declared_policies = {p for specs in OPENBAO_MAP.values() for s in specs for p in s.get("policies", [])}
undeclared = []
for key, role in sorted(roles.items()):
if key in mapped:
continue
where = f"{key[0]}/{key[1]}"
policies = set(role.get("token_policies", []))
claims = role.get("bound_claims") or {}
human = role.get("role_type") == "oidc"
if policies & ADMIN_POLICIES:
undeclared.append(finding(where, "railiance-platform", "non-conformant", "SI-11",
f"undeclared role grants admin policies {sorted(policies & ADMIN_POLICIES)}"
+ ("" if claims.get(AAL2) == "aal2" else " without binding aal2")))
elif policies & declared_policies:
undeclared.append(finding(where, "railiance-platform", "unknown", "SI-11",
f"undeclared role shares declared policies {sorted(policies & declared_policies)}"))
elif human and claims.get(AAL2) != "aal2":
undeclared.append(finding(where, "railiance-platform", "unknown", "SI-11",
f"undeclared human OIDC role without aal2 (groups {claims.get('groups')}, "
f"policies {sorted(policies)})"))
return results, undeclared
def check_postgresql(meta: dict, decls: dict[str, dict]) -> tuple[list[dict], list[dict]]:
roles = {r["name"]: r for r in meta["roles"]}
owners = {d["name"]: d["owner"] for d in meta["databases"]}
doc = decls.get("postgresql") or {}
results, declared_names = [], set()
for ident in doc.get("identities", []):
name, owner = ident["name"], ident.get("owner", doc.get("owner"))
declared_names.add(name)
role, findings = roles.get(name), []
if role is None:
findings.append(finding(name, owner, "unknown", "observed", "role not in the export"))
else:
for flag in ("superuser", "create_role", "create_db", "replication", "bypass_rls"):
if role.get(flag):
findings.append(finding(name, owner, "non-conformant", "grants", f"role has {flag}"))
if ident.get("interactive_login") is False and role.get("member_of"):
findings.append(finding(name, owner, "unknown", "grants", f"member of {role['member_of']}"))
for grant in ident.get("grants", []):
if grant["mode"] == "ADMINISTER" and owners.get(grant["scope"]) != name:
findings.append(finding(name, owner, "non-conformant", "grants",
f"declared owner of {grant['scope']}, observed {owners.get(grant['scope'])}"))
verdict = ("non-conformant" if any(f["verdict"] == "non-conformant" for f in findings)
else "unknown" if findings else "conformant")
results.append({"service": "postgresql", "evidence": "postgresql", "identity": name, "owner": owner, "verdict": verdict,
"observed_as": [f"{meta['cluster']}/{name}"], "findings": findings})
undeclared = []
for name, role in sorted(roles.items()):
if name in declared_names:
continue
flags = [f for f in ("superuser", "create_role", "create_db", "replication", "bypass_rls") if role.get(f)]
owns = sorted(db for db, o in owners.items() if o == name)
undeclared.append(finding(f"{meta['cluster']}/{name}", "railiance-platform", "unknown", "SI-11",
f"undeclared login role (flags {flags or 'none'}, owns {owns or 'nothing'})"))
return results, undeclared
def run(openbao: pathlib.Path, postgresql: pathlib.Path, now: dt.datetime, max_age_days: int) -> dict:
decls = {d["service"]: d for d in (yaml.safe_load(p.read_text()) for p in sorted(DECLARATIONS.glob("*.yaml")))}
max_age = dt.timedelta(days=max_age_days)
ob_meta, ob_fresh = load_metadata(openbao, "openbao", now, max_age)
pg_meta, pg_fresh = load_metadata(postgresql, "postgresql", now, max_age)
ob_results, ob_undeclared = check_openbao(ob_meta, decls)
pg_results, pg_undeclared = check_postgresql(pg_meta, decls)
results = ob_results + pg_results
stale = [f["service"] for f in (ob_fresh, pg_fresh) if not f["fresh"]]
if stale: # stale evidence cannot establish conformance
for r in results:
if r["verdict"] == "conformant" and r["evidence"] in stale:
r["verdict"] = "unknown"
r["findings"].append(finding(r["identity"], r["owner"], "unknown", "freshness", "evidence is stale"))
counts: dict[str, int] = {}
for r in results:
counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1
return {"as_of": now.isoformat(timespec="seconds").replace("+00:00", "Z"), "freshness": [ob_fresh, pg_fresh],
"summary": counts, "identities": results, "undeclared": ob_undeclared + pg_undeclared,
"undeclared_summary": {"non-conformant": sum(f["verdict"] == "non-conformant" for f in ob_undeclared + pg_undeclared),
"unknown": sum(f["verdict"] == "unknown" for f in ob_undeclared + pg_undeclared)}}
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--openbao", type=pathlib.Path, required=True)
ap.add_argument("--postgresql", type=pathlib.Path, required=True)
ap.add_argument("--now", help="ISO time for the freshness bound (default: now)")
ap.add_argument("--max-age-days", type=int, default=7)
ap.add_argument("--json", action="store_true")
args = ap.parse_args(argv)
now = dt.datetime.fromisoformat(args.now.replace("Z", "+00:00")) if args.now else dt.datetime.now(dt.UTC)
report = run(args.openbao, args.postgresql, now, args.max_age_days)
if args.json:
print(json.dumps(report, indent=2, sort_keys=True))
else:
for f in report["freshness"]:
print(f"evidence {f['service']}: collected {f['collected_at']}, {f['age_hours']} h old, "
f"{'fresh' if f['fresh'] else 'STALE'} (bound {f['max_age_days']} d)")
for r in report["identities"]:
print(f"{r['service']}/{r['identity']}: {r['verdict']} [owner {r['owner']}]")
for f in r["findings"]:
print(f" - {f['verdict']} {f['rule']}: {f['detail']}")
for f in report["undeclared"]:
print(f"undeclared {f['identity']}: {f['verdict']} {f['rule']} [owner {f['owner']}]: {f['detail']}")
print(f"summary: {report['summary']}; undeclared: {report['undeclared_summary']} (as of {report['as_of']})")
nonconf = report["summary"].get("non-conformant", 0) + report["undeclared_summary"]["non-conformant"]
return 1 if nonconf else 0
if __name__ == "__main__":
sys.exit(main())

145
tools/service-identity/tests/test_observe.py Normal file
View file

@ -0,0 +1,145 @@
"""Live conformance check against synthetic owner metadata (NK-WP-0054-T04)."""
from __future__ import annotations
import copy
import datetime as dt
import importlib.util
import json
import pathlib
import pytest
HERE = pathlib.Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location("observe", HERE.parent / "observe.py")
O = importlib.util.module_from_spec(spec)
spec.loader.exec_module(O)
NOW = dt.datetime(2026, 11, 7, 12, tzinfo=dt.UTC)
COLLECTED = "2026-11-05T12:00:00Z"
def oidc(name, policies, groups, aal2=True, ttl=900, max_ttl=900, rules=()):
claims = {"groups": groups} | ({"/assurance/level": "aal2"} if aal2 else {})
return {"mount": "netkingdom", "name": name, "role_type": "oidc", "token_policies": policies,
"bound_claims": claims, "token_ttl_seconds": ttl, "token_max_ttl_seconds": max_ttl,
"effective_complete": True, "effective_rules": list(rules)}
def kv(path, *caps):
return {"path": path, "capabilities": list(caps), "wildcard": False}
AGENT_RULES = [kv("platform/data/workloads/x", "deny"), kv("platform/metadata/workloads/x", "read")]
def openbao_doc():
roles = [
oidc("platform-admin", ["platform-admin", "operator-custody"], ["net-kingdom-admins"], ttl=3600, max_ttl=3600),
oidc("credential-broker-keycape-client-secret-generate",
["credential-broker-keycape-client-secret-generate-issuer"], ["net-kingdom-admins"]),
oidc("fluid-telegram-hall-of-helix-workload-kv", ["workload-kv-fluid-telegram-hall-of-helix-operator"],
["fluid-telegram-operators"], max_ttl=1800,
rules=[kv(f"platform/data/workloads/ft/{i}", "create", "read", "update") for i in range(4)]),
{"mount": "keycape-services", "name": "secrets-engine-openbao", "role_type": "jwt",
"token_policies": ["secrets-engine-login-self"], "bound_subject": "service:secrets-engine",
"token_ttl_seconds": 300, "token_max_ttl_seconds": 300, "effective_rules": []},
{"mount": "netkingdom", "name": "coding-agent", "role_type": "jwt",
"token_policies": ["agent-high-risk-boundary", "workload-kv-read-issue-core-runtime"],
"token_ttl_seconds": 900, "token_max_ttl_seconds": 900, "effective_complete": True,
"effective_rules": AGENT_RULES},
{"mount": "approle", "name": "coding-agent-railiance-platform", "role_type": "approle",
"token_policies": ["agent-high-risk-boundary", "workload-kv-read-issue-core-runtime"],
"token_ttl_seconds": 900, "token_max_ttl_seconds": 900, "effective_complete": True,
"effective_rules": AGENT_RULES},
{"mount": "kubernetes", "name": "net-kingdom-bind-executor-svc-authelia-write", "role_type": "kubernetes",
"bound_service_account_names": ["nk-bind-executor"], "bound_service_account_namespaces": ["sso"],
"token_policies": ["x"], "effective_rules": []},
]
return {"schema": O.SCHEMA, "service": "openbao", "value_safe": True, "source_kind": "live",
"collected_at": COLLECTED, "openbao": {"auth_mounts": [], "policies": [], "roles": roles}}
def pg_doc():
role = {"name": "privacyidea", "can_login": True, "superuser": False, "create_role": False, "create_db": False,
"replication": False, "bypass_rls": False, "member_of": []}
return {"schema": O.SCHEMA, "service": "postgresql", "value_safe": True, "source_kind": "live",
"collected_at": COLLECTED,
"postgresql": {"cluster": "pg", "roles": [role],
"databases": [{"name": "privacyidea_db", "owner": "privacyidea"}]}}
def run(tmp_path, ob=None, pg=None, now=NOW):
a, b = tmp_path / "ob.json", tmp_path / "pg.json"
a.write_text(json.dumps(ob or openbao_doc()))
b.write_text(json.dumps(pg or pg_doc()))
return O.run(a, b, now, 7)
def verdicts(report):
return {r["identity"]: r["verdict"] for r in report["identities"]}
def test_matching_metadata_is_conformant(tmp_path):
report = run(tmp_path)
assert set(verdicts(report).values()) == {"conformant"}, report
assert report["undeclared"] == []
def test_missing_aal2_binding_is_non_conformant(tmp_path):
ob = openbao_doc()
del ob["openbao"]["roles"][0]["bound_claims"]["/assurance/level"]
assert verdicts(run(tmp_path, ob=ob))["netkingdom-platform-admin"] == "non-conformant"
def test_undeclared_admin_path_without_aal2_is_non_conformant(tmp_path):
ob = openbao_doc()
alias = copy.deepcopy(ob["openbao"]["roles"][0])
alias.update(mount="keycape", token_policies=["platform-admin"])
del alias["bound_claims"]["/assurance/level"]
ob["openbao"]["roles"].append(alias)
report = run(tmp_path, ob=ob)
[f] = report["undeclared"]
assert f["identity"] == "keycape/platform-admin" and f["verdict"] == "non-conformant" and f["owner"]
assert O.main(["--openbao", str(tmp_path / "ob.json"), "--postgresql", str(tmp_path / "pg.json"),
"--now", "2026-11-07T12:00:00Z"]) == 1
def test_unbounded_max_ttl_is_unknown_not_a_pass(tmp_path):
ob = openbao_doc()
ob["openbao"]["roles"][0]["token_max_ttl_seconds"] = 0
assert verdicts(run(tmp_path, ob=ob))["netkingdom-platform-admin"] == "unknown"
def test_agent_data_read_is_non_conformant(tmp_path):
ob = openbao_doc()
ob["openbao"]["roles"][4]["effective_rules"] = [kv("platform/data/workloads/x", "read")]
assert verdicts(run(tmp_path, ob=ob))["coding-agent"] == "non-conformant"
def test_missing_role_is_unknown(tmp_path):
ob = openbao_doc()
ob["openbao"]["roles"].pop()
assert verdicts(run(tmp_path, ob=ob))["net-kingdom-bounded-executor"] == "unknown"
def test_postgres_privilege_flag_and_undeclared_role(tmp_path):
pg = pg_doc()
pg["postgresql"]["roles"][0]["create_db"] = True
pg["postgresql"]["roles"].append({"name": "postgres", "superuser": True, "member_of": []})
report = run(tmp_path, pg=pg)
assert verdicts(report)["privacyidea"] == "non-conformant"
assert [f["identity"] for f in report["undeclared"]] == ["pg/postgres"]
def test_stale_evidence_cannot_be_conformant(tmp_path):
report = run(tmp_path, now=NOW + dt.timedelta(days=30))
assert set(verdicts(report).values()) == {"unknown"}
assert not any(f["fresh"] for f in report["freshness"])
def test_refuses_metadata_not_marked_value_safe(tmp_path):
ob = openbao_doc()
ob["value_safe"] = False
with pytest.raises(SystemExit):
run(tmp_path, ob=ob)