#!/usr/bin/env python3 """Export surviving ext4 directories from an UNMOUNTED damaged LV using debugfs. Does not read file contents or write to the source. Unreachable trees must be supplemented manually; no software can infer names below missing dir metadata. """ import argparse import json import os import re import stat import subprocess import sys from pathlib import Path, PurePosixPath DEFAULT_LV = "/dev/mapper/plex--new-plex--new" DEFAULT_OUTPUT = "/mnt/recovery/plex-directory-tree.jsonl" INODE_RE = re.compile(r"\bInode:\s*(\d+)\s+Type:\s*([^\s]+)") MODE_RE = re.compile(r"\bMode:\s*([0-7]+)") OWNER_RE = re.compile(r"\bUser:\s*(\d+)\s+Group:\s*(\d+)") def fail(message): sys.exit(f"ERROR: {message}") def normalize_path(value): if not value.startswith("/"): fail(f"Source directories must be absolute filesystem paths: {value!r}") # Avoid guessing about special path syntax; '/' is meaningful within debugfs. parts = value.split("/") if any(part in (".", "..") for part in parts): fail(f"Source paths must not contain '.' or '..': {value!r}") return "/" + "/".join(p for p in parts if p) def quote_debugfs(value): return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' def debugfs(lv, command): env = os.environ.copy() env["DEBUGFS_PAGER"] = "cat" return subprocess.run( ["debugfs", "-c", "-R", command, lv], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, errors="replace", ) def get_inode(lv, path): r = debugfs(lv, f"stat {quote_debugfs(path)}") match = INODE_RE.search(r.stdout) if not match or match.group(2) != "directory": return None, r mode_match = MODE_RE.search(r.stdout) owner_match = OWNER_RE.search(r.stdout) return { "inode": int(match.group(1)), "mode": int(mode_match.group(1), 8) if mode_match else None, "uid": int(owner_match.group(1)) if owner_match else None, "gid": int(owner_match.group(2)) if owner_match else None, }, r def list_dir(lv, inode): r = debugfs(lv, f"ls -p <{inode}>") entries = [] bad_lines = [] for raw in r.stdout.splitlines(): if not raw.startswith("/"): continue # debugfs ls -p: /inode/mode/uid/gid/name/size/ # Linux names cannot contain a slash. The last empty field is expected. fields = raw.split("/") if len(fields) < 7: bad_lines.append(raw) continue try: ino = int(fields[1]) mode = int(fields[2], 8) uid = int(fields[3]) gid = int(fields[4]) except ValueError: bad_lines.append(raw) continue name = fields[5] if name in (".", "..") or ino == 0: continue entries.append((ino, mode, uid, gid, name)) # debugfs always writes a version banner to stderr; strip that banner. stderr = "\n".join( line for line in r.stderr.splitlines() if not line.startswith("debugfs ") ).strip() return entries, stderr, bad_lines def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source", action="append", default=[], help="Existing source directory to traverse; repeatable (default: /)") parser.add_argument("--include", action="append", default=[], help="Add an expected directory even if its inode is unreadable; repeatable") parser.add_argument("--lv", default=DEFAULT_LV) parser.add_argument("--output", default=DEFAULT_OUTPUT, help="Write JSONL plus a .txt folder-path list and .warnings.txt") args = parser.parse_args() if os.geteuid() != 0: fail("Run capture as root so debugfs can access the block device.") ro = subprocess.run(["blockdev", "--getro", args.lv], capture_output=True, text=True) if ro.returncode or ro.stdout.strip() != "1": fail("Source LV must be kernel read-only. Set it read-only before capture.") output = Path(args.output).absolute() if output.suffix != ".jsonl": fail("--output must end with .jsonl (two companion files are generated).") if str(output).startswith("/mnt/recovery/") and not os.path.ismount("/mnt/recovery"): fail("/mnt/recovery isn't mounted. Refusing to write the snapshot to an unmounted recovery drive.") output.parent.mkdir(parents=True, exist_ok=True) # A temporary JSONL is renamed only when the traversal is complete. tmp = Path(str(output) + ".incomplete") txt_path = output.with_suffix(".txt") warning_path = output.with_suffix(".warnings.txt") sources = [normalize_path(s) for s in (args.source or ["/"])] additions = [normalize_path(s) for s in args.include] records = {} warnings = [] visited = set() pending = [] def add_record(path, inode=None, mode=None, uid=None, gid=None, origin="debugfs"): if path == "/": return if path not in records: records[path] = { "path": path, "inode": inode, "mode": mode, "uid": uid, "gid": gid, "origin": origin, } elif origin == "debugfs": records[path].update( inode=inode, mode=mode, uid=uid, gid=gid, origin=origin, ) for path in additions: add_record(path, origin="manual") for path in sources: info, r = get_inode(args.lv, path) if info is None: warnings.append(f"UNREADABLE_SOURCE {path}: {r.stderr.strip()} {r.stdout.strip()}") add_record(path, origin="unreadable-source") continue add_record(path, **info) pending.append((info["inode"], path)) while pending: inode, current_path = pending.pop() if inode in visited: continue visited.add(inode) entries, stderr, bad_lines = list_dir(args.lv, inode) if stderr: warnings.append(f"DIR_READ_WARNING {current_path} (inode {inode}): {stderr}") for raw in bad_lines: warnings.append(f"UNPARSEABLE_ENTRY {current_path}: {raw}") for child_ino, mode, uid, gid, name in entries: path = current_path.rstrip("/") + "/" + name if stat.S_ISDIR(mode): add_record(path, inode=child_ino, mode=stat.S_IMODE(mode), uid=uid, gid=gid) pending.append((child_ino, path)) elif stat.S_IFMT(mode) == 0: warnings.append( f"AMBIGUOUS_ENTRY {path} (inode {child_ino}): bad/missing type; " "might be a lost directory. Manually include if required." ) if len(visited) % 200 == 0: print(f"Visited {len(visited):,} directories; found {len(records):,} paths", flush=True) with tmp.open("w", encoding="utf-8") as f: f.write(json.dumps({"_meta": { "format": "plex-directory-tree-v1", "lv": args.lv, "sources": sources, "manual_includes": additions, "warnings": len(warnings), "note": "Only reachable ext4 metadata is exported; check warnings and missing share roots." }}, ensure_ascii=False) + "\n") for path in sorted(records): f.write(json.dumps(records[path], ensure_ascii=False) + "\n") f.flush() os.fsync(f.fileno()) tmp.replace(output) txt_path.write_text("".join(p + "\n" for p in sorted(records)), encoding="utf-8") warning_path.write_text("\n".join(warnings) + ("\n" if warnings else ""), encoding="utf-8") print(f"Exported {len(records):,} directories ({len(visited):,} visited)") print(f"Manifest: {output}\nPaths: {txt_path}\nWarnings: {warning_path}") if warnings: print(f"WARNING: {len(warnings)} warnings / ambiguous entries. Review before rebuilding.") if __name__ == "__main__": main()