net-kingdom/tests/test_isolated_privacyidea_mfa.py

"""Existing-token MFA by user and realm on a restored synthetic privacyIDEA + LLDAP pair."""
import importlib.util
import json
import os
from pathlib import Path
import re
import shutil
import subprocess

import pytest
from test_identity_bundle import stage  # noqa: F401  (fixture reuse)
from test_isolated_lldap_recovery import PASSWORD, retained_directory  # noqa: F401  (fixture reuse)
from test_isolated_privacyidea_recovery import retained_state  # noqa: F401  (fixture reuse)

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('pi_mfa', ROOT / 'tools/bootstrap-recovery/isolated-privacyidea-mfa.py')
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
BASE = 'dc=example,dc=test'

SETUP = r'''
import contextlib,io,json,logging,sys
logging.disable(logging.CRITICAL)
v=json.load(sys.stdin)
with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
    from privacyidea.app import create_app
    from privacyidea.lib.resolver import save_resolver
    from privacyidea.lib.realm import set_realm,set_default_realm
    from privacyidea.lib.user import User
    from privacyidea.lib.token import init_token,assign_token,get_tokens
    app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True)
    with app.app_context():
        save_resolver({'resolver':'lldap-coulomb','type':'ldapresolver','LDAPURI':'ldap://127.0.0.1:3890',
            'LDAPBASE':'ou=people,'+v['base'],'BINDDN':'uid=admin,ou=people,'+v['base'],'BINDPW':v['password'],
            'LOGINNAMEATTRIBUTE':'uid','LDAPSEARCHFILTER':'(uid=*)','LDAPFILTER':'(&(uid=%s))',
            'USERINFO':json.dumps({'username':'uid','email':'mail'}),
            'UIDTYPE':'uid','SCOPE':'SUBTREE','AUTHTYPE':'Simple','TIMEOUT':'5','CACHE_TIMEOUT':'0',
            'SIZELIMIT':'500','NOREFERRALS':'True','TLS_VERIFY':'False'})
        set_realm('coulomb',[{'name':'lldap-coulomb','priority':1}])
        set_default_realm('coulomb')
        owner=User('admin','coulomb')
        assert owner.exist()
        if v['kind']=='totp':
            token=init_token({'type':'totp','genkey':1,'pin':'synthetic-pin','serial':'TOTPFIXTURE01'})
            assign_token('TOTPFIXTURE01',owner)
            serial='TOTPFIXTURE01'
        else:
            serial=get_tokens()[0].token.serial
            assign_token(serial,owner)
        otp=get_tokens(serial=serial)[0].get_otp()[2]
print(json.dumps({'serial':serial,'otp':otp}))
'''


@pytest.fixture(params=['hotp', 'totp'])
def owned_state(request, tmp_path, retained_state, retained_directory):  # noqa: F811
    """Dump of a database whose resolver names the in-cluster address, as production does."""
    root, database, worker = retained_state
    shutil.copytree(tmp_path / 'data/lldap', root / 'data/lldap')
    shutil.copytree(tmp_path / 'resources/sso', root / 'resources/sso')
    directory = database + '-fixture-dir'
    try:
        m.start_directory(directory, database, root)
        m.run(['docker', 'exec', worker, 'python3', '-c', m.WAIT_LDAP], timeout=90)
        raw = m.run(['docker', 'exec', '-i', worker, 'python3', '-c', SETUP],
                    input=json.dumps({'base': BASE, 'password': PASSWORD, 'kind': request.param}).encode(), timeout=120)
        made = json.loads(raw)
        assert m.psql(database, "UPDATE resolverconfig SET \"Value\"='" + m.CLUSTER_LDAP + "' WHERE \"Key\"='LDAPURI'") == 'UPDATE 1'
        (root / 'databases/privacyidea.dump').write_bytes(m.run(
            ['docker', 'exec', database, 'pg_dump', '-U', 'postgres', '-Fc', 'recovery']))
        yield root, made['otp'], request.param
    finally:
        m.pi.cleanup([directory])


def passcode(otp, kind):
    return ('synthetic-pin' if kind == 'totp' else '') + otp


def test_launcher_preflight_entry_point_exists_and_launcher_calls_it():
    # The launcher swallows exceptions into one message; a missing attribute once hid here.
    assert callable(m.preflight)
    launcher = (ROOT / 'tools/bootstrap-recovery/run-isolated-privacyidea-mfa.sh').read_text()
    assert 'isolated-privacyidea-mfa.py' in launcher and 'm.preflight()' in launcher


