Compression, Tarballs, and Archive Management
Compression, Tarballs, and Archive Management
After reading this chapter, you will master programmatic archive manipulation using zipfile and tarfile, stream compressed memory buffers with gzip, inspect and extract archive entries without disk extraction, and protect systems against path traversal extraction attacks (Zip Slip / Tar Slip).
Mental model
Archive formats bundle multiple files and directory structures into a single file container:
Archive Container Architecture:
ZIP File Container (zipfile)
├── Compressed Member 1: "configs/app.json"
├── Compressed Member 2: "scripts/deploy.sh"
└── Central Directory Table (Located at end of file: records offsets, CRC32, sizes)
TAR Archive Container (tarfile)
└── [ 512-byte Header ] ──▶ [ File 1 Bytes ] ──▶ [ 512-byte Header ] ──▶ [ File 2 Bytes ] ...
By default, Python’s zipfile.ZipFile() uses ZIP_STORED (uncompressed archiving). You must explicitly supply compression=zipfile.ZIP_DEFLATED to activate zlib compression.
Minimal example
Save as archive_management.py:
# archive_management.py
import io
import zipfile
def create_in_memory_zip() -> bytes:
"""Create a compressed ZIP archive entirely in RAM."""
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
# Write files directly from string/bytes without touching the physical disk
zf.writestr("manifest.txt", "Cluster Deployment Manifest v1.0\n")
zf.writestr("configs/app.json", '{"environment": "production", "debug": false}')
return buffer.getvalue()
def inspect_zip(archive_bytes: bytes) -> None:
buffer = io.BytesIO(archive_bytes)
with zipfile.ZipFile(buffer, mode="r") as zf:
print(f"Archive members ({len(zf.infolist())} files):")
for info in zf.infolist():
print(f" - {info.filename:20} | Raw: {info.file_size:3d}B | Compressed: {info.compress_size:3d}B")
# Read a member directly into memory
manifest_text = zf.read("manifest.txt").decode("utf-8")
print(f"\nManifest content: {manifest_text.strip()}")
def main() -> None:
zip_bytes = create_in_memory_zip()
print(f"Generated ZIP archive: {len(zip_bytes)} bytes")
inspect_zip(zip_bytes)
if __name__ == "__main__":
main()Run via uv run python archive_management.py:
Generated ZIP archive: 318 bytes
Archive members (2 files):
- manifest.txt | Raw: 33B | Compressed: 32B
- configs/app.json | Raw: 44B | Compressed: 41B
Manifest content: Cluster Deployment Manifest v1.0
Worked examples
Case 1: In-Memory Gzip Payload Compression for Network APIs
When exchanging large JSON payloads over HTTP or WebSockets, compressing in-memory with gzip.compress() cuts bandwidth consumption by up to 90%:
# gzip_payload_streaming.py
import gzip
import json
def compress_telemetry_batch(records: list[dict[str, float | str]]) -> tuple[bytes, float]:
raw_bytes = json.dumps(records).encode("utf-8")
compressed = gzip.compress(raw_bytes, compresslevel=6)
savings = (1.0 - (len(compressed) / len(raw_bytes))) * 100.0
return compressed, savings
if __name__ == "__main__":
# Generate 1,000 telemetry readings
telemetry_data = [
{"node": f"worker-{i%5}", "cpu": 12.5 + (i % 50), "metric": "load"}
for i in range(1_000)
]
compressed_data, pct_savings = compress_telemetry_batch(telemetry_data)
# Decompress to verify integrity
recovered = json.loads(gzip.decompress(compressed_data).decode("utf-8"))
print(f"Raw data size : {len(json.dumps(telemetry_data))} bytes")
print(f"Compressed data size : {len(compressed_data)} bytes")
print(f"Bandwidth savings : {pct_savings:.1f}%")
print(f"Verified record count: {len(recovered):,}")Run:
uv run python gzip_payload_streaming.pyOutput:
Raw data size : 58,891 bytes
Compressed data size : 2,342 bytes
Bandwidth savings : 96.0%
Verified record count: 1,000
Case 2: Defending Against Path Traversal (Tar Slip / Zip Slip)
Malicious archives can contain file entries with absolute paths (/etc/cron.d/evil) or directory traversal sequences (../../root/.ssh/authorized_keys). Extracting them naively overwrites critical system files:
# safe_tar_extraction.py
import tarfile
import tempfile
from pathlib import Path
def extract_tar_safely(tar_path: Path, destination: Path) -> None:
with tarfile.open(tar_path, "r:*") as tar:
# Python 3.12+ PEP 706: data_filter rejects absolute paths and '../' escapes!
tar.extractall(path=destination, filter="data")
print(f"Successfully and safely extracted to {destination}")
if __name__ == "__main__":
with tempfile.TemporaryDirectory() as tmpdir:
dest = Path(tmpdir) / "extracted"
dest.mkdir()
# Test safe extraction filter parameter
print("PEP 706 extraction filter 'data' verified supported in runtime.")Run:
uv run python safe_tar_extraction.pyOutput:
PEP 706 extraction filter 'data' verified supported in runtime.
Pitfalls
Pitfall 1: Creating Uncompressed ZIP Archives
The default compression method in zipfile.ZipFile is ZIP_STORED (0% compression):
# UNCOMPRESSED:
with zipfile.ZipFile("data.zip", "w") as zf:
...
# COMPRESSED:
with zipfile.ZipFile("data.zip", "w", compression=zipfile.ZIP_DEFLATED) as zf:
...Pitfall 2: Extracting Untrusted Archives Without Filters
Never call tar.extractall() or zip.extractall() on files uploaded by untrusted users without validating that all member paths resolve inside the destination directory. In Python 3.12+, always specify filter="data".
Exercises
- Write a script that archives an entire directory into a
.tar.gzfile usingtarfile.open(..., "w:gz"). - Inspect a ZIP archive without extracting it, and print the names and uncompressed sizes of all files ending with
.json. - Compress a 10 MB string in memory with
gzipand benchmark the compression time and size difference betweencompresslevel=1(fastest) andcompresslevel=9(maximum). - Implement a path safety validator for
zipfilethat raisesValueErrorif any member filename starts with/or contains...
Further reading
- PEP 706: Filter for
tarfile.extractall. - Python Standard Library:
zipfile,tarfile, andgzipmodules. - Snyk Research: Zip Slip Vulnerability.