NK-WP-0049-T42: isolated existing-token MFA stage, fixture-tested

isolated-privacyidea-mfa.py restores the privacyIDEA database in the accepted
network-none stage and runs privacyIDEA's own check_serial_pass three times in
separate processes: wrong key, wrong code, the owner's code. The serial list is
terminal-only; receipts carry booleans, categories and a serial hash. A first
fixture run was vacuous (PRIVACYIDEA_CONFIGFILE overrode the wrong-key config);
the probe now fixes the variable and asserts the key path. Eight native fixture
tests pass (HOTP, TOTP with PIN, negatives, encrypted bundle). The user-to-token
resolver is not proven here and is recorded as such. T19 awaits the founder run.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 328925@bnt-lap001
Assistant-Session: 14b795fd-c632-4c01-8d28-e7e490c7b0ed

tests/test_isolated_privacyidea_mfa.py

@@ -0,0 +1,163 @@
+"""Existing-token MFA on a restored synthetic privacyIDEA database (disposable fixtures only)."""
+import importlib.util
+import json
+import os
+from pathlib import Path
+
+import pytest
+from test_identity_bundle import stage  # 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)
+
+CODE = r'''
+import contextlib,io,json,logging
+logging.disable(logging.CRITICAL)
+with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
+    from privacyidea.app import create_app
+    from privacyidea.lib.token import get_tokens
+    app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True)
+    with app.app_context():
+        out=[]
+        for t in get_tokens():
+            res,pin,otp,_=t.get_otp()
+            out.append({'serial':t.token.serial,'otp':otp})
+print(json.dumps(out))
+'''
+
+
+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_serial_pattern_refuses_shell_and_path_characters():
+    for bad in ('', 'a b', '../x', 'a;b', 'x' * 65):
+        assert not m.SERIAL.fullmatch(bad)
+    assert m.SERIAL.fullmatch('HOTP0001A2B3')
+
+
+def retained_codes(worker):
+    return {x['serial']: x['otp'] for x in json.loads(m.run(['docker', 'exec', worker, 'python3', '-c', CODE]))}
+
+
+def test_native_existing_token_passes_and_negatives_are_refused(retained_state):
+    root, _, worker = retained_state
+    codes = retained_codes(worker)
+    serial, otp = sorted(codes.items())[0]
+    result = {}
+    m.restore_and_check(root, result, lambda tokens: (serial, otp))
+    assert result['existing_token_mfa_proven'] is True, result
+    assert result['wrong_key_refused'] and result['wrong_code_refused']
+    assert result['existing_token_native_check_accepted']
+    assert result['retained_tokens_inventoried'] == 2 and result['selected_token_type'] == 'hotp'
+    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 serial not in text and otp not in text
+
+
+def test_native_wrong_code_alone_never_proves_mfa(retained_state):
+    root, _, worker = retained_state
+    serial, otp = sorted(retained_codes(worker).items())[0]
+    result = {}
+    with pytest.raises(m.bundle.Refused):
+        m.restore_and_check(root, result, lambda tokens: (serial, m.different_code(otp)))
+    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_serial_is_refused_before_any_check(retained_state):
+    root, _, _ = retained_state
+    result = {}
+    with pytest.raises(m.bundle.Refused):
+        m.restore_and_check(root, result, lambda tokens: ('NOSUCHTOKEN', '123456'))
+    assert result['failed_stage'] == 'owner_selection'
+    assert 'wrong_key_refused' not in result and result['container_cleanup_verified']
+
+
+def test_native_malformed_original_key_does_not_prove_mfa(retained_state):
+    root, _, worker = retained_state
+    serial, otp = sorted(retained_codes(worker).items())[0]
+    (root / 'data/privacyidea/enckey').write_bytes(b'synthetic-malformed-key')
+    result = {}
+    with pytest.raises(m.bundle.Refused):
+        m.restore_and_check(root, result, lambda tokens: (serial, otp))
+    assert result['existing_token_mfa_proven'] is False and result['container_cleanup_verified']
+
+
+def test_encrypted_bundle_to_native_mfa_and_tmpfs_cleanup(retained_state, stage, tmp_path, monkeypatch, capsys):  # noqa: F811
+    import shutil
+    import sys
+    root, _, worker = retained_state
+    serial, otp = sorted(retained_codes(worker).items())[0]
+    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')
+    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_choice', lambda tokens: (serial, otp))
+    # main() binds restore_and_check's callback by name at call time.
+    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 False
+    assert result['container_cleanup_verified'] and result['temporary_plaintext_cleanup_verified']
+    assert not result['production_contacted'] and not result['secret_values_exported']
+    text = capsys.readouterr().out + receipt.read_text()
+    assert serial not in text and otp not in text
+
+
+ADD_TOTP = r'''
+import contextlib,io,json,logging
+logging.disable(logging.CRITICAL)
+with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
+    from privacyidea.app import create_app
+    from privacyidea.lib.token import init_token
+    app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True)
+    with app.app_context():
+        token=init_token({'type':'totp','genkey':1,'pin':'synthetic-pin','serial':'TOTPFIXTURE01'})
+print(json.dumps({'serial':'TOTPFIXTURE01'}))
+'''
+
+
+def test_native_totp_token_with_pin_passes_within_the_check_sequence(retained_state):
+    root, database, worker = retained_state
+    m.run(['docker', 'exec', worker, 'python3', '-c', ADD_TOTP])
+    (root / 'databases/privacyidea.dump').write_bytes(m.run(
+        ['docker', 'exec', database, 'pg_dump', '-U', 'postgres', '-Fc', 'recovery']))
+    code = retained_codes(worker)['TOTPFIXTURE01']
+    result = {}
+    m.restore_and_check(root, result, lambda tokens: ('TOTPFIXTURE01', 'synthetic-pin' + code))
+    assert result['selected_token_type'] == 'totp' and result['retained_tokens_inventoried'] == 3
+    assert result['existing_token_mfa_proven'] is True, result
+    assert result['wrong_code_refusal'] in ('wrong_value', 'wrong_pin', 'other_refusal')