net-kingdom/tests/test_lldap_group_maintenance.py

"""Credential boundaries, exact group effects and honest diagnostic outcomes."""
import contextlib
import copy
import importlib.util
import io
import json
import os
from pathlib import Path
import subprocess
import sys
from unittest.mock import Mock, patch

import pytest

ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location('lldap_admin', ROOT / 'sso-mfa/k8s/lldap/lldap_admin.py')
admin = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(admin)


@pytest.mark.parametrize('script', ['lldap/bootstrap-users.sh', 'lldap/manage-group-members.sh', 'verify-t07.sh'])
def test_legacy_arguments_never_source_credentials_or_invoke_external_tools(tmp_path, script):
    marker = tmp_path / 'sourced'
    calls = tmp_path / 'calls'
    root = tmp_path / 'private'
    (root / 'lldap').mkdir(parents=True, mode=0o700)
    (root / 'lldap/secrets.env').write_text(f'LLDAP_LDAP_USER_PASS=synthetic-private-marker\ntouch "{marker}"\n')
    commands = tmp_path / 'bin'
    commands.mkdir()
    for name in ('curl', 'kubectl'):
        executable = commands / name
        executable.write_text(f'#!/bin/sh\nprintf called >> "{calls}"\nexit 1\n')
        executable.chmod(0o700)
    args = ['http://127.0.0.1:1', str(root)]
    if 'manage-group' in script:
        args = ['add', 'test-user', 'activity-core-operators', *args]
    env = dict(os.environ, PATH=str(commands) + ':' + os.environ['PATH'])
    for key in ('LLDAP_ADMIN_PASS', 'BASH_ENV'):
        env.pop(key, None)
    result = subprocess.run(['bash', '-x', str(ROOT / 'sso-mfa/k8s' / script), *args],
                            env=env, capture_output=True, text=True, timeout=10)
    assert result.returncode != 0
    assert not marker.exists() and not calls.exists()
    assert 'synthetic-private-marker' not in result.stdout + result.stderr


class Directory:
    def __init__(self):
        self.state = {'users': ['agent-codex', 'break-glass', 'tegwick'], 'groups': {
            'lldap_admin': {'id': 1, 'members': ['tegwick']},
            'net-kingdom-admins': {'id': 2, 'members': ['break-glass']},
            'activity-core-operators': {'id': 3, 'members': ['tegwick']},
            'tenant:one:users': {'id': 4, 'members': ['agent-codex']},
        }}
        self.writes = []
        self.login = 0
        self.reads = 0
        self.drop_write = False
        self.extra_effect = False
        self.drift_at = None
        self.malformed = None

    def _login(self):
        self.login += 1
        return 'fixture-token'

    def _gql(self, token, query, variables):
        assert query == 'query { users { id } groups { id displayName users { id } } }'
        self.reads += 1
        if self.reads == self.drift_at:
            self.state['groups']['tenant:one:users']['members'] = []
        if self.malformed is not None:
            return self.malformed
        return {'users': [{'id': u} for u in self.state['users']], 'groups': [
            {'id': g['id'], 'displayName': name, 'users': [{'id': u} for u in g['members']]}
            for name, g in self.state['groups'].items()]}

    def _ensure_group(self, token, groups, name):
        self.writes.append(('create', name))
        if not self.drop_write:
            self.state['groups'][name] = {'id': max(g['id'] for g in self.state['groups'].values()) + 1, 'members': []}

    def _add_group(self, token, user, gid):
        self._write('add', user, gid)

    def _remove_group(self, token, user, gid):
        self._write('remove', user, gid)

    def _write(self, action, user, gid):
        self.writes.append((action, user, gid))
        if self.drop_write:
            return
        group = next(g for g in self.state['groups'].values() if g['id'] == gid)
        members = set(group['members'])
        (members.add if action == 'add' else members.discard)(user)
        group['members'] = sorted(members)
        if self.extra_effect:
            self.state['groups']['tenant:one:users']['members'] = []


def change(provider, **kwargs):
    return admin.maintain(provider, 'https://lldap.coulomb.social', 'membership', **kwargs)


