tegwick
a17aa7edc1
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
159 lines
8.4 KiB
Python
159 lines
8.4 KiB
Python
"""Contained provider fixtures: exact quota, conditional publication and retention."""
|
|
from datetime import datetime,timedelta,timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import urllib.error
|
|
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')
|
|
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m);return m
|
|
|
|
|
|
o=load('identity-offsite');r=load('identity-retention');setup=load('identity-credential-setup');runtime=load('identity-runtime')
|
|
TOKEN='fixturefixturefixture'
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
def __init__(self,url,data=b'',status=200,version='"generation"'):
|
|
super().__init__(data);self.url=url;self.status=status
|
|
self.headers={'ETag':version,'Content-Length':str(len(data))}
|
|
|
|
|
|
class Storage:
|
|
def __init__(self):self.objects={};self.calls=[];self.free=10*1024**3;self.tamper=False
|
|
def open(self,req,timeout):
|
|
self.calls.append((req.get_method(),req.full_url,dict(req.headers)))
|
|
name=req.full_url.removeprefix(o.BASE)
|
|
if req.get_method()=='PROPFIND':
|
|
return Response(req.full_url,f'<d:multistatus xmlns:d="DAV:"><d:quota-available-bytes>{self.free}</d:quota-available-bytes></d:multistatus>'.encode(),207)
|
|
if req.get_method()=='PUT':
|
|
assert req.get_header('If-none-match')=='*'
|
|
if name in self.objects:raise urllib.error.HTTPError(req.full_url,412,'exists',{},None)
|
|
value=req.data if isinstance(req.data,bytes) else req.data.read()
|
|
self.objects[name]=value
|
|
return Response(req.full_url,status=201)
|
|
if name not in self.objects:raise urllib.error.HTTPError(req.full_url,404,'missing',{},None)
|
|
body=self.objects[name]
|
|
if req.get_method()=='GET' and self.tamper:body=b'X'*len(body)
|
|
if req.get_method()=='DELETE':
|
|
assert req.get_header('If-match')=='"generation"'
|
|
del self.objects[name];return Response(req.full_url,status=204)
|
|
return Response(req.full_url,body)
|
|
|
|
|
|
@pytest.fixture
|
|
def source(tmp_path):
|
|
name='net-kingdom-identity-20261016T000000Z-0123456789ab.tar.age';path=tmp_path/name
|
|
path.write_bytes(b'age-encryption.org/v1\n'+b'ciphertext-fixture'*10);path.chmod(0o600)
|
|
record={'status':'encrypted_bundle_created','helper_cleanup_verified':True,'temporary_plaintext_cleanup_verified':True,
|
|
'ciphertext_sha256':o.digest(path),'ciphertext_bytes':path.stat().st_size,'finished_at':'2026-10-16T00:00:30+00:00'}
|
|
return path,record
|
|
|
|
|
|
def test_conditional_upload_exact_readback_and_index_commit(source,tmp_path):
|
|
path,capture=source;server=Storage();target=tmp_path/'fetched.age'
|
|
receipt=o.publish(server,TOKEN,path,capture,target)
|
|
assert receipt['status']=='readback_verified' and target.read_bytes()==path.read_bytes()
|
|
assert receipt['verified_index_published']
|
|
assert len(server.objects)==2
|
|
assert not any(method=='DELETE' for method,_,_ in server.calls)
|
|
assert TOKEN not in json.dumps(receipt)
|
|
assert all(url.startswith(o.BASE) for _,url,_ in server.calls)
|
|
|
|
|
|
def test_quota_shortfall_refuses_before_any_provider_write(source,tmp_path):
|
|
path,capture=source;server=Storage();server.free=o.HEADROOM+path.stat().st_size-1
|
|
with pytest.raises(o.Refused):o.publish(server,TOKEN,path,capture,tmp_path/'fetched.age')
|
|
assert not server.objects and not any(method=='PUT' for method,_,_ in server.calls)
|
|
|
|
|
|
def test_existing_archive_never_overwritten(source,tmp_path):
|
|
path,capture=source;server=Storage();server.objects[path.name]=b'existing-protected-object'
|
|
with pytest.raises(o.Refused):o.publish(server,TOKEN,path,capture,tmp_path/'fetched.age')
|
|
assert server.objects[path.name]==b'existing-protected-object'
|
|
|
|
|
|
def test_corrupted_readback_never_publishes_success_index(source,tmp_path):
|
|
path,capture=source;server=Storage();server.tamper=True
|
|
with pytest.raises(o.Refused):o.publish(server,TOKEN,path,capture,tmp_path/'fetched.age')
|
|
assert list(server.objects)==[path.name]
|
|
|
|
|
|
@pytest.mark.parametrize('key',['helper_cleanup_verified','temporary_plaintext_cleanup_verified'])
|
|
def test_unverified_capture_cleanup_prevents_provider_access(source,tmp_path,key):
|
|
path,capture=source;capture[key]=False;server=Storage()
|
|
with pytest.raises(o.Refused):o.publish(server,TOKEN,path,capture,tmp_path/'fetched.age')
|
|
assert not server.calls
|
|
|
|
|
|
def points(count=30):
|
|
now=datetime(2026,10,15,tzinfo=timezone.utc)
|
|
return [{'artifact':f'point-{i}','stamp':now-timedelta(days=i)} for i in range(count)]
|
|
|
|
|
|
def test_retention_preserves_recovered_point_newest_daily_and_weekly():
|
|
rows=points();delete,keep=r.select(rows,'point-29')
|
|
assert 'point-29' in keep and 'point-0' in keep
|
|
assert all(f'point-{i}' in keep for i in range(7))
|
|
assert len(keep)>=8 and not (keep&{p['artifact'] for p in delete})
|
|
assert {p['artifact'] for p in delete}|keep=={p['artifact'] for p in rows}
|
|
|
|
|
|
def test_missing_independently_recovered_point_refuses_all_expiration():
|
|
with pytest.raises(r.offsite.Refused):r.select(points(),'lost-point')
|
|
|
|
|
|
@pytest.mark.parametrize('action,number',[('publisher',48),('retention',49)])
|
|
def test_proposed_ccr_cannot_create_provider_share_or_reader(monkeypatch,action,number):
|
|
path=setup.PLATFORM/f'credential-change-requests/CCR-2026-{number:04d}-net-kingdom-backup-{action}.yaml'
|
|
from types import SimpleNamespace
|
|
import yaml
|
|
proposed=yaml.safe_load(path.read_text());proposed['status']='proposed';proposed['review']['comments']=[]
|
|
# The negative fixture must stay proposed after the real owner approves a CCR.
|
|
original_load=setup.load
|
|
monkeypatch.setattr(setup,'load',lambda name,p:SimpleNamespace(validate_ccr=lambda _: (proposed,[],[])) if name=='credential_change' else original_load(name,p))
|
|
monkeypatch.setattr(setup.inventory,'caller_credentials',lambda:pytest.fail('unapproved request reached credential read'))
|
|
with pytest.raises(setup.Refused):setup.run(action,hashlib.sha256(path.read_bytes()).hexdigest(),{})
|
|
|
|
|
|
def test_runtime_refuses_gate_without_actual_independent_restore(tmp_path):
|
|
path=tmp_path/'accepted.json';path.write_text(json.dumps({'schema':'netkingdom.identity-recovery.v1','status':'prepared'}));path.chmod(0o600)
|
|
with pytest.raises(runtime.offsite.Refused):runtime.accepted(path)
|
|
|
|
|
|
def test_isolation_preserves_dns_peers_and_tests_immutable_target(monkeypatch):
|
|
from copy import deepcopy
|
|
iso=load('identity-isolation');iso.capture.KUBE=['kubectl','--kubeconfig','/reviewed/kubeconfig']
|
|
original={'podSelector':{},'policyTypes':['Egress'],'egress':[{'ports':[{'protocol':'UDP','port':53}]}]}
|
|
def get(ns,kind,name):
|
|
if name=='kube-system':return {'metadata':{'uid':iso.capture.prepare.CLUSTER_UID}}
|
|
return {'metadata':{'name':name,'uid':ns+'-uid','resourceVersion':'7'},'spec':deepcopy(original)}
|
|
def command(args,payload=None):
|
|
if 'get' in args:return json.dumps({'items':[{'metadata':{'labels':{'app':'service'}}}]}).encode()
|
|
patch=json.loads(args[args.index('-p')+1]);assert patch[2]['value']==original
|
|
wanted=deepcopy(original);wanted['podSelector']=iso.SELECTOR
|
|
return json.dumps({'metadata':{'uid':args[1]+'-uid'},'spec':wanted}).encode()
|
|
monkeypatch.setattr(iso.capture,'get',get);monkeypatch.setattr(iso.capture,'command',command)
|
|
packet=iso.plan();receipt={'attempted':[],'verified':[]};iso.apply(packet,receipt,True)
|
|
assert receipt['status']=='server_dry_run_passed' and len(receipt['verified'])==2
|
|
packet['policies'][0]['uid']='foreign'
|
|
with pytest.raises(iso.capture.Refused):iso.apply(packet,{'attempted':[],'verified':[]})
|
|
|
|
|
|
def test_bootstrap_dispatcher_handoff_help_and_missing_checkout(tmp_path):
|
|
import os,subprocess
|
|
bootstrap=ROOT.parent/'railiance-bootstrap'
|
|
if not bootstrap.is_dir():pytest.skip('bootstrap integration checkout absent')
|
|
help_run=subprocess.run(['bash',str(bootstrap/'bin/railiance'),'identity-recovery','--help'],capture_output=True)
|
|
assert help_run.returncode==0 and b'isolated archive verifier' not in help_run.stderr
|
|
assert b'--transfer-receipt' in help_run.stdout
|
|
env=dict(os.environ,NETKINGDOM_RECOVERY_ROOT=str(tmp_path/'missing'))
|
|
failure=subprocess.run(['bash',str(bootstrap/'bin/railiance'),'identity-recovery','--transfer-receipt','missing','--receipt','new'],env=env,capture_output=True)
|
|
assert failure.returncode==1 and b'Missing reviewed NetKingdom' in failure.stderr
|