"""Native pinned PI/PostgreSQL recovery with synthetic token/audit material.""" import base64 import importlib.util import json import os from pathlib import Path import secrets import subprocess import textwrap import uuid import pytest from test_identity_bundle import stage from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location('pi_recovery', ROOT / 'tools/bootstrap-recovery/isolated-privacyidea.py') m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) def test_archived_legacy_setting_cannot_execute_expressions(tmp_path): folder = tmp_path / 'resources/mfa/configmap';folder.mkdir(parents=True) value = {'metadata': {'namespace': 'mfa', 'name': 'privacyidea-cfg'}, 'data': {'pi.cfg': 'PI_CHECK_OLD_SIGNATURES = __import__("os").system("false")'}} (folder / 'privacyidea-cfg.json').write_text(json.dumps(value)) with pytest.raises(m.bundle.Refused): m.legacy_setting(tmp_path) def test_archived_literal_legacy_setting_is_retained(tmp_path): folder = tmp_path / 'resources/mfa/configmap';folder.mkdir(parents=True) value = {'metadata': {'namespace': 'mfa', 'name': 'privacyidea-cfg'}, 'data': {'pi.cfg': 'import os\nPI_CHECK_OLD_SIGNATURES = True'}} (folder / 'privacyidea-cfg.json').write_text(json.dumps(value)) assert m.legacy_setting(tmp_path) is True @pytest.fixture def retained_state(tmp_path): if os.environ.get('NK49_RUN_ISOLATED_PRIVACYIDEA_TEST') != '1': pytest.skip('explicit disposable pinned native-image fixture opt-in') m.preflight() os.umask(0o077) root = tmp_path / 'retained' keys = root / 'data/privacyidea';keys.mkdir(parents=True) key = rsa.generate_private_key(public_exponent=65537, key_size=2048) (keys / 'private.pem').write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption())) (keys / 'public.pem').write_bytes(key.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)) (keys / 'enckey').write_bytes(os.urandom(96)) resource = root / 'resources/mfa/secret';resource.mkdir(parents=True) (resource / 'privacyidea-config.json').write_text(json.dumps({ 'metadata': {'namespace': 'mfa', 'name': 'privacyidea-config'}, 'data': {name: base64.b64encode(secrets.token_bytes(32).hex().encode()).decode() for name in ('PI_SECRET_KEY', 'PI_PEPPER')}})) (root / 'databases').mkdir() name = 'nk49-pi-fixture-' + uuid.uuid4().hex database = name + '-db';worker = name + '-worker' setup = r''' import contextlib,io,logging,json logging.disable(logging.CRITICAL) with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()): from privacyidea.app import create_app from privacyidea.models import db from privacyidea.lib.token import init_token from privacyidea.lib.auditmodules.sqlaudit import Audit,LogEntry app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True) with app.app_context(): db.create_all() audit=Audit(app.config) LogEntry.__table__.create(audit.engine,checkfirst=True) for _ in range(2): token=init_token({'type':'hotp','genkey':1,'pin':''}) audit=Audit(app.config) audit.log({'action':'synthetic retained fixture','success':True,'info':'synthetic only'}) audit.finalize_log() print(json.dumps({'synthetic_state_created':True})) ''' try: m.start_database(database) m.start_worker(worker, database, root) safe_setup = ('import traceback,sys,json\ntry:\n' + textwrap.indent(setup, ' ') + '\nexcept Exception as error:\n' ' frame=traceback.extract_tb(error.__traceback__)[-1]\n' ' original=getattr(error,"orig",None)\n' ' text=str(original).lower()\n' ' categories=[x for x in ("connection refused","permission denied","read-only file system","no pg_hba.conf entry","password authentication failed","could not translate host name") if x in text]\n' ' print(json.dumps({"type":type(error).__name__,"file":frame.filename.split("/")[-1],"line":frame.lineno,"sqlstate":getattr(original,"pgcode",None),"categories":categories,"frames":[f.name for f in traceback.extract_tb(error.__traceback__)]}))\n' ' sys.exit(1)\n') result = subprocess.run(['docker', 'exec', worker, 'python3', '-c', safe_setup], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=90) assert result.returncode == 0, result.stdout.decode() assert json.loads(result.stdout)['synthetic_state_created'] dump = m.run(['docker', 'exec', database, 'pg_dump', '-U', 'postgres', '-Fc', 'recovery']) (root / 'databases/privacyidea.dump').write_bytes(dump) yield root, database, worker finally: m.cleanup([worker, database]) def test_native_existing_token_material_and_audit_history_survive_restore(retained_state): root, _, _ = retained_state result = {} m.native_checks(root, result) assert result['status'] == 'native_checks_completed' assert result['retained_tokens_with_native_decryption'] == 2 assert result['different_key_cannot_reproduce_material_count'] == 2 assert result['historical_audit_rows'] == result['historical_signed_rows'] == 2 assert result['historical_invalid_signatures'] == result['historical_unsigned_rows'] == 0 assert result['audit_signature_continuity_proven'] assert result['new_audit_entry_signed_verified'] and result['wrong_audit_public_key_refused'] assert result['tampered_entry_refused'] and result['container_cleanup_verified'] assert result['container_isolation_verified_before_private_import'] assert not result['existing_token_mfa_proven'] and not result['event_completeness_claimed'] @pytest.mark.parametrize('problem', ['tampered', 'unsigned']) def test_native_history_exceptions_reported_without_hiding_with_new_keys(retained_state, problem): root, database, worker = retained_state mutate = r''' import contextlib,io,logging,sys logging.disable(logging.CRITICAL) with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()): from privacyidea.app import create_app from privacyidea.lib.auditmodules.sqlaudit import Audit,LogEntry app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True) with app.app_context(): audit=Audit(app.config) row=audit.session.query(LogEntry).first() row.signature=None if sys.argv[1]=='unsigned' else 'rsa_sha256_pss:'+('00'*256) audit.session.commit() ''' m.run(['docker', 'exec', worker, 'python3', '-c', mutate, problem]) (root / 'databases/privacyidea.dump').write_bytes(m.run( ['docker', 'exec', database, 'pg_dump', '-U', 'postgres', '-Fc', 'recovery'])) result = {} m.native_checks(root, result) assert result['historical_invalid_signatures'] == int(problem == 'tampered') assert result['historical_unsigned_rows'] == int(problem == 'unsigned') assert not result['audit_signature_continuity_proven'] assert result['new_audit_entry_signed_verified'] and result['container_cleanup_verified'] def test_native_malformed_original_encryption_key_fails_and_cleans_up(retained_state): root, _, _ = retained_state (root / 'data/privacyidea/enckey').write_bytes(b'synthetic-malformed-key') result = {} with pytest.raises(m.bundle.Refused): m.native_checks(root, result) assert result['failed_stage'] == 'retained_token_decryption' assert result['container_cleanup_verified'] assert not result.get('retained_token_material_decrypted_inside_process') def test_encrypted_bundle_to_native_checks_and_tmpfs_cleanup(retained_state, stage, tmp_path, monkeypatch, capsys): import shutil import sys root, _, _ = retained_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') 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)) receipt = tmp_path / 'receipt.json' monkeypatch.setattr(sys, 'argv', ['isolated-privacyidea', '--transfer-receipt', str(transfer), '--receipt', str(receipt)]) assert m.main() == 0 result = json.loads(receipt.read_text()) assert result['status'] == 'isolated_privacyidea_checked' and result['audit_signature_continuity_proven'] assert result['container_cleanup_verified'] and result['temporary_plaintext_cleanup_verified'] assert result['all_manifest_hashes_verified'] and result['retained_tokens_with_native_decryption'] == 2 assert not result['production_contacted'] and not result['secret_values_exported'] assert not result['existing_token_mfa_proven'] assert 'rsa_sha256_pss:' not in capsys.readouterr().out + receipt.read_text()