Create Reproducible gzip Files with Python: Control Headers and Verify Hashes

Compress the same CSV twice and its decompressed contents may match while the .gz SHA-256 changes. That creates unnecessary changes in backup deduplication, build caches, and artifact comparisons.

This tutorial uses Python’s standard library to control gzip headers, stream into a new file, and verify both compressed bytes and decompressed contents. It is for beginner-to-intermediate readers working with completed ordinary files that remain unchanged during processing.

1. Method: control headers, then compare results

A gzip header can contain a timestamp and an original filename. The RFC 1952 field definitions specify that MTIME = 0 omits a timestamp and that the FNAME flag controls the filename field. Explicitly changing these fields reproduces different compressed bytes without waiting for the clock:

python3 - <<'PY_HEADER'
import gzip
from io import BytesIO

payload = b"id,value\n1,alpha\n"

def pack(name, stamp):
    buffer = BytesIO()
    with gzip.GzipFile(filename=name, mode="wb", fileobj=buffer,
                       mtime=stamp, compresslevel=9) as file:
        file.write(payload)
    return buffer.getvalue()

a = pack("a.csv", 1700000000)
b = pack("a.csv", 1700000001)
c = pack("b.csv", 1700000000)
assert a != b and a != c
assert gzip.decompress(a) == gzip.decompress(b) == gzip.decompress(c) == payload
print("OK: same content, different gzip bytes")
PY_HEADER

Expect OK: same content, different gzip bytes. Compare original bytes or decompressed hashes to identify source content; compare the .gz itself to establish byte-for-byte artifact equality. The dataset SHA-256 manifest tutorial can help record input versions.

We fix three parameters: filename="", mtime=0, and compresslevel=9. The reproducibility scope is identical input bytes, script, and Python/zlib environment. Retest after upgrading the compression implementation; this is not a permanent hash agreement across versions.

2. Prepare the environment and input

Use Python 3.9+ with the existing gzip and zlib standard library modules. No third-party packages are required. Commands target Linux bash or zsh; the examples were verified on Python 3.11.4.

1. Create a new demonstration directory and run subsequent commands there. Choose a different directory name when repeating the experiment:

python3 --version
python3 -c 'import platform, zlib; print(platform.python_version(), zlib.ZLIB_RUNTIME_VERSION)'
mkdir reproducible-gzip-demo
cd reproducible-gzip-demo
printf 'id,value\n1,alpha\n2,beta\n' > first.csv
cp first.csv second.csv

2. Keep the printed Python and runtime zlib versions when comparing machines. The official zlib documentation describes the runtime version constant.
3. Save the next section as reproducible_gzip.py. The two input files have identical contents but different names, allowing us to check that filenames do not enter the artifact.

3. Implementation: stream compression and refuse overwrites

"""Create a reproducible gzip in a new file. Python 3.9+."""
import argparse
import gzip
from pathlib import Path
import shutil
import sys
import zlib


def compress_file(source, target):
    created = False
    try:
        with source.open("rb") as src:
            with target.open("xb") as raw:
                created = True
                with gzip.GzipFile(
                    filename="", mode="wb", fileobj=raw,
                    mtime=0, compresslevel=9,
                ) as packed:
                    shutil.copyfileobj(src, packed, length=1024 * 1024)
    except BaseException:
        if created:
            target.unlink(missing_ok=True)
        raise


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("source", type=Path)
    parser.add_argument("target", type=Path)
    args = parser.parse_args()
    try:
        compress_file(args.source, args.target)
    except (OSError, zlib.error) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print("OK: gzip complete")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

fileobj supplies the destination stream; an empty filename excludes the destination name from the header. mtime=0 removes time variation. The GzipFile documentation also explains that closing the gzip wrapper leaves the supplied file object open, which is why the code uses nested context managers.

shutil.copyfileobj copies in 1 MiB chunks rather than loading the entire source. Exclusive xb creation rejects an existing destination and leaves the source unchanged. On a caught exception or Ctrl+C, the function attempts to remove the incomplete output it created; it does not delete a previously existing destination.

This assumes trusted directories with no concurrent replacement of the destination. The output is visible before compression finishes, and a forced kill or power loss can leave a partial file. For delivery to other processes, adapt the atomic write pattern: stage beside the destination and publish only after a successful close.

