Prepare isolated privacyIDEA token and audit recovery checks
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
This commit is contained in:
parent
0e298813c7
commit
1ab5d53bd7
8 changed files with 762 additions and 8 deletions
196
tests/test_isolated_privacyidea_recovery.py
Normal file
196
tests/test_isolated_privacyidea_recovery.py
Normal file
|
|
@ -0,0 +1,196 @@
|
|||
"""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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue