railiance-platform/scripts/openbao_operator_loopback_callback.py

#!/usr/bin/env python3
"""Silent contained callback update; preserve the existing administrator role."""
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile

ROLE = 'auth/netkingdom/role/platform-admin'
CALLBACK = 'http://127.0.0.1:18200/ui/vault/auth/netkingdom/oidc/callback'
WRITABLE = (
    'role_type', 'user_claim', 'user_claim_json_pointer', 'groups_claim',
    'bound_claims', 'bound_claims_type', 'bound_audiences', 'bound_subject',
    'claim_mappings', 'oidc_scopes', 'allowed_redirect_uris',
    'clock_skew_leeway', 'expiration_leeway', 'not_before_leeway', 'max_age',
    'verbose_oidc_logging', 'token_ttl', 'token_max_ttl', 'token_explicit_max_ttl',
    'token_policies', 'token_bound_cidrs', 'token_no_default_policy',
    'token_num_uses', 'token_period', 'token_type', 'policies', 'ttl', 'max_ttl',
    'period', 'num_uses',
)
PRESERVED = (
    'role_type', 'user_claim', 'groups_claim', 'bound_claims', 'claim_mappings',
    'oidc_scopes', 'token_policies', 'policies', 'token_ttl', 'ttl',
)


class Refused(Exception):
    pass


def require_attended():
    if Path.home().parent.name != '.warden-attended-login' or os.getenv('BAO_TOKEN') or os.getenv('VAULT_TOKEN'):
        raise Refused('attended_envelope_required')


def payload(role):
    return {key: role[key] for key in WRITABLE if key in role and role[key] is not None}


def preserved(role):
    return {key: role.get(key) for key in PRESERVED}


def read_role():
    result = subprocess.run(['bao', 'read', '-format=json', ROLE],
                            capture_output=True, check=True, timeout=30)
    role = json.loads(result.stdout)['data']
    policies = role.get('token_policies') or role.get('policies') or []
    if (role.get('role_type') != 'oidc'
            or 'platform-admin' not in policies
            or not isinstance(role.get('allowed_redirect_uris'), list)
            or not all(isinstance(uri, str) for uri in role['allowed_redirect_uris'])):
        raise Refused('unexpected_role')
    return role


def write_role(desired):
    handle = tempfile.NamedTemporaryFile('w', encoding='utf-8', delete=False)
    try:
        json.dump(desired, handle)
        handle.close()
        subprocess.run(['bao', 'write', ROLE, '@' + handle.name],
                       capture_output=True, check=True, timeout=30)
    finally:
        Path(handle.name).unlink(missing_ok=True)


def update(read=read_role, write=write_role):
    original = read()
    if CALLBACK in original['allowed_redirect_uris']:
        return False
    desired = dict(payload(original),
                   allowed_redirect_uris=original['allowed_redirect_uris'] + [CALLBACK])
    if preserved(read()) != preserved(original):
        raise Refused('role_changed_before_write')
    write(desired)
    after = read()
    if CALLBACK not in after.get('allowed_redirect_uris', []):
        raise Refused('readback_callback_missing')
    if preserved(after) != preserved(original):
        raise Refused('readback_settings_changed')
    if set(original['allowed_redirect_uris']) - set(after['allowed_redirect_uris']):
        raise Refused('readback_existing_callback_dropped')
    return True


def write_receipt(path, status, **extra):
    from datetime import datetime, timezone
    path = Path(path)
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    fd = os.open(path, flags, 0o600)
    body = {
        'schema': 'railiance-platform.openbao-loopback-callback.v1',
        'observed_at': datetime.now(timezone.utc).isoformat(),
        'role': ROLE, 'callback': CALLBACK, '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 classify(error):
    if isinstance(error, Refused):
        return str(error)
    if isinstance(error, subprocess.CalledProcessError):
        command = error.cmd[1] if isinstance(error.cmd, list) and len(error.cmd) > 1 else ''
        return 'bao_write_failed' if command == 'write' else 'bao_read_failed'
    return 'contained_operation_failed'


def parse(argv):
    receipt = None
    check_only = False
    args = list(argv)
    while args:
        if args[0] == '--receipt':
            if len(args) < 2:
                raise SystemExit(2)
            receipt = args[1]
            args = args[2:]
        elif args[0] == '--check-only':
            check_only = True
            args = args[1:]
        else:
            raise SystemExit(2)
    return receipt, check_only


def main(argv):
    receipt, check_only = parse(argv)
    try:
        if check_only:
            present = CALLBACK in read_role()['allowed_redirect_uris']
            if receipt:
                write_receipt(receipt, 'present' if present else 'absent')
            return 0 if present else 3
        require_attended()
        changed = update()
        if receipt:
            write_receipt(receipt, 'applied' if changed else 'already_present', changed=changed)
        return 0
    except Exception as error:
        if receipt:
            try:
                write_receipt(receipt, classify(error))
            except Exception:
                pass
        return 1


if __name__ == '__main__':
    raise SystemExit(main(sys.argv[1:]))