railiance-platform/tests/test_openbao_operator_loopback_callback.py

import copy
import importlib.util
from pathlib import Path
import pytest

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


def role():
    return {'role_type': 'oidc', 'token_policies': ['platform-admin'],
            'allowed_redirect_uris': ['https://existing.example/callback'],
            'token_ttl': 900, 'bound_claims': {'groups': ['custom-admins']},
            'claim_mappings': {'email': 'email'}, 'token_max_ttl': 1800}


def test_preserves_all_other_settings():
    current = role()
    original = copy.deepcopy(current)
    writes = []
    def write(value):
        writes.append(value)
        current.update(value)
    assert m.update(lambda: copy.deepcopy(current), write)
    assert len(writes) == 1
    assert current == dict(original, allowed_redirect_uris=original['allowed_redirect_uris'] + [m.CALLBACK])
    assert not m.update(lambda: copy.deepcopy(current), write)
    assert len(writes) == 1


def test_write_payload_omits_read_only_fields():
    current = dict(role(), request_id='synthetic', lease_duration=0)
    writes = []
    def write(value):
        writes.append(value)
        current.update(value)
    assert m.update(lambda: copy.deepcopy(current), write)
    assert 'request_id' not in writes[0]
    assert 'lease_duration' not in writes[0]
    assert writes[0]['allowed_redirect_uris'][-1] == m.CALLBACK


def test_readback_allows_normalized_extra_fields():
    current = role()
    def read():
        return dict(current, lease_duration=0)
    def write(value):
        current.update(value)
    assert m.update(read, write)
    assert m.CALLBACK in current['allowed_redirect_uris']


def test_observed_concurrent_change_prevents_write():
    reads = iter([role(), dict(role(), token_ttl=300)])
    with pytest.raises(m.Refused, match='role_changed_before_write'):
        m.update(lambda: next(reads), lambda _: pytest.fail('must not write'))


def test_failed_readback_is_not_success():
    with pytest.raises(m.Refused, match='readback_callback_missing'):
        m.update(role, lambda _: None)


@pytest.mark.parametrize('mutation', [{'role_type': 'jwt'}, {'token_policies': ['other']}, {'allowed_redirect_uris': 'bad'}])
def test_unexpected_live_role_refused(monkeypatch, mutation):
    import json
    import subprocess
    payload = dict(role(), **mutation)
    monkeypatch.setattr(m.subprocess, 'run', lambda *a, **kw: subprocess.CompletedProcess(a, 0, json.dumps({'data': payload}).encode()))
    with pytest.raises(m.Refused, match='unexpected_role'):
        m.read_role()


def test_receipt_records_apply_and_failure(tmp_path, monkeypatch):
    import json
    receipt = tmp_path / 'loopback.json'
    monkeypatch.setattr(m, 'require_attended', lambda: None)
    monkeypatch.setattr(m, 'update', lambda: True)
    assert m.main(['--receipt', str(receipt)]) == 0
    assert json.loads(receipt.read_text())['status'] == 'applied'
    failed = tmp_path / 'failed.json'
    def boom():
        raise m.Refused('bao_write_failed')
    monkeypatch.setattr(m, 'update', boom)
    assert m.main(['--receipt', str(failed)]) == 1
    assert json.loads(failed.read_text())['status'] == 'bao_write_failed'


def test_attended_wrapper_selects_sitting_requester_lane():
    spec = importlib.util.spec_from_file_location(
        'attended', Path(__file__).resolve().parents[1] / 'scripts/openbao-attended-exec.py')
    wrapper = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(wrapper)
    command = Path(__file__).resolve().parents[1] / 'scripts/prove-sitting-requester-exchange.sh'
    lane, args = wrapper.parse_argv(['--lane', 'informed-decision-sitting-requester-login', str(command)])
    assert lane == 'informed-decision-sitting-requester-login'
    assert args[0] == str(command.resolve())


def test_attended_wrapper_pins_operator_tunnel_address(monkeypatch):
    spec = importlib.util.spec_from_file_location(
        'attended', Path(__file__).resolve().parents[1] / 'scripts/openbao-attended-exec.py')
    wrapper = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(wrapper)
    monkeypatch.setenv('BAO_ADDR', 'https://bao.coulomb.social')
    monkeypatch.setenv('VAULT_ADDR', 'https://bao.coulomb.social')
    monkeypatch.setenv('BAO_TOKEN', 'must-not-propagate')
    env = wrapper.contained_env()
    assert env['BAO_ADDR'] == wrapper.TUNNEL
    assert env['VAULT_ADDR'] == wrapper.TUNNEL
    assert 'BAO_TOKEN' not in env
    assert 'VAULT_TOKEN' not in env
    import os
    launcher = str(Path(__file__).resolve().parents[1] / 'scripts/operator-browser')
    assert env['PATH'].split(os.pathsep)[0] == launcher


def test_attended_wrapper_requires_absolute_existing_executable(tmp_path, monkeypatch):
    import os
    spec = importlib.util.spec_from_file_location(
        'attended', Path(__file__).resolve().parents[1] / 'scripts/openbao-attended-exec.py')
    wrapper = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(wrapper)
    missing = tmp_path / 'missing.sh'
    with pytest.raises(SystemExit, match='missing'):
        wrapper.reviewed_command([str(missing)])
    command = tmp_path / 'owner.sh'
    command.write_text('#!/bin/sh\n')
    with pytest.raises(SystemExit, match='not executable'):
        wrapper.reviewed_command([str(command)])
    command.chmod(0o755)
    monkeypatch.chdir(tmp_path)
    assert wrapper.reviewed_command(['owner.sh']) == [str(command.resolve())]


def test_silent_entrypoint_on_command_failure(tmp_path):
    import os
    import subprocess
    executable = tmp_path / 'bao'
    executable.write_text('#!/bin/sh\necho SECRET_CANARY >&2\nexit 1\n')
    executable.chmod(0o755)
    result = subprocess.run([str(Path(m.__file__).with_name('openbao-apply-operator-loopback-callback.sh'))],
        env=dict(os.environ, PATH=str(tmp_path) + ':' + os.environ['PATH']), capture_output=True)
    assert result.returncode != 0
    assert result.stdout == result.stderr == b''