Cybersécurité augmentée par graphe de connaissances

@prefix : <http://security.example.org/ontology#> .
@prefix stix: <https://oasis-open.org/cti/stix/v2.1/> .

# Entités cyber
:IP a owl:Class .
:Host a owl:Class .
:Utilisateur a owl:Class .
:Processus a owl:Class .
:Fichier a owl:Class .
:Alerte a owl:Class .
:Incident a owl:Class .
:AttaquePattern a owl:Class ; rdfs:subClassOf stix:AttackPattern .
:IOC a owl:Class .  # Indicator of Compromise

# Relations temporelles et logiques
:aPourIPSource a owl:ObjectProperty .
:aPourIPDestination a owl:ObjectProperty .
:estExécutéSur a owl:ObjectProperty .
:estLancéPar a owl:ObjectProperty .
:aCréé a owl:ObjectProperty .
:estLiéÀIncident a owl:ObjectProperty .
:précèdeTemporellement a owl:ObjectProperty .
:estUneÉtapeDe a owl:ObjectProperty .

# Règle : Détection de campagne
[ (:aPourIPSource ?a1 ?ip), (:aPourIPSource ?a2 ?ip), (?a1 != ?a2),
  (:estUneÉtapeDe ?a1 ?pattern), (:estUneÉtapeDe ?a2 ?pattern)
  -> (:appartientÀCampagne ?a1 ?campagne), (:appartientÀCampagne ?a2 ?campagne) ]

# Règle : Prédiction prochaine étape
[ (:appartientÀCampagne ?a ?c), (:estUneÉtapeDe ?a ?e),
  (:aPourÉtapeSuivante ?e ?next)
  -> (:préditProchaineÉtape ?c ?next) ]

class SecurityKnowledgeGraph:
    def __init__(self, kg, vector_db, llm):
        self.kg = kg
        self.vector_db = vector_db
        self.llm = llm
        self.alert_queue = []
    
    def ingest_alert(self, alert):
        """Ingestion d'une alerte depuis SIEM, EDR, Firewall"""
        
        # 1. Normalisation
        normalized = self.normalize_alert(alert)
        
        # 2. Insertion dans le graphe
        alert_id = self.insert_alert(normalized)
        
        # 3. Corrélation avec alertes précédentes
        related = self.find_related_alerts(normalized)
        
        if related:
            # Lie l'alerte à un incident existant
            incident = self.attach_to_incident(alert_id, related)
            
            # 4. Prédiction de la prochaine étape
            next_steps = self.predict_next_attack_steps(incident)
            
            if next_steps:
                self.raise_proactive_alert(next_steps)
        else:
            # Nouvel incident potentiel
            self.create_new_incident(alert_id)
        
        # 5. Enrichissement avec Threat Intelligence
        self.enrich_with_threat_intel(alert_id)
        
        return alert_id
    
    def find_related_alerts(self, alert):
        """Trouve les alertes liées par IP, utilisateur, host, pattern"""
        query = f"""
            SELECT ?alert WHERE {{
                {{
                    ?alert :aPourIPSource "{alert['src_ip']}" .
                }} UNION {{
                    ?alert :aPourIPDestination "{alert['dst_ip']}" .
                }} UNION {{
                    ?alert :estExécutéSur "{alert['host']}" .
                }} UNION {{
                    ?alert :estLancéPar "{alert['user']}" .
                }}
                FILTER(?alert != :alert_{alert['id']})
                FILTER(?date > NOW() - "1h"^^xsd:duration)
            }}
        """
        return self.kg.query(query)
    
    def predict_next_attack_steps(self, incident_id):
        """Prédiction de la prochaine étape d'attaque via graphe + LLM"""
        incident = self.kg.get_incident(incident_id)
        
        # Récupération du pattern d'attaque détecté
        pattern = incident['attack_pattern']
        
        # Requête du graphe pour les étapes suivantes connues
        next_steps = self.kg.query(f"""
            SELECT ?step WHERE {{
                :{pattern} :aPourÉtapeSuivante ?step .
            }}
        """)
        
        if next_steps:
            return next_steps
        
        # Fallback : LLM pour prédiction probabiliste
        prediction = self.llm.predict(f"""
            Pattern d'attaque détecté: {pattern}
            Étapes déjà observées: {incident['observed_steps']}
            Quelle est la prochaine étape la plus probable ?
        """)
        
        return prediction
    
    def generate_incident_report(self, incident_id):
        """Génération automatisée de rapport d'incident"""
        incident = self.kg.get_incident(incident_id)
        
        report = self.llm.generate(f"""
            Génère un rapport d'incident cyber pour la direction:
            
            Incident ID: {incident_id}
            Détection: {incident['first_seen']}
            Pattern: {incident['attack_pattern']}
            Alertes associées: {incident['alerts']}
            Entités compromises: {incident['compromised_assets']}
            Chronologie: {incident['timeline']}
            Actions recommandées: {incident['recommended_actions']}
            
            Format: Synthèse exécutive + détails techniques + recommandations
        """)
        
        return report
    
    def enrich_with_threat_intel(self, alert_id):
        """Enrichissement avec OSINT et Threat Intelligence feeds"""
        alert = self.kg.get_alert(alert_id)
        
        # Vérification IP malveillante
        ip_reputation = self.check_ip_reputation(alert['src_ip'])
        if ip_reputation['malicious']:
            self.kg.insert(f"""
                :alert_{alert_id} :estLiéÀThreatIntel :ti_{ip_reputation['threat_id']} ;
                :niveauConfiance {ip_reputation['confidence']} .
            """)
        
        # Vérification hash de fichier
        if alert.get('file_hash'):
            hash_reputation = self.check_hash_reputation(alert['file_hash'])
            if hash_reputation['malicious']:
                self.kg.insert(f"""
                    :alert_{alert_id} :indicateurCompromis :{alert['file_hash']} .
                """)