Add attended prune of retired bao.coulomb.social callbacks from platform-admin
Read-preserve-write via the reviewed loopback helper; keeps the tunnel callback and every other setting; also captures non-secret OIDC mount config. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 150322@bnt-lap001 Assistant-Session: 16a7b788-374e-4915-a1df-fc87ffd9a5e4
This commit is contained in:
parent
85f6a14892
commit
7f25af3cb1
3 changed files with 166 additions and 0 deletions
4
scripts/openbao-platform-admin-callback-prune.sh
Executable file
4
scripts/openbao-platform-admin-callback-prune.sh
Executable file
|
|
@ -0,0 +1,4 @@
|
||||||
|
#!/usr/bin/env bash
|
||||||
|
# Silent child for the governed attended login; results go to the receipt only.
|
||||||
|
set -euo pipefail
|
||||||
|
exec python3 "$(dirname "$0")/openbao_platform_admin_callback_prune.py" "$@" >/dev/null 2>&1
|
||||||
104
scripts/openbao_platform_admin_callback_prune.py
Executable file
104
scripts/openbao_platform_admin_callback_prune.py
Executable file
|
|
@ -0,0 +1,104 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Silent contained removal of the retired public callbacks from platform-admin.
|
||||||
|
|
||||||
|
Removes exactly the two bao.coulomb.social callbacks retired by RPF-WP-0025-T03
|
||||||
|
from auth/netkingdom/role/platform-admin. Every other role setting is kept,
|
||||||
|
using the reviewed read/payload/write helpers of the loopback-callback script.
|
||||||
|
Refuses unless the tunnel callback is present and stays present. Writes
|
||||||
|
nothing if the retired callbacks are already gone. One non-secret receipt; no
|
||||||
|
output.
|
||||||
|
"""
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
import importlib.util
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
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
|
||||||
|
KEEP = loopback.CALLBACK # http://127.0.0.1:18200/ui/vault/auth/netkingdom/oidc/callback
|
||||||
|
RETIRED = (
|
||||||
|
'https://bao.coulomb.social/ui/vault/auth/netkingdom/oidc/callback',
|
||||||
|
'https://bao.coulomb.social/ui/vault/auth/keycape/oidc/callback',
|
||||||
|
)
|
||||||
|
Refused = loopback.Refused
|
||||||
|
MOUNTS = ('netkingdom', 'keycape')
|
||||||
|
CONFIG_FIELDS = ('oidc_discovery_url', 'oidc_client_id', 'default_role', 'bound_issuer',
|
||||||
|
'oidc_response_mode', 'oidc_response_types', 'jwt_supported_algs',
|
||||||
|
'provider_config', 'namespace_in_state')
|
||||||
|
|
||||||
|
|
||||||
|
def read_mount_configs():
|
||||||
|
"""Non-secret OIDC mount settings; OpenBao never returns oidc_client_secret."""
|
||||||
|
import subprocess
|
||||||
|
configs = {}
|
||||||
|
for mount in MOUNTS:
|
||||||
|
try:
|
||||||
|
result = subprocess.run(['bao', 'read', '-format=json', 'auth/%s/config' % mount],
|
||||||
|
capture_output=True, check=True, timeout=30)
|
||||||
|
data = json.loads(result.stdout)['data']
|
||||||
|
configs[mount] = {key: data[key] for key in CONFIG_FIELDS if key in data}
|
||||||
|
except Exception:
|
||||||
|
configs[mount] = {'error': 'read_failed'}
|
||||||
|
return configs
|
||||||
|
|
||||||
|
|
||||||
|
def prune(read=loopback.read_role, write=loopback.write_role):
|
||||||
|
original = read()
|
||||||
|
uris = original['allowed_redirect_uris']
|
||||||
|
if KEEP not in uris:
|
||||||
|
raise Refused('tunnel_callback_missing')
|
||||||
|
remaining = [uri for uri in uris if uri not in RETIRED]
|
||||||
|
if remaining == uris:
|
||||||
|
return False, uris
|
||||||
|
desired = dict(loopback.payload(original), allowed_redirect_uris=remaining)
|
||||||
|
if loopback.preserved(read()) != loopback.preserved(original):
|
||||||
|
raise Refused('role_changed_before_write')
|
||||||
|
write(desired)
|
||||||
|
after = read()
|
||||||
|
if sorted(after.get('allowed_redirect_uris', [])) != sorted(remaining):
|
||||||
|
raise Refused('readback_callbacks_mismatch')
|
||||||
|
if loopback.preserved(after) != loopback.preserved(original):
|
||||||
|
raise Refused('readback_settings_changed')
|
||||||
|
return True, after['allowed_redirect_uris']
|
||||||
|
|
||||||
|
|
||||||
|
def write_receipt(path, status, **extra):
|
||||||
|
fd = os.open(Path(path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||||||
|
body = {
|
||||||
|
'schema': 'railiance-platform.openbao-callback-prune.v1',
|
||||||
|
'observed_at': datetime.now(timezone.utc).isoformat(),
|
||||||
|
'role': ROLE, 'retired': list(RETIRED), 'status': status,
|
||||||
|
'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 main(argv):
|
||||||
|
if len(argv) != 2 or argv[0] != '--receipt':
|
||||||
|
return 2
|
||||||
|
receipt = argv[1]
|
||||||
|
try:
|
||||||
|
loopback.require_attended()
|
||||||
|
changed, uris = prune()
|
||||||
|
write_receipt(receipt, 'pruned' if changed else 'already_pruned',
|
||||||
|
changed=changed, allowed_redirect_uris=uris,
|
||||||
|
mount_configs=read_mount_configs())
|
||||||
|
return 0
|
||||||
|
except Exception as error:
|
||||||
|
try:
|
||||||
|
write_receipt(receipt, loopback.classify(error), changed=False)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return 1
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
raise SystemExit(main(sys.argv[1:]))
|
||||||
58
tests/test_openbao_platform_admin_callback_prune.py
Normal file
58
tests/test_openbao_platform_admin_callback_prune.py
Normal file
|
|
@ -0,0 +1,58 @@
|
||||||
|
import copy
|
||||||
|
import importlib.util
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
spec = importlib.util.spec_from_file_location(
|
||||||
|
'prune', Path(__file__).resolve().parents[1] / 'scripts/openbao_platform_admin_callback_prune.py')
|
||||||
|
m = importlib.util.module_from_spec(spec)
|
||||||
|
spec.loader.exec_module(m)
|
||||||
|
|
||||||
|
DECLARED = json.loads((Path(__file__).resolve().parents[1] / 'openbao/auth/netkingdom-platform-admin-role.json').read_text())
|
||||||
|
|
||||||
|
|
||||||
|
def live(role):
|
||||||
|
state = {'role': copy.deepcopy(role)}
|
||||||
|
writes = []
|
||||||
|
|
||||||
|
def write(value):
|
||||||
|
writes.append(value)
|
||||||
|
state['role'].update(value)
|
||||||
|
return state, writes, (lambda: copy.deepcopy(state['role'])), write
|
||||||
|
|
||||||
|
|
||||||
|
def test_prunes_only_the_two_retired_callbacks_and_keeps_settings():
|
||||||
|
role = dict(DECLARED, allowed_redirect_uris=list(DECLARED['allowed_redirect_uris']) + list(m.RETIRED)
|
||||||
|
if not set(m.RETIRED) <= set(DECLARED['allowed_redirect_uris']) else list(DECLARED['allowed_redirect_uris']))
|
||||||
|
state, writes, read, write = live(role)
|
||||||
|
changed, uris = m.prune(read, write)
|
||||||
|
assert changed and len(writes) == 1
|
||||||
|
assert not set(m.RETIRED) & set(uris)
|
||||||
|
assert m.KEEP in uris and 'http://localhost:8250/oidc/callback' in uris
|
||||||
|
assert state['role']['token_policies'] == ['platform-admin', 'operator-custody']
|
||||||
|
assert state['role']['bound_claims'] == {'groups': ['net-kingdom-admins']}
|
||||||
|
|
||||||
|
|
||||||
|
def test_idempotent_when_already_pruned():
|
||||||
|
role = dict(DECLARED, allowed_redirect_uris=[u for u in DECLARED['allowed_redirect_uris'] if u not in m.RETIRED])
|
||||||
|
_, writes, read, write = live(role)
|
||||||
|
assert m.prune(read, write)[0] is False and writes == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_refuses_without_tunnel_callback():
|
||||||
|
role = dict(DECLARED, allowed_redirect_uris=list(m.RETIRED))
|
||||||
|
_, writes, read, write = live(role)
|
||||||
|
with pytest.raises(m.Refused, match='tunnel_callback_missing'):
|
||||||
|
m.prune(read, write)
|
||||||
|
assert writes == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_refuses_when_readback_drops_other_settings():
|
||||||
|
role = copy.deepcopy(DECLARED)
|
||||||
|
calls = iter([copy.deepcopy(role), copy.deepcopy(role),
|
||||||
|
dict(copy.deepcopy(role), token_ttl=60,
|
||||||
|
allowed_redirect_uris=[u for u in role['allowed_redirect_uris'] if u not in m.RETIRED])])
|
||||||
|
with pytest.raises(m.Refused, match='readback_settings_changed'):
|
||||||
|
m.prune(lambda: next(calls), lambda value: None)
|
||||||
Loading…
Add table
Add a link
Reference in a new issue