def test_launcher_preflight_snippet_passes_against_the_pinned_target():
    launcher = (ROOT / 'tools/bootstrap-recovery/run-isolated-privacyidea-mfa.sh').read_text()
    snippet = re.search(r"python3 - <<'PY'\n(.*?)\nPY\n", launcher, re.S).group(1)
    done = subprocess.run(['python3', '-c', snippet], cwd=ROOT, capture_output=True, text=True, timeout=120)
    assert done.returncode == 0 and done.stdout.startswith('OK: exact bnt-lap001'), done.stdout + done.stderr


def test_wrong_code_keeps_shape_but_changes_value():
    assert m.different_code('123456') == '123450'
    assert m.different_code('123450') == '123451'
    assert len(m.different_code('pin123456')) == len('pin123456')


def test_login_and_realm_patterns_refuse_shell_and_path_characters():
    for bad in ('', 'a b', '../x', 'a;b', 'x' * 129):
        assert not m.LOGIN.fullmatch(bad)
    assert m.LOGIN.fullmatch('first.last@example.test')
    assert not m.REALM.fullmatch('a/b') and m.REALM.fullmatch('coulomb')


def test_resolver_rewrite_accepts_only_the_archived_cluster_address(monkeypatch):
    calls = []

    def fake(database, statement):
        calls.append(statement)
        return 'UPDATE 2' if statement.startswith('UPDATE') else '\n'.join([m.CLUSTER_LDAP] * 2)
    monkeypatch.setattr(m, 'psql', fake)
    assert m.rewrite_resolvers('db') == 2 and calls[1].startswith('UPDATE')
    monkeypatch.setattr(m, 'psql', lambda d, s: 'ldap://elsewhere.example:389')
    with pytest.raises(m.bundle.Refused):
        m.rewrite_resolvers('db')
    monkeypatch.setattr(m, 'psql', lambda d, s: '')
    with pytest.raises(m.bundle.Refused):
        m.rewrite_resolvers('db')
    monkeypatch.setattr(m, 'psql', lambda d, s: 'UPDATE 1' if s.startswith('UPDATE') else m.CLUSTER_LDAP + '\n' + m.CLUSTER_LDAP)
    with pytest.raises(m.bundle.Refused):
        m.rewrite_resolvers('db')


def test_native_existing_token_passes_by_user_and_negatives_are_refused(owned_state):
    root, otp, kind = owned_state
    result = {}
    m.restore_and_check(root, result, lambda realms, default: ('admin', default), lambda summary: passcode(otp, kind))
    assert result['existing_token_mfa_proven'] is True, result
    assert result['wrong_key_refused'] and result['wrong_code_refused'] and result['existing_token_native_check_accepted']
    assert result['user_to_token_resolver_proven'] and result['user_resolved_through_restored_resolver_and_directory']
    assert result['ldap_resolvers_rewritten_to_loopback'] == 1 and result['user_usable_token_types'] == [kind]
    assert result['container_cleanup_verified'] and result['container_isolation_verified_before_private_import']
    assert result['only_loopback_interface_verified'] and result['external_routes_absent']
    text = json.dumps(result)
    assert otp not in text and 'admin' not in text.replace('"selected_user_sha256"', '')


def test_native_wrong_code_alone_never_proves_mfa(owned_state):
    root, otp, kind = owned_state
    result = {}
    with pytest.raises(m.bundle.Refused):
        m.restore_and_check(root, result, lambda realms, default: ('admin', default),
                            lambda summary: m.different_code(passcode(otp, kind)))
    assert result['failed_stage'] == 'acceptance'
    assert result['existing_token_mfa_proven'] is False and not result['existing_token_native_check_accepted']
    assert result['container_cleanup_verified']


def test_native_unknown_user_is_refused_before_any_code_is_requested(owned_state):
    root, _, _ = owned_state
    result = {}
    asked = []
    with pytest.raises(m.bundle.Refused):
        m.restore_and_check(root, result, lambda realms, default: ('nosuchuser', default),
                            lambda summary: asked.append(1) or '123456')
    assert result['failed_stage'] == 'user_lookup_through_restored_resolver' and not asked
    assert result['user_resolved_through_restored_resolver_and_directory'] is False
    assert result['container_cleanup_verified']


