railiance-platform/tests/test_openbao_platform_admin_aal2.py

"""RPF-WP-0065-T04 aal2 rollout script against a fake bao. No live access."""
import json
import os
from pathlib import Path
import stat
import subprocess
import sys

import pytest

ROOT = Path(__file__).resolve().parents[1]
WRAPPER = ROOT / 'scripts/openbao-platform-admin-aal2.sh'
DECLARED = json.loads((ROOT / 'openbao/auth/netkingdom-platform-admin-role.json').read_text())
EXPECTED = {'groups': ['net-kingdom-admins'], '/assurance/level': 'aal2'}

# The live role before T04: group binding only, no assurance mapping.
LIVE = {k: v for k, v in DECLARED.items()}
LIVE['bound_claims'] = {'groups': ['net-kingdom-admins']}
LIVE['claim_mappings'] = {'email': 'email', 'preferred_username': 'username'}
LIVE['max_age'] = 0

FAKE_BAO = r'''#!/usr/bin/env python3
import json, os, sys
state_path = os.environ['FAKE_BAO_STATE']
state = json.load(open(state_path))
state.setdefault('calls', []).append(sys.argv[1:])
args = [a for a in sys.argv[1:] if a != '-format=json']
def save():
    json.dump(state, open(state_path, 'w'))
if state.get('fail') in args:
    save(); sys.stderr.write('Code: 403. Errors:\n* permission denied\n'); sys.exit(2)
if args[:2] == ['read', 'auth/token/lookup-self']:
    save(); print(json.dumps({'data': {'id': 'hvs.SECRETTOKENVALUE00000', 'accessor': 'ACCESSORVALUE',
        'policies': ['default', 'operator-custody', 'platform-admin'], 'path': 'auth/netkingdom/oidc/callback',
        'meta': state['meta']}})); sys.exit(0)
if args[:2] == ['read', 'auth/netkingdom/role/platform-admin']:
    save(); print(json.dumps({'data': state['role']})); sys.exit(0)
if args[:2] == ['write', 'auth/netkingdom/role/platform-admin']:
    body = json.load(open(args[2][1:]))
    if state.get('ignore_bound'):
        body.pop('bound_claims', None)
    state['role'].update(body); state['writes'] = state.get('writes', 0) + 1
    save(); print('Success! Data written to: auth/netkingdom/role/platform-admin'); sys.exit(0)
save(); sys.stderr.write('unsupported\n'); sys.exit(1)
'''


@pytest.fixture
def env(tmp_path):
    home = tmp_path / '.warden-attended-login' / 'session'
    home.mkdir(parents=True)
    bindir = tmp_path / 'bin'
    bindir.mkdir()
    bao = bindir / 'bao'
    bao.write_text(FAKE_BAO)
    bao.chmod(bao.stat().st_mode | stat.S_IEXEC)
    state = tmp_path / 'state.json'
    state.write_text(json.dumps({'role': LIVE, 'meta': {'role': 'platform-admin', 'email': 'x@example.org',
                                                        'username': 'admin'}}))
    base = {k: v for k, v in os.environ.items() if k not in ('BAO_TOKEN', 'VAULT_TOKEN')}
    base.update(HOME=str(home), PATH='%s:%s' % (bindir, os.environ['PATH']), FAKE_BAO_STATE=str(state))
    return tmp_path, state, base


def run(env, *args):
    _, _, base = env
    proc = subprocess.run([str(WRAPPER), *args], capture_output=True, env=base)
    assert proc.stdout == b'' and proc.stderr == b''  # warden fails closed on any output
    return proc.returncode


def st(env):
    return json.loads(env[1].read_text())


def patch(env, **values):
    state = st(env)
    state.update(values)
    env[1].write_text(json.dumps(state))


def receipt(path):
    assert stat.S_IMODE(path.stat().st_mode) == 0o600
    text = path.read_text()
    for value in ('SECRETTOKENVALUE', 'ACCESSORVALUE', 'x@example.org'):
        assert value not in text
    return json.loads(text)


def test_declaration_maps_assurance_and_binds_aal2():
    assert DECLARED['bound_claims'] == EXPECTED
    assert DECLARED['claim_mappings']['/assurance/level'] == 'assurance_level'


def test_full_rollout_prepare_apply_verify_is_silent(env):
    tmp = env[0]
    assert run(env, 'prepare', '--receipt', str(tmp / 'p.json')) == 0
    p = receipt(tmp / 'p.json')
    assert p['status'] == 'prepared' and p['backup_role']['bound_claims'] == {'groups': ['net-kingdom-admins']}
    role = st(env)['role']
    assert role['bound_claims'] == {'groups': ['net-kingdom-admins']}  # prepare does not restrict logins
    assert role['claim_mappings']['/assurance/level'] == 'assurance_level'

    patch(env, meta={'role': 'platform-admin', 'assurance_level': 'aal2'})
    assert run(env, 'apply', '--receipt', str(tmp / 'a.json')) == 0
    a = receipt(tmp / 'a.json')
    assert a['status'] == 'applied' and a['bound_claims'] == EXPECTED
    assert a['gate_token']['assurance_level'] == 'aal2'
    assert st(env)['role']['bound_claims'] == EXPECTED

    assert run(env, 'verify', '--receipt', str(tmp / 'v.json')) == 0
    assert receipt(tmp / 'v.json')['status'] == 'verified'
    assert all(call[:2] != ['token', 'revoke'] for call in st(env)['calls'])  # warden revokes, not us


