railiance-platform/tests/test_openbao_platform_admin_check.py

import importlib.util
import json
from pathlib import Path

import pytest

spec = importlib.util.spec_from_file_location(
    'check', Path(__file__).resolve().parents[1] / 'scripts/openbao_platform_admin_check.py')
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)

DECLARED = m.POLICY_FILE.read_text(encoding='utf-8')
FULL = {p: ['create', 'delete', 'list', 'read', 'sudo', 'update'] for p in m.SELF_PATHS}


def token(**extra):
    return dict({'policies': ['default', 'platform-admin'], 'identity_policies': None,
                 'ttl': 3500, 'creation_ttl': 3600, 'type': 'service',
                 'path': 'auth/netkingdom/oidc/callback'}, **extra)


def test_matching_policy_and_capabilities_pass():
    result = m.assess({'role_type': 'oidc'}, DECLARED + '\n', token(), FULL)
    assert result['policy']['matches_declared']
    assert result['self_revocation_permitted']
    assert result['token']['has_default_policy']


def test_missing_self_capabilities_are_reported():
    caps = {'auth/token/revoke-self': ['deny'], 'auth/token/lookup-self': ['read']}
    result = m.assess({}, 'path "x" {}', token(policies=['platform-admin']), caps)
    assert not result['policy']['matches_declared']
    assert result['self_capabilities_missing'] == ['auth/token/revoke-self']
    assert not result['token']['has_default_policy']


def test_bao_json_places_format_flag_before_positionals(monkeypatch):
    seen = []

    class Done:
        stdout = b'{"data": {}}'

    monkeypatch.setattr(m.subprocess, 'run', lambda cmd, **kw: seen.append(cmd) or Done())
    m.bao_json('read', 'auth/token/lookup-self')
    m.bao_json('write', 'sys/capabilities-self', 'paths=a,b')
    assert seen == [['bao', 'read', '-format=json', 'auth/token/lookup-self'],
                    ['bao', 'write', '-format=json', 'sys/capabilities-self', 'paths=a,b']]


def test_token_id_is_never_kept(monkeypatch):
    monkeypatch.setattr(m, 'bao_json', lambda *a: {'data': {'id': 'SECRET', 'accessor': 'ACC', 'policies': ['default']}})
    kept = m.read_token()
    assert 'SECRET' not in json.dumps(kept) and 'ACC' not in json.dumps(kept)


def test_apply_only_when_drifted_and_verified(monkeypatch):
    live = {'rules': 'path "old" {}'}
    writes = []
    monkeypatch.setattr(m, 'read_role', lambda: {})
    monkeypatch.setattr(m, 'read_policy', lambda: live['rules'])
    monkeypatch.setattr(m, 'read_token', token)
    monkeypatch.setattr(m, 'read_capabilities', lambda: FULL)
    monkeypatch.setattr(m, 'apply_policy', lambda: writes.append(1) or live.update(rules=DECLARED))
    result, changed = m.run(apply=True)
    assert changed and writes == [1] and result['policy']['matches_declared']
    _, changed = m.run(apply=True)
    assert not changed and writes == [1]


def test_refuses_outside_attended_envelope(tmp_path, monkeypatch):
    monkeypatch.setenv('HOME', str(tmp_path))
    receipt = tmp_path / 'r.json'
    assert m.main(['--receipt', str(receipt)]) == 1
    assert json.loads(receipt.read_text())['status'] == 'attended_envelope_required'


def test_failed_step_is_recorded_and_others_continue(monkeypatch):
    def denied():
        raise m.subprocess.CalledProcessError(
            2, ['bao'], stderr=b'Error\nURL: GET http://127.0.0.1:18200/v1/auth/token/lookup-self\nCode: 403. Errors:\n\n* permission denied\n')
    monkeypatch.setattr(m, 'read_role', lambda: {'role_type': 'oidc'})
    monkeypatch.setattr(m, 'read_policy', lambda: DECLARED)
    monkeypatch.setattr(m, 'read_token', denied)
    monkeypatch.setattr(m, 'read_capabilities', lambda: FULL)
    result, changed = m.run()
    assert not changed
    assert result['steps']['token_lookup_self'] == {
        'ok': False, 'error': 'exit_2',
        'detail': ['URL: GET http://127.0.0.1:18200/v1/auth/token/lookup-self',
                   'Code: 403. Errors:', '* permission denied']}
    assert result['steps']['read_role']['ok'] and result['policy']['matches_declared']


def test_local_error_line_is_kept_with_token_shapes_redacted():
    error = m.subprocess.CalledProcessError(
        1, ['bao'], stderr=b'error looking up token hvs.CAESIabcdefghijklmnopqrstu: bad\n')
    detail = m.error_summary(error)['detail']
    assert detail == ['error looking up token [redacted]: bad']


def test_attached_rules_include_default(monkeypatch):
    monkeypatch.setattr(m, 'read_rules', lambda name: 'rules-' + name)
    rules = m.read_attached_rules({'token_policies': ['platform-admin', 'operator-custody']})
    assert rules == {'default': 'rules-default', 'operator-custody': 'rules-operator-custody',
                     'platform-admin': 'rules-platform-admin'}