def test_native_malformed_original_key_does_not_prove_mfa(owned_state):
    root, otp, kind = owned_state
    (root / 'data/privacyidea/enckey').write_bytes(b'synthetic-malformed-key')
    result = {}
    with pytest.raises(m.bundle.Refused):
        m.restore_and_check(root, result, lambda realms, default: ('admin', default), lambda summary: passcode(otp, kind))
    assert not result.get('existing_token_mfa_proven') and not result.get('existing_token_native_check_accepted')
    assert result['failed_stage'] and result['container_cleanup_verified']


def test_encrypted_bundle_to_native_mfa_and_tmpfs_cleanup(owned_state, stage, tmp_path, monkeypatch, capsys):  # noqa: F811
    import sys
    root, otp, kind = owned_state
    for key in ('enckey', 'private.pem', 'public.pem'):
        shutil.copyfile(root / 'data/privacyidea' / key, stage / 'data/privacyidea' / key)
    shutil.copyfile(root / 'databases/privacyidea.dump', stage / 'databases/privacyidea.dump')
    for source in (root / 'data/lldap').iterdir():
        if source.is_file():
            shutil.copyfile(source, stage / 'data/lldap' / source.name)
    for kind_, name in (('deployment', 'lldap'), ('secret', 'lldap-secrets')):
        target = stage / 'resources/sso' / kind_ / (name + '.json')
        value = json.loads(target.read_text())
        value.update({k: v for k, v in json.loads((root / 'resources/sso' / kind_ / (name + '.json')).read_text()).items()
                      if k != 'metadata'})
        target.write_text(json.dumps(value))
    resource = stage / 'resources/mfa/secret/privacyidea-config.json'
    value = json.loads(resource.read_text())
    value['data'] = json.loads((root / 'resources/mfa/secret/privacyidea-config.json').read_text())['data']
    resource.write_text(json.dumps(value))
    configmap = stage / 'resources/mfa/configmap/privacyidea-cfg.json'
    value = json.loads(configmap.read_text())
    value['data'] = {'pi.cfg': 'PI_CHECK_OLD_SIGNATURES = False\n'}
    configmap.write_text(json.dumps(value))
    key = tmp_path / 'synthetic-age.key'
    m.run(['age-keygen', '-o', str(key)])
    recipient = m.run(['age-keygen', '-y', str(key)]).decode().strip()
    profile = m.bundle.profile()
    profile['public_recipient'] = recipient
    monkeypatch.setattr(m.bundle, 'profile', lambda: profile)
    ciphertext = tmp_path / 'net-kingdom-identity-20261016T063159Z-0123456789ab.tar.age'
    m.bundle.seal(stage, ciphertext, {'synthetic_native_fixture': True})
    transfer = tmp_path / 'transfer.json'
    transfer.write_text(json.dumps({'schema': 'netkingdom.identity-offsite.v1', 'status': 'readback_verified',
        'provider': 'Storage Share', 'folder': 'Backup/railiance-backups/net-kingdom-identity',
        'artifact': ciphertext.name, 'ciphertext_bytes': ciphertext.stat().st_size,
        'ciphertext_sha256': m.local.offline.offsite.digest(ciphertext), 'etag': '"synthetic"',
        'fetched_path': str(ciphertext), 'independent_owner_retrieval_verified': True}))
    monkeypatch.setattr(m.local.verify.custody, 'private_key_fd', lambda: os.open(key, os.O_RDONLY))
    monkeypatch.setattr(m, 'owner_user', lambda realms, default: ('admin', default))
    monkeypatch.setattr(m, 'owner_code', lambda summary: passcode(otp, kind))
    receipt = tmp_path / 'receipt.json'
    monkeypatch.setattr(sys, 'argv', ['isolated-privacyidea-mfa', '--transfer-receipt', str(transfer),
                                      '--receipt', str(receipt)])
    assert m.main() == 0, receipt.read_text()
    result = json.loads(receipt.read_text())
    assert result['status'] == 'isolated_privacyidea_mfa_verified' and result['existing_token_mfa_proven']
    assert result['user_to_token_resolver_proven'] is True
    assert result['container_cleanup_verified'] and result['temporary_plaintext_cleanup_verified']
    assert not result['production_contacted'] and not result['secret_values_exported']
    assert otp not in capsys.readouterr().out + receipt.read_text()