"""Verify and restore a versioned SQLite/file backup set into disposable resources.""" from __future__ import annotations import argparse, hashlib, json, os, shutil, sqlite3, tarfile, tempfile, time from pathlib import Path, PurePosixPath FORMAT = 'recorder-backup-v1' MAX_ARCHIVE = 32 * 1024 * 1024 MAX_FILES = 1000 class CheckError(Exception): pass def digest(path): h=hashlib.sha256() with Path(path).open('rb') as f: for block in iter(lambda:f.read(65536),b''): h.update(block) return h.hexdigest() def safe_name(name): p=PurePosixPath(name) if not name or '\\' in name or p.is_absolute() or any(x in ('..','.') for x in name.split('/')) or ':' in name or '\x00' in name: raise CheckError('Unsafe archive path: '+repr(name)) return p def load_manifest(bundle): p=bundle/'manifest.json' if p.is_symlink() or p.stat().st_size>1024*1024: raise CheckError('Invalid manifest file') m=json.loads(p.read_text()) if m.get('format')!=FORMAT or not isinstance(m.get('snapshot_id'),str) or not m['snapshot_id']: raise CheckError('Unsupported manifest format or snapshot ID') if not isinstance(m.get('files'),list) or len(m['files'])>MAX_FILES: raise CheckError('Invalid file list') names=set() for f in m['files']: safe_name(f['path']) if f['path'] in names: raise CheckError('Duplicate manifest path') names.add(f['path']) if not isinstance(f.get('bytes'),int) or f['bytes']<0: raise CheckError('Invalid file size') if len(f.get('sha256',''))!=64: raise CheckError('Invalid file hash') return m def run(bundle, report_path=None, work_parent=None, interrupt_at=None, stage_hook=None): """Only delete a TemporaryDirectory created by this invocation. Never accept a restore target.""" started=time.perf_counter(); bundle=Path(bundle).resolve() report={'format':'restore-report-v1','passed':False,'engine':'SQLite','sqlite_version':sqlite3.sqlite_version, 'source':str(bundle),'checks':[],'timings_seconds':{},'restore_succeeded':False,'application_checks_passed':False} owned=None def check(name,fn): t=time.perf_counter() try: value=fn(); report['checks'].append({'check':name,'passed':True,'detail':value}) return value except Exception as e: report['checks'].append({'check':name,'passed':False,'detail':str(e)}); raise finally: report['timings_seconds'][name]=round(time.perf_counter()-t,6) def stop(stage): if stage_hook is not None: stage_hook(stage, report.get('temporary_directory')) if interrupt_at==stage: raise InterruptedError('Injected interruption after '+stage) try: m=check('manifest',lambda:load_manifest(bundle)) report['snapshot_id']=m['snapshot_id']; report['manifest_sha256']=digest(bundle/'manifest.json') # Keep the manifest out of the compact check log; full manifest is in the source bundle. report['checks'][-1]['detail']={'snapshot_id':m['snapshot_id'],'files':len(m['files'])} def input_hashes(): for name,field in [('database.sqlite','database_sha256'),('files.tar','archive_sha256')]: p=bundle/name if p.is_symlink() or not p.is_file(): raise CheckError('Missing/unsafe input '+name) if p.stat().st_size>MAX_ARCHIVE: raise CheckError('Input size limit exceeded') if digest(p)!=m[field]: raise CheckError('Checksum mismatch: '+name) return 'Database and archive digests match the manifest' check('input_checksums',input_hashes); stop('checksums') owned=tempfile.TemporaryDirectory(prefix='recovery-drill-',dir=work_parent) work=Path(owned.name); report['temporary_directory']=str(work) files=work/'files'; files.mkdir() def extract(): expected={f['path']:f for f in m['files']}; seen=set(); total=0 with tarfile.open(bundle/'files.tar','r:') as tf: members=tf.getmembers() if len(members)>MAX_FILES: raise CheckError('Too many archive entries') for entry in members: safe_name(entry.name) if not entry.isfile(): raise CheckError('Only regular archive files are permitted') if entry.name in seen: raise CheckError('Duplicate archive entry') if entry.name not in expected: raise CheckError('Unlisted archive file: '+entry.name) total+=entry.size if total>MAX_ARCHIVE: raise CheckError('Expanded archive limit exceeded') if entry.size!=expected[entry.name]['bytes']: raise CheckError('Size mismatch: '+entry.name) target=files.joinpath(*PurePosixPath(entry.name).parts);target.parent.mkdir(parents=True,exist_ok=True) source=tf.extractfile(entry) if source is None: raise CheckError('Unreadable archive member') with source, target.open('xb') as out: shutil.copyfileobj(source,out) if digest(target)!=expected[entry.name]['sha256']: raise CheckError('File digest mismatch: '+entry.name) seen.add(entry.name) if seen!=set(expected): raise CheckError('Archive is missing manifest files: '+', '.join(sorted(set(expected)-seen))) return {'files':len(seen),'bytes':total} check('file_restore',extract); stop('files') restored=work/'restored.sqlite' def restore(): src=sqlite3.connect((bundle/'database.sqlite').as_uri()+'?mode=ro',uri=True) dst=sqlite3.connect(restored) try: src.backup(dst) finally: src.close();dst.close() report['restore_succeeded']=True return 'Copied source pages through SQLite backup API into a new database' check('database_restore',restore);stop('restore') def probes(): db=sqlite3.connect(restored.as_uri()+'?mode=ro',uri=True);db.execute('PRAGMA query_only=ON') try: if db.execute('PRAGMA integrity_check').fetchall()!=[('ok',)]: raise CheckError('SQLite integrity check failed') if db.execute('PRAGMA foreign_key_check').fetchall(): raise CheckError('Foreign-key violation') snapshot=db.execute("SELECT value FROM metadata WHERE key='snapshot_id'").fetchone() if snapshot!=(m['snapshot_id'],): raise CheckError('Database/manifest snapshot mismatch') expected={f['path']:f for f in m['files']};refs=db.execute('SELECT path,sha256,bytes FROM documents ORDER BY path').fetchall() if {r[0] for r in refs}!=set(expected): raise CheckError('Database references differ from manifest file set') for name,sha,size in refs: if (sha,size)!=(expected[name]['sha256'],expected[name]['bytes']): raise CheckError('Database/file metadata mismatch: '+name) records=db.execute('SELECT COUNT(*) FROM visits').fetchone()[0] if records!=m['expected_visits']: raise CheckError('Visit count differs from the declared application contract') invalid=db.execute("SELECT COUNT(*) FROM visits WHERE status NOT IN ('open','complete') OR length(asset)=0").fetchone()[0] if invalid: raise CheckError('Application visit invariant failed') return {'visits':records,'documents':len(refs),'integrity':'ok','snapshot_match':True} finally: db.close() check('application_probes',probes);report['application_checks_passed']=True;stop('probes') report['passed']=True except (Exception,KeyboardInterrupt) as e: report['error']=f'{type(e).__name__}: {e}' finally: t=time.perf_counter() if owned: owned.cleanup() report['cleanup_succeeded']=not owned or not Path(owned.name).exists() report['timings_seconds']['cleanup']=round(time.perf_counter()-t,6) report['timings_seconds']['total']=round(time.perf_counter()-started,6) report['passed']=report['passed'] and report['cleanup_succeeded'] if report_path: p=Path(report_path);p.parent.mkdir(parents=True,exist_ok=True);p.write_text(json.dumps(report,indent=2)+'\n') return report def main(): p=argparse.ArgumentParser(description=__doc__);p.add_argument('bundle',type=Path);p.add_argument('--report',type=Path,default=Path('results/latest.json'));args=p.parse_args() r=run(args.bundle,args.report) print(('PASS' if r['passed'] else 'FAIL')+' '+r.get('snapshot_id','unknown snapshot')) for c in r['checks']: print((' OK ' if c['passed'] else ' FAIL ')+c['check']+': '+str(c['detail'])) if 'error' in r: print(r['error']) print('Report:',args.report) return 0 if r['passed'] else 1 if __name__=='__main__': raise SystemExit(main())