#!/usr/bin/env python3 """Silent contained aal2 rollout for auth/netkingdom/role/platform-admin (RPF-WP-0065-T04). Each attended mode runs as the one reviewed child of `warden access openbao-platform-admin-login --exec -- ...`, writes exactly one non-secret 0600 receipt (O_EXCL) to the caller-given path and prints nothing. Any refusal or mismatch exits 1 after recording a status in the receipt. prepare --receipt R back up the live role into R, then add only the claim mapping /assurance/level -> assurance_level (no new binding; logins are not restricted yet) apply --receipt R gate: this login's token metadata must show role=platform-admin and assurance_level=aal2; then write the declared role and read it back verify --receipt R role equals the declaration and this fresh login is aal2 with platform-admin + operator-custody rollback --backup B --receipt R restore the role recorded by `prepare` in B rollback-body --backup B --out F local only (no bao): write the backed-up role as a bao write body for the docs/openbao.md break-glass path Only role configuration (no secrets in an OIDC role) and the token's policy set, role name and assurance level are recorded: never a token, accessor, email or username. """ from datetime import datetime, timezone import importlib.util import json import os from pathlib import Path import subprocess import sys HERE = Path(__file__).resolve().parent _spec = importlib.util.spec_from_file_location('loopback', HERE / 'openbao_operator_loopback_callback.py') loopback = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(loopback) ROLE = loopback.ROLE Refused = loopback.Refused DECLARED_FILE = HERE.parent / 'openbao/auth/netkingdom-platform-admin-role.json' EXPECTED_BOUND = {'groups': ['net-kingdom-admins'], '/assurance/level': 'aal2'} MAPPING = ('/assurance/level', 'assurance_level') REQUIRED_POLICIES = {'platform-admin', 'operator-custody'} SCHEMA = 'railiance-platform.openbao-platform-admin-aal2.v1' LEGACY_ALIASES = {'policies': 'token_policies', 'ttl': 'token_ttl'} def declared(): return json.loads(DECLARED_FILE.read_text(encoding='utf-8')) def norm(value): if isinstance(value, list): return sorted(norm(item) for item in value) if all(isinstance(i, str) for i in value) else [norm(i) for i in value] if isinstance(value, dict): return {key: norm(item) for key, item in value.items()} return value def drift(want, live): """Keys of `want` whose live value differs. Legacy aliases may be folded by OpenBao.""" out = [] for key, value in want.items(): if key in live: if norm(live[key]) != norm(value): out.append(key) elif key in LEGACY_ALIASES and norm(live.get(LEGACY_ALIASES[key])) == norm(value): continue elif value not in ([], {}, '', 0, None, False): out.append(key) return sorted(out) def bound_ok(role): return norm(role.get('bound_claims')) == norm(EXPECTED_BOUND) def mapped(role): return (role.get('claim_mappings') or {}).get(MAPPING[0]) == MAPPING[1] def token_facts(): """Value-free view of this session: policies, role name, assurance level.""" result = subprocess.run(['bao', 'read', '-format=json', 'auth/token/lookup-self'], capture_output=True, check=True, timeout=30) data = json.loads(result.stdout)['data'] meta = data.get('meta') or {} policies = set(data.get('policies') or []) | set(data.get('identity_policies') or []) return {'policies': sorted(policies), 'role': meta.get('role'), 'assurance_level': meta.get(MAPPING[1]), 'path': data.get('path')} def gate(facts): if facts['role'] != 'platform-admin': raise Refused('gate_wrong_role') if facts['assurance_level'] != 'aal2': raise Refused('gate_login_not_aal2') def prepare(read=loopback.read_role, write=loopback.write_role): original = read() if mapped(original): return original, False mappings = dict(original.get('claim_mappings') or {}, **{MAPPING[0]: MAPPING[1]}) write(dict(loopback.payload(original), claim_mappings=mappings)) after = read() if not mapped(after): raise Refused('readback_mapping_missing') rest = {k: v for k, v in loopback.payload(original).items() if k != 'claim_mappings'} if drift(rest, after): raise Refused('readback_settings_changed') return original, True def apply(read=loopback.read_role, write=loopback.write_role, facts=token_facts): before = read() if not mapped(before): raise Refused('not_prepared') observed = facts() gate(observed) want = declared() if not bound_ok(want) or not mapped(want): raise Refused('declaration_unexpected') changed = bool(drift(want, before)) if changed: write(dict(loopback.payload(before), **want)) after = read() if drift(want, after) or not bound_ok(after): raise Refused('readback_mismatch') return before, observed, changed, after def verify(read=loopback.read_role, facts=token_facts): role = read() observed = facts() if drift(declared(), role) or not bound_ok(role): raise Refused('role_not_declared') gate(observed) if not REQUIRED_POLICIES <= set(observed['policies']): raise Refused('token_policies_missing') return role, observed def load_backup(path): body = json.loads(Path(path).read_text(encoding='utf-8')) role = body.get('backup_role') if body.get('schema') != SCHEMA or body.get('mode') != 'prepare' or not isinstance(role, dict): raise Refused('backup_receipt_invalid') if role.get('role_type') != 'oidc' or bound_ok(role): raise Refused('backup_role_unexpected') return role def rollback_body(backup): body = loopback.payload(backup) body.setdefault('claim_mappings', {}) return body def rollback(backup, read=loopback.read_role, write=loopback.write_role): want = rollback_body(backup) write(want) after = read() if drift(want, after): raise Refused('readback_mismatch') return after def write_receipt(path, mode, status, **extra): fd = os.open(Path(path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) body = { 'schema': SCHEMA, 'mode': mode, 'status': status, 'role_path': ROLE, 'observed_at': datetime.now(timezone.utc).isoformat(), 'expected_bound_claims': EXPECTED_BOUND, 'credential_values_emitted': False, **extra, } with os.fdopen(fd, 'w', encoding='utf-8') as handle: json.dump(body, handle, indent=2, sort_keys=True) handle.write('\n') def parse(argv): if not argv: raise SystemExit(2) mode, opts, args = argv[0], {}, list(argv[1:]) while args: if args[0] in ('--receipt', '--backup', '--out') and len(args) > 1: opts[args[0][2:]] = args[1] args = args[2:] else: raise SystemExit(2) need = {'prepare': {'receipt'}, 'apply': {'receipt'}, 'verify': {'receipt'}, 'rollback': {'receipt', 'backup'}, 'rollback-body': {'backup', 'out'}} if mode not in need or set(opts) != need[mode]: raise SystemExit(2) return mode, opts def run(mode, opts): if mode == 'prepare': original, changed = prepare() return 'prepared' if changed else 'already_prepared', { 'backup_role': loopback.payload(original), 'changed': changed} if mode == 'apply': before, observed, changed, after = apply() return 'applied' if changed else 'already_applied', { 'before_role': loopback.payload(before), 'gate_token': observed, 'changed': changed, 'bound_claims': after.get('bound_claims')} if mode == 'verify': role, observed = verify() return 'verified', {'token': observed, 'bound_claims': role.get('bound_claims')} after = rollback(load_backup(opts['backup'])) return 'rolled_back', {'bound_claims': after.get('bound_claims'), 'claim_mappings': after.get('claim_mappings')} def main(argv): mode, opts = parse(argv) if mode == 'rollback-body': try: body = rollback_body(load_backup(opts['backup'])) fd = os.open(Path(opts['out']), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, 'w', encoding='utf-8') as handle: json.dump(body, handle, indent=2, sort_keys=True) handle.write('\n') return 0 except Exception: return 1 if Path(opts['receipt']).exists(): return 1 # never touch OpenBao when the evidence could not be written try: loopback.require_attended() status, extra = run(mode, opts) write_receipt(opts['receipt'], mode, status, **extra) return 0 except Exception as error: try: write_receipt(opts['receipt'], mode, loopback.classify(error)) except Exception: pass return 1 if __name__ == '__main__': raise SystemExit(main(sys.argv[1:]))