ulises/intelligence/rag_engine.py

# © 2026 Antonio Rodriguez Martinez / TONJ EU
# All rights reserved. No use, copy, or distribution without written permission.
# Legal actions built to execute.

"""
rag_engine.py — Motor RAG de tendencias soberanas (ulises.sty)

4 análisis multi-fuente que producen el "golpe de mesa" ante cualquier cliente EU:
  1. regulatory_drift      — convergencia EUR-Lex + ENISA → artículos AI Act bajo presión
  2. pqc_migration_urgency — HNDL countdown + NVD CVEs → urgencia migración post-cuántica
  3. compliance_gap_score  — SBOM cripto vs AI Act + NIS2 → gap accionable por artículo
  4. threat_act_correlation — mapa amenazas activas → artículos AI Act activados + acciones
"""
from __future__ import annotations

import asyncio
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

log = logging.getLogger("ulises.intelligence.rag_engine")

# ── Lookup tables ────────────────────────────────────────────────────────────

# Mapa artículo AI Act → descripción corta (usado en gap scoring y correlation)
_AI_ACT_ARTICLES: Dict[str, str] = {
    "9":  "Risk management system — lifecycle risk identification and mitigation",
    "13": "Transparency — deployers must understand system outputs and limitations",
    "14": "Human oversight — ability to monitor, intervene and interrupt the system",
    "17": "Quality management system — documented policies, procedures, instructions",
    "43": "Conformity assessment — third-party or self-assessment for high-risk AI",
    "62": "Serious incident reporting — 15-day notification to market surveillance",
    "72": "GPAI transparency — technical documentation for general-purpose AI models",
}

_NIS2_ARTICLES: Dict[str, str] = {
    "21": "Cybersecurity risk management — cryptographic measures, MFA, encryption",
    "23": "Incident reporting — significant incidents to NIS authority within 24h/72h",
}

# Mapa categoría de amenaza → artículos que activa
_THREAT_TO_ARTICLES: Dict[str, List[str]] = {
    "supply_chain":          ["9", "17", "NIS2-21"],
    "biometric_data":        ["9", "14", "62"],
    "ai_model_poisoning":    ["9", "13", "17"],
    "inference_attack":      ["9", "13", "NIS2-21"],
    "critical_infrastructure": ["9", "14", "62", "NIS2-21"],
    "no_audit_trail":        ["9", "13", "62"],
    "no_human_oversight":    ["14"],
    "quantum_crypto_risk":   ["9", "17", "NIS2-21"],
    "data_exfiltration":     ["9", "NIS2-21", "NIS2-23"],
    "deepfake_impersonation": ["52", "13"],
    "gpai_misuse":           ["72", "13", "14"],
    "border_control_ai":     ["9", "14", "62"],
    "healthcare_ai":         ["9", "13", "14", "62"],
    "financial_scoring":     ["13", "17", "43"],
}

# Artículos EU referenciados en publicaciones EUR-Lex recientes (base offline)
_EURLEX_ARTICLE_PRESSURE: Dict[str, int] = {
    "9": 14, "13": 11, "14": 18, "17": 9,
    "43": 6, "62": 7, "72": 15, "NIS2-21": 12,
}

# Algoritmos cripto considerados cuánticamente vulnerables
_QUANTUM_VULNERABLE: List[str] = [
    "RSA-1024", "RSA-2048", "RSA-4096",
    "EC-256", "EC-384", "ECDSA", "ECDH",
    "DH-2048", "DSA",
]

# Scores PQC requeridos por artículo (umbral mínimo de conformidad)
_PQC_THRESHOLD_BY_ARTICLE = {
    "9": 60,   # Risk management exige inventario cripto actualizado
    "17": 50,  # QMS exige política de gestión cripto documentada
    "NIS2-21": 70,  # NIS2 exige medidas criptográficas actualizadas
}


# ── 1. Regulatory Drift ──────────────────────────────────────────────────────

