# © 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, }