tegwick
1ab5d53bd7
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
196 lines
11 KiB
Python
196 lines
11 KiB
Python
"""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()
|