tegwick
27b4ea6933
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
169 lines
7.7 KiB
Python
169 lines
7.7 KiB
Python
"""No-send role observation, credential containment and exact subject binding."""
|
|
|
|
import copy
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
'reset_observer', ROOT / 'sso-mfa/k8s/lldap/reset-authority-observe.py')
|
|
observer = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(observer)
|
|
|
|
|
|
class Directory:
|
|
def __init__(self, subject_roles=(), bind_roles=('lldap_strict_readonly',)):
|
|
self.roles = {'admin': ['lldap_admin'], 'svc-authelia': list(bind_roles),
|
|
'private-person': list(subject_roles), 'tegwick': list(subject_roles)}
|
|
self.requests = []
|
|
self.transform = lambda value: value
|
|
|
|
def __call__(self, request, timeout):
|
|
payload = json.loads(request.data)
|
|
self.requests.append((request.full_url, payload))
|
|
if request.full_url.endswith('/auth/simple/login'):
|
|
assert payload == {'username': 'admin', 'password': 'private-password-marker'}
|
|
return io.BytesIO(b'{"token":"private-token-marker"}')
|
|
assert request.full_url == 'https://lldap.coulomb.social/api/graphql'
|
|
assert payload['query'] == observer.QUERY and not payload['query'].startswith('mutation')
|
|
assert request.get_header('Authorization') == 'Bearer private-token-marker'
|
|
uid = payload['variables']['id']
|
|
value = {'data': {'user': {'id': uid,
|
|
'groups': [{'displayName': g} for g in self.roles[uid]]}}}
|
|
return io.BytesIO(json.dumps(self.transform(value)).encode())
|
|
|
|
|
|
@pytest.mark.parametrize('roles', [[], ['lldap_admin'], ['lldap_password_manager'],
|
|
['private-application-admin-group']])
|
|
def test_exact_classification_only_and_no_mutations(roles):
|
|
directory = Directory(roles)
|
|
result = observer.Observer('https://lldap.coulomb.social', directory).observe(
|
|
'private-person', 'private-password-marker')
|
|
assert [r[1].get('variables', {}).get('id') for r in directory.requests] == [
|
|
None, 'admin', 'svc-authelia', 'private-person']
|
|
assert result['selected_subject']['builtin_roles']['lldap_admin'] == ('lldap_admin' in roles)
|
|
assert result['selected_subject']['application_authority_assessed'] is False
|
|
assert not result['membership_changed'] and not result['password_reset_attempted']
|
|
assert not result['grant_approved'] and not result['application_access_verified']
|
|
raw = json.dumps(result)
|
|
for private in ['private-person', 'private-password-marker', 'private-token-marker',
|
|
'private-application-admin-group']:
|
|
assert private not in raw
|
|
assert len(result['sources']) == 4
|
|
|
|
|
|
def test_ambiguous_bind_roles_reported_without_effective_permission_claim():
|
|
directory = Directory(bind_roles=['lldap_strict_readonly', 'lldap_password_manager'])
|
|
result = observer.Observer('https://lldap.coulomb.social', directory).observe(
|
|
'private-person', 'private-password-marker')
|
|
assert result['bind_has_multiple_builtin_roles']
|
|
assert not result['grant_approved']
|
|
|
|
|
|
@pytest.mark.parametrize('defect', ['wrong-subject', 'missing', 'duplicate-group',
|
|
'malformed-group', 'error-body', 'extra-fields'])
|
|
def test_untrusted_responses_refuse_without_reporting_observation(defect):
|
|
directory = Directory()
|
|
|
|
def transform(value):
|
|
value = copy.deepcopy(value)
|
|
if defect == 'wrong-subject':
|
|
value['data']['user']['id'] = 'different-person'
|
|
elif defect == 'missing':
|
|
value['data']['user'] = None
|
|
elif defect == 'duplicate-group':
|
|
value['data']['user']['groups'] *= 2
|
|
elif defect == 'malformed-group':
|
|
value['data']['user']['groups'] = [{'displayName': 'unsafe\nvalue'}]
|
|
elif defect == 'error-body':
|
|
value['errors'] = [{'message': 'private-provider-body'}]
|
|
else:
|
|
value['data']['user']['email'] = 'private@example.test'
|
|
return value
|
|
|
|
directory.transform = transform
|
|
with pytest.raises(observer.Refused):
|
|
observer.Observer('https://lldap.coulomb.social', directory).observe(
|
|
'private-person', 'private-password-marker')
|
|
|
|
|
|
@pytest.mark.parametrize('subject', ['', 'Bad User', 'alice"}{mutation', 'a' * 129])
|
|
def test_invalid_subject_refuses_before_authentication(subject):
|
|
directory = Directory()
|
|
with pytest.raises(observer.Refused):
|
|
observer.Observer('https://lldap.coulomb.social', directory).observe(
|
|
subject, 'private-password-marker')
|
|
assert not directory.requests
|
|
|
|
|
|
@pytest.mark.parametrize('defect', ['existing', 'symlink', 'public-parent', 'in-repository'])
|
|
def test_unsafe_receipt_refused_before_private_prompts(tmp_path, monkeypatch, defect):
|
|
tmp_path.chmod(0o700)
|
|
path = tmp_path / 'receipt.json'
|
|
if defect == 'existing':
|
|
path.write_text('preserve')
|
|
elif defect == 'symlink':
|
|
path.symlink_to(tmp_path / 'missing')
|
|
elif defect == 'public-parent':
|
|
tmp_path.chmod(0o755)
|
|
else:
|
|
(tmp_path / '.git').mkdir()
|
|
monkeypatch.setattr(observer.sys.stdin, 'isatty', lambda: True)
|
|
monkeypatch.setattr(observer.sys.stderr, 'isatty', lambda: True)
|
|
monkeypatch.setattr(observer.getpass, 'getpass', lambda prompt: pytest.fail('private prompt'))
|
|
assert observer.main(['--receipt', str(path)]) == 1
|
|
if defect == 'existing':
|
|
assert path.read_text() == 'preserve'
|
|
|
|
|
|
def test_nonterminal_refuses_before_private_prompt(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(observer.sys.stdin, 'isatty', lambda: False)
|
|
monkeypatch.setattr(observer.getpass, 'getpass', lambda prompt: pytest.fail('private prompt'))
|
|
assert observer.main(['--receipt', str(tmp_path / 'receipt.json')]) == 1
|
|
|
|
|
|
@pytest.mark.parametrize('endpoint', ['https://attacker.invalid',
|
|
'https://lldap.coulomb.social@attacker.invalid'])
|
|
def test_origin_refuses_before_private_prompt(tmp_path, monkeypatch, endpoint):
|
|
tmp_path.chmod(0o700)
|
|
monkeypatch.setattr(observer.sys.stdin, 'isatty', lambda: True)
|
|
monkeypatch.setattr(observer.sys.stderr, 'isatty', lambda: True)
|
|
monkeypatch.setattr(observer.getpass, 'getpass', lambda prompt: pytest.fail('private prompt'))
|
|
assert observer.main(['--receipt', str(tmp_path / 'receipt.json'),
|
|
'--endpoint', endpoint]) == 1
|
|
|
|
|
|
@pytest.mark.parametrize('fail', [False, True])
|
|
def test_cli_private_receipt_and_sanitized_failure(tmp_path, monkeypatch, capsys, fail):
|
|
tmp_path.chmod(0o700)
|
|
path = tmp_path / 'receipt.json'
|
|
monkeypatch.setattr(observer.sys.stdin, 'isatty', lambda: True)
|
|
monkeypatch.setattr(observer.sys.stderr, 'isatty', lambda: True)
|
|
prompts = []
|
|
def private_input(prompt):
|
|
prompts.append(prompt)
|
|
return 'private-password-marker'
|
|
monkeypatch.setattr(observer.getpass, 'getpass', private_input)
|
|
directory = Directory()
|
|
monkeypatch.setattr(observer.transport, 'Transport', lambda base: directory)
|
|
if fail:
|
|
def broken(value):
|
|
raise RuntimeError('private-provider-body private-token-marker')
|
|
directory.transform = broken
|
|
assert observer.main(['--receipt', str(path)]) == int(fail)
|
|
assert len(prompts) == 1
|
|
captured = capsys.readouterr()
|
|
for private in ['private-person', 'private-password-marker', 'private-token-marker',
|
|
'private-provider-body']:
|
|
assert private not in captured.out + captured.err
|
|
if fail:
|
|
assert not path.exists()
|
|
else:
|
|
assert directory.requests[-1][1]['variables']['id'] == 'tegwick'
|
|
assert stat.S_IMODE(path.stat().st_mode) == 0o600
|
|
assert json.loads(path.read_text())['status'] == 'current_reset_roles_observed'
|