"""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()