230 lines
7.7 KiB
Python
230 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Independent post-apply verification for the non-work recovery import.
|
|
|
|
Read-only checks:
|
|
- all 18 staging sources still exist and retain approved SHA-256 hashes
|
|
- all 14 COPY_NEW destinations exist and match approved SHA-256 hashes
|
|
- all 4 SKIP_EXACT_DUPLICATE target files still exist and match
|
|
- recovery folders contain exactly the 14 planned files and no extras
|
|
- apply report contains 14 COPIED_AND_VERIFIED + 4 SKIPPED_VERIFIED_DUPLICATE
|
|
- no apply failures are recorded
|
|
|
|
No files or Syncthing settings are modified.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
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")
|
|
|
|
|
|
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 read_tsv(path: Path) -> list[dict[str, str]]:
|
|
with path.open("r", encoding="utf-8", newline="") as f:
|
|
return list(csv.DictReader(f, delimiter="\t"))
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--manifest", required=True, type=Path)
|
|
ap.add_argument("--apply-report", required=True, type=Path)
|
|
args = ap.parse_args()
|
|
|
|
manifest = args.manifest.resolve()
|
|
apply_report = args.apply_report.resolve()
|
|
|
|
rows = read_tsv(manifest)
|
|
report_rows = read_tsv(apply_report)
|
|
|
|
problems: list[str] = []
|
|
|
|
print("=== INDEPENDENT POST-APPLY VERIFICATION ===")
|
|
print(f"Manifest rows: {len(rows)}")
|
|
print(f"Apply report rows: {len(report_rows)}")
|
|
print()
|
|
|
|
if len(rows) != 18:
|
|
problems.append(f"Expected 18 manifest rows, found {len(rows)}.")
|
|
if len(report_rows) != 18:
|
|
problems.append(f"Expected 18 apply-report rows, found {len(report_rows)}.")
|
|
|
|
# Verify apply report statuses independently.
|
|
report_counts = Counter(r.get("apply_status", "") for r in report_rows)
|
|
expected_report = {
|
|
"COPIED_AND_VERIFIED": 14,
|
|
"SKIPPED_VERIFIED_DUPLICATE": 4,
|
|
}
|
|
|
|
for status, expected in expected_report.items():
|
|
actual = report_counts.get(status, 0)
|
|
if actual != expected:
|
|
problems.append(
|
|
f"Apply report status {status}: expected {expected}, found {actual}."
|
|
)
|
|
|
|
unexpected_statuses = {
|
|
k: v for k, v in report_counts.items()
|
|
if k not in expected_report and v
|
|
}
|
|
if unexpected_statuses:
|
|
problems.append(
|
|
"Unexpected/non-success apply statuses: "
|
|
+ ", ".join(f"{k}={v}" for k, v in sorted(unexpected_statuses.items()))
|
|
)
|
|
|
|
# Verify sources, destinations, and duplicate targets.
|
|
verified_sources = 0
|
|
verified_destinations = 0
|
|
verified_duplicates = 0
|
|
planned_destinations: set[Path] = set()
|
|
|
|
for i, row in enumerate(rows, 1):
|
|
action = row["dry_run_action"].strip()
|
|
expected = row["sha256"].strip().lower()
|
|
source = Path(row["source_path"])
|
|
|
|
# Source must still be intact.
|
|
if not source.is_file():
|
|
problems.append(f"Source missing: {source}")
|
|
else:
|
|
actual = sha256_file(source)
|
|
if actual.lower() != expected:
|
|
problems.append(
|
|
f"Source hash mismatch: {source}\n"
|
|
f" expected {expected}\n"
|
|
f" actual {actual}"
|
|
)
|
|
else:
|
|
verified_sources += 1
|
|
|
|
if action == "COPY_NEW":
|
|
dest = Path(row["proposed_destination"])
|
|
planned_destinations.add(dest.resolve(strict=False))
|
|
|
|
if not dest.is_file():
|
|
problems.append(f"Copied destination missing: {dest}")
|
|
print(f"[{i:>2}/18] COPY MISSING {row['filename']}")
|
|
continue
|
|
|
|
actual = sha256_file(dest)
|
|
if actual.lower() != expected:
|
|
problems.append(
|
|
f"Destination hash mismatch: {dest}\n"
|
|
f" expected {expected}\n"
|
|
f" actual {actual}"
|
|
)
|
|
print(f"[{i:>2}/18] COPY HASH_FAIL {row['filename']}")
|
|
else:
|
|
verified_destinations += 1
|
|
print(f"[{i:>2}/18] COPY VERIFIED {row['filename']}")
|
|
|
|
elif action == "SKIP_EXACT_DUPLICATE":
|
|
target_text = row["proposed_destination"].strip()
|
|
targets = [Path(p) for p in target_text.split(" | ") if p]
|
|
if not targets:
|
|
problems.append(
|
|
f"No duplicate target listed for {row['relative_path']}"
|
|
)
|
|
print(f"[{i:>2}/18] SKIP NO_TARGET {row['filename']}")
|
|
continue
|
|
|
|
row_ok = True
|
|
for target in targets:
|
|
if not target.is_file():
|
|
problems.append(f"Duplicate target missing: {target}")
|
|
row_ok = False
|
|
continue
|
|
actual = sha256_file(target)
|
|
if actual.lower() != expected:
|
|
problems.append(
|
|
f"Duplicate target hash mismatch: {target}\n"
|
|
f" expected {expected}\n"
|
|
f" actual {actual}"
|
|
)
|
|
row_ok = False
|
|
|
|
if row_ok:
|
|
verified_duplicates += 1
|
|
print(f"[{i:>2}/18] SKIP VERIFIED {row['filename']}")
|
|
else:
|
|
print(f"[{i:>2}/18] SKIP VERIFY_FAIL {row['filename']}")
|
|
|
|
else:
|
|
problems.append(f"Unexpected action in manifest: {action!r}")
|
|
|
|
# Recovery folder contents must equal exactly the 14 planned destinations.
|
|
print()
|
|
print("=== RECOVERY FOLDER INVENTORY ===")
|
|
|
|
actual_files: set[Path] = set()
|
|
for base in (PRIVATE_BASE, INBOX_BASE):
|
|
if not base.is_dir():
|
|
problems.append(f"Recovery folder missing: {base}")
|
|
print(f"{base}: MISSING")
|
|
continue
|
|
|
|
files = sorted(
|
|
p for p in base.rglob("*")
|
|
if p.is_file() and not p.is_symlink()
|
|
)
|
|
print(f"{base}: {len(files)} file(s)")
|
|
actual_files.update(p.resolve(strict=False) for p in files)
|
|
|
|
extra = sorted(actual_files - planned_destinations)
|
|
missing = sorted(planned_destinations - actual_files)
|
|
|
|
if extra:
|
|
problems.append(
|
|
"Unexpected extra files in recovery folders:\n "
|
|
+ "\n ".join(str(p) for p in extra)
|
|
)
|
|
if missing:
|
|
problems.append(
|
|
"Planned files missing from recovery folders:\n "
|
|
+ "\n ".join(str(p) for p in missing)
|
|
)
|
|
|
|
print()
|
|
print("=== SUMMARY ===")
|
|
print(f"Sources verified: {verified_sources}/18")
|
|
print(f"Copied destinations verified: {verified_destinations}/14")
|
|
print(f"Skipped duplicates verified: {verified_duplicates}/4")
|
|
print(f"Recovery files found: {len(actual_files)}/14")
|
|
print(f"Extra recovery files: {len(extra)}")
|
|
print(f"Problems: {len(problems)}")
|
|
|
|
if problems:
|
|
print()
|
|
print("POST-APPLY VERIFY FAILED — KEEP SYNCTHING PAUSED.")
|
|
for item in problems:
|
|
print(" -", item)
|
|
return 2
|
|
|
|
print()
|
|
print("POST-APPLY VERIFY PASS")
|
|
print("The 14 copied files and 4 skipped duplicates are independently verified.")
|
|
print("All 18 staging source files remain intact.")
|
|
print("No unexpected files exist in the two recovery folders.")
|
|
print("This script made no filesystem or Syncthing changes.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|