def test_bootstrap_preview_apply_and_idempotence_preserve_existing_state():
    provider = Directory()
    before = copy.deepcopy(provider.state)
    preview = admin.maintain(provider, 'https://lldap.coulomb.social', 'bootstrap')
    assert preview['changes'] == [{'create_group': 'net-kingdom-users'}]
    assert provider.writes == []
    result = admin.maintain(provider, 'https://lldap.coulomb.social', 'bootstrap', apply=True, expected_digest=preview['snapshot_sha256'])
    assert result['result'] == 'verified'
    assert provider.state['users'] == before['users']
    for name, group in before['groups'].items():
        assert provider.state['groups'][name] == group
    assert provider.state['groups']['net-kingdom-users']['members'] == []
    provider.writes.clear()
    preview = admin.maintain(provider, 'https://lldap.coulomb.social', 'bootstrap')
    admin.maintain(provider, 'https://lldap.coulomb.social', 'bootstrap', apply=True, expected_digest=preview['snapshot_sha256'])
    assert provider.writes == []


@pytest.mark.parametrize('action,user', [('add', 'agent-codex'), ('remove', 'tegwick')])
def test_membership_changes_only_exact_member(action, user):
    provider = Directory()
    before = copy.deepcopy(provider.state)
    kwargs = {'action': action, 'group': 'activity-core-operators', 'user': user}
    preview = change(provider, **kwargs)
    assert provider.writes == []
    result = change(provider, **kwargs, apply=True, expected_digest=preview['snapshot_sha256'])
    assert result['result'] == 'verified'
    assert (user in provider.state['groups']['activity-core-operators']['members']) == (action == 'add')
    for name in set(before['groups']) - {'activity-core-operators'}:
        assert provider.state['groups'][name] == before['groups'][name]
    provider.writes.clear()
    preview = change(provider, **kwargs)
    change(provider, **kwargs, apply=True, expected_digest=preview['snapshot_sha256'])
    assert provider.writes == []


@pytest.mark.parametrize('group', ['net-kingdom-admins', 'net-kingdom-users', 'netkingdom-suspended', 'tenant-one-users', 'lldap_admin'])
@pytest.mark.parametrize('action', ['add', 'remove', 'list'])
def test_reserved_groups_refuse_before_login(group, action):
    provider = Directory()
    with pytest.raises(Exception):
        change(provider, action=action, group=group, user=None if action == 'list' else 'agent-codex')
    assert provider.login == 0 and provider.writes == []


@pytest.mark.parametrize('kind', ['changed_state', 'changed_action', 'changed_origin', 'second_read_drift', 'missing_group', 'missing_user'])
def test_stale_or_ambiguous_requests_never_mutate(kind):
    provider = Directory()
    kwargs = {'action': 'add', 'group': 'activity-core-operators', 'user': 'agent-codex'}
    preview = change(provider, **kwargs)
    base = 'https://lldap.coulomb.social'
    if kind == 'changed_state': provider.state['groups']['tenant:one:users']['members'] = []
    if kind == 'changed_action': kwargs['action'] = 'remove'
    if kind == 'changed_origin': base = 'http://127.0.0.1:10000'
    if kind == 'second_read_drift': provider.drift_at = 3
    if kind == 'missing_group': kwargs['group'] = 'missing-group'
    if kind == 'missing_user': kwargs['user'] = 'missing-user'
    with pytest.raises(Exception):
        admin.maintain(provider, base, 'membership', **kwargs, apply=True, expected_digest=preview['snapshot_sha256'])
    assert provider.writes == []


@pytest.mark.parametrize('failure', ['drop_write', 'extra_effect'])
def test_failed_or_broader_write_cannot_report_success(failure):
    provider = Directory()
    kwargs = {'action': 'add', 'group': 'activity-core-operators', 'user': 'agent-codex'}
    preview = change(provider, **kwargs)
    setattr(provider, failure, True)
    with pytest.raises(admin.Refused):
        change(provider, **kwargs, apply=True, expected_digest=preview['snapshot_sha256'])
    assert len(provider.writes) == 1  # No blind rollback/retry.


def test_list_is_nonmutating_and_verification_does_not_claim_acceptance():
    provider = Directory()
    listing = change(provider, action='list', group='activity-core-operators')
    assert listing['members'] == ['tegwick']
    result = admin.maintain(provider, 'https://lldap.coulomb.social', 'verify')
    assert result['native_acceptance'] == 'unverified'
    assert result['baseline_groups_present']['net-kingdom-users'] is False
    assert 'independent_recovery' in result['remaining']
    assert provider.writes == []


def test_groups_preview_lists_display_names_for_one_uid_without_writes():
    provider = Directory()
    result = admin.maintain(provider, 'https://lldap.coulomb.social', 'groups', user='tegwick')
    assert result == {'result': 'groups_observed', 'uid': 'tegwick',
                      'groups': ['activity-core-operators', 'lldap_admin'], 'native_acceptance': 'unverified'}
    assert provider.writes == []


