验证(08-23-isolated-backup-restore-drill): 固化隔离恢复runner合同

scripts/qa/validate-offhost-shadow-restore-attachments.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+import base64
+import hashlib
+import os
+import pathlib
+import sys
+
+
+def main() -> int:
+    if len(sys.argv) != 2:
+        return 2
+
+    root = os.path.realpath(sys.argv[1])
+    total = 0
+    mismatch = 0
+    s3_count = 0
+
+    for raw in sys.stdin:
+        raw = raw.rstrip("\n")
+        if not raw:
+            continue
+        parts = raw.split("\t")
+        if len(parts) != 3:
+            mismatch += 1
+            continue
+
+        total += 1
+        try:
+            storage_path = base64.b64decode(parts[0], validate=True).decode("utf-8")
+        except (ValueError, UnicodeDecodeError):
+            mismatch += 1
+            continue
+
+        if storage_path.startswith("s3:"):
+            s3_count += 1
+            mismatch += 1
+            continue
+
+        pure_path = pathlib.PurePosixPath(storage_path)
+        if not storage_path or pure_path.is_absolute() or ".." in pure_path.parts:
+            mismatch += 1
+            continue
+
+        target = os.path.realpath(os.path.join(root, *pure_path.parts))
+        if not target.startswith(root + os.sep) or not os.path.isfile(target):
+            mismatch += 1
+            continue
+
+        expected_size, expected_sha = parts[1], parts[2].lower()
+        if expected_size and str(os.path.getsize(target)) != expected_size:
+            mismatch += 1
+            continue
+
+        if expected_sha:
+            digest = hashlib.sha256()
+            with open(target, "rb") as handle:
+                for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+                    digest.update(chunk)
+            if digest.hexdigest() != expected_sha:
+                mismatch += 1
+
+    print(total, mismatch, s3_count)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())