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