7496d9fab5
A relative scripts path can fail to spawn after a successful contained OIDC session, which Warden then revokes. The wrapper now resolves the command first; T03 records that this attempt did not write the role. Assistant: grok Assistant-Session: 01a0a23b-3bf0-7341-b4e5-9dc05f72573a
80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
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_observed_concurrent_change_prevents_write():
|
|
reads = iter([role(), dict(role(), token_ttl=300)])
|
|
with pytest.raises(ValueError):
|
|
m.update(lambda: next(reads), lambda _: pytest.fail('must not write'))
|
|
|
|
|
|
def test_failed_readback_is_not_success():
|
|
with pytest.raises(ValueError):
|
|
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(ValueError):
|
|
m.read_role()
|
|
|
|
|
|
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''
|