54 lines
2.4 KiB
Python
54 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Protocol v1 ZIP extractor. Exact entries; unknown files remain in the parent."""
|
|
import json
|
|
import pathlib
|
|
import stat
|
|
import sys
|
|
import zipfile
|
|
|
|
def run(request):
|
|
if request["protocol"] != 1:
|
|
raise ValueError("unsupported protocol")
|
|
source = pathlib.Path(request["input_dir"])
|
|
output = pathlib.Path(request["output_dir"])
|
|
files, warnings = [], []
|
|
budget = request["workspace_bytes"] // 2
|
|
used = 0
|
|
archives = 0
|
|
for path in sorted(source.rglob("*")):
|
|
if not path.is_file() or path.is_symlink():
|
|
continue
|
|
if not zipfile.is_zipfile(path):
|
|
warnings.append(f"Opaque input retained in parent: {path.relative_to(source)}")
|
|
continue
|
|
archives += 1
|
|
prefix = pathlib.PurePosixPath(str(path.relative_to(source)) + ".entries")
|
|
with zipfile.ZipFile(path) as archive:
|
|
for entry in archive.infolist():
|
|
name = pathlib.PurePosixPath(entry.filename)
|
|
if name.is_absolute() or ".." in name.parts or "\\" in entry.filename or not name.parts:
|
|
raise ValueError("unsafe ZIP entry path")
|
|
if stat.S_ISLNK(entry.external_attr >> 16):
|
|
warnings.append(f"ZIP symlink left opaque in parent: {entry.filename}")
|
|
continue
|
|
dest = output.joinpath(prefix, name)
|
|
if entry.is_dir():
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
with archive.open(entry) as reader, dest.open("xb") as writer:
|
|
while chunk := reader.read(1024 * 1024):
|
|
used += len(chunk)
|
|
if used > budget:
|
|
raise ValueError("ZIP expansion exceeds workspace allowance")
|
|
writer.write(chunk)
|
|
files.append(dest.relative_to(output).as_posix())
|
|
if not archives:
|
|
raise ValueError("no supported ZIP inputs found")
|
|
return {"protocol": 1, "layer": "extracted", "coverage": "partial" if warnings else "complete", "files": files, "warnings": warnings}
|
|
|
|
if __name__ == "__main__":
|
|
request = json.loads(pathlib.Path(sys.argv[1]).read_text())
|
|
result = run(request)
|
|
pathlib.Path(request["result_file"]).write_text(json.dumps(result))
|