async def regulatory_drift(
    eurlex_updates: Optional[List[Dict]] = None,
    enisa_threats: Optional[List[Dict]] = None,
) -> Dict[str, Any]:
    """
    Detecta qué artículos del AI Act están bajo presión regulatoria activa.
    Cruza publicaciones EUR-Lex recientes con amenazas ENISA para identificar
    convergencias ('drift events') — señal de que el regulador está endureciendo
    la interpretación de ese artículo.
    """
    article_pressure = dict(_EURLEX_ARTICLE_PRESSURE)

    # Sumar presión de amenazas ENISA sobre artículos
    threats = enisa_threats or []
    for threat in threats:
        scenario = threat.get("scenario", "").lower() + threat.get("domain", "").lower()
        for category, articles in _THREAT_TO_ARTICLES.items():
            if any(kw in scenario for kw in category.split("_")):
                for art in articles:
                    article_pressure[art] = article_pressure.get(art, 0) + 2

    # Sumar presión de actualizaciones EUR-Lex
    for update in (eurlex_updates or []):
        title = update.get("title", "").lower() + update.get("summary", "").lower()
        for art_num in _AI_ACT_ARTICLES:
            if f"article {art_num}" in title or f"art. {art_num}" in title:
                article_pressure[art_num] = article_pressure.get(art_num, 0) + 3

    # Ordenar por presión descendente
    ranked = sorted(article_pressure.items(), key=lambda x: x[1], reverse=True)
    top_articles = ranked[:5]

    # Drift events: artículos con presión combinada superior a umbral
    drift_threshold = 15
    drift_events = [
        {
            "article": art,
            "intensity": score,
            "description": _AI_ACT_ARTICLES.get(art, _NIS2_ARTICLES.get(art.replace("NIS2-", ""), ""))
        }
        for art, score in ranked if score >= drift_threshold
    ]

    return {
        "analysis": "regulatory_drift",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "article_pressure_heatmap": dict(ranked),
        "top_pressure_articles": [a for a, _ in top_articles],
        "drift_events": drift_events,
        "drift_event_count": len(drift_events),
        "interpretation": (
            f"{len(drift_events)} artículo(s) bajo presión regulatoria activa. "
            f"Art. {top_articles[0][0]} es el más expuesto (intensidad {top_articles[0][1]})."
            if drift_events else
            "Sin drift regulatorio significativo detectado en fuentes actuales."
        ),
    }


# ── 2. PQC Migration Urgency ─────────────────────────────────────────────────

async def pqc_migration_urgency(
    nvd_cves: Optional[List[Dict]] = None,
    pqc_score: Optional[float] = None,
    criticality: str = "HIGH",
    data_sensitivity_years: int = 10,
) -> Dict[str, Any]:
    """
    Calcula el índice de urgencia de migración post-cuántica (0-100).
    Combina: HNDL countdown (Harvest-Now-Decrypt-Later) + CVEs activos en
    algoritmos cuánticamente vulnerables + score PQC actual del activo.
    """
    # HNDL: estimación de años hasta amenaza cuántica relevante
    # NIST estima primer CRQC ~2030-2035; urgencia aumenta con sensibilidad de datos
    years_to_quantum = max(0, 8 - max(0, data_sensitivity_years - 5))
    hndl_base_urgency = min(100, int((10 - years_to_quantum) * 10))

    # CVEs en algoritmos vulnerables
    cves = nvd_cves or []
    quantum_cves = [
        c for c in cves
        if any(alg.lower() in c.get("description", "").lower() for alg in _QUANTUM_VULNERABLE)
    ]
    cve_urgency = min(30, len(quantum_cves) * 5)

    # Penalización por criticidad
    criticality_factor = {"CRITICAL": 1.2, "HIGH": 1.0, "MEDIUM": 0.8, "LOW": 0.5}.get(
        criticality.upper(), 1.0
    )

    # Score PQC actual: inversamente proporcional a la urgencia
    pqc_penalty = 0
    if pqc_score is not None:
        pqc_penalty = max(0, int((100 - pqc_score) * 0.3))

    raw_urgency = (hndl_base_urgency + cve_urgency + pqc_penalty) * criticality_factor
    urgency_index = min(100, int(raw_urgency))

    # Nivel semáforo
    if urgency_index >= 75:
        urgency_level = "CRITICAL"
        action = "Iniciar migración PQC inmediatamente — ventana de acción < 18 meses"
    elif urgency_index >= 50:
        urgency_level = "HIGH"
        action = "Planificar migración PQC en 2025-2026 — prioridad en sistemas críticos"
    elif urgency_index >= 25:
        urgency_level = "MEDIUM"
        action = "Inventariar algoritmos vulnerables y preparar plan de migración"
    else:
        urgency_level = "LOW"
        action = "Monitorizar evolución NIST PQC y EUR-Lex — revisión anual suficiente"

    # Milestones de migración NIST PQC
    current_year = datetime.now().year
    milestones = [
        {"year": current_year, "action": "Crypto-SBOM completo del stack"},
        {"year": current_year + 1, "action": "Piloto ML-KEM-768 en servicios no críticos"},
        {"year": current_year + 2, "action": "Migración ML-KEM + ML-DSA en sistemas críticos"},
        {"year": current_year + 3, "action": "Deprecación total de RSA/ECDH en infraestructura pública"},
    ]

    return {
        "analysis": "pqc_migration_urgency",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "urgency_index": urgency_index,
        "urgency_level": urgency_level,
        "years_to_quantum_threat": years_to_quantum,
        "hndl_base_urgency": hndl_base_urgency,
        "quantum_vulnerable_cves": len(quantum_cves),
        "cve_details": quantum_cves[:3],
        "current_pqc_score": pqc_score,
        "recommended_action": action,
        "nist_pqc_milestones": milestones,
        "algorithms_to_replace": _QUANTUM_VULNERABLE[:6],
        "target_algorithms": ["ML-KEM-768", "ML-DSA-65", "SLH-DSA-128s"],
    }


