compress-lib 0.8.0

DEFLATE, zlib, and gzip for Rust services: streaming and sans-IO with bounded memory, hard limits against decompression bombs, WebSocket permessage-deflate (RFC 7692) support, no unsafe code, and zero third-party dependencies.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

"""Interoperation with C zlib through CPython's zlib and gzip modules.

    python3 tests/interop/check.py write DIR   # CPython's streams for compress-lib
    python3 tests/interop/check.py read DIR    # check compress-lib's streams

compress-lib's side is `cargo run --example interop -- read|write DIR`.
"""

import gzip
import os
import sys
import zlib

WBITS = {"raw": -15, "zlib": 15, "gzip": 31}


def samples():
    words = b"the of and stream window header deflate inflate gzip zlib message server ".split()
    x = 0x12345678
    out = []
    for n in (0, 1, 100, 5000, 70000):
        buf = bytearray()
        while len(buf) < n:
            x ^= (x << 13) & 0xFFFFFFFF
            x ^= x >> 17
            x ^= (x << 5) & 0xFFFFFFFF
            buf += words[x % len(words)] + b" "
        out.append(bytes(buf[:n]))
    out.append(bytes(range(256)) * 300)
    out.append(os.urandom(0) + bytes((i * 7919) & 0xFF for i in range(40000)))
    return out


def write(d):
    os.makedirs(d, exist_ok=True)
    for i, s in enumerate(samples()):
        open(os.path.join(d, f"plain-{i}"), "wb").write(s)
        for fmt, wbits in WBITS.items():
            for level in (0, 1, 6, 9):
                c = zlib.compressobj(level, zlib.DEFLATED, wbits)
                open(os.path.join(d, f"py-{fmt}-{level}-{i}"), "wb").write(c.compress(s) + c.flush())
        # A two-member gzip file, as `cat a.gz b.gz` makes.
        open(os.path.join(d, f"py-gzip-members-{i}"), "wb").write(gzip.compress(s) + gzip.compress(s[::-1]))
    print("wrote", len(os.listdir(d)), "files")


def read(d):
    checked = 0
    for name in sorted(os.listdir(d)):
        if not name.startswith("ours-"):
            continue
        _, fmt, rest = name.split("-", 2)
        index = rest.rsplit("-", 1)[1]
        plain = open(os.path.join(d, f"plain-{index}"), "rb").read()
        data = open(os.path.join(d, name), "rb").read()
        got = zlib.decompress(data, WBITS[fmt])
        if got != plain:
            sys.exit(f"{name}: C zlib decoded {len(got)} bytes, expected {len(plain)}")
        if fmt == "gzip" and gzip.decompress(data) != plain:
            sys.exit(f"{name}: gzip module disagrees")
        checked += 1
    if checked == 0:
        sys.exit("no compress-lib streams found")
    print("C zlib decoded", checked, "compress-lib streams")


if __name__ == "__main__":
    {"write": write, "read": read}[sys.argv[1]](sys.argv[2])