"""InFinBot one-shot verified SQLite backup; retains all backups by default. Python 3.8+ on macOS/Linux. No broker access, ledger initialization or scheduling. Explicit --db-path and --backup-dir required. Deletion requires --retain --prune. """ import argparse from contextlib import closing, contextmanager from datetime import datetime, timezone import fcntl import hashlib import json import math import os from pathlib import Path import re import sqlite3 import stat import tempfile import time import uuid class BackupBlocked(Exception): pass NAME = re.compile(r'(?:foresixa|infinbot)_backup_\d{8}T\d{12}Z_[0-9a-f]{32}\.db\Z') def digest(path): result = hashlib.sha256() with path.open('rb') as stream: for block in iter(lambda: stream.read(1024 * 1024), b''): result.update(block) return result.hexdigest() def regular_private(path): info = path.lstat() return (stat.S_ISREG(info.st_mode) and info.st_uid == os.geteuid() and info.st_mode & 0o077 == 0 and info.st_nlink == 1) def sync_directory(path): descriptor = os.open(str(path), os.O_RDONLY) try: os.fsync(descriptor) finally: os.close(descriptor) def check_deadline(deadline): if time.monotonic() >= deadline: raise BackupBlocked('TIME_LIMIT_REACHED') def integrity(connection, deadline): connection.set_progress_handler(lambda: int(time.monotonic() >= deadline), 1000) try: check_deadline(deadline) rows = connection.execute('PRAGMA integrity_check').fetchall() if rows != [('ok',)]: raise BackupBlocked('BACKUP_INTEGRITY_FAILED') finally: connection.set_progress_handler(None, 0) @contextmanager def vault_lock(root, deadline): descriptor = os.open(str(root / '.foresixa_backup.lock'), os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600) locked = False try: info = os.fstat(descriptor) if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077 or info.st_nlink != 1): raise BackupBlocked('UNSAFE_VAULT_LOCK') while True: check_deadline(deadline) try: fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) locked = True break except BlockingIOError: time.sleep(0.05) yield finally: if locked: fcntl.flock(descriptor, fcntl.LOCK_UN) os.close(descriptor) def publish_json(root, target, payload): descriptor, name = tempfile.mkstemp(prefix='.foresixa_manifest_', dir=str(root)) temporary = Path(name) try: with os.fdopen(descriptor, 'w', encoding='utf-8') as stream: json.dump(payload, stream, indent=2, allow_nan=False) stream.write('\n') stream.flush() os.fsync(stream.fileno()) # Atomic publication without overwriting an existing file. os.link(str(temporary), str(target)) temporary.unlink() sync_directory(root) finally: if temporary.exists(): temporary.unlink() class ForesixaBackupRotationEngine: def __init__(self, source_db, backup_root, *, retain=None, prune=False, timeout=30): self.source_db = Path(source_db).expanduser().absolute() self.backup_root = Path(backup_root).expanduser().absolute() self.retain, self.prune, self.timeout = retain, prune, timeout def _prepare(self): if type(self.prune) is not bool: raise BackupBlocked('INVALID_PRUNE_OPTION') if self.retain is not None and (type(self.retain) is not int or self.retain < 1): raise BackupBlocked('RETENTION_MUST_BE_A_POSITIVE_SNAPSHOT_COUNT') if self.prune and self.retain is None: raise BackupBlocked('PRUNE_REQUIRES_EXPLICIT_RETENTION') if isinstance(self.timeout, bool) or not isinstance(self.timeout, (float, int)) or not math.isfinite(self.timeout) or not 0 < self.timeout <= 300: raise BackupBlocked('INVALID_TIME_LIMIT') if not self.source_db.exists(): raise BackupBlocked('SOURCE_MISSING') info = self.source_db.lstat() if not stat.S_ISREG(info.st_mode): raise BackupBlocked('SOURCE_MUST_BE_A_REGULAR_FILE_WITHOUT_SYMLINK') with self.source_db.open('rb') as stream: if stream.read(16) != b'SQLite format 3\x00': raise BackupBlocked('SOURCE_IS_NOT_AN_EXISTING_SQLITE_DATABASE') self.source_db = self.source_db.resolve(strict=True) if not self.backup_root.exists(): self.backup_root.mkdir(mode=0o700, parents=True, exist_ok=True) info = self.backup_root.lstat() if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077): raise BackupBlocked('BACKUP_DIRECTORY_MUST_BE_OWNER_PRIVATE_WITHOUT_SYMLINK') self.backup_root = self.backup_root.resolve(strict=True) def _copy(self, target, deadline): descriptor, name = tempfile.mkstemp(prefix='.foresixa_pending_', suffix='.db', dir=str(self.backup_root)) os.close(descriptor) temporary = Path(name) try: with closing(sqlite3.connect(self.source_db.as_uri() + '?mode=ro', uri=True, timeout=1)) as source: source.execute('PRAGMA query_only=ON') with closing(sqlite3.connect(str(temporary), timeout=1)) as destination: destination.execute('PRAGMA synchronous=FULL') source.backup(destination, pages=20, progress=lambda status, remaining, total: check_deadline(deadline), sleep=0.05) destination.execute('PRAGMA journal_mode=DELETE') integrity(destination, deadline) # Reopen the completed snapshot independently before publication. with closing(sqlite3.connect(temporary.as_uri() + '?mode=ro', uri=True, timeout=1)) as restored: integrity(restored, deadline) with temporary.open('rb') as stream: os.fsync(stream.fileno()) os.link(str(temporary), str(target)) temporary.unlink() sync_directory(self.backup_root) finally: # Remove only this run's unfinished temporary files. for suffix in ('', '-wal', '-shm', '-journal'): unfinished = Path(str(temporary) + suffix) if unfinished.exists(): unfinished.unlink() def _retention(self, deadline, current_backup): if self.retain is None: return [], [], [] eligible, skipped = [], [] for path in self.backup_root.iterdir(): if not NAME.fullmatch(path.name): continue metadata = Path(str(path) + '.json') try: check_deadline(deadline) if path == self.source_db or not regular_private(path) or not regular_private(metadata): raise ValueError('Unmanaged file') with metadata.open('rb') as stream: raw = stream.read(16385) if len(raw) > 16384: raise ValueError('Oversized metadata') manifest = json.loads(raw) if (manifest.get('format') != 'FORESIXA_SQLITE_BACKUP_V1' or manifest.get('source_db') != str(self.source_db) or manifest.get('backup_filename') != path.name or manifest.get('sha256') != digest(path)): raise ValueError('Unmanaged or altered backup') with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=1)) as connection: integrity(connection, deadline) eligible.append(path) except BackupBlocked: raise except (OSError, ValueError, AttributeError, sqlite3.Error): skipped.append(path.name) # Preserve this run's verified snapshot even if the host clock moved back. eligible.sort(key=lambda path: (path == current_backup, path.name), reverse=True) candidates = eligible[self.retain:] removed = [] if self.prune: for path in candidates: check_deadline(deadline) path.unlink() self.deleted_backups.append(path.name) Path(str(path) + '.json').unlink() removed.append(path.name) if removed: sync_directory(self.backup_root) return [path.name for path in candidates], removed, sorted(skipped) def execute_backup(self): self.deleted_backups = [] report = {'service': 'InFinBot', 'status': 'BACKUP_BLOCKED', 'source_db': str(self.source_db), 'backup_directory': str(self.backup_root), 'retention_snapshot_count': self.retain, 'prune_enabled': self.prune, 'source_opened_read_only': False, 'backup_verified': False, 'credential_files_copied': False, 'scheduler_installed': False, 'source_provenance': 'NOT_CHECKED', 'current_host_ledger_verified': False} try: self._prepare() deadline = time.monotonic() + self.timeout with vault_lock(self.backup_root, deadline): stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ') filename = 'infinbot_backup_' + stamp + '_' + uuid.uuid4().hex + '.db' target = self.backup_root / filename self._copy(target, deadline) report.update(backup_path=str(target), source_opened_read_only=True, backup_verified=True, status='BACKUP_VERIFIED') with closing(sqlite3.connect(target.as_uri() + '?mode=ro', uri=True, timeout=1)) as copied: fixture = copied.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='sandbox_core_metadata'").fetchone() report['source_provenance'] = ('SYNTHETIC_FIXTURE' if fixture else 'EXISTING_SQLITE_FILE_UNVERIFIED_PROVENANCE') manifest = {'format': 'FORESIXA_SQLITE_BACKUP_V1', 'source_db': str(self.source_db), 'backup_filename': filename, 'completed_at': datetime.now(timezone.utc).isoformat(), 'sha256': digest(target), 'size_bytes': target.stat().st_size, 'integrity_check': 'ok', 'reopened_read_only': True, 'source_provenance': report['source_provenance']} publish_json(self.backup_root, Path(str(target) + '.json'), manifest) report['sha256'] = manifest['sha256'] candidates, removed, skipped = self._retention(deadline, target) report.update(retention_candidates=candidates, deleted_backups=removed, skipped_unmanaged_backups=skipped) return report except BackupBlocked as error: report['reason'] = str(error) except Exception: report['reason'] = 'FILESYSTEM_OR_SQLITE_OPERATION_FAILED' if report['backup_verified']: report['status'] = 'BACKUP_VERIFIED_POSTPROCESSING_INCOMPLETE' report['deleted_backups'] = self.deleted_backups return report def execute_friday_backup_rotation(self): """One-shot compatibility method; this does not install a Friday schedule.""" return self.execute_backup() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--db-path', required=True, help='Actual existing persistent SQLite simulation ledger') parser.add_argument('--backup-dir', required=True, help='Owner-private backup directory, preferably another storage device') parser.add_argument('--retain', type=int, help='Keep this many verified managed snapshots; default keeps all') parser.add_argument('--prune', action='store_true', help='Delete older eligible snapshots after a verified backup; requires --retain') parser.add_argument('--timeout', type=float, default=30, help='Backup/retention SQLite processing time limit, 0 < seconds <= 300') args = parser.parse_args() result = ForesixaBackupRotationEngine(args.db_path, args.backup_dir, retain=args.retain, prune=args.prune, timeout=args.timeout).execute_backup() print(json.dumps(result, indent=2, allow_nan=False)) return 0 if result['status'] == 'BACKUP_VERIFIED' else 1 if __name__ == '__main__': raise SystemExit(main())