"""Private local-demo snapshots. No networking, extraction helpers or live imports. Only the twelve pinned local formats and up to 25 invented businesses are supported. Archives are checksummed, not signed/encrypted. Use only backups made from your own demo. """ from contextlib import ExitStack, contextmanager from functools import lru_cache from pathlib import Path import hashlib import io import json import os import re import secrets import sqlite3 import stat import tempfile import time import zipfile from PIL import Image from .data_lock import data_lock, DataBusy from .models import SiteContent, LegacySiteContent, Domain from .site_models import asset_ids from .store import Store, DEMO_BUSINESSES, uid, now, password_hash FORMAT='shs-local-recovery' VERSION=14 LEGACY_AUTH_POLICY='new-owner-passwords;no-sessions;no-portal-grants' AUTH_POLICY='new-account-passwords;no-sessions;no-invitations;no-resets;no-portal-grants' BUSINESSES={b['id']:b for b in DEMO_BUSINESSES} MAX_TOTAL=256*1024*1024 MAX_DATABASE=64*1024*1024 MAX_IMAGE=16*1024*1024 MAX_FILES=2100 DEFAULT_DATA=Path(__file__).resolve().parents[1]/'.local-data' class RecoveryError(ValueError): pass def require(condition,message): if not condition:raise RecoveryError(message) def local_path(value): path=Path(os.path.abspath(value)) for item in (*reversed(path.parents),path): require(not item.is_symlink(),'Symlinks are not supported for recovery paths.') return path def regular(path,limit): info=path.stat(follow_symlinks=False) require(stat.S_ISREG(info.st_mode) and info.st_nlink==1,'Recovery input must be a regular file, not a link.') require(info.st_size<=limit,'Recovery input exceeds the local-demo size limit.') def checksum(path): value=hashlib.sha256() with path.open('rb') as stream: for chunk in iter(lambda:stream.read(1024*1024),b''):value.update(chunk) return value.hexdigest() def private_write(path,data): descriptor=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600) with os.fdopen(descriptor,'wb') as stream: stream.write(data);stream.flush();os.fsync(stream.fileno()) def sync_directory(path): descriptor=os.open(path,os.O_RDONLY|os.O_DIRECTORY) try:os.fsync(descriptor) finally:os.close(descriptor) @contextmanager def read_db(path): connection=sqlite3.connect(path.as_uri()+'?mode=ro',uri=True,timeout=0) try: connection.execute('PRAGMA trusted_schema=OFF') connection.execute('PRAGMA query_only=ON') connection.setlimit(sqlite3.SQLITE_LIMIT_LENGTH,8*1024*1024) connection.setlimit(sqlite3.SQLITE_LIMIT_ATTACHED,0) deadline=time.monotonic()+10 connection.set_progress_handler(lambda:time.monotonic()>deadline,1000) yield connection finally:connection.close() @contextmanager def write_db(path): connection=sqlite3.connect(path) try: with connection:yield connection finally:connection.close() def schema(db): return db.execute("SELECT type,name,tbl_name,sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%' ORDER BY type,name").fetchall() @lru_cache(maxsize=14) def expected_schemas(format_version=VERSION): if format_version in (1,2,3,4,5,6,7,8,9,10,11,12,13): saved=json.loads((Path(__file__).resolve().parents[1]/('docs/recovery-schema-v'+str(format_version)+'.json')).read_text()) return [tuple(r) for r in saved['registry']], [tuple(r) for r in saved['tenant']] with tempfile.TemporaryDirectory(prefix='shs-schema-') as folder: root=Path(folder)/'data';Store(root) with read_db(root/'registry.sqlite3') as db:registry=schema(db) with read_db(root/'oak/business.sqlite3') as db:tenant=schema(db) return registry,tenant def check_db(db,expected): require(schema(db)==expected,'Database schema is unsupported or has unexpected tables, views, indexes or triggers.') require(db.execute('PRAGMA integrity_check').fetchall()==[('ok',)],'Database integrity check failed.') require(db.execute('PRAGMA foreign_key_check').fetchone() is None,'Database contains broken record relationships.') def hex_id(value): return isinstance(value,str) and re.fullmatch('[0-9a-f]{32}',value) is not None def tenant_summary(path,business_id,dataset_id,*,sanitised,format_version=VERSION): with read_db(path) as db: check_db(db,expected_schemas(format_version)[1]) if format_version<=12: from .customer_recovery_v12 import validate as validate_customers elif format_version==13: from .customer_recovery_v13 import validate as validate_customers else: from .customer_recovery import validate as validate_customers customer_identity=validate_customers(db,require) if format_version>=11 else {} from .documents import validate_records validate_records(db,legacy=format_version==1) require(db.execute('SELECT id,business_id,dataset_id FROM storage_identity').fetchall()==[(1,business_id,dataset_id)],'Business/dataset identity mismatch. Restore remapping is not allowed.') if sanitised:require(db.execute('SELECT COUNT(*) FROM customers WHERE portal_hash IS NOT NULL').fetchone()[0]==0,'Backup contains an active portal grant.') assets={row[0] for row in db.execute('SELECT id FROM assets')} require(all(hex_id(a) for a in assets),'Invalid owned image identity.') content=[v for row in db.execute('SELECT draft,published FROM site') for v in row] require(len(content)==2,'The saved website record is missing or duplicated.') content.extend(row[0] for row in db.execute('SELECT content FROM site_history')) for raw in content: if format_version>=3: require(json.loads(raw).get('content_version')==2,'Website content version is unsupported.') site=SiteContent.model_validate_json(raw) else:site=LegacySiteContent.model_validate_json(raw) references=asset_ids(site.model_dump()) require(references<=assets,'A saved website revision refers to an image outside its business.') for table in ('quotes','invoices'): for raw, in db.execute('SELECT snapshot FROM '+table): require(asset_ids(json.loads(raw))<=assets,'A saved document refers to an image outside its business.') tables=[row[1] for row in expected_schemas(format_version)[1] if row[0]=='table'] counts={table:db.execute('SELECT COUNT(*) FROM "'+table+'"').fetchone()[0] for table in tables} from .photo_recovery import validate as validate_photos media=validate_photos(db,require) if format_version>=5 else {} from .field_recovery import validate as validate_field field=validate_field(db,require) if format_version>=6 else {'logs':0,'timers':0,'source_claims':0} from .recurrence_recovery import validate as validate_recurrence recurrence=validate_recurrence(db,require) if format_version>=7 else {} from .portal_recovery import validate as validate_portal portal=validate_portal(db,require,sanitised=sanitised) if format_version>=8 else {} from .feedback_recovery import validate as validate_feedback feedback=validate_feedback(db,require,sanitised=sanitised) if format_version>=9 else {} from .workflow_recovery import validate as validate_workflow workflow=validate_workflow(db,require) if format_version>=10 else {} from .enquiry_recovery import validate as validate_enquiries if format_version>=12:validate_enquiries(db,require) return {'customer_identity':customer_identity,'workflow':workflow,'feedback':feedback,'counts':counts,'assets':sorted(assets),'photos':media,'field':field,'recurrence':recurrence,'portal':portal} def check_image(path): regular(path,MAX_IMAGE) try: with Image.open(path) as image: require(image.format=='PNG' and image.width<=1800 and image.height<=1800,'An owned image is not a supported local PNG.') image.verify() except (OSError, SyntaxError, ValueError, Image.DecompressionBombError): raise RecoveryError('An owned image is corrupt or is not a supported local PNG.') def check_photo(path,metadata): check_image(path) require(path.stat().st_size==metadata['bytes'] and checksum(path)==metadata['sha256'],'Private image checksum/size does not match its saved record.') with Image.open(path) as image: require(image.size==(metadata['width'],metadata['height']) and image.mode=='RGBA' and not image.info and getattr(image,'n_frames',1)==1,'Private image contains unexpected metadata or dimensions.') def check_metadata(meta): require(set(meta)=={'format','version','created_at','prototype_version','dataset_id','businesses','domains','auth_policy','files'},'Unexpected recovery manifest fields.') require(meta['format']==FORMAT and type(meta['version']) is int and (meta['version'],meta['prototype_version']) in ((1,'0.4.0'),(2,'0.5.0'),(3,'0.6.0'),(4,'0.7.0'),(5,'0.8.0'),(6,'0.9.0'),(7,'0.10.0'),(8,'0.11.0'),(9,'0.12.0'),(10,'0.13.0'),(11,'0.14.0'),(12,'0.19.0'),(12,'0.20.0'),(13,'0.24.0'),(14,'0.25.0')),'Unsupported recovery format/version.') require(meta['auth_policy']==(AUTH_POLICY if meta['version']>=4 else LEGACY_AUTH_POLICY) and hex_id(meta['dataset_id']),'Invalid recovery identity or authentication policy.') require(isinstance(meta['created_at'],str) and len(meta['created_at'])<=40,'Invalid snapshot timestamp.') require(isinstance(meta['businesses'],list) and (1<=len(meta['businesses'])<=25 if meta['version']>=4 else len(meta['businesses'])==2),'A complete supported business inventory is required.') from .store import valid_business_id from .identity import valid_slug found=set();slugs=set() for business in meta['businesses']: require(isinstance(business,dict) and set(business)=={'id','slug','name','enabled'},'Invalid business metadata.') bid=business['id'];require(valid_business_id(bid) and bid not in found and (meta['version']>=4 or bid in BUSINESSES),'Unexpected or duplicated business identity.') require(valid_slug(business['slug']) and business['slug'] not in slugs,'Invalid or duplicated business slug.') if meta['version']<4:require(business['slug']==BUSINESSES[bid]['slug'],'Cross-business slug mapping is not allowed.') slugs.add(business['slug']) require(isinstance(business['name'],str) and 1<=len(business['name'])<=120 and type(business['enabled']) is int and business['enabled'] in (0,1),'Invalid business settings.') found.add(bid) require(isinstance(meta['domains'],list) and len(meta['domains'])<=100,'Invalid domain simulator settings.') hosts=set() for domain in meta['domains']: require(isinstance(domain,dict) and set(domain)=={'hostname','business_id'},'Invalid simulated domain mapping.') Domain(hostname=domain['hostname']) require(domain['business_id'] in found and domain['hostname'] not in hosts,'Invalid/duplicate simulated domain mapping.') hosts.add(domain['hostname']) def registry_summary(path,meta,tenant_paths,*,sanitised): with read_db(path) as db: check_db(db,expected_schemas()[0]) businesses=[dict(zip(('id','slug','name','enabled'),row)) for row in db.execute('SELECT id,slug,name,enabled FROM businesses ORDER BY id')] require(businesses==sorted(meta['businesses'],key=lambda b:b['id']),'Business inventory and registry disagree.') require(db.execute('SELECT id,dataset_id FROM dataset_identity').fetchall()==[(1,meta['dataset_id'])],'Registry dataset identity mismatch.') domains=[dict(zip(('hostname','business_id'),row)) for row in db.execute('SELECT hostname,business_id FROM domains ORDER BY hostname')] require(domains==sorted(meta['domains'],key=lambda d:d['hostname']),'Domain inventory and registry disagree.') require(db.execute("SELECT key,value FROM registry_meta").fetchall()==[('memberships_initialized','1')],'Unsupported identity migration state.') if sanitised: require(db.execute("SELECT COUNT(*) FROM users WHERE password<>''").fetchone()[0]==0,'Backup must not contain credentials.') for table in ('sessions','access_tokens','registry_operations','limits'): require(db.execute('SELECT COUNT(*) FROM '+table).fetchone()[0]==0,'Backup must not contain sessions, tokens or identity retry results.') for userid,home,email,password,display,disabled in db.execute('SELECT id,business_id,email,password,display_name,disabled FROM users'): require((hex_id(userid) or userid in BUSINESSES) and home in tenant_paths and isinstance(email,str) and email==email.lower() and re.fullmatch(r'[^\s@]+@[^\s@]+\.[^\s@]+',email) is not None and len(email)<=180 and disabled in (0,1) and 1<=len(display)<=100,'Invalid private account identity.') require(db.execute('SELECT 1 FROM memberships WHERE user_id=?',(userid,)).fetchone() is not None,'An account has no membership history.') for mid,bid,user,role,enabled,version,rid in db.execute('SELECT id,business_id,user_id,role,enabled,version,resource_id FROM memberships'): require(hex_id(mid) and role in ('owner','staff') and enabled in (0,1) and type(version) is int and version>=1,'Invalid membership state.') if rid is not None: require(hex_id(rid),'Invalid resource link.') with read_db(tenant_paths[bid]) as tenant:require(tenant.execute('SELECT 1 FROM team_resources WHERE id=?',(rid,)).fetchone() is not None,'A membership resource is missing from its business.') for business in businesses: if business['enabled']: require(db.execute("SELECT 1 FROM memberships m JOIN users u ON u.id=m.user_id WHERE m.business_id=? AND m.role='owner' AND m.enabled=1 AND u.disabled=0",(business['id'],)).fetchone() is not None,'An enabled business needs an active owner.') for bid,path in tenant_paths.items(): with read_db(path) as tenant: for mid,user,rid in tenant.execute('SELECT membership_id,author_id,resource_id FROM staff_updates'): require(db.execute('SELECT 1 FROM memberships WHERE id=? AND user_id=? AND business_id=?',(mid,user,bid)).fetchone() is not None,'Work note has an invalid membership author.') require(tenant.execute('SELECT 1 FROM team_resources WHERE id=?',(rid,)).fetchone() is not None,'Work note resource is missing.') if meta['version']>=5: for table,actor in [('job_photos','uploaded_by'),('photo_history','actor_id'),('gallery_history','actor_id')]: cursor=tenant.execute('SELECT * FROM '+table);columns=[v[0] for v in cursor.description] for values in cursor: row=dict(zip(columns,values));mid,user=row['membership_id'],row[actor] if meta['version']>=12: from .enquiry_recovery import non_member_author if non_member_author(tenant,table,row):continue require(db.execute('SELECT 1 FROM memberships WHERE id=? AND user_id=? AND business_id=?',(mid,user,bid)).fetchone() is not None,'Photo or gallery history has an invalid membership author.') if meta['version']>=6: for table,actor in [('field_logs','author_id'),('field_sessions','author_id'),('field_log_history','actor_id'),('field_history','actor_id')]: for mid,user in tenant.execute('SELECT membership_id,'+actor+' FROM '+table): if mid is None and user is None and table=='field_history':continue require(db.execute('SELECT 1 FROM memberships WHERE id=? AND user_id=? AND business_id=?',(mid,user,bid)).fetchone() is not None,'Field history has an invalid membership author.') for (actor,) in tenant.execute('SELECT actor_id FROM source_history'): require(db.execute('SELECT 1 FROM memberships WHERE user_id=? AND business_id=?',(actor,bid)).fetchone() is not None,'Source reservation history has an invalid author.') if meta['version']>=11: for table in ('customer_record_history','customer_pair_history','customer_transfers','customer_completion_runs'): for actor, in tenant.execute('SELECT actor FROM '+table): if table=='customer_record_history' and actor=='initialisation':continue require(db.execute('SELECT 1 FROM memberships WHERE user_id=? AND business_id=?',(actor,bid)).fetchone() is not None,'Customer history has an invalid business author.') if meta['version']>=14: for actor,raw in tenant.execute('SELECT actor,plan FROM customer_merge_corrections'): a=json.loads(raw)['actor_access'] require(a['business_id']==bid and a['subject']==actor and db.execute('SELECT 1 FROM memberships WHERE id=? AND user_id=? AND business_id=? AND version>=?',(a['membership_id'],actor,bid,a['membership_version'])).fetchone() is not None,'Merge correction has an invalid business membership author.') if meta['version']>=10: for table in ('job_workflow_history','opportunity_settings_history','opportunity_history'): for actor, in tenant.execute('SELECT actor FROM '+table): if table=='job_workflow_history' and actor=='initialisation':continue require(db.execute('SELECT 1 FROM memberships WHERE user_id=? AND business_id=?',(actor,bid)).fetchone() is not None,'Operational history has an invalid business author.') if meta['version']>=9: for rid,cid,action,actor in tenant.execute('SELECT h.request_id,r.customer_id,h.action,h.actor FROM feedback_history h JOIN feedback_requests r ON r.id=h.request_id'): if action in ('submitted','consent.approved','consent.withdrawn'): require(actor in ('portal:'+cid,'feedback-token:'+rid),'Feedback customer author is outside its exact scope.') elif action=='recovery.revoked':require(actor=='recovery','Invalid feedback recovery author.') else:require(db.execute('SELECT 1 FROM memberships WHERE user_id=? AND business_id=?',(actor,bid)).fetchone() is not None,'Feedback history has an invalid business author.') if meta['version']>=8: for (actor,) in tenant.execute('SELECT actor FROM sales_settings_history'): require(db.execute('SELECT 1 FROM memberships WHERE user_id=? AND business_id=?',(actor,bid)).fetchone() is not None,'Document settings history has an invalid business author.') for cid,actor in tenant.execute('SELECT customer_id,actor FROM portal_events'): require(actor in ('recovery','portal-recovery','portal:'+cid) or db.execute('SELECT 1 FROM memberships WHERE user_id=? AND business_id=?',(actor,bid)).fetchone() is not None,'Portal history has an invalid business author.') for raw, in tenant.execute('SELECT evidence FROM quote_responses'): evidence=json.loads(raw) if evidence['mode']=='owner-recorded':require(db.execute('SELECT 1 FROM memberships WHERE user_id=? AND business_id=?',(evidence['actor'],bid)).fetchone() is not None,'Manual quote response has an invalid business author.') if meta['version']>=7: for table,actor in [('recurrence_rules','author_id'),('recurrence_history','actor_id'),('recurrence_runs','actor_id')]: for mid,user in tenant.execute('SELECT membership_id,'+actor+' FROM '+table): require(db.execute('SELECT 1 FROM memberships WHERE id=? AND user_id=? AND business_id=?',(mid,user,bid)).fetchone() is not None,'Recurrence review has an invalid membership author.') def backup(data_dir,output): root=local_path(data_dir);destination=local_path(output) require(root.is_dir(),'Choose an existing local-demo data folder.') require(destination.parent.is_dir(),'Create the backup destination folder first.') require(not destination.is_relative_to(root),'Save the backup outside the active data folder.') digest_path=destination.with_name(destination.name+'.sha256') require(not destination.exists() and not digest_path.exists(),'Backup or checksum already exists. Choose a new filename.') require(not (root/'RESTORE_IN_PROGRESS').exists(),'This recovery folder is incomplete.') with data_lock(root,exclusive=True),tempfile.TemporaryDirectory(prefix='.shs-backup-',dir=destination.parent) as temporary,ExitStack() as locks: stage=Path(temporary);paths=[root/'registry.sqlite3'] # Reserve every SQLite writer before reading any database. The runner # lock also keeps complete requests and immutable image writes quiescent. def reserve(path): local_path(path);regular(path,MAX_DATABASE) db=sqlite3.connect(path.as_uri()+'?mode=rw',uri=True,timeout=0) locks.callback(db.close) db.execute('BEGIN IMMEDIATE') reserve(paths[0]) with read_db(paths[0]) as db: check_db(db,expected_schemas()[0]) identities=db.execute('SELECT id,dataset_id FROM dataset_identity').fetchall() require(len(identities)==1 and identities[0][0]==1 and hex_id(identities[0][1]),'Open and stop the updated local app once before making its first backup.') dataset_id=identities[0][1] businesses=[dict(zip(('id','slug','name','enabled'),row)) for row in db.execute('SELECT id,slug,name,enabled FROM businesses ORDER BY id')] domains=[dict(zip(('hostname','business_id'),row)) for row in db.execute('SELECT hostname,business_id FROM domains ORDER BY hostname')] meta={'format':FORMAT,'version':VERSION,'created_at':now(),'prototype_version':'0.25.0','dataset_id':dataset_id,'businesses':businesses,'domains':domains,'auth_policy':AUTH_POLICY,'files':{}} check_metadata(meta) business_ids=[b['id'] for b in businesses] for bid in business_ids:reserve(root/bid/'business.sqlite3') registry_summary(root/'registry.sqlite3',meta,{bid:root/bid/'business.sqlite3' for bid in business_ids},sanitised=False) def record(path,bid,kind): regular(path,MAX_DATABASE if kind in ('database','identity') else MAX_IMAGE) meta['files'][path.relative_to(stage).as_posix()]={'bytes':path.stat().st_size,'sha256':checksum(path),'business_id':bid,'kind':kind} registry_copy=stage/'registry.sqlite3' with read_db(root/'registry.sqlite3') as source_db,write_db(registry_copy) as copy:source_db.backup(copy) with write_db(registry_copy) as copy: copy.execute('PRAGMA secure_delete=ON') copy.execute("UPDATE users SET password=''") for table in ('sessions','access_tokens','registry_operations','limits'):copy.execute('DELETE FROM '+table) copy.commit();copy.execute('VACUUM') registry_copy.chmod(0o600);record(registry_copy,None,'identity') for bid in business_ids: source=root/bid/'business.sqlite3';folder=stage/'businesses'/bid;folder.mkdir(parents=True,mode=0o700) target=folder/'business.sqlite3' with read_db(source) as source_db,write_db(target) as copy: source_db.backup(copy) target.chmod(0o600) summary=tenant_summary(target,bid,dataset_id,sanitised=False) with write_db(target) as copy: copy.execute('PRAGMA secure_delete=ON') copy.execute('UPDATE customers SET portal_hash=NULL') from .portal_recovery import sanitise sanitise(copy) from .feedback_recovery import sanitise as sanitise_feedback sanitise_feedback(copy) copy.commit();copy.execute('VACUUM') record(target,bid,'database') for aid in summary['assets']: image_path=local_path(root/bid/'assets'/(aid+'.png'));check_image(image_path) assets=folder/'assets';assets.mkdir(exist_ok=True,mode=0o700) copied=assets/image_path.name;private_write(copied,image_path.read_bytes());check_image(copied);record(copied,bid,'image') for pid,metadata in summary['photos'].items(): source=local_path(root/bid/'photos'/(pid+'.png'));check_photo(source,metadata) folder_photos=folder/'photos';folder_photos.mkdir(exist_ok=True,mode=0o700) copied=folder_photos/source.name;private_write(copied,source.read_bytes());check_photo(copied,metadata);record(copied,bid,'job_image') require(len(meta['files'])<=MAX_FILES and sum(f['bytes'] for f in meta['files'].values())<=MAX_TOTAL,'Backup exceeds the bounded local-demo size limit.') private_write(stage/'manifest.json',(json.dumps(meta,indent=2)+'\n').encode()) temporary_archive=stage/'snapshot.zip' with zipfile.ZipFile(temporary_archive,'x',compression=zipfile.ZIP_DEFLATED) as archive: for name in ['manifest.json',*sorted(meta['files'])]:archive.write(stage/name,name) temporary_archive.chmod(0o600) with temporary_archive.open('rb') as stream:os.fsync(stream.fileno()) sha=checksum(temporary_archive) verify(temporary_archive,sha) # Hard-link publication is atomic and never overwrites an existing file. private_write(digest_path,(sha+'\n').encode()) try:os.link(temporary_archive,destination) except BaseException: digest_path.unlink();raise sync_directory(destination.parent) return {'archive':str(destination),'sha256':sha,'businesses':sorted(business_ids),'files':len(meta['files']),'authentication':AUTH_POLICY} def no_duplicate_keys(pairs): value={} for key,item in pairs: require(key not in value,'Manifest contains duplicate keys.') value[key]=item return value def unpack_validated(archive_path,sha256,stage): path=local_path(archive_path);regular(path,MAX_TOTAL) descriptor=os.open(path,os.O_RDONLY|os.O_NOFOLLOW) with os.fdopen(descriptor,'rb') as source: info=os.fstat(source.fileno()) require(stat.S_ISREG(info.st_mode) and info.st_nlink==1,'Backup must be a regular private file.') raw=source.read(MAX_TOTAL+1) require(len(raw)<=MAX_TOTAL,'Backup exceeds the local-demo size limit.') require(isinstance(sha256,str) and re.fullmatch('[a-f0-9]{64}',sha256) and hashlib.sha256(raw).hexdigest()==sha256,'Backup SHA-256 does not match. Use the checksum saved when this backup was made.') # Hash and parse the same bounded bytes so another process cannot swap the # file between checksum verification and archive inspection. with zipfile.ZipFile(io.BytesIO(raw)) as archive: entries=archive.infolist();names=[i.filename for i in entries] require(len(entries)<=MAX_FILES+1 and len(names)==len(set(names)),'Too many or duplicate archive entries.') require('manifest.json' in names,'Recovery manifest is missing.') require(sum(i.file_size for i in entries)<=MAX_TOTAL+2*1024*1024,'Archive is too large when unpacked.') for entry in entries: require(not entry.is_dir() and stat.S_IFMT(entry.external_attr>>16) in (0,stat.S_IFREG) and not entry.flag_bits&1 and entry.compress_type in (zipfile.ZIP_STORED,zipfile.ZIP_DEFLATED),'Archive contains an unsafe entry type.') require(archive.getinfo('manifest.json').file_size<=2*1024*1024,'Manifest is too large.') meta=json.loads(archive.read('manifest.json'),object_pairs_hook=no_duplicate_keys) require(isinstance(meta,dict),'Invalid manifest.') check_metadata(meta) require(isinstance(meta['files'],dict) and set(names)=={'manifest.json',*meta['files']},'Archive inventory does not match the manifest.') for name,details in meta['files'].items(): match=re.fullmatch(r'businesses/(oak|harbour|[a-f0-9]{32})/(business\.sqlite3|(?:assets|photos)/[a-f0-9]{32}\.png)',name) if name=='registry.sqlite3' and meta['version']>=4:bid=None;kind='identity' else: require(match is not None and match[1] in {b['id'] for b in meta['businesses']},'Unsafe archive path or unexpected file.') bid=match[1];kind='database' if match[2]=='business.sqlite3' else 'job_image' if match[2].startswith('photos/') else 'image' require(kind!='job_image' or meta['version']>=5,'Private photos require recovery format 5.') entry=archive.getinfo(name) require(isinstance(details,dict) and set(details)=={'bytes','sha256','business_id','kind'},'Invalid file metadata.') require(details['business_id']==bid and details['kind']==kind,'Cross-business file mapping is not allowed.') require(type(details['bytes']) is int and 0=4: require('registry.sqlite3' in meta['files'],'The identity registry is missing.') registry_summary(stage/'registry.sqlite3',meta,{b['id']:stage/'businesses'/b['id']/'business.sqlite3' for b in meta['businesses']},sanitised=True) return meta,summaries def verify(archive,sha256): with tempfile.TemporaryDirectory(prefix='shs-verify-') as temporary: meta,summaries=unpack_validated(archive,sha256,Path(temporary)) return {'valid':True,'sha256':sha256,'snapshot_at':meta['created_at'],'businesses':summaries,'authentication':meta['auth_policy']} def publish_fresh(stage,target): # Never replace or merge into any existing destination, even an empty one. target.mkdir(mode=0o700) created_files=[];created_dirs=[target] try: marker=target/'RESTORE_IN_PROGRESS';private_write(marker,b'Incomplete recovery. Do not start this folder.\n');created_files.append(marker) for source in sorted(stage.rglob('*')): relative=source.relative_to(stage) if str(relative)=='.data.lock':continue dest=target/relative if source.is_dir():dest.mkdir(mode=0o700);created_dirs.append(dest) else:os.link(source,dest);created_files.append(dest) for directory in reversed(created_dirs):sync_directory(directory) marker.unlink();created_files.remove(marker);sync_directory(target);sync_directory(target.parent) except BaseException: # Remove only files/directories created here; never recursively delete an # existing or concurrently populated target. for path in reversed(created_files): try:path.unlink() except OSError:pass for path in reversed(created_dirs): try:path.rmdir() except OSError:pass raise def restore(archive,sha256,target,*,confirm_fresh=False): require(confirm_fresh is True,'Explicitly confirm restoration into a new folder.') target=local_path(target) require(not target.exists(),'Restore target already exists. Choose a completely new folder.') require(target.parent.is_dir(),'Create the parent folder first.') require(not target.is_relative_to(DEFAULT_DATA),'The normal demo data folder is protected. Choose a separate recovery folder.') require(not any((p/'registry.sqlite3').exists() for p in target.parents),'Do not restore inside another demo data folder.') with tempfile.TemporaryDirectory(prefix='.shs-restore-',dir=target.parent) as temporary: stage=Path(temporary);unpacked=stage/'validated';unpacked.mkdir(mode=0o700) meta,summaries=unpack_validated(archive,sha256,unpacked) recovered=stage/'recovered';store=Store(recovered,seed_demo=False) logins=['LOCAL RECOVERY — FICTIONAL DATA ONLY','New passwords for this restored copy. Keep this file private.','Old passwords, sessions, invitations, resets and portal codes do not work here.',''] if meta['version']>=4: with read_db(unpacked/'registry.sqlite3') as source,write_db(recovered/'registry.sqlite3') as destination:source.backup(destination) with store.registry() as db: new_dataset=uid();db.execute('UPDATE dataset_identity SET dataset_id=?',(new_dataset,)) if meta['version']<4: for business in meta['businesses']: bid=business['id'] db.execute('INSERT INTO businesses(id,slug,name,enabled) VALUES(?,?,?,?)',(bid,business['slug'],business['name'],business['enabled'])) db.execute('INSERT INTO users(id,business_id,email,password,display_name) VALUES(?,?,?,?,?)',(bid,bid,BUSINESSES[bid]['email'],'','Recovered local owner')) db.execute('INSERT INTO memberships(id,business_id,user_id,role,created_at) VALUES(?,?,?,?,?)',(uid(),bid,bid,'owner',now())) for domain in meta['domains']:db.execute('INSERT INTO domains VALUES(?,?)',(domain['hostname'],domain['business_id'])) for table in ('sessions','access_tokens','registry_operations','limits'):db.execute('DELETE FROM '+table) for user in db.execute('SELECT id,email,display_name,disabled FROM users ORDER BY id').fetchall(): active=db.execute('SELECT 1 FROM memberships m JOIN businesses b ON b.id=m.business_id WHERE m.user_id=? AND m.enabled=1 AND b.enabled=1',(user['id'],)).fetchone() new_password='Recovered-'+secrets.token_urlsafe(24) if active and not user['disabled'] else None db.execute('UPDATE users SET password=? WHERE id=?',(password_hash(new_password) if new_password else '',user['id'])) if new_password:logins.extend([user['display_name'],user['email'],new_password,'']) from .identity import identity_audit identity_audit(db,None,None,sha256,'recovery.identities_revoked') for business in meta['businesses']: bid=business['id'];folder=unpacked/'businesses'/bid (recovered/bid/'assets').mkdir(parents=True,mode=0o700) with read_db(folder/'business.sqlite3') as source,write_db(recovered/bid/'business.sqlite3') as destination: source.backup(destination) destination.execute('UPDATE storage_identity SET dataset_id=?',(new_dataset,)) destination.execute('UPDATE customers SET portal_hash=NULL,portal_version=portal_version+1') destination.execute('INSERT INTO audit(action,subject,created_at) VALUES(?,?,?)',('recovery.restored_portals_revoked',sha256,now())) for aid in summaries[bid]['assets']:private_write(recovered/bid/'assets'/(aid+'.png'),(folder/'assets'/(aid+'.png')).read_bytes()) (recovered/bid/'photos').mkdir(mode=0o700) for pid in summaries[bid]['photos']:private_write(recovered/bid/'photos'/(pid+'.png'),(folder/'photos'/(pid+'.png')).read_bytes()) # Legacy snapshots migrate only inside this private fresh copy. Store(recovered,seed_demo=False) from .portal_recovery import restored as revoke_restored_portals revoke_restored_portals(store) from .feedback_recovery import restored as revoke_feedback revoke_feedback(store) from .enquiry_recovery import restored as suspend_enquiry_photos suspend_enquiry_photos(store) from .fieldwork import interrupt_open interrupt_open(store,'restored copy; no clock resumes automatically') report={'restored':True,'source_format':meta['version'],'recovery_format':VERSION,'source_sha256':sha256,'snapshot_at':meta['created_at'],'restored_at':now(),'new_dataset_id':new_dataset,'businesses':summaries,'authentication':AUTH_POLICY,'browser_verified':False} private_write(recovered/'RESTORE-REPORT.json',(json.dumps(report,indent=2)+'\n').encode()) private_write(recovered/'RESTORED-LOGIN.txt',('\n'.join(logins)+'\n').encode()) for file in recovered.rglob('*'): if file.is_file():file.chmod(0o600) # Revalidate the rebuilt identity and all original relationships before # publishing any destination. Passwords are never printed or archived. for bid in (b['id'] for b in meta['businesses']):tenant_summary(recovered/bid/'business.sqlite3',bid,new_dataset,sanitised=True) registry_summary(recovered/'registry.sqlite3',{**meta,'dataset_id':new_dataset}, {b['id']:recovered/b['id']/'business.sqlite3' for b in meta['businesses']},sanitised=False) publish_fresh(recovered,target) return {'restored':True,'target':str(target),'report':str(target/'RESTORE-REPORT.json'),'new_owner_logins':str(target/'RESTORED-LOGIN.txt'),'authentication':AUTH_POLICY}