4. Run and verify the artifacts

1. View help and create two gzip files:

python3 reproducible_gzip.py --help
python3 reproducible_gzip.py first.csv first.gz
python3 reproducible_gzip.py second.csv second.gz
sha256sum first.gz second.gz

Each compression prints OK: gzip complete. Both sha256sum hashes should match, with different filenames beside them.

2. Check the header, compressed bytes, and full decompressed contents:

python3 - <<'PY_CHECK'
import gzip
import hashlib
from pathlib import Path

first = Path("first.gz").read_bytes()
second = Path("second.gz").read_bytes()
assert first == second
assert first[:3] == b"\x1f\x8b\x08"
assert first[3] & 0x08 == 0  # No FNAME field.
assert first[4:8] == b"\x00" * 4  # MTIME = 0.
assert gzip.decompress(first) == Path("first.csv").read_bytes()
assert gzip.decompress(second) == Path("second.csv").read_bytes()
print("OK: equal gzip bytes, zero timestamp, no filename, exact round trip")
print("sha256:", hashlib.sha256(first).hexdigest())
PY_CHECK

The header assertions apply to gzip files produced by this script; they are not a complete validator for arbitrary gzip files. gzip.decompress() reads the small demonstration files into memory. For full reads of large logs, see the gzip log integrity tutorial.

Keep both checks: identical corrupt artifacts can share a hash, while checking only decompressed contents misses unnecessary changes in the header.

5. Test failures and boundaries

1. Try writing the existing destination again. An if condition handles the expected failure even in scripts using set -e:

if python3 reproducible_gzip.py first.csv first.gz; then
  printf 'UNEXPECTED: existing output was accepted\n' >&2
else
  printf 'exit=%s\n' "$?"
fi

Expect ERROR: and a file-exists explanation on stderr, followed by exit=1. The original first.gz remains unchanged. Success returns 0, file or compression failures return 1, and invalid CLI arguments return 2.

2. Save this as check_gzip.py and run python3 check_gzip.py. Its automatically cleaned temporary directory tests empty input, UTF-8, multi-chunk binary data, repeated runs, identical source/destination paths, and missing input:

"""Run beside reproducible_gzip.py, using only disposable test files."""
import gzip
from pathlib import Path
import subprocess
import sys
from tempfile import TemporaryDirectory

script = Path("reproducible_gzip.py").resolve()

def run(source, target):
    return subprocess.run(
        [sys.executable, str(script), str(source), str(target)],
        capture_output=True, text=True, timeout=30,
    )

with TemporaryDirectory(prefix="gzip-check-") as directory:
    root = Path(directory)
    for number, payload in enumerate((b"", "中文\n".encode("utf-8"),
                                      bytes(range(256)) * 9000)):
        source = root / f"input {number}.bin"
        target = root / f"output {number}.gz"
        other = root / f"other {number}.gz"
        source.write_bytes(payload)
        assert run(source, target).returncode == 0
        assert run(source, other).returncode == 0
        saved = target.read_bytes()
        assert saved == other.read_bytes()
        assert gzip.decompress(saved) == payload
        assert source.read_bytes() == payload
        result = run(source, target)
        assert result.returncode == 1 and not result.stdout
        assert target.read_bytes() == saved
        result = run(source, source)
        assert result.returncode == 1
        assert source.read_bytes() == payload
    missing = root / "missing.bin"
    output = root / "missing.gz"
    result = run(missing, output)
    assert result.returncode == 1 and not output.exists()
    assert "ERROR:" in result.stderr
print("OK: empty, Unicode, multi-chunk binary, overwrite and missing-input checks")

Expect the final line OK: empty, Unicode, multi-chunk binary, overwrite and missing-input checks. These tests also check that sources remain unchanged and existing destinations are preserved.

6. Troubleshooting

Avoid depending on version-specific defaults. The current gzip changelog records gzip.compress() changing its default mtime to 0 in 3.14 and its default compression level to 6 in 3.15. This tutorial uses streaming GzipFile with all three parameters explicit.

Summary

Fix the timestamp, filename policy, and compression level, then pin and record the runtime environment. Compare compressed bytes and fully decompress the result to check the original contents. Use overwrite protection and failure exits so automation does not mistake an incomplete artifact for a reproducible result.