@pytest.mark.parametrize('kwargs', [{'user': 'nobody'}, {'user': 'Bad uid'}, {'user': None},
                                    {'user': 'tegwick', 'group': 'lldap_admin'},
                                    {'user': 'tegwick', 'apply': True, 'expected_digest': '0' * 64}])
def test_groups_preview_refuses_unknown_or_malformed_requests(kwargs):
    provider = Directory()
    with pytest.raises(admin.Refused):
        admin.maintain(provider, 'https://lldap.coulomb.social', 'groups', **kwargs)
    assert provider.writes == []


def test_groups_cli_has_no_apply_and_prints_only_diagnosis_fields():
    out = io.StringIO()
    with patch.object(admin, 'private_password', return_value='fixture'), patch.object(admin, 'LLDAPProvisioner', return_value=Directory()), contextlib.redirect_stdout(out):
        assert admin.main(['groups', '--root', '/fixture', '--user', 'agent-codex']) == 0
    assert json.loads(out.getvalue()) == {'groups': ['tenant:one:users'], 'native_acceptance': 'unverified',
                                          'result': 'groups_observed', 'uid': 'agent-codex'}
    with contextlib.redirect_stderr(io.StringIO()):
        assert admin.main(['groups', '--root', '/fixture', '--user', 'agent-codex', '--apply']) == 1


@pytest.mark.parametrize('payload', [{}, {'users': [], 'groups': None}, {'users': [{'id': 'same'}, {'id': 'same'}], 'groups': []},
    {'users': [], 'groups': [{'id': True, 'displayName': 'a', 'users': []}]},
    {'users': [], 'groups': [{'id': 1, 'displayName': 'a', 'users': [{'id': 'unknown'}]}]},
    {'users': [], 'groups': [{'id': 1, 'displayName': 'a', 'users': []}, {'id': 1, 'displayName': 'b', 'users': []}]}])
def test_malformed_directory_snapshot_refused(payload):
    provider = Directory()
    provider.malformed = payload
    with pytest.raises(admin.Refused):
        admin.maintain(provider, 'https://lldap.coulomb.social', 'bootstrap')
    assert provider.writes == []


@pytest.mark.parametrize('url', ['http://lldap.coulomb.social', 'https://evil.test', 'http://localhost:17170',
                                'https://lldap.coulomb.social@evil.test', 'http://127.0.0.1:80/path',
                                'http://127.0.0.1:80?secret=marker', 'https://lldap.coulomb.social/#marker'])
def test_unapproved_origin_refused_before_private_input(url):
    with patch.object(admin, 'private_password') as private, contextlib.redirect_stderr(io.StringIO()):
        assert admin.main(['bootstrap', '--root', '/not-read', '--lldap-url', url]) == 1
    private.assert_not_called()


def test_verification_cli_always_returns_incomplete_for_structure_only():
    with patch.object(admin, 'private_password', return_value='fixture'), patch.object(admin, 'LLDAPProvisioner', return_value=Directory()), contextlib.redirect_stdout(io.StringIO()):
        assert admin.main(['verify', '--root', '/fixture']) == 2


def test_private_env_is_literal_not_executable(tmp_path):
    root = tmp_path / 'private'
    root.mkdir(mode=0o700)
    (root / 'lldap').mkdir(mode=0o700)
    marker = tmp_path / 'executed'
    value = f'$(touch {marker})'
    source = root / 'lldap/secrets.env'
    source.write_text('LLDAP_LDAP_USER_PASS=' + value + '\n')
    source.chmod(0o600)
    assert admin.private_password(root) == value
    assert not marker.exists()
    source.chmod(0o644)
    with pytest.raises(Exception): admin.private_password(root)


def test_provider_errors_are_not_printed():
    output = io.StringIO()
    with patch.object(admin, 'private_password', side_effect=RuntimeError('synthetic-secret-marker')), contextlib.redirect_stderr(output):
        assert admin.main(['bootstrap', '--root', '/fixture']) == 1
    assert 'synthetic-secret-marker' not in output.getvalue()


