68 行
1.7 KiB
Python
可执行文件
68 行
1.7 KiB
Python
可执行文件
#!/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())
|