"""Recovery drills use disposable invented data; never the ordinary demo folder.""" from pathlib import Path import hashlib import io import json import shutil import sqlite3 import stat import subprocess import sys import zipfile import pytest from fastapi.testclient import TestClient from PIL import Image from backend.app import create_app from backend.data_lock import data_lock,DataBusy from backend.recovery import backup,verify,restore,checksum,RecoveryError,DEFAULT_DATA,MAX_DATABASE from backend.store import DEMO_BUSINESSES,digest from test_boundaries import client,write,job,quote,invoice,portal,ORIGIN from test_materials import catalogue,add,material from test_workforce import person,entry from test_scheduling import booking def image_bytes(colour): data=io.BytesIO();Image.new('RGB',(12,12),colour).save(data,'PNG');return data.getvalue() def table_rows(root,bid): db=sqlite3.connect(root/bid/'business.sqlite3');db.row_factory=sqlite3.Row try: names=[r[0] for r in db.execute("SELECT name FROM sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%'")] return {name:[dict(r) for r in db.execute('SELECT * FROM '+name+' ORDER BY rowid')] for name in names} finally:db.close() def new_login(app,target,bid): lines=(target/'RESTORED-LOGIN.txt').read_text().splitlines();business=next(b for b in DEMO_BUSINESSES if b['id']==bid) password=lines[lines.index(business['email'])+1] c=TestClient(app,headers=ORIGIN);r=c.post('/api/login',json={'email':business['email'],'password':password}) assert r.status_code==200,r.text;c.headers['x-csrf-token']=r.json()['csrf'];return c,password def populate(root): app=create_app(root);results={} for n,b in enumerate(DEMO_BUSINESSES): c=client(app,n);j=job(c,name='Alex Fictional',email='same@customer.example');q=quote(c,j);i=invoice(c,q) assert write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':2500,'note':'Invented payment'}).status_code==200 cat=catalogue(c,name='Recovery timber');assert add(c,cat,j).status_code==200 assert write(c,'/api/catalogue/'+cat['id'],{'version':1,'name':'Recovery timber updated','cost_price':'19.99'},method='put').status_code==200 material(c,j,name='Received screws',quantity='1',cost_price='12',status='received',ordered_date='2026-10-01',received_date='2026-10-02') p=person(c);t=entry(c,j,p);cal=booking(c,j,p) assert write(c,'/api/team/'+p['id'],{'version':1,'name':'Jamie Demo','role':'Installer','hourly_cost':'25.00','colour':'#245648'},method='put').status_code==200 images=[] for colour in (('green','yellow') if n==0 else ('blue','red')): r=c.post('/api/assets',content=image_bytes(colour));assert r.status_code==200;images.append(r.json()['id']) site=c.get('/api/site').json();site['draft']['logo_id']=images[0] assert write(c,'/api/site',method='put',data={'version':1,'content':site['draft']}).status_code==200 assert write(c,'/api/site/publish',{'version':2}).status_code==200 site=c.get('/api/site').json();site['draft']['logo_id']=images[1];site['draft']['tagline']='Unpublished fictional draft' assert write(c,'/api/site',method='put',data={'version':3,'content':site['draft']}).status_code==200 assert write(c,'/api/domains',{'hostname':b['id']+'.test'}).status_code==200 pc,code=portal(app,c,j,slug=b['slug']) original_password='Original-private-demo-'+b['id']+'-2026!' assert write(c,'/api/password',{'current_password':b['password'],'new_password':original_password}).status_code==200 c=TestClient(app,headers=ORIGIN);login=c.post('/api/login',json={'email':b['email'],'password':original_password});assert login.status_code==200;c.headers['x-csrf-token']=login.json()['csrf'] results[b['id']]={'client':c,'portal':pc,'code':code,'source_password':original_password,'job':j,'invoice':i,'catalogue':cat,'time':t,'booking':cal,'images':images,'before':table_rows(root,b['id'])} return app,results def test_full_two_business_restore_after_simulated_loss(tmp_path): root=tmp_path/'original-fictional';app,records=populate(root) (root/'unrelated.txt').write_text('Do not include this unrelated file') (root/'.env').write_text('FAKE_UNRELATED_SECRET=not-for-the-backup') archive=tmp_path/'private-snapshot.zip';saved=backup(root,archive);checked=verify(archive,saved['sha256']) assert checked['valid'] and {b:len(v['assets']) for b,v in checked['businesses'].items()}=={'oak':2,'harbour':2} with zipfile.ZipFile(archive) as z: assert len(z.namelist())==8 and 'registry.sqlite3' in z.namelist() and all('.env' not in name for name in z.namelist()) body=b''.join(z.read(name) for name in z.namelist()) for record in records.values(): assert digest(record['code']).encode() not in body assert record['code'].encode() not in body for business in DEMO_BUSINESSES: assert records[business['id']]['client'].get('/api/session').status_code==200 assert records[business['id']]['portal'].get('/api/portal/'+business['slug']+'/me').status_code==200 original_assets={bid:{aid:(root/bid/'assets'/(aid+'.png')).read_bytes() for aid in row['images']} for bid,row in records.items()} # Deletion is restricted to this test-created disposable source. shutil.rmtree(root);assert not root.exists() target=tmp_path/'recovered-fictional';result=restore(archive,saved['sha256'],target,confirm_fresh=True) assert result['restored'] and not (target/'RESTORE_IN_PROGRESS').exists() with sqlite3.connect(target/'registry.sqlite3') as registry: assert registry.execute('SELECT COUNT(*) FROM sessions').fetchone()[0]==0 assert registry.execute('SELECT COUNT(*) FROM limits').fetchone()[0]==0 reopened=create_app(target);fresh={};passwords=[] for business in DEMO_BUSINESSES: bid=business['id'];before=records[bid]['before'];after=table_rows(target,bid) for table in before: if table=='customers': expected=[{**row,'portal_hash':None,'portal_version':row['portal_version']+1} for row in before[table]] assert after[table]==expected elif table=='portal_tokens':assert after[table]==[] elif table=='portal_accounts': assert [{k:v for k,v in r.items() if k not in ('password','state','version','updated_at')} for r in after[table]]==[{k:v for k,v in r.items() if k not in ('password','state','version','updated_at')} for r in before[table]] assert all(r['state']=='revoked' and r['password']=='' and r['version']==old['version']+1 for r,old in zip(after[table],before[table],strict=True)) elif table=='portal_events': assert after[table][:len(before[table])]==before[table] assert all(r['action']=='recovery.revoked' for r in after[table][len(before[table]):]) elif table=='storage_identity':assert after[table][0]['business_id']==bid and after[table][0]['dataset_id']!=before[table][0]['dataset_id'] elif table=='audit':assert after[table][:-1]==before[table] and after[table][-1]['action']=='recovery.restored_portals_revoked' else:assert after[table]==before[table],table for aid,content in original_assets[bid].items():assert (target/bid/'assets'/(aid+'.png')).read_bytes()==content old=TestClient(reopened,headers=ORIGIN);old.cookies.update(records[bid]['client'].cookies) assert old.get('/api/session').status_code==401 old.cookies.update(records[bid]['portal'].cookies) assert old.get('/api/portal/'+business['slug']+'/me').status_code==401 assert old.post('/api/portal/'+business['slug']+'/login',json={'code':records[bid]['code']}).status_code==401 assert old.post('/api/login',json={'email':business['email'],'password':business['password']}).status_code==401 assert old.post('/api/login',json={'email':business['email'],'password':records[bid]['source_password']}).status_code==401 fresh[bid],password=new_login(reopened,target,bid);passwords.append(password) document=fresh[bid].get('/api/documents/invoice/'+records[bid]['invoice']['id']).json() assert document['total_pence']==10000 and document['paid_pence']==2500 assert fresh[bid].get('/api/jobs/'+records[bid]['job']['id']+'/costs').json()['labour']['cost_pence']==3000 assert len(fresh[bid].get('/api/calendar?start_date=2026-11-01&end_date=2026-11-03').json()['items'])==1 assert fresh[bid].get('/api/demo/resolve-host?hostname='+bid+'.test').json()['slug']==business['slug'] assert passwords[0]!=passwords[1] for bid,other in [('oak','harbour'),('harbour','oak')]: assert fresh[bid].get('/api/jobs/'+records[other]['job']['id']).status_code==404 assert fresh[bid].get('/api/documents/invoice/'+records[other]['invoice']['id']).status_code==404 assert fresh[bid].get('/api/assets/'+records[other]['images'][0]).status_code==404 assert fresh[bid].get('/api/time/'+records[other]['time']['id']+'/history').status_code==404 assert fresh[bid].get('/api/calendar/'+records[other]['booking']['id']+'/history').status_code==404 public=TestClient(reopened,headers=ORIGIN) for business in DEMO_BUSINESSES: bid=business['id'] assert public.get('/api/public/'+business['slug']+'/assets/'+records[bid]['images'][0]).status_code==200 assert public.get('/api/public/'+business['slug']+'/assets/'+records[bid]['images'][1]).status_code==404 assert stat.S_IMODE(archive.stat().st_mode)==0o600 and stat.S_IMODE(target.stat().st_mode)==0o700 assert all(stat.S_IMODE(p.stat().st_mode)==0o600 for p in target.rglob('*') if p.is_file()) with sqlite3.connect(target/'registry.sqlite3') as db: assert db.execute("SELECT COUNT(*) FROM sessions WHERE kind='portal'").fetchone()[0]==0 second=backup(target,tmp_path/'second-generation.zip');assert verify(second['archive'],second['sha256'])['valid'] proof={'scope':'local invented data only','source_deleted_before_restore':True,'businesses':checked['businesses'],'preserved':['customers/jobs and exact links','quote/invoice snapshots and payments','Materials/catalogue snapshots and audits','labour entries/rates and history','calendar bookings and history','website draft/publication/history','owned image bytes and publication boundaries','retry-operation ledger','reserved .test mappings'],'revoked':['old owner passwords','all old staff and portal sessions','all portal grants/codes'],'new_passwords_distinct':True,'cross_business_access_rejected':True,'second_generation_backup_verified':True,'browser_checks_performed':False} (tmp_path/'recovery-drill.json').write_text(json.dumps(proof,indent=2)+'\n') @pytest.fixture def snapshot(tmp_path): root=tmp_path/'fictional';create_app(root);archive=tmp_path/'backup.zip';saved=backup(root,archive) return root,archive,saved['sha256'] def rewrite_archive(source,destination,mutate): with zipfile.ZipFile(source) as original:members={i.filename:original.read(i) for i in original.infolist()} manifest=json.loads(members['manifest.json']);mutate(members,manifest) members['manifest.json']=(json.dumps(manifest)+'\n').encode() with zipfile.ZipFile(destination,'w',compression=zipfile.ZIP_DEFLATED) as z: for name,body in members.items():z.writestr(name,body) return checksum(destination) @pytest.mark.parametrize('path',['../outside','/absolute','businesses/oak/../harbour/evil','businesses\\oak\\evil','registry.sqlite3']) def test_unsafe_paths_rejected_without_creating_target(snapshot,tmp_path,path): _,archive,_=snapshot;bad=tmp_path/'unsafe.zip' def mutate(files,meta): files[path]=b'bad';meta['files'][path]={'bytes':3,'sha256':hashlib.sha256(b'bad').hexdigest(),'business_id':'oak','kind':'image'} sha=rewrite_archive(archive,bad,mutate);target=tmp_path/'must-not-exist' with pytest.raises(RecoveryError):restore(bad,sha,target,confirm_fresh=True) assert not target.exists() and not (tmp_path/'outside').exists() @pytest.mark.parametrize('kind',['swap_businesses','wrong_file_owner','wrong_slug','missing_database','unknown_version','wrong_member_hash','corrupt_sqlite','extra_schema','broken_foreign_key','portal_grant','oversize_metadata']) def test_corrupt_or_mapped_archives_rejected(snapshot,tmp_path,kind): _,archive,_=snapshot;bad=tmp_path/'damaged.zip' def mutate(files,meta): oak='businesses/oak/business.sqlite3';harbour='businesses/harbour/business.sqlite3' if kind=='swap_businesses':files[oak],files[harbour]=files[harbour],files[oak] elif kind=='wrong_file_owner':meta['files'][oak]['business_id']='harbour' elif kind=='wrong_slug':meta['businesses'][0]['slug']='oak-and-vale' if meta['businesses'][0]['id']=='harbour' else 'harbour-homeworks' elif kind=='missing_database':del files[oak];del meta['files'][oak] elif kind=='unknown_version':meta['version']=999 elif kind=='wrong_member_hash':meta['files'][oak]['sha256']='0'*64 elif kind=='oversize_metadata':meta['files'][oak]['bytes']=MAX_DATABASE+1 elif kind=='corrupt_sqlite':files[oak]=b'not a database' else: dbfile=tmp_path/'edited.sqlite3';dbfile.write_bytes(files[oak]) with sqlite3.connect(dbfile) as db: if kind=='extra_schema':db.execute('CREATE TRIGGER unexpected AFTER INSERT ON customers BEGIN DELETE FROM jobs; END') elif kind=='broken_foreign_key':db.execute("INSERT INTO jobs(id,customer_id,service,message,status,source,created_at) VALUES('bad','missing','Test','Test','new','test','2026-10-01')") elif kind=='portal_grant':db.execute("INSERT INTO customers(id,name,email,phone,address,portal_hash,created_at) VALUES('bad','Test','','','','revived-grant','2026-10-01')") files[oak]=dbfile.read_bytes() for name in (oak,harbour): if name in files and kind not in ('wrong_member_hash','oversize_metadata'): meta['files'][name]['bytes']=len(files[name]);meta['files'][name]['sha256']=hashlib.sha256(files[name]).hexdigest() sha=rewrite_archive(archive,bad,mutate) with pytest.raises((RecoveryError,sqlite3.Error)):restore(bad,sha,tmp_path/'recovered',confirm_fresh=True) assert not (tmp_path/'recovered').exists() def test_duplicate_symlink_and_crc_inputs_rejected(snapshot,tmp_path): _,archive,sha=snapshot with pytest.raises(RecoveryError):verify(archive,'0'*64) duplicate=tmp_path/'duplicate.zip';shutil.copyfile(archive,duplicate) with zipfile.ZipFile(duplicate,'a') as z: with pytest.warns(UserWarning):z.writestr('manifest.json','{}') with pytest.raises(RecoveryError):verify(duplicate,checksum(duplicate)) linkzip=tmp_path/'link.zip' with zipfile.ZipFile(archive) as source,zipfile.ZipFile(linkzip,'w') as target: for info in source.infolist(): if info.filename.endswith('oak/business.sqlite3'):info.external_attr=(stat.S_IFLNK|0o777)<<16 target.writestr(info,source.read(info.filename)) with pytest.raises(RecoveryError):verify(linkzip,checksum(linkzip)) truncated=tmp_path/'truncated.zip';truncated.write_bytes(archive.read_bytes()[:-30]) with pytest.raises(zipfile.BadZipFile):verify(truncated,checksum(truncated)) def test_missing_or_corrupt_owned_images_fail(tmp_path): root=tmp_path/'source';app=create_app(root);c=client(app);r=c.post('/api/assets',content=image_bytes('green'));aid=r.json()['id'] archive=tmp_path/'snapshot.zip';backup(root,archive) name='businesses/oak/assets/'+aid+'.png' for kind in ('missing','invalid'): def mutate(files,meta): if kind=='missing':del files[name];del meta['files'][name] else: files[name]=b'invalid image';meta['files'][name]['bytes']=len(files[name]);meta['files'][name]['sha256']=hashlib.sha256(files[name]).hexdigest() bad=tmp_path/(kind+'.zip');sha=rewrite_archive(archive,bad,mutate) with pytest.raises((RecoveryError,OSError)):restore(bad,sha,tmp_path/('target-'+kind),confirm_fresh=True) assert not (tmp_path/('target-'+kind)).exists() image=root/'oak/assets'/(aid+'.png');image.unlink();image.symlink_to(tmp_path/'unrelated.png');(tmp_path/'unrelated.png').write_bytes(image_bytes('blue')) with pytest.raises(RecoveryError):backup(root,tmp_path/'unsafe-source.zip') assert not (tmp_path/'unsafe-source.zip').exists() def test_fresh_target_and_explicit_confirmation_guards(snapshot,tmp_path): root,archive,sha=snapshot;target=tmp_path/'existing';target.mkdir();sentinel=target/'keep.txt';sentinel.write_text('keep') for destination in (target,root,root/'inside',DEFAULT_DATA,DEFAULT_DATA/'nested'): with pytest.raises(RecoveryError):restore(archive,sha,destination,confirm_fresh=True) assert sentinel.read_text()=='keep' with pytest.raises(RecoveryError):restore(archive,sha,tmp_path/'new') assert not (tmp_path/'new').exists() link=tmp_path/'linked-parent';link.symlink_to(target,target_is_directory=True) with pytest.raises(RecoveryError):restore(archive,sha,link/'new',confirm_fresh=True) with pytest.raises(RecoveryError):backup(root,archive) def test_running_lock_and_sqlite_writer_prevent_snapshot(snapshot,tmp_path): root,archive,sha=snapshot with data_lock(root): with pytest.raises(DataBusy):backup(root,tmp_path/'busy.zip') assert not (tmp_path/'busy.zip').exists() with sqlite3.connect(root/'harbour/business.sqlite3') as writer: writer.execute('BEGIN IMMEDIATE') with pytest.raises(sqlite3.OperationalError):backup(root,tmp_path/'writer.zip') assert not (tmp_path/'writer.zip').exists() app=create_app(root);c=client(app) with data_lock(root,exclusive=True): assert c.get('/api/workspace').status_code==503 with pytest.raises(DataBusy):create_app(root) assert c.get('/api/workspace').status_code==200 def test_process_lock_release_and_incomplete_restore_guard(snapshot,tmp_path): root,archive,sha=snapshot process=subprocess.Popen([sys.executable,'-c','from backend.data_lock import data_lock; from pathlib import Path; import sys\nwith data_lock(Path(sys.argv[1])):\n print("ready",flush=True);sys.stdin.readline()',str(root)],stdin=subprocess.PIPE,stdout=subprocess.PIPE,text=True) try: assert process.stdout.readline().strip()=='ready' with pytest.raises(DataBusy):backup(root,tmp_path/'running.zip') finally:process.communicate('\n',timeout=5) assert backup(root,tmp_path/'stopped.zip')['sha256'] incomplete=tmp_path/'partial';incomplete.mkdir();(incomplete/'RESTORE_IN_PROGRESS').write_text('interrupted') with pytest.raises(ValueError):create_app(incomplete) assert not (incomplete/'registry.sqlite3').exists() def test_cli_safe_summary_failure_and_no_password_output(snapshot,tmp_path): _,archive,sha=snapshot base=[sys.executable,'recovery.py'] verified=subprocess.run([*base,'verify','--archive',str(archive),'--sha256',sha],capture_output=True,text=True,timeout=15) assert verified.returncode==0 and json.loads(verified.stdout)['valid'] target=tmp_path/'cli-restored' denied=subprocess.run([*base,'restore','--archive',str(archive),'--sha256',sha,'--target',str(target)],capture_output=True,text=True,timeout=15) assert denied.returncode==2 and not target.exists() restored=subprocess.run([*base,'restore','--archive',str(archive),'--sha256',sha,'--target',str(target),'--confirm-fresh-target'],capture_output=True,text=True,timeout=15) assert restored.returncode==0 and json.loads(restored.stdout)['restored'] assert 'Recovered-' not in restored.stdout+restored.stderr def test_committed_wal_pages_are_included_consistently(tmp_path): root=tmp_path/'wal-data';app=create_app(root);c=client(app) keeper=sqlite3.connect(root/'oak/business.sqlite3') try: assert keeper.execute('PRAGMA journal_mode=WAL').fetchone()[0]=='wal' keeper.execute('PRAGMA wal_autocheckpoint=0') keeper.execute('BEGIN') keeper.execute('SELECT COUNT(*) FROM jobs').fetchone() j=job(c,name='WAL Fictional') assert (root/'oak/business.sqlite3-wal').stat().st_size>0 saved=backup(root,tmp_path/'wal-backup.zip') target=tmp_path/'wal-restored';restore(saved['archive'],saved['sha256'],target,confirm_fresh=True) owner,_=new_login(create_app(target),target,'oak') assert owner.get('/api/jobs/'+j['id']).json()['name']=='WAL Fictional' finally:keeper.close() def test_interrupted_publication_cleans_only_new_target(snapshot,tmp_path,monkeypatch): import backend.recovery as recovery _,archive,sha=snapshot;target=tmp_path/'interrupted';original=recovery.os.link;calls=0 def fail_midway(source,destination): nonlocal calls calls+=1 if calls==3:raise OSError('Simulated interrupted local publication') return original(source,destination) monkeypatch.setattr(recovery.os,'link',fail_midway) with pytest.raises(OSError,match='Simulated interrupted'):restore(archive,sha,target,confirm_fresh=True) assert not target.exists() assert archive.exists() def test_additive_v03_upgrade_keeps_records_and_rejects_mixed_datasets(tmp_path): root=tmp_path/'old-demo';app=create_app(root);c=client(app);j=job(c) before=table_rows(root,'oak') for bid in ('oak','harbour'): with sqlite3.connect(root/bid/'business.sqlite3') as db:db.execute('DROP TABLE storage_identity') with sqlite3.connect(root/'registry.sqlite3') as db:db.execute('DROP TABLE dataset_identity') upgraded=create_app(root) assert client(upgraded).get('/api/jobs/'+j['id']).json()['name']=='Alex Demo' after=table_rows(root,'oak') assert all(after[k]==v for k,v in before.items() if k!='storage_identity') other=tmp_path/'other-data';create_app(other) shutil.copyfile(other/'oak/business.sqlite3',root/'oak/business.sqlite3') with pytest.raises(ValueError,match='different folder or dataset'):create_app(root) with pytest.raises(RecoveryError,match='identity mismatch'):backup(root,tmp_path/'mixed.zip') assert not (tmp_path/'mixed.zip').exists()