@pytest.fixture
def http_directory():
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
    import threading

    state = {'mode': 'ok', 'calls': [], 'redirect_hits': 0, 'directory': Directory()}

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def do_POST(self):
            payload = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
            state['calls'].append((self.path, payload))
            if self.path == '/redirected':
                state['redirect_hits'] += 1
            if state['mode'] == 'redirect':
                self.send_response(307)
                self.send_header('Location', '/redirected')
                self.end_headers()
                return
            if state['mode'] == 'denied':
                self.send_response(403)
                self.end_headers()
                self.wfile.write(b'upstream-secret-marker')
                return
            if state['mode'] == 'oversize': raw = b' ' * (admin.MAX_BODY + 1)
            elif state['mode'] == 'duplicate': raw = b'{"token":"first", "token":"second"}'
            elif state['mode'] == 'invalid': raw = b'upstream-secret-marker'
            elif self.path == '/auth/simple/login': raw = b'{"token":"synthetic-token"}'
            else:
                assert self.headers['Authorization'] == 'Bearer synthetic-token'
                if state['mode'] == 'graphql_error': result = {'errors': [{'message': 'upstream-secret-marker'}]}
                else: result = {'data': state['directory']._gql('', payload['query'], payload.get('variables'))}
                raw = json.dumps(result).encode()
            self.send_response(200)
            self.send_header('Content-Type', 'application/json')
            self.send_header('Content-Length', str(len(raw)))
            self.end_headers()
            try: self.wfile.write(raw)
            except (BrokenPipeError, ConnectionResetError): pass

    server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield f'http://127.0.0.1:{server.server_port}', state
    finally:
        server.shutdown()
        server.server_close()
        thread.join()


def private_root(tmp_path):
    root = tmp_path / 'private'
    root.mkdir(mode=0o700)
    (root / 'lldap').mkdir(mode=0o700)
    source = root / 'lldap/secrets.env'
    source.write_text('LLDAP_LDAP_USER_PASS=synthetic-private-marker\n')
    source.chmod(0o600)
    return root


@pytest.mark.parametrize('mode', ['redirect', 'denied', 'oversize', 'duplicate', 'invalid', 'graphql_error'])
def test_http_failure_does_not_forward_credentials_or_render_provider_body(tmp_path, http_directory, mode):
    base, state = http_directory
    state['mode'] = mode
    output, error = io.StringIO(), io.StringIO()
    with contextlib.redirect_stdout(output), contextlib.redirect_stderr(error):
        code = admin.main(['bootstrap', '--root', str(private_root(tmp_path)), '--lldap-url', base])
    assert code == 1
    assert state['redirect_hits'] == 0
    assert 'synthetic-private-marker' not in output.getvalue() + error.getvalue()
    assert 'upstream-secret-marker' not in output.getvalue() + error.getvalue()
    assert json.loads(error.getvalue())['result'] == 'unverified'


def test_loopback_http_uses_explicit_origin_and_ignores_ambient_proxy(tmp_path, http_directory):
    base, state = http_directory
    output = io.StringIO()
    with patch.dict(os.environ, {'http_proxy': 'http://127.0.0.1:1', 'HTTP_PROXY': 'http://127.0.0.1:1', 'no_proxy': '', 'NO_PROXY': ''}), contextlib.redirect_stdout(output):
        code = admin.main(['bootstrap', '--root', str(private_root(tmp_path)), '--lldap-url', base])
    assert code == 0
    assert json.loads(output.getvalue())['result'] == 'preview'
    assert len(state['calls']) == 2
    assert state['directory'].writes == []


@pytest.mark.parametrize('defect', ['symlink_file', 'hardlink_file', 'public_parent', 'duplicate_field', 'git_root'])
def test_private_input_boundary_refuses_unsafe_custody(tmp_path, defect):
    root = private_root(tmp_path)
    source = root / 'lldap/secrets.env'
    if defect == 'symlink_file':
        original = root / 'lldap/original'
        source.rename(original)
        source.symlink_to(original)
    elif defect == 'hardlink_file':
        os.link(source, root / 'lldap/alias')
    elif defect == 'public_parent':
        root.chmod(0o755)
    elif defect == 'duplicate_field':
        source.write_text(source.read_text() + 'LLDAP_LDAP_USER_PASS=second\n')
    elif defect == 'git_root':
        (root / '.git').mkdir()
        (root / '.git/HEAD').write_text('ref: refs/heads/main\n')
    with patch.object(admin, 'LLDAPProvisioner') as provider, contextlib.redirect_stderr(io.StringIO()):
        assert admin.main(['bootstrap', '--root', str(root)]) == 1
    provider.assert_not_called()


@pytest.mark.parametrize('extra', [['--apply'], ['--expected-digest', '0' * 64], ['--apply', '--expected-digest', 'not-a-digest']])
def test_apply_binding_errors_refuse_before_credential_read(extra):
    with patch.object(admin, 'private_password') as private, contextlib.redirect_stderr(io.StringIO()):
        assert admin.main(['bootstrap', '--root', '/not-read', *extra]) == 1
    private.assert_not_called()