tegwick
c07e6ebe34
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
203 lines
8.9 KiB
Python
203 lines
8.9 KiB
Python
"""Exact membership effects, durable interruption recovery and attended CLI boundaries."""
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
'reset_transition', ROOT / 'sso-mfa/k8s/lldap/reset-role-transition.py')
|
|
change = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(change)
|
|
DIGEST = 'sha256:' + 'a' * 64
|
|
|
|
|
|
def baseline():
|
|
return {'users': ['admin', 'other', change.TARGET, change.SUBJECT], 'groups': {
|
|
change.OLD: {'id': 1, 'members': [change.TARGET]},
|
|
change.NEW: {'id': 2, 'members': []},
|
|
'lldap_admin': {'id': 3, 'members': ['admin']},
|
|
'tenant:one:users': {'id': 4, 'members': [change.SUBJECT]}}}
|
|
|
|
|
|
class Directory:
|
|
def __init__(self, state=None, failure=None, when=1):
|
|
self.state = state or baseline()
|
|
self.calls = []
|
|
self.failure, self.when = failure, when
|
|
|
|
def snapshot(self):
|
|
return copy.deepcopy(self.state)
|
|
|
|
def membership(self, action, group_id):
|
|
self.calls.append((action, group_id))
|
|
failing = len(self.calls) == self.when
|
|
if failing and self.failure == 'before':
|
|
raise RuntimeError('private-provider-error')
|
|
group = next(g for g in self.state['groups'].values() if g['id'] == group_id)
|
|
if action == 'add':
|
|
group['members'].append(change.TARGET)
|
|
group['members'].sort()
|
|
else:
|
|
group['members'].remove(change.TARGET)
|
|
if failing and self.failure == 'drift':
|
|
self.state['groups']['tenant:one:users']['members'].insert(0, 'other')
|
|
if failing and self.failure == 'after':
|
|
raise RuntimeError('private-provider-error')
|
|
|
|
|
|
@pytest.fixture
|
|
def journal(tmp_path):
|
|
tmp_path.chmod(0o700)
|
|
value = change.Journal(tmp_path / 'attempt')
|
|
try:
|
|
yield value
|
|
finally:
|
|
value.close()
|
|
|
|
|
|
def test_grant_and_rollback_preserve_all_other_memberships_and_receipt_history(journal):
|
|
directory = Directory()
|
|
result = change.transition(directory, journal, DIGEST, 'grant')
|
|
assert result['status'] == 'password_manager_role_verified'
|
|
assert result['other_directory_memberships_unchanged']
|
|
assert directory.state == change.expected_state(baseline(), change.NEW)
|
|
assert directory.calls == [('remove', 1), ('add', 2)]
|
|
original_receipt = journal.read()
|
|
# Rollback preserves productive changes made after the grant, including subject changes.
|
|
directory.state['groups']['tenant:one:users']['members'].insert(0, 'other')
|
|
directory.state['groups']['lldap_admin']['members'].append(change.SUBJECT)
|
|
before_rollback = directory.snapshot()
|
|
result = change.transition(directory, journal, DIGEST, 'rollback')
|
|
assert result['status'] == 'read_only_role_restored_verified'
|
|
assert directory.state == change.expected_state(before_rollback, change.OLD)
|
|
history = [json.loads(p.read_text()) for p in journal.path.glob('history-*.json')]
|
|
assert original_receipt in history
|
|
with pytest.raises(change.Refused, match='cannot_replay'):
|
|
change.transition(directory, journal, DIGEST, 'rollback')
|
|
|
|
|
|
@pytest.mark.parametrize('failure,when', [('before', 1), ('after', 1), ('after', 2),
|
|
('drift', 1), ('drift', 2)])
|
|
def test_ambiguous_effects_recover_only_readonly_and_preserve_foreign_changes(journal, failure, when):
|
|
directory = Directory(failure=failure, when=when)
|
|
with pytest.raises(change.Refused, match='uncertain'):
|
|
change.transition(directory, journal, DIGEST, 'grant')
|
|
assert journal.read()['status'] == 'uncertain_requires_attended_recovery'
|
|
with pytest.raises(change.Refused, match='not_replay'):
|
|
change.transition(directory, journal, DIGEST, 'grant')
|
|
directory.failure = None
|
|
before = directory.snapshot()
|
|
result = change.transition(directory, journal, DIGEST, 'recover')
|
|
assert result['status'] == 'read_only_role_restored_verified'
|
|
assert directory.state == change.expected_state(before, change.OLD)
|
|
assert 'private-provider-error' not in json.dumps(journal.read())
|
|
|
|
|
|
@pytest.mark.parametrize('defect', ['foreign-role', 'already-granted', 'no-admin',
|
|
'accepted-admin', 'missing-subject', 'missing-group'])
|
|
def test_unreviewed_baseline_refuses_before_any_journal_or_write(journal, defect):
|
|
state = baseline()
|
|
if defect == 'foreign-role':
|
|
state['groups']['tenant:one:users']['members'].insert(0, change.TARGET)
|
|
elif defect == 'already-granted':
|
|
state = change.expected_state(state, change.NEW)
|
|
elif defect == 'no-admin':
|
|
state['groups']['lldap_admin']['members'] = []
|
|
elif defect == 'accepted-admin':
|
|
state['groups']['lldap_admin']['members'].append(change.SUBJECT)
|
|
elif defect == 'missing-subject':
|
|
state['users'].remove(change.SUBJECT)
|
|
else:
|
|
del state['groups'][change.NEW]
|
|
directory = Directory(state)
|
|
with pytest.raises(change.Refused):
|
|
change.transition(directory, journal, DIGEST, 'grant')
|
|
assert not directory.calls and journal.read() is None
|
|
|
|
|
|
@pytest.mark.parametrize('defect', ['digest', 'group-id', 'foreign-role'])
|
|
def test_recovery_refuses_changed_binding(journal, defect):
|
|
directory = Directory(failure='after')
|
|
with pytest.raises(change.Refused):
|
|
change.transition(directory, journal, DIGEST, 'grant')
|
|
if defect == 'group-id':
|
|
directory.state['groups'][change.NEW]['id'] = 99
|
|
if defect == 'foreign-role':
|
|
directory.state['groups']['lldap_admin']['members'].append(change.TARGET)
|
|
directory.calls.clear()
|
|
with pytest.raises(change.Refused):
|
|
change.transition(directory, journal, DIGEST + ('x' if defect == 'digest' else ''), 'recover')
|
|
assert not directory.calls
|
|
|
|
|
|
@pytest.mark.parametrize('prior', [False, True])
|
|
def test_partial_local_journal_write_is_preserved_and_recovery_never_grants(journal, prior):
|
|
directory = Directory(failure='after')
|
|
if prior:
|
|
with pytest.raises(change.Refused):
|
|
change.transition(directory, journal, DIGEST, 'grant')
|
|
partial = journal.path / 'journal.next'
|
|
partial.write_bytes(b'{"incomplete')
|
|
partial.chmod(0o600)
|
|
directory.failure = None
|
|
result = change.transition(directory, journal, DIGEST, 'recover')
|
|
assert result['status'] == 'read_only_role_restored_verified'
|
|
assert not partial.exists()
|
|
assert b'{"incomplete' in [p.read_bytes() for p in journal.path.glob('history-*.json')]
|
|
|
|
|
|
def test_exclusive_lock_and_completed_grant_refuse_reentry(journal):
|
|
with pytest.raises(change.Refused, match='active'):
|
|
change.Journal(journal.path)
|
|
directory = Directory()
|
|
change.transition(directory, journal, DIGEST, 'grant')
|
|
directory.calls.clear()
|
|
with pytest.raises(change.Refused, match='rollback_for_completed'):
|
|
change.transition(directory, journal, DIGEST, 'recover')
|
|
with pytest.raises(change.Refused, match='not_replay'):
|
|
change.transition(directory, journal, DIGEST, 'grant')
|
|
assert not directory.calls
|
|
|
|
|
|
@pytest.mark.parametrize('defect', ['symlink', 'hardlink', 'public'])
|
|
def test_unsafe_existing_journal_refuses(journal, tmp_path, defect):
|
|
path = journal.path / 'journal.json'
|
|
if defect == 'symlink':
|
|
path.symlink_to(tmp_path / 'missing')
|
|
else:
|
|
path.write_text('{}')
|
|
path.chmod(0o600 if defect == 'hardlink' else 0o644)
|
|
if defect == 'hardlink':
|
|
os.link(path, tmp_path / 'alias')
|
|
with pytest.raises(change.Refused):
|
|
change.transition(Directory(), journal, DIGEST, 'grant')
|
|
|
|
|
|
@pytest.mark.parametrize('defect', ['no-terminal', 'wrong-digest', 'runtime-drift'])
|
|
def test_cli_refuses_before_private_input(monkeypatch, defect):
|
|
monkeypatch.setattr(change, 'proposal', lambda: ({'runtime': {}, 'scope': {}}, DIGEST))
|
|
monkeypatch.setattr(change.sys.stdin, 'isatty', lambda: defect != 'no-terminal')
|
|
monkeypatch.setattr(change.sys.stderr, 'isatty', lambda: True)
|
|
monkeypatch.setattr(change, 'observe_runtime', lambda: {'changed': True})
|
|
monkeypatch.setattr(change.getpass, 'getpass', lambda prompt: pytest.fail('private prompt'))
|
|
assert change.main(['--execute', '--expect-digest',
|
|
DIGEST + ('x' if defect == 'wrong-digest' else '')]) == 1
|
|
|
|
|
|
def test_recovery_allows_unready_peers_but_not_changed_runtime_identity(monkeypatch):
|
|
expected = {'authelia': {'uid': 'a', 'ready': True},
|
|
'lldap': {'uid': 'b', 'ready': True}, 'config': 'pinned'}
|
|
actual = copy.deepcopy(expected)
|
|
actual['authelia']['ready'] = False
|
|
monkeypatch.setattr(change, 'observe_runtime', lambda: copy.deepcopy(actual))
|
|
with pytest.raises(change.Refused, match='runtime_changed'):
|
|
change.verify_runtime(expected)
|
|
change.verify_runtime(expected, recovering=True)
|
|
actual['lldap']['uid'] = 'replacement'
|
|
with pytest.raises(change.Refused, match='runtime_changed'):
|
|
change.verify_runtime(expected, recovering=True)
|