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