"""Deterministic archive primitives shared by host and optional plugin builders.""" import gzip import io import tarfile from pathlib import PurePosixPath def tarball(files): names = [name for name, _, _ in files] if len(set(names)) != len(names): raise ValueError("Duplicate package member") if any(PurePosixPath(n).is_absolute() or ".." in PurePosixPath(n).parts for n in names): raise ValueError("Unsafe package member") stream = io.BytesIO() with tarfile.open(fileobj=stream, mode="w", format=tarfile.GNU_FORMAT) as archive: directories = { str(parent) for name in names for parent in PurePosixPath(name).parents if str(parent) != "." } for name in sorted(directories): item = tarfile.TarInfo(name + "/") item.type, item.mode = tarfile.DIRTYPE, 0o755 item.uname = item.gname = "root" archive.addfile(item) for name, data, mode in sorted(files): item = tarfile.TarInfo(name) item.size, item.mode, item.uid, item.gid = len(data), mode, 0, 0 item.uname = item.gname = "root" archive.addfile(item, io.BytesIO(data)) return gzip.compress(stream.getvalue(), mtime=0) def ar_member(name, data): header = f"{name + '/':<16}{0:<12}{0:<6}{0:<6}{'100644':<8}{len(data):<10}`\n".encode() if len(header) != 60: raise ValueError("Invalid archive header") return header + data + (b"\n" if len(data) % 2 else b"") def package(controls, files): return ( b"!\n" + ar_member("debian-binary", b"2.0\n") + ar_member("control.tar.gz", tarball(controls)) + ar_member("data.tar.gz", tarball(files)) )