def test_apply_refused_without_aal2_login_and_writes_nothing(env):
    tmp = env[0]
    assert run(env, 'prepare', '--receipt', str(tmp / 'p.json')) == 0
    writes = st(env)['writes']
    patch(env, meta={'role': 'platform-admin', 'assurance_level': 'aal1'})
    assert run(env, 'apply', '--receipt', str(tmp / 'a.json')) == 1
    assert receipt(tmp / 'a.json')['status'] == 'gate_login_not_aal2'
    assert st(env)['writes'] == writes and st(env)['role']['bound_claims'] == {'groups': ['net-kingdom-admins']}


def test_apply_refused_before_prepare(env):
    tmp = env[0]
    patch(env, meta={'role': 'platform-admin', 'assurance_level': 'aal2'})
    assert run(env, 'apply', '--receipt', str(tmp / 'a.json')) == 1
    assert receipt(tmp / 'a.json')['status'] == 'not_prepared'
    assert 'writes' not in st(env)


def test_apply_readback_mismatch_exits_nonzero(env):
    tmp = env[0]
    assert run(env, 'prepare', '--receipt', str(tmp / 'p.json')) == 0
    patch(env, meta={'role': 'platform-admin', 'assurance_level': 'aal2'}, ignore_bound=True)
    assert run(env, 'apply', '--receipt', str(tmp / 'a.json')) == 1
    assert receipt(tmp / 'a.json')['status'] == 'readback_mismatch'


def test_verify_fails_on_unapplied_role(env):
    tmp = env[0]
    patch(env, meta={'role': 'platform-admin', 'assurance_level': 'aal2'})
    assert run(env, 'verify', '--receipt', str(tmp / 'v.json')) == 1
    assert receipt(tmp / 'v.json')['status'] == 'role_not_declared'


def test_rollback_restores_backup_with_fresh_login(env):
    tmp = env[0]
    assert run(env, 'prepare', '--receipt', str(tmp / 'p.json')) == 0
    patch(env, meta={'role': 'platform-admin', 'assurance_level': 'aal2'})
    assert run(env, 'apply', '--receipt', str(tmp / 'a.json')) == 0
    assert run(env, 'rollback', '--backup', str(tmp / 'p.json'), '--receipt', str(tmp / 'r.json')) == 0
    r = receipt(tmp / 'r.json')
    assert r['status'] == 'rolled_back' and r['bound_claims'] == {'groups': ['net-kingdom-admins']}
    assert '/assurance/level' not in st(env)['role']['claim_mappings']


def test_rollback_refuses_a_non_backup_receipt(env):
    tmp = env[0]
    patch(env, meta={'role': 'platform-admin', 'assurance_level': 'aal2'})
    assert run(env, 'prepare', '--receipt', str(tmp / 'p.json')) == 0
    assert run(env, 'apply', '--receipt', str(tmp / 'a.json')) == 0
    assert run(env, 'rollback', '--backup', str(tmp / 'a.json'), '--receipt', str(tmp / 'r.json')) == 1
    assert receipt(tmp / 'r.json')['status'] == 'backup_receipt_invalid'


def test_rollback_body_is_local_and_private(env, monkeypatch):
    tmp = env[0]
    assert run(env, 'prepare', '--receipt', str(tmp / 'p.json')) == 0
    calls = len(st(env)['calls'])
    _, _, base = env
    base = dict(base, HOME=str(tmp))  # outside the envelope on purpose
    proc = subprocess.run([str(WRAPPER), 'rollback-body', '--backup', str(tmp / 'p.json'),
                           '--out', str(tmp / 'body.json')], capture_output=True, env=base)
    assert proc.returncode == 0 and proc.stdout == proc.stderr == b''
    body = receipt(tmp / 'body.json')
    assert body['bound_claims'] == {'groups': ['net-kingdom-admins']}
    assert len(st(env)['calls']) == calls


def test_refuses_outside_attended_envelope(env):
    tmp = env[0]
    _, _, base = env
    proc = subprocess.run([str(WRAPPER), 'prepare', '--receipt', str(tmp / 'p.json')],
                          capture_output=True, env=dict(base, HOME=str(tmp)))
    assert proc.returncode == 1 and proc.stdout == proc.stderr == b''
    assert receipt(tmp / 'p.json')['status'] == 'attended_envelope_required'


def test_existing_receipt_is_never_overwritten(env):
    tmp = env[0]
    (tmp / 'p.json').write_text('{}')
    assert run(env, 'prepare', '--receipt', str(tmp / 'p.json')) == 1
    assert (tmp / 'p.json').read_text() == '{}'
    assert 'calls' not in st(env)


def test_bad_arguments_exit_2(env):
    assert run(env, 'apply') == 2
    assert run(env, 'destroy', '--receipt', '/dev/null') == 2