From 4bad7ca4acb59609b290c3f1fd695dc535697ab1 Mon Sep 17 00:00:00 2001 From: Nikhil Iyer Date: Thu, 2 Jul 2026 11:04:57 -0400 Subject: [PATCH 1/3] Add option for full extraction of passwordless backup --- backuplens.py | 237 ++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 210 insertions(+), 27 deletions(-) diff --git a/backuplens.py b/backuplens.py index 8fcc2dc..6ee9521 100644 --- a/backuplens.py +++ b/backuplens.py @@ -17,12 +17,51 @@ import plistlib import platform import shutil +import sqlite3 +from contextlib import contextmanager from datetime import datetime __version__ = "1.0.0" APP_NAME = "BackupLens" +class PlainBackup: + """Read-only access to a non-encrypted iOS backup. + + Mirrors the subset of EncryptedBackup's interface used by this app + (manifest_db_cursor / extract_file) so callers don't need to branch + on whether a backup is encrypted. + """ + + def __init__(self, *, backup_directory): + self._backup_directory = backup_directory + self._manifest_db_path = os.path.join(backup_directory, "Manifest.db") + + @contextmanager + def manifest_db_cursor(self): + conn = sqlite3.connect(self._manifest_db_path) + try: + cur = conn.cursor() + yield cur + cur.close() + finally: + conn.close() + + def extract_file(self, *, relative_path, domain, output_filename): + with self.manifest_db_cursor() as cur: + cur.execute( + "SELECT fileID FROM Files WHERE relativePath=? AND domain=? " + "AND flags=1 LIMIT 1", + (relative_path, domain), + ) + row = cur.fetchone() + if not row: + raise FileNotFoundError(relative_path) + file_id = row[0] + src = os.path.join(self._backup_directory, file_id[:2], file_id) + shutil.copy2(src, output_filename) + + class BackupLens: """Main application class for BackupLens.""" @@ -160,19 +199,20 @@ def _build_ui(self): row2 = ttk.Frame(conn_frame) row2.pack(fill="x", pady=3) - ttk.Label(row2, text="Password:").pack(side="left") + ttk.Label(row2, text="Password (if encrypted):").pack(side="left") self.pass_var = tk.StringVar() self.pass_entry = ttk.Entry(row2, textvariable=self.pass_var, show="*", width=40) self.pass_entry.pack(side="left", padx=8) self.pass_entry.bind("", lambda e: self._decrypt()) - self.decrypt_btn = ttk.Button(row2, text="Decrypt & Open", + self.decrypt_btn = ttk.Button(row2, text="Open Backup", command=self._decrypt, style="Safe.TButton") self.decrypt_btn.pack(side="left", padx=8) self.status_var = tk.StringVar( - value="Select a backup folder and enter your password to begin." + value="Select a backup folder to begin. Enter a password " + "only if the backup is encrypted." ) status_bar = ttk.Label(conn_frame, textvariable=self.status_var, style="Status.TLabel") @@ -213,6 +253,8 @@ def _build_ui(self): command=self._extract_selected).pack(side="right", padx=6) ttk.Button(toolbar, text="Extract All in View", command=self._extract_all_view).pack(side="right", padx=2) + ttk.Button(toolbar, text="Extract Entire Backup", + command=self._extract_full_backup).pack(side="right", padx=6) cols = ("domain", "path", "size", "modified") self.file_tree = ttk.Treeview(right_frame, columns=cols, @@ -268,6 +310,19 @@ def _browse_folder(self): # ── Decryption ─────────────────────────────────────────── + @staticmethod + def _is_backup_encrypted(backup_dir): + """Inspect Manifest.plist to determine if the backup is encrypted.""" + manifest_plist_path = os.path.join(backup_dir, "Manifest.plist") + try: + with open(manifest_plist_path, "rb") as f: + manifest = plistlib.load(f) + return bool(manifest.get("IsEncrypted", False)) + except Exception: + # If we can't tell, assume encrypted so the user isn't + # silently handed a decryption failure with no explanation. + return True + def _decrypt(self): backup_dir = self.path_var.get().strip() passphrase = self.pass_var.get().strip() @@ -275,43 +330,57 @@ def _decrypt(self): if not backup_dir or not os.path.isdir(backup_dir): messagebox.showerror("Error", "Please select a valid backup folder.") return - if not passphrase: - messagebox.showerror("Error", - "Please enter the backup encryption password.") + + encrypted = self._is_backup_encrypted(backup_dir) + if encrypted and not passphrase: + messagebox.showerror( + "Error", + "This backup is encrypted. Please enter the backup " + "encryption password.", + ) return self.decrypt_btn.configure(state="disabled") - self.status_var.set("Decrypting... this may take a moment.") + self.status_var.set( + "Decrypting... this may take a moment." if encrypted + else "Opening backup..." + ) self.root.update_idletasks() threading.Thread(target=self._decrypt_thread, - args=(backup_dir, passphrase), daemon=True).start() + args=(backup_dir, passphrase, encrypted), + daemon=True).start() def _query_manifest(self, callback, *args): """Execute a callback with a manifest DB cursor (context-managed).""" with self.backup.manifest_db_cursor() as cur: return callback(cur, *args) - def _decrypt_thread(self, backup_dir, passphrase): - try: - from iphone_backup_decrypt import EncryptedBackup - except ImportError: - self.root.after(0, lambda: ( - self.decrypt_btn.configure(state="normal"), - messagebox.showerror( - "Missing Dependency", - "The 'iphone_backup_decrypt' package is required.\n\n" - "Install it by running:\n" - " pip install iphone_backup_decrypt", - ), - self.status_var.set("Missing dependency. See error above."), - )) - return + def _decrypt_thread(self, backup_dir, passphrase, encrypted): + if encrypted: + try: + from iphone_backup_decrypt import EncryptedBackup + except ImportError: + self.root.after(0, lambda: ( + self.decrypt_btn.configure(state="normal"), + messagebox.showerror( + "Missing Dependency", + "The 'iphone_backup_decrypt' package is required " + "for encrypted backups.\n\n" + "Install it by running:\n" + " pip install iphone_backup_decrypt", + ), + self.status_var.set("Missing dependency. See error above."), + )) + return try: - self.backup = EncryptedBackup( - backup_directory=backup_dir, passphrase=passphrase - ) + if encrypted: + self.backup = EncryptedBackup( + backup_directory=backup_dir, passphrase=passphrase + ) + else: + self.backup = PlainBackup(backup_directory=backup_dir) self.backup_dir = backup_dir # Clear password from the UI after successful decryption self.root.after(0, lambda: self.pass_var.set("")) @@ -338,7 +407,7 @@ def load_initial(cur): def _on_decrypt_success(self, domains, total_files): self.decrypt_btn.configure(state="normal") self.status_var.set( - f"Decrypted! {total_files:,} files across {len(domains)} domains." + f"Loaded! {total_files:,} files across {len(domains)} domains." ) self.domain_tree.delete(*self.domain_tree.get_children()) @@ -620,6 +689,120 @@ def _extract_thread(self, file_ids, dest): self.root.after(0, lambda: self.status_var.set(msg)) self.root.after(0, lambda: messagebox.showinfo("Done", msg)) + def _extract_full_backup(self): + """Extract every file in the backup, preserving domain/relativePath. + + Unlike Extract Selected / Extract All in View, this reads straight + from the manifest instead of the (10,000-row-capped) UI file list, + so nothing is left out for large backups. + """ + if not self.backup: + messagebox.showinfo("Info", "Open a backup first.") + return + dest = filedialog.askdirectory( + title="Select Output Folder for Full Backup Extraction" + ) + if not dest: + return + if not messagebox.askyesno( + "Confirm", + "This will extract every file in the backup into " + "domain/relativePath folders (e.g. " + "HomeDomain/Library/SMS/sms.db), matching the layout tools " + "like iphone-backup-tools expect. This may take a while and " + "use significant disk space. Continue?", + ): + return + self.status_var.set("Extracting entire backup...") + self.root.update_idletasks() + threading.Thread(target=self._extract_full_thread, args=(dest,), + daemon=True).start() + + def _extract_full_thread(self, dest): + real_dest = os.path.realpath(dest) + + try: + with self.backup.manifest_db_cursor() as cur: + cur.execute( + "SELECT fileID, domain, relativePath FROM Files " + "WHERE flags=1" + ) + rows = cur.fetchall() + except Exception as e: + self.root.after(0, lambda: self.status_var.set(f"Error: {e}")) + self.root.after(0, lambda: messagebox.showerror( + "Error", f"Failed to read backup manifest:\n{e}" + )) + return + + total = len(rows) + extracted = 0 + errors = 0 + skipped = 0 + + for i, (file_id, domain, rel_path) in enumerate(rows, start=1): + try: + if not rel_path: + skipped += 1 + continue + + out_path = os.path.join(dest, domain, rel_path) + + # Path traversal protection — ensure output stays inside dest + real_out = os.path.realpath(out_path) + if not real_out.startswith(real_dest + os.sep): + skipped += 1 + continue + + os.makedirs(os.path.dirname(out_path), exist_ok=True) + + try: + self.backup.extract_file( + relative_path=rel_path, domain=domain, + output_filename=out_path, + ) + extracted += 1 + except Exception: + src = os.path.join(self.backup_dir, file_id[:2], file_id) + if os.path.exists(src): + shutil.copy2(src, out_path) + extracted += 1 + else: + errors += 1 + except Exception: + errors += 1 + + if i % 200 == 0 or i == total: + done = i + self.root.after(0, lambda d=done, t=total: self.status_var.set( + f"Extracting entire backup... {d:,}/{t:,} files" + )) + + msg = f"Extracted {extracted:,} of {total:,} files to {dest}" + if errors: + msg += f" ({errors} errors)" + if skipped: + msg += f" ({skipped} skipped)" + + sms_path = os.path.join(dest, "HomeDomain", "Library", "SMS", "sms.db") + addr_path = os.path.join( + dest, "HomeDomain", "Library", "AddressBook", "AddressBook.sqlitedb" + ) + if os.path.exists(sms_path): + msg += ( + "\n\nTo browse messages with iphone-backup-tools:\n" + f' python message_viewer.py "{sms_path}"' + ) + if os.path.exists(addr_path): + msg += f' --addressbook "{addr_path}"' + + self.root.after(0, lambda: self.status_var.set( + f"Extracted {extracted:,} of {total:,} files to {dest}" + )) + self.root.after(0, lambda: messagebox.showinfo( + "Extraction Complete", msg + )) + # ── Helpers ────────────────────────────────────────────── @staticmethod From 624f69ebdce161cf6f3a377751b33dc23a16069d Mon Sep 17 00:00:00 2001 From: Nikhil Iyer Date: Thu, 9 Jul 2026 00:20:24 -0400 Subject: [PATCH 2/3] Make plain backups open readonly --- backuplens.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/backuplens.py b/backuplens.py index 6ee9521..adf59f9 100644 --- a/backuplens.py +++ b/backuplens.py @@ -35,11 +35,13 @@ class PlainBackup: def __init__(self, *, backup_directory): self._backup_directory = backup_directory - self._manifest_db_path = os.path.join(backup_directory, "Manifest.db") + _manifest_db_path = os.path.expanduser(os.path.join(backup_directory, "Manifest.db")) + self._manifest_db_uri = f"file:{_manifest_db_path}?mode=ro&immutable=1" + @contextmanager def manifest_db_cursor(self): - conn = sqlite3.connect(self._manifest_db_path) + conn = sqlite3.connect(self._manifest_db_uri, uri=True) try: cur = conn.cursor() yield cur From 7f1afb2274eaaca9fed0ea6a418adcbcabbbc10d Mon Sep 17 00:00:00 2001 From: Nikhil Iyer Date: Wed, 8 Jul 2026 23:50:54 -0400 Subject: [PATCH 3/3] Allow backups to be FUSE mounted --- backuplens.py | 119 +++++++++++++++++++++++ backuplens_fuse.py | 231 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 350 insertions(+) create mode 100644 backuplens_fuse.py diff --git a/backuplens.py b/backuplens.py index adf59f9..be5fe0a 100644 --- a/backuplens.py +++ b/backuplens.py @@ -18,9 +18,14 @@ import platform import shutil import sqlite3 +import subprocess +import tempfile +import time from contextlib import contextmanager from datetime import datetime +FUSE_SUPPORTED_PLATFORMS = ("Linux", "Darwin") + __version__ = "1.0.0" APP_NAME = "BackupLens" @@ -87,8 +92,14 @@ def __init__(self, root): self.backup_dir = None self.file_list = [] + self.fuse_mountpoint = None + self.fuse_cache_dir = None + self.fuse_thread = None + self.mount_btn = None + self._apply_style() self._build_ui() + self.root.protocol("WM_DELETE_WINDOW", self._on_close) self._auto_detect_backup() # ── Theming ────────────────────────────────────────────── @@ -257,6 +268,12 @@ def _build_ui(self): command=self._extract_all_view).pack(side="right", padx=2) ttk.Button(toolbar, text="Extract Entire Backup", command=self._extract_full_backup).pack(side="right", padx=6) + if platform.system() in FUSE_SUPPORTED_PLATFORMS: + self.mount_btn = ttk.Button( + toolbar, text="Mount as Filesystem (FUSE)", + command=self._toggle_fuse_mount, + ) + self.mount_btn.pack(side="right", padx=6) cols = ("domain", "path", "size", "modified") self.file_tree = ttk.Treeview(right_frame, columns=cols, @@ -805,6 +822,108 @@ def _extract_full_thread(self, dest): "Extraction Complete", msg )) + # ── FUSE mount ─────────────────────────────────────────── + + def _toggle_fuse_mount(self): + if self.fuse_mountpoint: + self._unmount_fuse() + else: + self._mount_fuse() + + def _mount_fuse(self): + if not self.backup: + messagebox.showinfo("Info", "Open a backup first.") + return + try: + import backuplens_fuse + except Exception as e: + messagebox.showerror( + "Missing Dependency", + "Mounting requires the 'fusepy' package and libfuse.\n\n" + "Install with:\n pip install fusepy\n\n" + "Linux: libfuse2/libfuse3 is usually preinstalled.\n" + "macOS: install macFUSE from https://osxfuse.github.io/\n\n" + f"Details: {e}", + ) + return + + mountpoint = filedialog.askdirectory( + title="Select an EMPTY folder to mount the backup on" + ) + if not mountpoint: + return + if os.listdir(mountpoint): + messagebox.showerror( + "Error", "The mount point folder must be empty." + ) + return + + cache_dir = tempfile.mkdtemp(prefix="backuplens_fuse_cache_") + mount_time = time.time() + + def run(): + try: + backuplens_fuse.mount( + self.backup, mountpoint, cache_dir, mount_time, + foreground=True, + ) + except Exception as e: + self.root.after(0, lambda: messagebox.showerror( + "Mount Failed", str(e) + )) + finally: + self.root.after(0, self._on_fuse_stopped) + + self.fuse_mountpoint = mountpoint + self.fuse_cache_dir = cache_dir + self.fuse_thread = threading.Thread(target=run, daemon=True) + self.fuse_thread.start() + self.mount_btn.configure(text="Unmount Filesystem") + self.status_var.set(f"Mounted at {mountpoint}") + + def _unmount_syscall(self, mountpoint): + if platform.system() == "Darwin": + subprocess.run(["umount", mountpoint], check=True) + else: + cmd = shutil.which("fusermount3") or shutil.which("fusermount") \ + or "umount" + subprocess.run([cmd, "-u", mountpoint], check=True) + + def _unmount_fuse(self): + if not self.fuse_mountpoint: + return + self.status_var.set("Unmounting...") + self.root.update_idletasks() + try: + self._unmount_syscall(self.fuse_mountpoint) + except Exception as e: + messagebox.showerror("Unmount Failed", str(e)) + self.status_var.set("Unmount failed. See error above.") + return + # The blocking FUSE() call in the mount thread returns once the + # kernel finishes the unmount; _on_fuse_stopped resets UI state then. + + def _on_fuse_stopped(self): + cache_dir = self.fuse_cache_dir + self.fuse_mountpoint = None + self.fuse_cache_dir = None + self.fuse_thread = None + if cache_dir: + shutil.rmtree(cache_dir, ignore_errors=True) + if self.mount_btn: + self.mount_btn.configure(text="Mount as Filesystem (FUSE)") + self.status_var.set("Filesystem unmounted.") + + def _on_close(self): + if self.fuse_mountpoint: + try: + self._unmount_syscall(self.fuse_mountpoint) + except Exception: + pass + if self.fuse_cache_dir: + shutil.rmtree(self.fuse_cache_dir, ignore_errors=True) + self.root.destroy() + # ── Helpers ────────────────────────────────────────────── @staticmethod diff --git a/backuplens_fuse.py b/backuplens_fuse.py new file mode 100644 index 0000000..d979c7e --- /dev/null +++ b/backuplens_fuse.py @@ -0,0 +1,231 @@ +""" +Read-only FUSE view over an opened iOS backup. + +Presents the backup as a / tree — the same layout +`backuplens.py`'s full extraction produces — but without copying the whole +backup up front. Each file is decrypted/copied into a local cache directory +the first time something opens it, then served from that cached copy. + +Linux/macOS only (needs libfuse + the `fusepy` package). +""" + +from __future__ import annotations + +import errno +import os +import plistlib +import stat +import threading + +from fuse import FuseOSError, FUSE, Operations + + +class _Dir: + __slots__ = ("children",) + + def __init__(self): + self.children = {} + + +class _FileEntry: + __slots__ = ("file_id", "domain", "relative_path", "size", "mtime") + + def __init__(self, file_id, domain, relative_path, size, mtime): + self.file_id = file_id + self.domain = domain + self.relative_path = relative_path + self.size = size + self.mtime = mtime + + +def _parse_size_mtime(file_blob): + size = 0 + mtime = None + if file_blob: + try: + meta = plistlib.loads(file_blob) + objects = meta.get("$objects", []) + if isinstance(objects, list) and len(objects) > 1: + obj1 = objects[1] + if isinstance(obj1, dict): + size = obj1.get("Size", 0) or 0 + for obj in (objects if isinstance(objects, list) else []): + if isinstance(obj, dict) and "LastModified" in obj: + mtime = obj["LastModified"] + break + except Exception: + pass + return size, mtime + + +class BackupFS(Operations): + """`backup` must expose `manifest_db_cursor()` and `extract_file(...)`, + matching both `backuplens.PlainBackup` and + `iphone_backup_decrypt.EncryptedBackup`. + """ + + def __init__(self, backup, cache_dir, mount_time): + self.backup = backup + self.cache_dir = cache_dir + self.mount_time = mount_time + self.root = _Dir() + self._file_locks_guard = threading.Lock() + self._file_locks = {} + self._build_tree() + + def _build_tree(self): + with self.backup.manifest_db_cursor() as cur: + cur.execute( + "SELECT fileID, domain, relativePath, file FROM Files " + "WHERE flags=1 AND relativePath IS NOT NULL " + "AND relativePath != ''" + ) + rows = cur.fetchall() + + for file_id, domain, rel_path, file_blob in rows: + if not domain or not rel_path: + continue + size, mtime = _parse_size_mtime(file_blob) + parts = [domain] + [p for p in rel_path.split("/") if p] + node = self.root + collided = False + for part in parts[:-1]: + child = node.children.setdefault(part, _Dir()) + if not isinstance(child, _Dir): + # A file and a directory both claim this name — keep + # whichever showed up first and drop this entry. + collided = True + break + node = child + if not collided: + node.children[parts[-1]] = _FileEntry( + file_id, domain, rel_path, size, mtime + ) + + def _lookup(self, path): + if path in ("/", ""): + return self.root + node = self.root + for part in path.strip("/").split("/"): + if not isinstance(node, _Dir) or part not in node.children: + return None + node = node.children[part] + return node + + def _ensure_cached(self, entry): + cache_path = os.path.join(self.cache_dir, entry.file_id) + if os.path.exists(cache_path): + return cache_path + with self._file_locks_guard: + lock = self._file_locks.setdefault(entry.file_id, threading.Lock()) + with lock: + if not os.path.exists(cache_path): + tmp_path = f"{cache_path}.part" + self.backup.extract_file( + relative_path=entry.relative_path, + domain=entry.domain, + output_filename=tmp_path, + ) + os.replace(tmp_path, cache_path) + return cache_path + + # ── FUSE operations ────────────────────────────────────── + + def getattr(self, path, fh=None): + node = self._lookup(path) + if node is None: + raise FuseOSError(errno.ENOENT) + if isinstance(node, _Dir): + return { + "st_mode": stat.S_IFDIR | 0o500, + "st_nlink": 2, + "st_size": 0, + "st_ctime": self.mount_time, + "st_mtime": self.mount_time, + "st_atime": self.mount_time, + } + mtime = node.mtime if node.mtime is not None else self.mount_time + return { + "st_mode": stat.S_IFREG | 0o400, + "st_nlink": 1, + "st_size": node.size, + "st_ctime": mtime, + "st_mtime": mtime, + "st_atime": mtime, + } + + def readdir(self, path, fh): + node = self._lookup(path) + if not isinstance(node, _Dir): + raise FuseOSError(errno.ENOTDIR) + return ["." , ".."] + list(node.children.keys()) + + def open(self, path, flags): + node = self._lookup(path) + if node is None or isinstance(node, _Dir): + raise FuseOSError(errno.ENOENT) + if (flags & os.O_ACCMODE) != os.O_RDONLY: + raise FuseOSError(errno.EROFS) + try: + cache_path = self._ensure_cached(node) + except FileNotFoundError: + raise FuseOSError(errno.ENOENT) + except Exception: + raise FuseOSError(errno.EIO) + return os.open(cache_path, os.O_RDONLY) + + def read(self, path, size, offset, fh): + os.lseek(fh, offset, os.SEEK_SET) + return os.read(fh, size) + + def release(self, path, fh): + os.close(fh) + return 0 + + def statfs(self, path): + st = os.statvfs(self.cache_dir) + return { + key: getattr(st, key) + for key in ( + "f_bavail", "f_bfree", "f_blocks", "f_bsize", "f_favail", + "f_ffree", "f_files", "f_flag", "f_frsize", "f_namemax", + ) + } + + # Read-only filesystem — refuse anything that would mutate the backup. + def write(self, path, data, offset, fh): + raise FuseOSError(errno.EROFS) + + def create(self, path, mode, fi=None): + raise FuseOSError(errno.EROFS) + + def unlink(self, path): + raise FuseOSError(errno.EROFS) + + def mkdir(self, path, mode): + raise FuseOSError(errno.EROFS) + + def rmdir(self, path): + raise FuseOSError(errno.EROFS) + + def truncate(self, path, length, fh=None): + raise FuseOSError(errno.EROFS) + + def chmod(self, path, mode): + raise FuseOSError(errno.EROFS) + + def chown(self, path, uid, gid): + raise FuseOSError(errno.EROFS) + + def rename(self, old, new): + raise FuseOSError(errno.EROFS) + + +def mount(backup, mountpoint, cache_dir, mount_time, foreground=True): + """Blocking call — mount `backup` at `mountpoint`. + + Meant to be run on a background thread. Returns once the filesystem is + unmounted (e.g. via `fusermount -u`/`umount` on `mountpoint`). + """ + fs = BackupFS(backup, cache_dir, mount_time) + FUSE(fs, mountpoint, nothreads=False, foreground=foreground, ro=True)