"""Actual SQLite/age/RSA fixtures and hostile archive/custody boundaries.""" import base64 import importlib.util import io import json import os from pathlib import Path import sqlite3 import subprocess import tarfile import time import uuid from copy import deepcopy import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa ROOT=Path(__file__).resolve().parents[1] def load(name): spec=importlib.util.spec_from_file_location(name,ROOT/f'tools/bootstrap-recovery/{name}.py') m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m);return m b=load('identity-bundle');files=load('identity-files');capture=load('capture-identity') @pytest.fixture def stage(tmp_path): os.umask(0o077);root=tmp_path/'payload';root.mkdir(mode=0o700) key=rsa.generate_private_key(public_exponent=65537,key_size=2048) private=key.private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.TraditionalOpenSSL,serialization.NoEncryption()) public=key.public_key().public_bytes(serialization.Encoding.PEM,serialization.PublicFormat.SubjectPublicKeyInfo) for c in b.profile()['components']: ns=c['namespace'];name=c['deployment'] d={'kind':'Deployment','metadata':{'name':name,'namespace':ns,'uid':c['deployment_uid']},'spec':{}} b.save(root,f'resources/{ns}/deployment/{name}.json',json.dumps(d).encode()) for name in c['secret_refs']: data={'fixture':base64.b64encode(b'synthetic-private-value-canary').decode()} if name=='keycape-config':data={'key.pem':base64.b64encode(private).decode(),'config.yaml':base64.b64encode(b'issuer: https://isolated.invalid').decode()} b.save(root,f'resources/{ns}/secret/{name}.json',json.dumps({'kind':'Secret','metadata':{'name':name,'namespace':ns},'data':data}).encode()) for name in c['configmap_refs']: b.save(root,f'resources/{ns}/configmap/{name}.json',json.dumps({'kind':'ConfigMap','metadata':{'name':name,'namespace':ns},'data':{'fixture':'synthetic'}}).encode()) for component,name in [('lldap','users.db'),('authelia','db.sqlite3')]: path=root/f'data/{component}/{name}';path.parent.mkdir(mode=0o700,parents=True) with sqlite3.connect(path) as db:db.execute('CREATE TABLE retained(value TEXT)');db.execute("INSERT INTO retained VALUES ('fixture-preserved')") path.chmod(0o600) b.save(root,'data/keycape/policy.json',b'{"version":1}') for name,value in [('enckey',os.urandom(96)),('private.pem',private),('public.pem',public)]: b.save(root,'data/privacyidea/'+name,value) b.save(root,'databases/privacyidea.dump',b'PGDMP-synthetic-format-only') return root def test_complete_bundle_has_native_integrity_checks(stage): r=b.checks(stage) assert r['sqlite_integrity_passed'] and r['privacyidea_key_format_and_pair_passed'] assert r['native_identity_recovery_proven'] is False def test_real_age_seal_manifest_is_encrypted_and_plaintext_never_prints(stage,tmp_path,monkeypatch,capsys): key=tmp_path/'fixture-age.key' subprocess.run(['age-keygen','-o',str(key)],check=True,capture_output=True) recipient=subprocess.run(['age-keygen','-y',str(key)],check=True,capture_output=True).stdout.decode().strip() p=b.profile();p['public_recipient']=recipient;monkeypatch.setattr(b,'profile',lambda:p) artifact=tmp_path/'identity.tar.age';r=b.seal(stage,artifact,{'test_only':True}) assert artifact.stat().st_mode&0o077==0 assert b'synthetic-private-value-canary' not in artifact.read_bytes() result=subprocess.run(['age','-d','-i',str(key),str(artifact)],check=True,capture_output=True) with tarfile.open(fileobj=io.BytesIO(result.stdout)) as archive: manifest=json.load(archive.extractfile('manifest.json')) assert len(manifest['files'])==r['file_count'] assert 'data/privacyidea/enckey' in archive.getnames() assert 'data/lldap/users.db' in archive.getnames() plain=tmp_path/'decrypted.tar';plain.write_bytes(result.stdout) recovered=tmp_path/'recovered';recovered.mkdir(mode=0o700) restored,checks=b.restore(plain,recovered) assert checks['privacyidea_key_format_and_pair_passed'] assert (recovered/'data/privacyidea/enckey').read_bytes()==(stage/'data/privacyidea/enckey').read_bytes() assert 'synthetic-private-value-canary' not in json.dumps(r)+capsys.readouterr().out with pytest.raises(b.Refused):b.seal(stage,artifact,{}) @pytest.mark.parametrize('defect',['missing-key','bad-sqlite','bad-audit-pair','wrong-deployment','private-age-key','symlink','hardlink','unknown-payload']) def test_incomplete_unsafe_and_mismatched_inputs_refuse(stage,tmp_path,defect): if defect=='missing-key':(stage/'data/privacyidea/enckey').unlink() if defect=='bad-sqlite':(stage/'data/lldap/users.db').write_bytes(b'broken') if defect=='bad-audit-pair': key=rsa.generate_private_key(public_exponent=65537,key_size=2048) (stage/'data/privacyidea/public.pem').write_bytes(key.public_key().public_bytes(serialization.Encoding.PEM,serialization.PublicFormat.SubjectPublicKeyInfo)) if defect=='wrong-deployment': p=stage/'resources/sso/deployment/keycape.json';r=json.loads(p.read_text());r['metadata']['uid']='replacement';p.write_text(json.dumps(r)) if defect=='private-age-key':(stage/'data/keycape/poison').write_bytes(b'AGE-SECRET-KEY-' + b'fixture') if defect=='symlink':(stage/'data/keycape/linked').symlink_to(stage/'data/privacyidea/enckey') if defect=='hardlink':os.link(stage/'data/privacyidea/enckey',stage/'data/keycape/linked') if defect=='unknown-payload':(stage/'unreviewed').write_bytes(b'not-allowed') with pytest.raises(Exception): b.checks(stage);b.inventory(stage) @pytest.mark.parametrize('name,kind',[('../escape','file'),('/absolute','file'),('data/keycape/valid','symlink')]) def test_tar_import_refuses_traversal_and_links(tmp_path,name,kind): raw=io.BytesIO() with tarfile.open(fileobj=raw,mode='w') as archive: member=tarfile.TarInfo(name);member.size=1 if kind=='symlink':member.type=tarfile.SYMTYPE;member.linkname='/etc/passwd' archive.addfile(member,io.BytesIO(b'x')) with pytest.raises(b.Refused):b.import_files(raw.getvalue(),tmp_path,'sso') def test_native_sqlite_snapshot_keeps_committed_wal_rows_and_source_unchanged(tmp_path): root=tmp_path/'sources';root.mkdir();scratch=tmp_path/'scratch';scratch.mkdir() for component in ['lldap','authelia','keycape']:(root/component).mkdir() for component,name in [('lldap','users.db'),('authelia','db.sqlite3')]: with sqlite3.connect(root/component/name) as con: con.execute('CREATE TABLE retained(value TEXT)');con.execute("INSERT INTO retained VALUES ('native-snapshot')") (root/'keycape/policy.json').write_text('{}') live=sqlite3.connect(root/'lldap/users.db');live.execute('PRAGMA journal_mode=WAL') live.execute("INSERT INTO retained VALUES ('committed-wal')");live.commit() try: result=files.capture('sso',root,scratch) restored=tmp_path/'restored.sqlite';restored.write_bytes(result['data/lldap/users.db']) with sqlite3.connect(restored) as con:assert con.execute('SELECT count(*) FROM retained').fetchone()==(2,) assert list(scratch.iterdir())==[] assert not any(name.endswith(('-wal','-shm')) for name in result) finally:live.close() def test_helper_manifest_has_no_writable_service_mount_or_credentials(): objects=capture.manifests('fixture') for obj in objects: if obj['kind']=='NetworkPolicy': assert obj['spec']['egress']==[] and obj['spec']['ingress']==[] else: s=obj['spec'];assert s['automountServiceAccountToken'] is False assert 'fsGroup' not in s['securityContext'] assert s['activeDeadlineSeconds']==900 assert all(v['persistentVolumeClaim']['readOnly'] for v in s['volumes'] if 'persistentVolumeClaim' in v) assert not any('secret' in v for v in s['volumes']) def test_cleanup_delete_carries_exact_uid_precondition(monkeypatch): calls=[];monkeypatch.setattr(capture,'command',lambda args,payload=None,**kw:calls.append((args,payload)) or b'') capture.delete_created({'kind':'Pod','metadata':{'namespace':'sso','name':'nk49-capture-fixture','uid':'exact-created-uid'}}) assert json.loads(calls[0][1])['preconditions']=={'uid':'exact-created-uid'} @pytest.mark.parametrize('defect',['writable-mount','extra-container','service-token','different-image','host-network']) def test_mutated_helper_is_refused_before_private_exec(defect): wanted=next(o for o in capture.manifests('fixture') if o['kind']=='Pod') actual=deepcopy(wanted);spec=actual['spec'] if defect=='writable-mount':spec['containers'][0]['volumeMounts'][0]['readOnly']=False if defect=='extra-container':spec['containers'].append(deepcopy(spec['containers'][0])) if defect=='service-token':spec['automountServiceAccountToken']=True if defect=='different-image':spec['containers'][0]['image']='unreviewed:latest' if defect=='host-network':spec['hostNetwork']=True with pytest.raises(capture.Refused):capture.validate_helper(actual,wanted) def test_namespace_dns_allow_is_excluded_only_for_capture_helpers(): selector={'matchExpressions':[{'key':'net-kingdom/capture','operator':'DoesNotExist'}]} assert capture.selects(selector,{'app.kubernetes.io/name':'authelia'}) assert not capture.selects(selector,{'net-kingdom/capture':'fixture'}) def test_unknown_create_outcome_is_checkpointed_and_never_claimed_clean(tmp_path,monkeypatch): monkeypatch.setattr(capture.bundle,'profile',lambda:{'components':[]}) monkeypatch.setattr(capture,'command',lambda *args,**kwargs:(_ for _ in ()).throw(TimeoutError())) r={'started_at':'fixture'};checkpoints=[] p={'run_id':'fixture','create_resources':[capture.manifests('fixture')[0]]} with pytest.raises(capture.Refused):capture.capture(p,tmp_path/'unused.age',r,'fixture',lambda:checkpoints.append(deepcopy(r))) assert r['helper_cleanup_verified'] is False and r['unverified_create_outcomes'] assert any(c.get('unverified_create_outcomes') for c in checkpoints) @pytest.mark.skipif(os.environ.get('NK49_RUN_DISPOSABLE_POSTGRES_TEST')!='1',reason='explicit local Docker fixture opt-in') def test_full_new_archive_decrypt_extract_native_database_restore_and_cleanup(stage,tmp_path,monkeypatch): verify=load('verify-identity');fixture='nk49-bundle-fixture-'+uuid.uuid4().hex def command(args,**kwargs): return subprocess.run(args,check=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=120,**kwargs).stdout try: command(['docker','run','--detach','--pull=never','--name',fixture,'--network','none', '--tmpfs','/var/lib/postgresql/data:rw,size=268435456','--env','POSTGRES_HOST_AUTH_METHOD=trust', verify.pg.IMAGE,'postgres','-c','listen_addresses=']) for _ in range(60): probe=subprocess.run(['docker','exec',fixture,'pg_isready','-U','postgres'],capture_output=True) if probe.returncode==0:break time.sleep(1) command(['docker','exec','-i',fixture,'psql','-X','-U','postgres','-v','ON_ERROR_STOP=1'], input=b'CREATE TABLE token(id integer); INSERT INTO token VALUES (1),(2); CREATE TABLE resolver(id integer); CREATE TABLE realm(id integer);') (stage/'databases/privacyidea.dump').write_bytes(command(['docker','exec',fixture,'pg_dump','-U','postgres','-Fc','postgres'])) key=tmp_path/'disposable-age.key';command(['age-keygen','-o',str(key)]) recipient=command(['age-keygen','-y',str(key)]).decode().strip() p=b.profile();p['public_recipient']=recipient monkeypatch.setattr(b,'profile',lambda:p);monkeypatch.setattr(verify.bundle,'profile',lambda:p) artifact=tmp_path/'net-kingdom-identity-20261016T000000Z-0123456789ab.tar.age' c=b.seal(stage,artifact,{'test_only':True}) transfer={'schema':'netkingdom.identity-offsite.v1','status':'readback_verified', 'independent_owner_retrieval_verified':True,'fetched_path':str(artifact),'artifact':artifact.name, 'ciphertext_sha256':c['ciphertext_sha256'],'ciphertext_bytes':c['ciphertext_bytes']} r={};fd=os.open(key,os.O_RDONLY) try:verify.verify(transfer,fd,r) finally:os.close(fd) assert r['status']=='identity_archive_verified' assert r['database_restore']['aggregate_row_counts']['token']==2 assert r['all_manifest_hashes_verified'] and r['wrong_key_refused'] assert r['container_cleanup_verified'] and r['temporary_plaintext_cleanup_verified'] assert not r['full_service_authentication_proven'] finally:subprocess.run(['docker','rm','--force','--volumes',fixture],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=30) def test_runtime_host_requires_owner_and_exact_machine_binding(monkeypatch): import hashlib,socket module=load('identity-bundle') current=module.profile();current['runtime_host']={'hostname':'verified-node','machine_id_sha256':hashlib.sha256(b'fixture-machine').hexdigest()} monkeypatch.setattr(module,'profile',lambda:current) monkeypatch.setattr(module.os,'geteuid',lambda:0);monkeypatch.setattr(socket,'gethostname',lambda:'verified-node') original=Path.read_bytes monkeypatch.setattr(Path,'read_bytes',lambda p:b'fixture-machine' if str(p)=='/etc/machine-id' else original(p)) module.require_runtime_host() monkeypatch.setattr(socket,'gethostname',lambda:'foreign-node') with pytest.raises(module.Refused):module.require_runtime_host() monkeypatch.setattr(socket,'gethostname',lambda:'verified-node');monkeypatch.setattr(module.os,'geteuid',lambda:1000) with pytest.raises(module.Refused):module.require_runtime_host()