#!/usr/bin/env python3 """Dry-run or rebuild empty folder tree on an explicitly mounted new filesystem.""" import argparse import json import os import stat import subprocess import sys from pathlib import Path def fail(msg): sys.exit(f"ERROR: {msg}") def normalize_source_path(raw, strip_prefix=None): if not isinstance(raw, str) or not raw.startswith("/") or "\x00" in raw: fail(f"Invalid source directory path: {raw!r}") # Never silently normalize parent traversals or hidden whitespace. if any(part in (".", "..") for part in raw.split("/")): fail(f"Refusing dangerous path: {raw!r}") path = "/" + "/".join(part for part in raw.split("/") if part) if strip_prefix: if path == strip_prefix: return "/" if path.startswith(strip_prefix + "/"): path = path[len(strip_prefix):] else: fail(f"Path does not start with --strip-prefix {strip_prefix}: {path}") return path def load_rows(input_path, strip_prefix): rows = {} is_jsonl = input_path.suffix.lower() == ".jsonl" with input_path.open(encoding="utf-8-sig") as f: for no, raw in enumerate(f, 1): line = raw.rstrip("\r\n") if not line or (not is_jsonl and line.startswith("#")): continue if is_jsonl: try: record = json.loads(line) except json.JSONDecodeError as e: fail(f"Invalid JSONL at line {no}: {e}") if "_meta" in record: continue path = record.get("path") else: # Accept output from capture-tree.py, find -type d, and tree -dfi # --noreport. Do not accept the diagram/box-drawing variant. if line.startswith(("│", "├", "└", "|", "`")): fail(f"Line {no} looks like an indented tree diagram, not full paths") record = {"path": line} path = line path = normalize_source_path(path, strip_prefix) if path == "/": continue # Never chmod/chown the target mount's root. if path in rows: old = rows[path] if old.get("mode") is None and record.get("mode") is not None: rows[path] = record else: rows[path] = record return rows def verify_mount(target, expected_source): if not target.is_absolute() or str(target) == "/" or target.is_symlink(): fail("--target must be an absolute, non-symlink mountpoint other than /.") if not target.is_dir() or not os.path.ismount(target): fail(f"{target} is NOT itself a mounted filesystem. Nothing was created.") if os.statvfs(target).f_flag & os.ST_RDONLY: fail(f"{target} is mounted read-only.") r = subprocess.run(["findmnt", "-n", "-o", "SOURCE,FSTYPE", "--mountpoint", str(target)], capture_output=True, text=True) if r.returncode != 0 or not r.stdout.strip(): fail(f"Could not confirm mount source for {target}") mount_source, fs_type = r.stdout.strip().split(None, 1) print(f"Verified mount: {target} <- {mount_source} ({fs_type})") if expected_source and mount_source != expected_source: fail(f"Mount source is {mount_source}, expected {expected_source}.") def guarded_directory(target, rel, do_apply): """Create each component, refusing existing symlinks and non-directories.""" here = target created = 0 for component in rel.parts: if component in ("", "."): continue if component == "..": fail("Unexpected .. while creating directories") here = here / component try: st = here.lstat() except FileNotFoundError: if do_apply: here.mkdir(mode=0o755) created += 1 continue if stat.S_ISLNK(st.st_mode): fail(f"Refusing to follow symlink: {here}") if not stat.S_ISDIR(st.st_mode): fail(f"Path exists but is not a directory: {here}") return created def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("tree", type=Path, help="capture-tree.py JSONL or one-full-path-per-line .txt") parser.add_argument("--target", type=Path, default=Path("/storage"), help="Mountpoint on newly rebuilt array (default: /storage)") parser.add_argument("--strip-prefix", default=None, help="Strip the OLD mountpoint from text entries, e.g. /storage") parser.add_argument("--apply", action="store_true", help="Actually create the folders. Default: dry run, no changes") parser.add_argument("--restore-metadata", action="store_true", help="With --apply, use captured numeric UID/GID and mode where known") parser.add_argument("--expect-source", default=None, help="With --apply, require findmnt to report this exact mount SOURCE") args = parser.parse_args() if args.restore_metadata and not args.apply: fail("--restore-metadata requires --apply") if not args.tree.is_file(): fail(f"Input file not found: {args.tree}") if args.strip_prefix: args.strip_prefix = normalize_source_path(args.strip_prefix) if args.strip_prefix == "/": fail("--strip-prefix / is redundant; omit the option") target = args.target.absolute() if target.is_symlink() or str(target) == "/": fail("Refusing a symlink target or filesystem root /.") if args.apply: if os.geteuid() != 0: fail("Run as root when applying directory creation.") verify_mount(target, args.expect_source) rows = load_rows(args.tree, args.strip_prefix) if not rows: fail("No directory paths found in input. Nothing to do.") # Parents of input paths are created automatically; source folder names are # treated as filesystem-relative, NOT as host paths. ordered = sorted(rows, key=lambda path: (path.count("/"), path)) counts = {"input": len(ordered), "created": 0, "existing": 0, "metadata": 0} print("MODE:", "APPLY" if args.apply else "DRY RUN") print("Target:", target) print("Input directories:", len(ordered)) print("Top-level directories:", ", ".join(sorted({p.split("/")[1] for p in ordered}))) print("Sample paths:") for path in ordered[:12]: print(f" {target / path.lstrip('/')}") if len(ordered) > 12: print(f" ... and {len(ordered)-12:,} more") if not args.apply: print("\nDRY RUN ONLY: no directories created. Review your tree, then add --apply.") return # Apply phase: create top-down, with explicit symlink checks. failures = [] for path in ordered: rel = Path(path.lstrip("/")) dest = target / rel existed = dest.is_dir() and not dest.is_symlink() try: guarded_directory(target, rel, do_apply=True) except (OSError, SystemExit) as e: failures.append(f"MKDIR {path}: {e}") print(f"FAILED {path}: {e}", file=sys.stderr) continue if existed: counts["existing"] += 1 else: counts["created"] += 1 if args.restore_metadata: # Work bottom-up so restrictive permissions can't block child creation. for path in reversed(ordered): info = rows[path] dest = target / path.lstrip("/") if not dest.is_dir() or dest.is_symlink(): continue uid, gid, mode = (info.get("uid"), info.get("gid"), info.get("mode")) if uid is None or gid is None or mode is None: continue try: os.chown(dest, int(uid), int(gid), follow_symlinks=False) os.chmod(dest, int(mode)) # symlinks explicitly rejected above counts["metadata"] += 1 except OSError as e: failures.append(f"METADATA {path}: {e}") print(f"METADATA FAILED {path}: {e}", file=sys.stderr) print(f"\nInput: {counts['input']:,}; newly created: {counts['created']:,}; " f"already present: {counts['existing']:,}; metadata restored: {counts['metadata']:,}; " f"failures: {len(failures):,}") if failures: sys.exit(1) if __name__ == "__main__": main()