205 lines
7.8 KiB
Python
Executable File
205 lines
7.8 KiB
Python
Executable File
#!/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()
|