# ── 3. Compliance Gap Score ───────────────────────────────────────────────────

async def compliance_gap_score(
    crypto_sbom: Optional[Dict] = None,
    pqc_score: Optional[float] = None,
    tls_result: Optional[Dict] = None,
    has_audit_trail: bool = True,
    has_human_oversight_docs: bool = False,
    has_risk_management_docs: bool = False,
) -> Dict[str, Any]:
    """
    Calcula el gap de conformidad AI Act + NIS2 artículo por artículo.
    Toma datos del inventario cripto de CiberIA y los cruza con los requisitos
    regulatorios aplicables. Devuelve puntuación de gap (0=perfecto, 100=crítico)
    y acciones específicas por artículo.
    """
    gaps: Dict[str, Dict[str, Any]] = {}

    # Art. 9 — Risk management: ¿hay inventario cripto? ¿Risk docs?
    art9_gap = 0
    art9_actions = []
    if not crypto_sbom:
        art9_gap += 40
        art9_actions.append("Generar Crypto-SBOM del sistema (ejecutar safe_inventory)")
    if not has_risk_management_docs:
        art9_gap += 35
        art9_actions.append("Documentar sistema de gestión de riesgos (ISO 31000 + AI Act Art. 9)")
    if pqc_score is not None and pqc_score < _PQC_THRESHOLD_BY_ARTICLE["9"]:
        art9_gap += 25
        art9_actions.append(f"Score PQC {pqc_score:.0f}/100 < umbral 60 — plan migración post-cuántica requerido")
    gaps["Art.9"] = {"gap_score": min(100, art9_gap), "actions": art9_actions, "severity": _severity(art9_gap)}

    # Art. 13 — Transparency: ¿TLS activo? ¿Protocolos estándar?
    art13_gap = 0
    art13_actions = []
    if tls_result:
        tls_version = tls_result.get("tls_version", "")
        if tls_version in ("TLSv1", "TLSv1.1", ""):
            art13_gap += 35
            art13_actions.append(f"Actualizar TLS a 1.3 — versión detectada: {tls_version or 'no detectada'}")
        cert_days = tls_result.get("cert_days_remaining", 999)
        if isinstance(cert_days, int) and cert_days < 30:
            art13_gap += 20
            art13_actions.append(f"Certificado expira en {cert_days} días — renovar inmediatamente")
    else:
        art13_gap += 15
        art13_actions.append("Ejecutar tls_crypto_audit para verificar configuración de transparencia")
    gaps["Art.13"] = {"gap_score": min(100, art13_gap), "actions": art13_actions, "severity": _severity(art13_gap)}

    # Art. 14 — Human oversight: ¿documentada la supervisión humana?
    art14_gap = 0 if has_human_oversight_docs else 60
    art14_actions = (
        [] if has_human_oversight_docs
        else ["Documentar procedimientos de supervisión humana y capacidad de interrupción del sistema"]
    )
    gaps["Art.14"] = {"gap_score": art14_gap, "actions": art14_actions, "severity": _severity(art14_gap)}

    # Art. 17 — Quality management: ¿QMS documentado?
    art17_gap = 0
    art17_actions = []
    if not has_risk_management_docs:
        art17_gap += 50
        art17_actions.append("Establecer QMS documentado (políticas, procedimientos, instrucciones escritas)")
    if pqc_score is not None and pqc_score < _PQC_THRESHOLD_BY_ARTICLE["17"]:
        art17_gap += 25
        art17_actions.append("Incluir política de gestión cripto en QMS — score PQC insuficiente")
    gaps["Art.17"] = {"gap_score": min(100, art17_gap), "actions": art17_actions, "severity": _severity(art17_gap)}

    # Art. 62 — Incident reporting: ¿cadena de auditoría activa?
    art62_gap = 0 if has_audit_trail else 70
    art62_actions = (
        [] if has_audit_trail
        else ["Activar cadena de auditoría inmutable — TONJ EU EvidenceSeal (SHA256 append-only)"]
    )
    gaps["Art.62"] = {"gap_score": art62_gap, "actions": art62_actions, "severity": _severity(art62_gap)}

    # NIS2 Art. 21 — Crypto measures
    nis2_gap = 0
    nis2_actions = []
    if pqc_score is not None and pqc_score < _PQC_THRESHOLD_BY_ARTICLE["NIS2-21"]:
        nis2_gap += 40
        nis2_actions.append(f"NIS2 exige medidas cripto actualizadas — score PQC {pqc_score:.0f}/100 < umbral 70")
    if tls_result and tls_result.get("tls_version", "") in ("TLSv1", "TLSv1.1", ""):
        nis2_gap += 35
        nis2_actions.append("TLS < 1.2 detectado — NIS2 Art. 21 requiere cifrado en tránsito actualizado")
    gaps["NIS2-Art.21"] = {"gap_score": min(100, nis2_gap), "actions": nis2_actions, "severity": _severity(nis2_gap)}

    # Score global de gap
    scores = [g["gap_score"] for g in gaps.values()]
    overall_gap = int(sum(scores) / len(scores))
    compliance_score = 100 - overall_gap

    critical_gaps = [art for art, data in gaps.items() if data["severity"] == "CRITICAL"]
    high_gaps = [art for art, data in gaps.items() if data["severity"] == "HIGH"]

    return {
        "analysis": "compliance_gap_score",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "overall_gap_score": overall_gap,
        "compliance_score": compliance_score,
        "compliance_level": _compliance_level(compliance_score),
        "gap_by_article": gaps,
        "critical_gaps": critical_gaps,
        "high_gaps": high_gaps,
        "total_actions_required": sum(len(g["actions"]) for g in gaps.values()),
        "certification_ready": compliance_score >= 75 and not critical_gaps,
    }


