407 lines
13 KiB
Python
407 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Controlled APPLY for the approved non-work recovery routing plan.
|
|
|
|
This script intentionally:
|
|
- copies only rows marked COPY_NEW
|
|
- never moves or deletes source files
|
|
- never overwrites an existing destination
|
|
- verifies all source hashes immediately before applying
|
|
- re-verifies all copied files after writing
|
|
- re-verifies SKIP_EXACT_DUPLICATE targets
|
|
- only writes inside the two approved recovery folders
|
|
- writes an apply report back into the staging MANIFESTS directory
|
|
|
|
Syncthing for 03_PRIVAT and 01_INBOX must already be paused before running.
|
|
|
|
Rollback model:
|
|
The two recovery folders were confirmed absent before APPLY. Therefore every
|
|
file created there belongs to this import. Do NOT delete them automatically on
|
|
failure; inspect the apply report first. The original staging/source files are
|
|
never deleted.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
import os
|
|
import shutil
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
|
|
CHUNK = 1024 * 1024
|
|
|
|
PRIVATE_BASE = Path("/data/03_PRIVAT/01_INBOX/Recovered_Work-PC_2026-09-14")
|
|
INBOX_BASE = Path("/data/01_INBOX/Recovered_Work-PC_2026-09-14")
|
|
ALLOWED_COPY_BASES = (PRIVATE_BASE, INBOX_BASE)
|
|
ALLOWED_SKIP_ROOTS = (Path("/data/03_PRIVAT"), Path("/data/01_INBOX"))
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
h = hashlib.sha256()
|
|
with path.open("rb") as f:
|
|
for chunk in iter(lambda: f.read(CHUNK), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def is_under(path: Path, root: Path) -> bool:
|
|
try:
|
|
path.resolve(strict=False).relative_to(root.resolve(strict=False))
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def read_manifest(path: Path) -> list[dict[str, str]]:
|
|
with path.open("r", encoding="utf-8", newline="") as f:
|
|
rows = list(csv.DictReader(f, delimiter="\t"))
|
|
|
|
if not rows:
|
|
raise RuntimeError("Routing manifest is empty.")
|
|
|
|
required = {
|
|
"effective_scope",
|
|
"relative_path",
|
|
"filename",
|
|
"source_path",
|
|
"sha256",
|
|
"audit_status",
|
|
"dry_run_action",
|
|
"proposed_destination",
|
|
"source_delete",
|
|
}
|
|
missing = required - set(rows[0])
|
|
if missing:
|
|
raise RuntimeError(
|
|
"Routing manifest is missing columns: "
|
|
+ ", ".join(sorted(missing))
|
|
)
|
|
return rows
|
|
|
|
|
|
def validate_before_mutation(rows: list[dict[str, str]]) -> list[str]:
|
|
problems: list[str] = []
|
|
seen_destinations: set[Path] = set()
|
|
counts = Counter(row["dry_run_action"].strip() for row in rows)
|
|
|
|
if len(rows) != 18:
|
|
problems.append(f"Expected 18 manifest rows, found {len(rows)}.")
|
|
if counts["COPY_NEW"] != 14:
|
|
problems.append(f"Expected 14 COPY_NEW rows, found {counts['COPY_NEW']}.")
|
|
if counts["SKIP_EXACT_DUPLICATE"] != 4:
|
|
problems.append(
|
|
"Expected 4 SKIP_EXACT_DUPLICATE rows, found "
|
|
f"{counts['SKIP_EXACT_DUPLICATE']}."
|
|
)
|
|
unexpected = set(counts) - {"COPY_NEW", "SKIP_EXACT_DUPLICATE"}
|
|
if unexpected:
|
|
problems.append(
|
|
"Unexpected actions present: " + ", ".join(sorted(unexpected))
|
|
)
|
|
|
|
# The approved recovery folders must still be absent at APPLY start.
|
|
for base in ALLOWED_COPY_BASES:
|
|
if os.path.lexists(base):
|
|
problems.append(
|
|
f"Approved recovery folder already exists before APPLY: {base}"
|
|
)
|
|
|
|
for row in rows:
|
|
rel = row["relative_path"]
|
|
action = row["dry_run_action"].strip()
|
|
source = Path(row["source_path"])
|
|
expected = row["sha256"].strip().lower()
|
|
destination_text = row["proposed_destination"].strip()
|
|
|
|
if row["source_delete"].strip().upper() != "NO":
|
|
problems.append(f"{rel}: source_delete is not NO.")
|
|
|
|
if not source.is_file():
|
|
problems.append(f"{rel}: source missing: {source}")
|
|
continue
|
|
|
|
try:
|
|
actual = sha256_file(source)
|
|
except Exception as exc:
|
|
problems.append(
|
|
f"{rel}: source hash error: {type(exc).__name__}: {exc}"
|
|
)
|
|
continue
|
|
|
|
if actual.lower() != expected:
|
|
problems.append(
|
|
f"{rel}: source hash mismatch; expected {expected}, got {actual}"
|
|
)
|
|
|
|
if action == "COPY_NEW":
|
|
if not destination_text:
|
|
problems.append(f"{rel}: COPY_NEW has no destination.")
|
|
continue
|
|
|
|
dest = Path(destination_text)
|
|
if not any(is_under(dest, base) for base in ALLOWED_COPY_BASES):
|
|
problems.append(
|
|
f"{rel}: destination outside approved recovery folders: {dest}"
|
|
)
|
|
continue
|
|
|
|
norm = dest.resolve(strict=False)
|
|
if norm in seen_destinations:
|
|
problems.append(f"{rel}: duplicate proposed destination: {dest}")
|
|
seen_destinations.add(norm)
|
|
|
|
if os.path.lexists(dest):
|
|
problems.append(f"{rel}: destination already exists: {dest}")
|
|
|
|
elif action == "SKIP_EXACT_DUPLICATE":
|
|
targets = [Path(p) for p in destination_text.split(" | ") if p]
|
|
if not targets:
|
|
problems.append(f"{rel}: duplicate target missing.")
|
|
continue
|
|
|
|
for target in targets:
|
|
if not any(is_under(target, root) for root in ALLOWED_SKIP_ROOTS):
|
|
problems.append(
|
|
f"{rel}: duplicate target outside approved roots: {target}"
|
|
)
|
|
continue
|
|
if not target.is_file():
|
|
problems.append(f"{rel}: duplicate target missing: {target}")
|
|
continue
|
|
try:
|
|
target_hash = sha256_file(target)
|
|
except Exception as exc:
|
|
problems.append(
|
|
f"{rel}: duplicate target hash error {target}: "
|
|
f"{type(exc).__name__}: {exc}"
|
|
)
|
|
continue
|
|
if target_hash.lower() != expected:
|
|
problems.append(
|
|
f"{rel}: duplicate target hash mismatch: {target}"
|
|
)
|
|
|
|
return problems
|
|
|
|
|
|
def copy_exclusive(src: Path, dest: Path) -> None:
|
|
"""
|
|
Copy bytes with exclusive creation so an existing file can never be
|
|
overwritten. Metadata is copied afterwards.
|
|
"""
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 'xb' maps to O_CREAT|O_EXCL semantics.
|
|
try:
|
|
with src.open("rb") as rf, dest.open("xb") as wf:
|
|
for chunk in iter(lambda: rf.read(CHUNK), b""):
|
|
wf.write(chunk)
|
|
wf.flush()
|
|
os.fsync(wf.fileno())
|
|
shutil.copystat(src, dest, follow_symlinks=False)
|
|
except Exception:
|
|
# If this invocation created an incomplete destination, remove only
|
|
# that incomplete file. Never touch a pre-existing file because 'xb'
|
|
# would have failed before opening it.
|
|
try:
|
|
if dest.exists():
|
|
dest.unlink()
|
|
except Exception:
|
|
pass
|
|
raise
|
|
|
|
|
|
def write_report(path: Path, rows: list[dict[str, str]]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fields = [
|
|
"effective_scope",
|
|
"relative_path",
|
|
"filename",
|
|
"source_path",
|
|
"destination_path",
|
|
"sha256",
|
|
"planned_action",
|
|
"apply_status",
|
|
"postcopy_sha256",
|
|
"error",
|
|
]
|
|
with path.open("w", encoding="utf-8", newline="") as f:
|
|
w = csv.DictWriter(f, fieldnames=fields, delimiter="\t")
|
|
w.writeheader()
|
|
w.writerows(rows)
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--manifest", required=True, type=Path)
|
|
ap.add_argument("--report", required=True, type=Path)
|
|
args = ap.parse_args()
|
|
|
|
manifest = args.manifest.resolve()
|
|
report = args.report.resolve()
|
|
rows = read_manifest(manifest)
|
|
|
|
print("=== NON-WORK CONTROLLED APPLY ===")
|
|
print(f"Manifest: {manifest}")
|
|
print(f"Report: {report}")
|
|
print()
|
|
|
|
problems = validate_before_mutation(rows)
|
|
if problems:
|
|
print("PRE-APPLY VALIDATION FAILED — NOTHING WAS COPIED")
|
|
for item in problems:
|
|
print(" -", item)
|
|
return 2
|
|
|
|
print("PRE-APPLY VALIDATION PASS")
|
|
print("Creating only the two approved recovery folders and copying 14 files.")
|
|
print("No source file will be moved or deleted.")
|
|
print()
|
|
|
|
results: list[dict[str, str]] = []
|
|
fatal = False
|
|
|
|
for i, row in enumerate(rows, 1):
|
|
action = row["dry_run_action"].strip()
|
|
src = Path(row["source_path"])
|
|
dest_text = row["proposed_destination"].strip()
|
|
expected = row["sha256"].strip().lower()
|
|
|
|
result = {
|
|
"effective_scope": row["effective_scope"],
|
|
"relative_path": row["relative_path"],
|
|
"filename": row["filename"],
|
|
"source_path": str(src),
|
|
"destination_path": dest_text,
|
|
"sha256": expected,
|
|
"planned_action": action,
|
|
"apply_status": "",
|
|
"postcopy_sha256": "",
|
|
"error": "",
|
|
}
|
|
|
|
if action == "SKIP_EXACT_DUPLICATE":
|
|
result["apply_status"] = "SKIPPED_VERIFIED_DUPLICATE"
|
|
results.append(result)
|
|
print(
|
|
f"[{i:>2}/{len(rows)}] SKIP {row['filename']} "
|
|
"(verified exact duplicate)"
|
|
)
|
|
continue
|
|
|
|
dest = Path(dest_text)
|
|
|
|
try:
|
|
copy_exclusive(src, dest)
|
|
post = sha256_file(dest)
|
|
result["postcopy_sha256"] = post
|
|
|
|
if post.lower() != expected:
|
|
result["apply_status"] = "POSTCOPY_HASH_MISMATCH"
|
|
result["error"] = (
|
|
f"Expected {expected}, got {post}; destination retained "
|
|
"for inspection."
|
|
)
|
|
fatal = True
|
|
print(
|
|
f"[{i:>2}/{len(rows)}] FAIL {row['filename']} "
|
|
"(post-copy hash mismatch)"
|
|
)
|
|
results.append(result)
|
|
break
|
|
|
|
result["apply_status"] = "COPIED_AND_VERIFIED"
|
|
print(
|
|
f"[{i:>2}/{len(rows)}] COPY {row['filename']} VERIFIED"
|
|
)
|
|
|
|
except Exception as exc:
|
|
result["apply_status"] = "COPY_ERROR"
|
|
result["error"] = f"{type(exc).__name__}: {exc}"
|
|
fatal = True
|
|
print(
|
|
f"[{i:>2}/{len(rows)}] FAIL {row['filename']} "
|
|
f"({type(exc).__name__})"
|
|
)
|
|
results.append(result)
|
|
break
|
|
|
|
results.append(result)
|
|
|
|
# If execution stopped early, record untouched remaining rows.
|
|
processed = len(results)
|
|
if processed < len(rows):
|
|
for row in rows[processed:]:
|
|
results.append({
|
|
"effective_scope": row["effective_scope"],
|
|
"relative_path": row["relative_path"],
|
|
"filename": row["filename"],
|
|
"source_path": row["source_path"],
|
|
"destination_path": row["proposed_destination"],
|
|
"sha256": row["sha256"],
|
|
"planned_action": row["dry_run_action"],
|
|
"apply_status": "NOT_ATTEMPTED_AFTER_FAILURE",
|
|
"postcopy_sha256": "",
|
|
"error": "",
|
|
})
|
|
|
|
write_report(report, results)
|
|
|
|
print()
|
|
print("=== APPLY SUMMARY ===")
|
|
counts = Counter(r["apply_status"] for r in results)
|
|
print(f"COPIED_AND_VERIFIED: {counts['COPIED_AND_VERIFIED']}")
|
|
print(
|
|
f"SKIPPED_VERIFIED_DUPLICATE: {counts['SKIPPED_VERIFIED_DUPLICATE']}"
|
|
)
|
|
print(f"COPY_ERROR: {counts['COPY_ERROR']}")
|
|
print(f"POSTCOPY_HASH_MISMATCH: {counts['POSTCOPY_HASH_MISMATCH']}")
|
|
print(
|
|
f"NOT_ATTEMPTED_AFTER_FAILURE:{counts['NOT_ATTEMPTED_AFTER_FAILURE']}"
|
|
)
|
|
print(f"Apply report: {report}")
|
|
|
|
if fatal:
|
|
print()
|
|
print("APPLY INCOMPLETE — KEEP SYNCTHING PAUSED.")
|
|
print("Do not delete recovery folders automatically; inspect the report first.")
|
|
return 3
|
|
|
|
# Final independent pass over all 14 created files.
|
|
verify_problems = []
|
|
for row in rows:
|
|
if row["dry_run_action"].strip() != "COPY_NEW":
|
|
continue
|
|
dest = Path(row["proposed_destination"])
|
|
expected = row["sha256"].strip().lower()
|
|
if not dest.is_file():
|
|
verify_problems.append(f"Missing after apply: {dest}")
|
|
continue
|
|
actual = sha256_file(dest)
|
|
if actual.lower() != expected:
|
|
verify_problems.append(
|
|
f"Hash mismatch after apply: {dest} expected {expected} got {actual}"
|
|
)
|
|
|
|
print()
|
|
if verify_problems:
|
|
print("FINAL VERIFY FAILED — KEEP SYNCTHING PAUSED.")
|
|
for item in verify_problems:
|
|
print(" -", item)
|
|
return 4
|
|
|
|
print("APPLY PASS")
|
|
print("14 new files were copied and SHA-256 verified.")
|
|
print("4 exact duplicates were skipped.")
|
|
print("0 source files were moved or deleted.")
|
|
print("KEEP SYNCTHING PAUSED until the separate post-apply verification step.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|