tegwick
a1939a7d06
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
129 lines
6.5 KiB
Python
129 lines
6.5 KiB
Python
"""Transfer binding failures and a real disposable PostgreSQL fixture restore."""
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
import pytest
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def load(name):
|
|
spec=importlib.util.spec_from_file_location(name,ROOT/f'tools/bootstrap-recovery/{name}.py')
|
|
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
fetch=load('fetch-privacyidea-backup')
|
|
drill=load('check-privacyidea-database')
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
def __init__(self,body=b'abc',etag='"same"',length='3',status=200):
|
|
super().__init__(body);self.status=status
|
|
self.url=fetch.owner.HOST+fetch.owner.BASE+fetch.OBJECT
|
|
self.headers={'ETag':etag,'Content-Length':length}
|
|
|
|
|
|
class Provider:
|
|
def __init__(self,responses):self.responses=iter(responses);self.requests=[]
|
|
def open(self,req,timeout):self.requests.append(req);return next(self.responses)
|
|
|
|
|
|
def test_transfer_binds_head_get_and_recheck(monkeypatch,tmp_path):
|
|
monkeypatch.setattr(fetch,'SIZE',3)
|
|
provider=Provider([Response(),Response(),Response()]);target=tmp_path/'ciphertext'
|
|
result=fetch.transfer(provider,'disposable-test-auth',target)
|
|
assert result['ciphertext_sha256']==hashlib.sha256(b'abc').hexdigest()
|
|
assert target.stat().st_mode&0o077==0
|
|
assert provider.requests[1].get_header('If-match')=='"same"'
|
|
|
|
|
|
@pytest.mark.parametrize('responses',[
|
|
[Response(etag='W/"weak"')],
|
|
[Response(),Response(etag='"changed"')],
|
|
[Response(),Response(body=b'ab')],
|
|
[Response(),Response(body=b'abcd')],
|
|
[Response(),Response(),Response(etag='"changed"')],
|
|
[Response(status=302)],
|
|
])
|
|
def test_provider_changes_truncation_oversize_and_redirect_refuse(monkeypatch,tmp_path,responses):
|
|
monkeypatch.setattr(fetch,'SIZE',3)
|
|
with pytest.raises(fetch.owner.Refused):fetch.transfer(Provider(responses),'fixture',tmp_path/'ciphertext')
|
|
|
|
|
|
def test_existing_ciphertext_is_preserved(monkeypatch,tmp_path):
|
|
monkeypatch.setattr(fetch,'SIZE',3);target=tmp_path/'ciphertext';target.write_bytes(b'existing')
|
|
with pytest.raises(FileExistsError):fetch.transfer(Provider([Response(),Response()]),'fixture',target)
|
|
assert target.read_bytes()==b'existing'
|
|
|
|
|
|
def test_transfer_digest_and_permissions_gate_private_key_use(monkeypatch,tmp_path):
|
|
monkeypatch.setattr(drill,'SIZE',3)
|
|
with tempfile.TemporaryDirectory(prefix='nk49-offhost-',dir='/tmp') as directory:
|
|
target=Path(directory)/'privacyidea.dump.age';target.write_bytes(b'abc');target.chmod(0o600)
|
|
r={'schema':'netkingdom.privacyidea-offhost-transfer.v1','status':'ciphertext_fetched',
|
|
'object':drill.OBJECT,'ciphertext_bytes':3,'object_unchanged_during_transfer':True,
|
|
'ciphertext_path':str(target),'ciphertext_sha256':hashlib.sha256(b'abc').hexdigest()}
|
|
receipt=tmp_path/'receipt.json';receipt.write_text(json.dumps(r))
|
|
assert drill.transfer_input(receipt)[0]==b'abc'
|
|
target.chmod(0o644)
|
|
with pytest.raises(drill.Refused):drill.transfer_input(receipt)
|
|
target.chmod(0o600);target.write_bytes(b'bad')
|
|
with pytest.raises(drill.Refused):drill.transfer_input(receipt)
|
|
|
|
|
|
@pytest.mark.skipif(os.environ.get('NK49_RUN_DISPOSABLE_POSTGRES_TEST')!='1',reason='explicit local Docker fixture opt-in')
|
|
def test_real_disposable_custom_dump_restore_and_readback(tmp_path,monkeypatch,capsys):
|
|
"""No provider/key/production access: three synthetic PI-shaped tables."""
|
|
fixture='nk49-fixture-'+uuid.uuid4().hex;restored='nk49-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',drill.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); INSERT INTO realm VALUES (1);')
|
|
dump=tmp_path/'fixture.dump'
|
|
dump.write_bytes(command(['docker','exec',fixture,'pg_dump','-U','postgres','-Fc','postgres']))
|
|
result=drill.restore(dump,restored)
|
|
assert result['aggregate_row_counts']=={'token':2,'resolver':0,'realm':1}
|
|
assert '16.' in result['pg_restore_version']
|
|
# Exercise the actual custody/decryption/negative/restore/cleanup path
|
|
# with a freshly created fixture key, never an operator key.
|
|
key=tmp_path/'fixture.key'
|
|
command(['age-keygen','-o',str(key)])
|
|
recipient=command(['age-keygen','-y',str(key)]).decode().strip()
|
|
blob=command(['age','-r',recipient],input=dump.read_bytes())
|
|
monkeypatch.setenv('NETKINGDOM_CUSTODY_AGE_KEY_FILE',str(key))
|
|
monkeypatch.setattr(drill,'RECIPIENT',recipient)
|
|
monkeypatch.setattr(drill,'SIZE',len(blob))
|
|
monkeypatch.setattr(drill,'transfer_input',lambda path:(blob,hashlib.sha256(blob).hexdigest()))
|
|
receipt=tmp_path/'drill.json'
|
|
monkeypatch.setattr('sys.argv',['check','--transfer-receipt','fixture','--receipt',str(receipt)])
|
|
assert drill.main()==0
|
|
r=json.loads(receipt.read_text())
|
|
assert r['wrong_key_refused'] and r['offhost_ciphertext_decryption_verified']
|
|
assert r['aggregate_row_counts']=={'token':2,'resolver':0,'realm':1}
|
|
assert r['container_cleanup_verified'] and r['temporary_plaintext_cleanup_verified']
|
|
assert r['native_identity_recovery_proven'] is False
|
|
assert 'AGE-SECRET-KEY-' not in capsys.readouterr().out
|
|
failed=tmp_path/'wrong-recipient.json'
|
|
monkeypatch.setattr(drill,'RECIPIENT','age1invalid')
|
|
monkeypatch.setattr('sys.argv',['check','--transfer-receipt','fixture','--receipt',str(failed)])
|
|
assert drill.main()==1
|
|
assert json.loads(failed.read_text())['status']=='failed'
|
|
finally:
|
|
for name in (fixture,restored):
|
|
subprocess.run(['docker','rm','--force','--volumes',name],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=30)
|