def _severity(gap: int) -> str:
    if gap >= 60:
        return "CRITICAL"
    if gap >= 35:
        return "HIGH"
    if gap >= 15:
        return "MEDIUM"
    return "LOW"


def _compliance_level(score: int) -> str:
    if score >= 85:
        return "COMPLIANT"
    if score >= 65:
        return "PARTIALLY_COMPLIANT"
    if score >= 40:
        return "NON_COMPLIANT"
    return "CRITICAL_NON_COMPLIANT"


# ── 4. Threat-Act Correlation ─────────────────────────────────────────────────

async def threat_act_correlation(
    threats: Optional[List[Dict]] = None,
    nvd_cves: Optional[List[Dict]] = None,
) -> Dict[str, Any]:
    """
    Mapea las amenazas activas (ENISA + NVD) a los artículos AI Act que activan.
    Produce una tabla accionable: amenaza → artículos → acciones de cumplimiento.
    """
    all_threats = list(threats or [])

    # Convertir CVEs a formato de amenaza
    for cve in (nvd_cves or [])[:10]:
        cvss = float(cve.get("cvss", 0))
        desc = cve.get("description", "").lower()
        category = _classify_cve(desc, cvss)
        all_threats.append({
            "id": cve.get("id", "CVE-UNKNOWN"),
            "scenario": desc[:200],
            "domain": "EU_CYBERSECURITY",
            "risk_level": "CRITICAL" if cvss >= 9.0 else ("HIGH" if cvss >= 7.0 else "MEDIUM"),
            "_category": category,
            "_cvss": cvss,
        })

    # Correlacionar amenazas con artículos
    article_trigger_count: Dict[str, int] = {}
    correlations = []

    for threat in all_threats:
        scenario = (threat.get("scenario", "") + threat.get("domain", "")).lower()
        category = threat.get("_category") or _classify_scenario(scenario)
        activated_articles = _THREAT_TO_ARTICLES.get(category, ["9"])

        for art in activated_articles:
            article_trigger_count[art] = article_trigger_count.get(art, 0) + 1

        compliance_actions = _actions_for_articles(activated_articles)
        correlations.append({
            "threat_id": threat.get("id", "UNKNOWN"),
            "risk_level": threat.get("risk_level", "MEDIUM"),
            "category": category,
            "activated_articles": activated_articles,
            "compliance_actions": compliance_actions,
        })

    # Ranking de artículos más activados
    ranked_articles = sorted(article_trigger_count.items(), key=lambda x: x[1], reverse=True)
    top_articles = ranked_articles[:5]

    # Nivel de alerta global
    critical_threats = sum(1 for t in all_threats if t.get("risk_level") == "CRITICAL")
    if critical_threats >= 3:
        alert_level = "RED"
    elif critical_threats >= 1:
        alert_level = "ORANGE"
    elif all_threats:
        alert_level = "YELLOW"
    else:
        alert_level = "GREEN"

    return {
        "analysis": "threat_act_correlation",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "alert_level": alert_level,
        "total_threats_analysed": len(all_threats),
        "critical_threats": critical_threats,
        "most_triggered_articles": [a for a, _ in top_articles],
        "article_trigger_counts": dict(ranked_articles),
        "correlations": correlations[:10],
        "immediate_actions": _priority_actions(top_articles),
    }


def _classify_cve(description: str, cvss: float) -> str:
    desc = description.lower()
    if "supply chain" in desc or "model weight" in desc:
        return "supply_chain"
    if "quantum" in desc or "rsa" in desc or "ecdsa" in desc:
        return "quantum_crypto_risk"
    if "exfiltration" in desc or "data theft" in desc:
        return "data_exfiltration"
    if "critical infrastructure" in desc:
        return "critical_infrastructure"
    if cvss >= 9.0:
        return "critical_infrastructure"
    return "inference_attack"


def _classify_scenario(scenario: str) -> str:
    if "supply chain" in scenario:
        return "supply_chain"
    if "biometric" in scenario:
        return "biometric_data"
    if "border" in scenario:
        return "border_control_ai"
    if "healthcare" in scenario or "patient" in scenario:
        return "healthcare_ai"
    if "financial" in scenario or "credit" in scenario:
        return "financial_scoring"
    if "critical infrastructure" in scenario:
        return "critical_infrastructure"
    if "audit" in scenario or "trail" in scenario:
        return "no_audit_trail"
    if "oversight" in scenario or "human" in scenario:
        return "no_human_oversight"
    if "quantum" in scenario or "rsa" in scenario:
        return "quantum_crypto_risk"
    return "inference_attack"


_ARTICLE_ACTIONS: Dict[str, str] = {
    "9": "Actualizar sistema de gestión de riesgos (Art. 9 AI Act)",
    "13": "Verificar mecanismos de transparencia y explicabilidad (Art. 13)",
    "14": "Revisar procedimientos de supervisión humana (Art. 14)",
    "17": "Auditar QMS — políticas y procedimientos actualizados (Art. 17)",
    "43": "Iniciar evaluación de conformidad (Art. 43)",
    "52": "Verificar obligaciones de transparencia frente al usuario (Art. 52)",
    "62": "Activar protocolo de reporte de incidentes graves (Art. 62 — 15 días)",
    "72": "Documentar modelo GPAI conforme Art. 72",
    "NIS2-21": "Implementar medidas NIS2 Art. 21 — MFA, cifrado, gestión de riesgos cripto",
    "NIS2-23": "Preparar capacidad de reporte NIS2 — incidente significativo 24h/72h",
}


def _actions_for_articles(articles: List[str]) -> List[str]:
    return [_ARTICLE_ACTIONS[a] for a in articles if a in _ARTICLE_ACTIONS]


def _priority_actions(top_articles: List[tuple]) -> List[str]:
    actions = []
    for art, count in top_articles[:3]:
        action = _ARTICLE_ACTIONS.get(art)
        if action:
            actions.append(f"[{count} amenazas] {action}")
    return actions


# ── Función principal: ejecuta los 4 análisis en paralelo ───────────────────

async def run_full_rag_analysis(
    eurlex_updates: Optional[List[Dict]] = None,
    enisa_threats: Optional[List[Dict]] = None,
    nvd_cves: Optional[List[Dict]] = None,
    pqc_score: Optional[float] = None,
    tls_result: Optional[Dict] = None,
    crypto_sbom: Optional[Dict] = None,
    criticality: str = "HIGH",
    has_audit_trail: bool = True,
) -> Dict[str, Any]:
    """Ejecuta los 4 análisis RAG en paralelo y devuelve resultado unificado."""
    drift, urgency, gap, correlation = await asyncio.gather(
        regulatory_drift(eurlex_updates, enisa_threats),
        pqc_migration_urgency(nvd_cves, pqc_score, criticality),
        compliance_gap_score(crypto_sbom, pqc_score, tls_result, has_audit_trail),
        threat_act_correlation(enisa_threats, nvd_cves),
    )
    return {
        "regulatory_drift": drift,
        "pqc_migration_urgency": urgency,
        "compliance_gap_score": gap,
        "threat_act_correlation": correlation,
    }