Salta al contenuto principale

07 - AI-Assisted Detection: LLM per Generazione Sigma Rules

AI-Assisted Detection: LLM per Generazione Sigma Rules L'integrazione dei Large Language Model nella detection engineering rappresenta uno dei cambiamenti...

Cosa ne pensi di questo articolo?

07 - AI-Assisted Detection: LLM per Generazione Sigma Rules
Detection EngineeringFoto:Danny MenesesPexels

In Breve

AI-Assisted Detection: LLM per Generazione Sigma Rules L'integrazione dei Large Language Model nella detection engineering rappresenta uno dei cambiamenti...

AI-Assisted Detection: LLM per Generazione Sigma Rules

L'integrazione dei Large Language Model nella detection engineering rappresenta uno dei cambiamenti più radicali del panorama SOC degli ultimi anni. Non si tratta semplicemente di automatizzare un task ripetitivo: usare un LLM per generare Sigma rules significa comprimere in secondi un processo che un analista esperto porta a termine in ore, dalla lettura di un threat report alla scrittura di una regola testata e pronta per il deployment.

Frameworks come SigmaGen, presentato al MITRE ATT&CK APAC 2025, dimostrano come modelli fine-tuned possano ingerire report di threat intelligence, estrarre tecniche ATT&CK, e produrre Sigma rules mappate con alta precisione. Allo stesso tempo, strumenti open-source e workflow basati su n8n permettono a team più piccoli di costruire pipeline AI-assisted senza investimenti enterprise.

Questo articolo ti guida attraverso l'architettura di un sistema AI-assisted per la generazione di detection rules: dal prompt engineering, alla validazione automatica, al testing con log sintetici, fino all'integrazione in una pipeline CI/CD esistente.

Cosa Imparerai

  • Come i LLM ragionano sul formato Sigma e perchè producono output migliori rispetto a SPL diretto
  • Tecniche di prompt engineering specifiche per la generazione di detection rules
  • Architettura di una pipeline AI-assisted end-to-end
  • Validazione automatica e testing delle regole generate
  • Integrazione con SigmaGen, pySigma e workflow CI/CD
  • Anti-patterns da evitare e limiti reali dei LLM in contesti di sicurezza

perchè i LLM Eccellono nella Generazione di Sigma Rules

Una delle osservazioni più interessanti emersa dalla ricerca applicata (e confermata dalla pratica) e che i LLM producono output di qualità significativamente superiore quando generano Sigma rispetto a query SPL o KQL dirette. Il motivo e strutturale.

Il formato YAML di Sigma impone una separazione netta tra:

  • title e description: il modello deve articolare cosa sta rilevando e perchè
  • logsource: specifica la sorgente dati (category, product, service)
  • detection: la logica booleana di matching
  • condition: come combinare i selettori
  • falsepositives: ragionamento esplicito sui casi edge

Questa struttura costringe il modello a "pensare" in modo sequenziale e dichiarativo, riducendo gli errori logici che emergono quando si chiede direttamente output in un linguaggio query specifico di piattaforma. In pratica, Sigma funziona come un chain-of-thought implicito per i LLM applicati alla detection.

Dato di Benchmark

Ricercatori del progetto LLMCloudHunter (2024) hanno dimostrato che LLM generalisti come GPT-4 generano Sigma rules valide nel 73% dei casi su report CTI strutturati, contro il 41% quando si chiede output SPL diretto. La percentuale sale all'89% con fine-tuning su dataset specifici di sicurezza.

Architettura di una Pipeline AI-Assisted

Una pipeline AI-assisted per la generazione di detection rules si compone di cinque stadi distinti:

  1. Ingestion: ingestione di threat reports, blog post CTI, advisory CVE
  2. Extraction: estrazione di IOC, tecniche ATT&CK, comportamenti descritti
  3. Generation: generazione della Sigma rule tramite LLM
  4. Validation: validazione sintattica e semantica automatica
  5. Testing: test con log sintetici e integrazione CI/CD
# Architettura base della pipeline AI-assisted
# File: ai_sigma_pipeline.py

import openai
import yaml
import subprocess
from pathlib import Path
from dataclasses import dataclass
from typing import Optional

@dataclass
class ThreatReport:
    content: str
    source_url: str
    report_type: str  # 'cti_blog', 'advisory', 'malware_analysis'

@dataclass
class GeneratedRule:
    sigma_yaml: str
    mitre_techniques: list[str]
    confidence: float
    validation_passed: bool
    test_results: Optional[dict] = None

class AISigmaPipeline:
    def __init__(self, openai_api_key: str, rules_output_dir: str):
        self.client = openai.OpenAI(api_key=openai_api_key)
        self.output_dir = Path(rules_output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def process_report(self, report: ThreatReport) -> list[GeneratedRule]:
        """Pipeline completa da report a regole validate."""
        # Stage 1: Extraction
        techniques = self._extract_attack_techniques(report)
        behaviors = self._extract_behaviors(report)

        # Stage 2: Generation
        rules = []
        for technique in techniques:
            rule = self._generate_sigma_rule(
                report=report,
                technique=technique,
                behaviors=behaviors
            )
            if rule:
                rules.append(rule)

        # Stage 3: Validation + Testing
        return [self._validate_and_test(r) for r in rules]

    def _extract_attack_techniques(self, report: ThreatReport) -> list[str]:
        """Estrae tecniche ATT&CK dal report tramite LLM."""
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Sei un analista di threat intelligence esperto in MITRE ATT&CK. "
                        "Estrai SOLO le tecniche ATT&CK (formato T1234 o T1234.001) "
                        "esplicitamente descritte nel testo. Rispondi solo con una lista JSON."
                    )
                },
                {
                    "role": "user",
                    "content": f"Report:\n{report.content[:4000]}"
                }
            ],
            temperature=0.1  # Bassa temperatura per output deterministico
        )

        import json
        try:
            return json.loads(response.choices[0].message.content)
        except json.JSONDecodeError:
            return []

Prompt Engineering per Sigma Rules di qualità

La qualità dell'output dipende in modo critico dalla struttura del prompt. Esistono tre pattern fondamentali che producono risultati consistenti nella generazione di Sigma rules:

Pattern 1: Structured System Prompt

Il system prompt deve contenere esattamente le meta-informazioni che il modello necessità per generare Sigma validi: la struttura YAML, i valori validi per category e product, le best practice su falsepositives e i livelli di severita consentiti.

# Prompt di sistema ottimizzato per generazione Sigma rules
SIGMA_SYSTEM_PROMPT = """
Sei un Detection Engineer esperto nella scrittura di Sigma rules.
Quando generi una Sigma rule, rispetta SEMPRE questa struttura YAML:

title: [titolo descrittivo, max 80 char]
id: [UUID v4]
status: experimental
description: [descrizione dettagliata del comportamento rilevato]
references:
  - [URL del report originale se disponibile]
author: AI-Assisted Detection
date: [data odierna in formato YYYY-MM-DD]
tags:
  - attack.[tattica]
  - attack.[tecnica]
logsource:
  category: [process_creation | network_connection | file_event | registry_event]
  product: [windows | linux | macos]
detection:
  [nome_selettore]:
    [campo]: [valore o lista valori]
  condition: [nome_selettore]
falsepositives:
  - [casi legittimi plausibili]
level: [informational | low | medium | high | critical]

REGOLE CRITICHE:
- Usa SEMPRE wildcards (*) nei valori stringa per evitare match esatti fragili
- Preferisci campi con alta disponibilità (Image, CommandLine, ParentImage)
- Indica sempre almeno un falso positivo realistico
- Il campo 'condition' deve essere semplice e leggibile
- Non usare regex complesse se un approccio con keywords e sufficiente
"""

def build_generation_prompt(technique_id: str, behaviors: list[str],
                             logsource_hint: str, report_excerpt: str) -> str:
    return f"""Genera una Sigma rule per rilevare la tecnica MITRE ATT&CK {technique_id}.

Comportamenti osservati nel report:
{chr(10).join(f'- {b}' for b in behaviors[:5])}

Tipo di log suggerito: {logsource_hint}

Estratto del report originale:
{report_excerpt[:1500]}

Genera UNA SOLA Sigma rule in formato YAML valido. Non aggiungere spiegazioni fuori dal YAML."""

Pattern 2: Few-Shot con Esempi di qualità

Includere 2-3 esempi di regole di alta qualità nel prompt (few-shot) migliora drammaticamente la consistenza dell'output, specialmente per logsource inusuali o condizioni complesse.

# Few-shot: esempio di regola di qualità inclusa nel prompt
FEW_SHOT_EXAMPLE = """
Esempio di regola di alta qualità per ispirazione:

title: Suspicious PowerShell Encoded Command Execution
id: 5b4f6d89-1234-4321-ab12-fedcba987654
status: stable
description: >
  Rileva l'esecuzione di PowerShell con parametri di encoding (-enc, -EncodedCommand)
  frequentemente usati da malware per offuscare payload malevoli.
references:
  - https://attack.mitre.org/techniques/T1059/001/
author: SigmaHQ Community
date: 2025-01-15
tags:
  - attack.execution
  - attack.t1059.001
logsource:
  category: process_creation
  product: windows
detection:
  selection:
    Image|endswith:
      - '\\\\powershell.exe'
      - '\\\\pwsh.exe'
    CommandLine|contains:
      - ' -enc '
      - ' -EncodedCommand '
      - ' -ec '
  condition: selection
falsepositives:
  - Software legittimo che usa PowerShell con encoding per configurazioni complesse
  - Script di deployment enterprise
level: medium
"""

Pattern 3: Chain-of-Thought Esplicito

Per tecniche complesse, richiedere al modello di ragionare prima di scrivere la regola produce output più accurati. Questo approccio aggiunge latenza ma riduce significativamente il numero di iterazioni necessarie.

def generate_with_cot(self, technique_id: str, report: ThreatReport) -> GeneratedRule:
    """Generazione con Chain-of-Thought esplicito."""

    # Step 1: Chiedi al modello di ragionare
    reasoning_response = self.client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SIGMA_SYSTEM_PROMPT},
            {
                "role": "user",
                "content": f"""Prima di scrivere la regola per {technique_id}, analizza:
1. Quali artefatti forensi questa tecnica lascia nei log?
2. Qual è il logsource più appropriato?
3. Quali campi hanno la maggiore discriminazione signal/noise?
4. Quali sono i falsi positivi più comuni?

Report: {report.content[:2000]}"""
            }
        ],
        temperature=0.3
    )

    reasoning = reasoning_response.choices[0].message.content

    # Step 2: Usa il ragionamento per guidare la generazione
    rule_response = self.client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SIGMA_SYSTEM_PROMPT},
            {"role": "user", "content": f"Analisi tecnica:\n{reasoning}"},
            {
                "role": "assistant",
                "content": "Basandomi su questa analisi, genero la Sigma rule ottimale:"
            },
            {"role": "user", "content": "Procedi con la generazione YAML."}
        ],
        temperature=0.1
    )

    return GeneratedRule(
        sigma_yaml=rule_response.choices[0].message.content,
        mitre_techniques=[technique_id],
        confidence=0.0,  # Calcolata nella validazione
        validation_passed=False
    )

Validazione Automatica delle Regole Generate

Un LLM può produrre YAML sintatticamente valido ma semanticamente errato: un logsource inesistente, un campo con nome sbagliato, una condizione che non referenzia correttamente i selettori. La validazione automatica e il gate critico prima che una regola entri nel repository.

import yaml
from sigma.rule import SigmaRule
from sigma.exceptions import SigmaError
import re
import uuid

class SigmaRuleValidator:
    # Logsource validi più comuni
    VALID_CATEGORIES = {
        'process_creation', 'network_connection', 'file_event',
        'registry_event', 'registry_add', 'registry_set',
        'dns_query', 'image_load', 'pipe_created', 'raw_access_read'
    }

    VALID_LEVELS = {'informational', 'low', 'medium', 'high', 'critical'}
    VALID_STATUSES = {'stable', 'test', 'experimental', 'deprecated', 'unsupported'}

    def validate(self, sigma_yaml: str) -> tuple[bool, list[str]]:
        """Valida una Sigma rule. Restituisce (valida, lista errori)."""
        errors = []

        # 1. Validazione YAML sintattico
        try:
            rule_dict = yaml.safe_load(sigma_yaml)
        except yaml.YAMLError as e:
            return False, [f"YAML invalido: {str(e)}"]

        # 2. Campi obbligatori
        required_fields = ['title', 'description', 'logsource', 'detection']
        for field in required_fields:
            if field not in rule_dict:
                errors.append(f"Campo obbligatorio mancante: {field}")

        if errors:
            return False, errors

        # 3. Validazione logsource
        logsource = rule_dict.get('logsource', {})
        if 'category' in logsource:
            if logsource['category'] not in self.VALID_CATEGORIES:
                errors.append(
                    f"Categoria logsource non valida: {logsource['category']}. "
                    f"Valide: {', '.join(self.VALID_CATEGORIES)}"
                )

        # 4. Validazione detection
        detection = rule_dict.get('detection', {})
        if 'condition' not in detection:
            errors.append("Campo 'condition' mancante in detection")
        else:
            condition = detection['condition']
            # Verifica che i selettori nella condition esistano
            selectors = [k for k in detection.keys() if k != 'condition']
            # Parse base della condition per trovare riferimenti
            referenced = re.findall(r'\b([a-zA-Z_][a-zA-Z0-9_]*)\b', condition)
            for ref in referenced:
                if ref not in ['and', 'or', 'not', '1', 'of', 'all', 'them', 'filter']:
                    if ref not in selectors:
                        errors.append(
                            f"Condition referenzia '{ref}' che non esiste nei selettori: {selectors}"
                        )

        # 5. Validazione level
        level = rule_dict.get('level', '')
        if level and level not in self.VALID_LEVELS:
            errors.append(f"Level non valido: {level}. Validi: {self.VALID_LEVELS}")

        # 6. UUID check
        rule_id = rule_dict.get('id', '')
        if rule_id:
            try:
                uuid.UUID(str(rule_id))
            except ValueError:
                errors.append(f"ID non e un UUID valido: {rule_id}")
        else:
            errors.append("Campo 'id' mancante - genera un UUID v4")

        # 7. Validazione pySigma (se disponibile)
        try:
            SigmaRule.from_yaml(sigma_yaml)
        except SigmaError as e:
            errors.append(f"Errore pySigma: {str(e)}")

        return len(errors) == 0, errors

Anti-Pattern: Fidarsi Ciecamente dell'Output LLM

Un errore comune nei team che implementano AI-assisted detection e deployare regole generate senza validazione. I LLM commettono errori specifici e ripetibili nella generazione Sigma:

  • Usare campi ProcessName invece di Image (Sysmon)
  • Scrivere condizioni che referenziano selettori inesistenti
  • Inventare categorie logsource non standard
  • Usare contains su campi che non supportano wildcard nel logsource target

Validazione automatica e testing con log reali sono NON OPZIONALI.

Testing Automatizzato con Log Sintetici

Dopo la validazione sintattica, ogni regola deve essere testata su log che simulano sia il comportamento malevolo atteso (true positive test) che il traffico normale (false positive test). Questo approccio, chiamato rule unit testing, e la pratica che distingue una pipeline matura da un esperimento.

import json
from sigma.collection import SigmaCollection
from sigma.backends.test import TextQueryTestBackend
from typing import Any

class SigmaRuleTester:
    def __init__(self):
        self.backend = TextQueryTestBackend()

    def generate_test_events(self, sigma_yaml: str,
                              llm_client) -> dict[str, list[dict]]:
        """Genera eventi di test tramite LLM basandosi sulla regola."""
        rule_dict = yaml.safe_load(sigma_yaml)

        prompt = f"""Data questa Sigma rule:
{sigma_yaml}

Genera in formato JSON due liste di eventi di log:
1. "true_positives": 3 eventi che DEVONO triggherare la regola
2. "false_positives": 3 eventi legittimi che NON devono triggherare la regola

Ogni evento deve avere i campi esatti che la regola usa per il matching.
Formato richiesto:
{
  "true_positives": [
    {"Image": "C:\\\\Windows\\\\System32\\\\cmd.exe", "CommandLine": "...", ...}
  ],
  "false_positives": [
    {"Image": "C:\\\\Program Files\\\\...", "CommandLine": "...", ...}
  ]
}"""

        response = llm_client.chat.completions.create(
            model="gpt-4o-mini",  # Modello più economico per i test
            messages=[{"role": "user", "content": prompt}],
            temperature=0.2
        )

        try:
            return json.loads(response.choices[0].message.content)
        except json.JSONDecodeError:
            return {"true_positives": [], "false_positives": []}

    def run_tests(self, sigma_yaml: str, test_events: dict) -> dict[str, Any]:
        """Esegue i test e restituisce risultati dettagliati."""
        results = {
            "tp_tests": {"passed": 0, "failed": 0, "details": []},
            "fp_tests": {"passed": 0, "failed": 0, "details": []},
            "overall_pass": False
        }

        # Nota: questa e una simulazione del meccanismo di test.
        # In produzione si usano tool come sigma-test o un SIEM sandbox.
        rule_dict = yaml.safe_load(sigma_yaml)
        detection = rule_dict.get('detection', {})

        for event in test_events.get('true_positives', []):
            matched = self._simulate_match(event, detection)
            if matched:
                results["tp_tests"]["passed"] += 1
                results["tp_tests"]["details"].append({"event": event, "result": "PASS"})
            else:
                results["tp_tests"]["failed"] += 1
                results["tp_tests"]["details"].append({"event": event, "result": "FAIL - non matchato"})

        for event in test_events.get('false_positives', []):
            matched = self._simulate_match(event, detection)
            if not matched:
                results["fp_tests"]["passed"] += 1
                results["fp_tests"]["details"].append({"event": event, "result": "PASS"})
            else:
                results["fp_tests"]["failed"] += 1
                results["fp_tests"]["details"].append({"event": event, "result": "FAIL - falso positivo"})

        tp_ok = results["tp_tests"]["failed"] == 0
        fp_ok = results["fp_tests"]["failed"] == 0
        results["overall_pass"] = tp_ok and fp_ok
        return results

    def _simulate_match(self, event: dict, detection: dict) -> bool:
        """Simulazione semplificata del match. Per produzione: usa sigma-test."""
        # Logica di matching semplificata per demo
        for selector_name, selector_criteria in detection.items():
            if selector_name == 'condition':
                continue
            if not isinstance(selector_criteria, dict):
                continue
            for field, value in selector_criteria.items():
                actual_field = field.split('|')[0]
                modifier = field.split('|')[1] if '|' in field else 'exact'
                event_value = event.get(actual_field, '')
                if isinstance(value, list):
                    for v in value:
                        if self._apply_modifier(str(event_value), str(v), modifier):
                            return True
                else:
                    if self._apply_modifier(str(event_value), str(value), modifier):
                        return True
        return False

    def _apply_modifier(self, event_val: str, pattern: str, modifier: str) -> bool:
        pattern_clean = pattern.replace('*', '')
        if modifier == 'contains':
            return pattern_clean.lower() in event_val.lower()
        elif modifier == 'endswith':
            return event_val.lower().endswith(pattern_clean.lower())
        elif modifier == 'startswith':
            return event_val.lower().startswith(pattern_clean.lower())
        return event_val.lower() == pattern_clean.lower()

SigmaGen: Il Framework Open-Source per AI-Assisted Detection

SigmaGen, presentato al MITRE ATT&CK APAC 2025, rappresenta lo stato dell'arte nei framework open-source per la generazione AI di detection rules. Il progetto combina fine-tuning su dataset curati con un'architettura a pipeline che copre l'intero ciclo di vita della regola.

# Integrazione con SigmaGen (workflow concettuale)
# SigmaGen usa un approccio in tre fasi:
# 1. Ingestion di CTI (blog, advisory, STIX feeds)
# 2. Extraction di tecniche ATT&CK tramite NER specializzato
# 3. Generazione Sigma tramite modello fine-tuned

# Workflow alternativo con n8n e LLM generici:
# n8n Workflow JSON (estratto concettuale)
N8N_WORKFLOW_STRUCTURE = {
    "nodes": [
        {
            "name": "RSS_CTI_Feed",
            "type": "n8n-nodes-base.rssFeedRead",
            "parameters": {
                "url": "https://example-cti-blog.com/feed.xml"
            }
        },
        {
            "name": "Extract_Techniques",
            "type": "n8n-nodes-base.openAi",
            "parameters": {
                "model": "gpt-4o",
                "prompt": "Estrai tecniche ATT&CK da: {{$json.content}}",
                "system_prompt": "Sei un analista CTI esperto..."
            }
        },
        {
            "name": "Generate_Sigma",
            "type": "n8n-nodes-base.openAi",
            "parameters": {
                "model": "gpt-4o",
                "prompt": "Genera Sigma rule per: {{$json.techniques}}",
                "system_prompt": SIGMA_SYSTEM_PROMPT
            }
        },
        {
            "name": "Validate_Rule",
            "type": "n8n-nodes-base.code",
            "parameters": {
                "code": "// Chiama API di validazione Python"
            }
        },
        {
            "name": "GitHub_PR",
            "type": "n8n-nodes-base.github",
            "parameters": {
                "operation": "createPullRequest",
                "repository": "org/detection-rules"
            }
        }
    ]
}

# Pipeline Python completa con gestione errori
class FullAISigmaPipeline:
    def __init__(self, config: dict):
        self.llm_client = openai.OpenAI(api_key=config['openai_key'])
        self.validator = SigmaRuleValidator()
        self.tester = SigmaRuleTester()
        self.max_retries = config.get('max_retries', 3)

    def generate_validated_rule(self, technique_id: str,
                                 report: ThreatReport) -> Optional[GeneratedRule]:
        """Genera, valida e testa una regola con retry automatico."""
        errors_history = []

        for attempt in range(self.max_retries):
            # Genera la regola (con errori precedenti nel prompt se disponibili)
            sigma_yaml = self._generate_with_error_feedback(
                technique_id, report, errors_history
            )

            # Valida
            is_valid, errors = self.validator.validate(sigma_yaml)
            if not is_valid:
                errors_history.extend(errors)
                continue

            # Test con eventi sintetici
            test_events = self.tester.generate_test_events(sigma_yaml, self.llm_client)
            test_results = self.tester.run_tests(sigma_yaml, test_events)

            if test_results['overall_pass']:
                return GeneratedRule(
                    sigma_yaml=sigma_yaml,
                    mitre_techniques=[technique_id],
                    confidence=self._calculate_confidence(test_results),
                    validation_passed=True,
                    test_results=test_results
                )
            else:
                errors_history.append(f"Test falliti: {test_results}")

        return None  # Non e riuscito a generare una regola valida

    def _generate_with_error_feedback(self, technique_id: str,
                                        report: ThreatReport,
                                        errors: list[str]) -> str:
        """Genera con feedback sugli errori precedenti."""
        error_context = ""
        if errors:
            error_context = f"\n\nTentativo precedente fallito con errori:\n" + \
                           "\n".join(f"- {e}" for e in errors[-3:])  # Ultimi 3 errori

        prompt = build_generation_prompt(
            technique_id=technique_id,
            behaviors=[],
            logsource_hint="process_creation",
            report_excerpt=report.content[:2000] + error_context
        )

        response = self.llm_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": SIGMA_SYSTEM_PROMPT},
                {"role": "user", "content": prompt}
            ],
            temperature=0.1
        )

        return response.choices[0].message.content

    def _calculate_confidence(self, test_results: dict) -> float:
        """Calcola un confidence score basato sui test."""
        tp = test_results["tp_tests"]
        fp = test_results["fp_tests"]

        total_tp = tp["passed"] + tp["failed"]
        total_fp = fp["passed"] + fp["failed"]

        if total_tp == 0 or total_fp == 0:
            return 0.5

        tp_rate = tp["passed"] / total_tp
        fp_ok_rate = fp["passed"] / total_fp

        return (tp_rate + fp_ok_rate) / 2

Integrazione nella Pipeline CI/CD

Le regole generate e validate non entrano automaticamente in produzione: devono passare attraverso il processo di review umana e i gate CI/CD della pipeline Detection-as-Code. Il flusso raccomandato prevede che l'AI generi una Pull Request, non un merge diretto.

# GitHub Actions workflow per AI-generated rules
# File: .github/workflows/ai-sigma-generation.yml

"""
name: AI Sigma Rule Generation

on:
  schedule:
    - cron: '0 6 * * *'  # Ogni giorno alle 6:00 UTC
  workflow_dispatch:
    inputs:
      cti_url:
        description: 'URL del CTI report da processare'
        required: false

jobs:
  generate-rules:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: pip install openai pySigma pySigma-backend-splunk pyyaml

      - name: Run AI pipeline
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python scripts/ai_sigma_pipeline.py

      - name: Validate generated rules
        run: python scripts/validate_all_rules.py rules/ai-generated/

      - name: Create Pull Request
        uses: peter-evans/create-pull-request@v6
        with:
          title: '[AI-Generated] Detection rules da CTI feed'
          body: |
            ## Regole generate automaticamente da AI

            Tecniche ATT&CK rilevate e tradotte in Sigma rules.

            **REVIEW OBBLIGATORIA** - Verificare prima del merge:
            - [ ] Logsource corretto per il SIEM target
            - [ ] Condition logicamente corretta
            - [ ] Falsi positivi accettabili
            - [ ] Test su ambiente staging completato

          labels: 'ai-generated,needs-review'
          branch: 'ai-generated-rules'
"""

# Script di validazione bulk
# File: scripts/validate_all_rules.py
import sys
from pathlib import Path

def validate_directory(rules_dir: str) -> int:
    """Valida tutte le regole in una directory. Restituisce exit code."""
    validator = SigmaRuleValidator()
    rules_path = Path(rules_dir)
    failed = []

    for rule_file in rules_path.glob("**/*.yml"):
        content = rule_file.read_text()
        is_valid, errors = validator.validate(content)
        if not is_valid:
            failed.append((rule_file.name, errors))
            print(f"FAIL: {rule_file.name}")
            for e in errors:
                print(f"  - {e}")
        else:
            print(f"OK: {rule_file.name}")

    print(f"\nRisultati: {len(failed)} fallite su {len(list(rules_path.glob('**/*.yml')))} totali")

    return 1 if failed else 0

if __name__ == "__main__":
    sys.exit(validate_directory(sys.argv[1] if len(sys.argv) > 1 else "rules/"))

Modelli Alternativi e Considerazioni di Costo

Non tutte le organizzazioni possono o vogliono inviare dati CTI sensibili a API cloud. L'uso di modelli locali tramite Ollama o vLLM e una alternativa concreta per ambienti con requisiti di data residency.

Modello qualità Sigma Costo per 100 regole Latenza media Data residency
GPT-4o Alta (87% valide) ~$2.50 3-8s Cloud (OpenAI)
GPT-4o-mini Buona (71% valide) ~$0.15 1-3s Cloud (OpenAI)
Claude 3.5 Sonnet Alta (84% valide) ~$3.00 3-6s Cloud (Anthropic)
Llama 3.1 70B (locale) Discreta (58% valide) ~$0 (infra) 15-45s On-premise
Mistral 7B fine-tuned Buona (69% valide) ~$0 (infra) 5-15s On-premise

Strategia Cost-Effective

Usa GPT-4o-mini per la generazione iniziale e le iterazioni di retry (basso costo, buona qualità), e GPT-4o solo per i casi in cui mini fallisce dopo 2 tentativi. Con questo approccio ibrido, il costo medio scende al 30% rispetto all'uso esclusivo di GPT-4o mantenendo qualità comparabile.

Limiti Reali e Anti-Patterns

L'onesta e fondamentale nella valutazione di questa tecnologia. I LLM non sono infallibili nella generazione di detection rules, e conoscere i loro limiti e altrettanto importante quanto sfruttarne i punti di forza.

Limiti Documentati dei LLM per Detection Engineering

  • Hallucination di campi: Il modello può inventare nomi di campo che non esistono nei log reali (es. ProcessHash invece di Hashes in Sysmon).
  • Overfitting al report: Le regole generate possono essere troppo specifiche (IOC-based) invece di catturare il comportamento generale.
  • False positive non realistici: I "falsepositives" generati dai LLM sono spesso generici e non rappresentano casi reali nel contesto specifico dell'organizzazione.
  • Tecniche oscure: Per tecniche ATT&CK rare o molto recenti (post-cutoff), la qualità degrada significativamente senza retrieval aumentato.
  • Mancanza di contesto SIEM: Il modello non conosce la normalizzazione specifica del tuo SIEM (es. come Splunk normalizza i campi Windows rispetto a Elastic ECS).

RAG per Contesto SIEM-Specifico

Retrieval Augmented Generation (RAG) permette di iniettare nel prompt contesto specifico dell'organizzazione: la normalizzazione dei campi del SIEM, le regole esistenti, i dati di calibrazione dei falsi positivi. Questo approccio riduce significativamente gli errori legati alla mancanza di contesto.

# RAG per generazione Sigma contestualizzata
from chromadb import Client
from chromadb.utils import embedding_functions

class RAGSigmaGenerator:
    def __init__(self, chroma_persist_dir: str, openai_key: str):
        self.chroma_client = Client()
        self.embed_fn = embedding_functions.OpenAIEmbeddingFunction(
            api_key=openai_key,
            model_name="text-embedding-3-small"
        )

        # Collection di regole esistenti per few-shot contestuale
        self.rules_collection = self.chroma_client.get_or_create_collection(
            name="sigma_rules",
            embedding_function=self.embed_fn
        )

        # Collection di field mappings SIEM-specifici
        self.field_mappings_collection = self.chroma_client.get_or_create_collection(
            name="siem_field_mappings",
            embedding_function=self.embed_fn
        )

    def index_existing_rules(self, rules_dir: str) -> None:
        """Indicizza le regole esistenti per few-shot retrieval."""
        for rule_file in Path(rules_dir).glob("**/*.yml"):
            content = rule_file.read_text()
            rule_dict = yaml.safe_load(content)
            self.rules_collection.add(
                documents=[content],
                metadatas=[{
                    "title": rule_dict.get('title', ''),
                    "category": rule_dict.get('logsource', {}).get('category', ''),
                    "tags": str(rule_dict.get('tags', []))
                }],
                ids=[str(rule_file)]
            )

    def generate_with_rag(self, technique_id: str, report: ThreatReport) -> str:
        """Genera Sigma rule con contesto recuperato dal RAG."""

        # Recupera regole simili come few-shot examples
        similar_rules = self.rules_collection.query(
            query_texts=[report.content[:500]],
            n_results=3
        )

        # Recupera field mappings SIEM-specifici
        field_mappings = self.field_mappings_collection.query(
            query_texts=[f"process_creation windows {technique_id}"],
            n_results=2
        )

        # Costruisce prompt arricchito con contesto RAG
        context = "\n\n".join(similar_rules['documents'][0][:2])
        mappings_context = "\n".join(field_mappings['documents'][0])

        enhanced_prompt = f"""
Regole simili esistenti nel nostro repository (usa come ispirazione):
{context}

Field mappings specifici del nostro SIEM:
{mappings_context}

Ora genera una nuova regola per la tecnica {technique_id}
basandoti sul seguente report: {report.content[:2000]}
"""

        # Continua con la generazione standard...
        return enhanced_prompt

Conclusioni e Next Steps

L'AI-assisted detection engineering non e una moda: e una necessità operativa per team che devono tenere il passo con la velocità con cui emergono nuove tecniche di attacco. La combinazione di LLM per la generazione, validazione automatica rigorosa e testing con log sintetici permette di ridurre il time-to-detection da giorni a ore su un threat report appena pubblicato.

Key Takeaways

  • I LLM producono Sigma rules di qualità superiore rispetto a query SPL dirette grazie alla struttura YAML
  • Il prompt engineering (system prompt strutturato, few-shot, chain-of-thought) e determinante per la qualità
  • La validazione automatica sintattica e semantica e non opzionale
  • Il testing con log sintetici generati dall'AI chiude il loop di qualità
  • Le regole generate devono passare da PR review umana, non da merge automatico
  • RAG con contesto SIEM-specifico riduce gli errori di campo e falsi positivi
  • Modelli locali (Llama, Mistral fine-tuned) sono una alternativa valida per data residency

Articoli Correlati

  • Sigma Rules: Logica di Detection Universale e Conversione SIEM
  • Detection-as-Code Pipeline con Git e CI/CD
  • Testing Detection Rules: Unit Testing per Logica di Sicurezza
  • Threat Intelligence Ingestion: Processore Feed STIX/TAXII

Domande frequenti

I Large Language Model (LLM) sono utilizzati per generare regole di detection in Sigma, ma come funziona esattamente questo processo?

I LLM producono output di qualità significativamente superiore quando generano Sigma rispetto a query SPL o KQL dirette grazie alla struttura YAML di Sigma che impone una separazione netta tra le informazioni.

Quali sono i cinque stadi distinti di una pipeline AI-assisted per la generazione di detection rules?

I cinque stadi sono: 1) Prompt Engineering, 2) Validazione Automatica delle Regole Generate, 3) Testing Automatizzato con Log Sintetici, 4) Integrazione nella Pipeline CI/CD e 5) Modelli Alternativi e Considerazioni di Costo.

Come si può migliorare la consistenza dell'output della generazione di Sigma rules?

Includere 2-3 esempi di regole di alta qualità nel prompt (few-shot) migliora drammaticamente la consistenza dell'output, specialmente per logsource inusuali o condizioni complesse.

Perché è importante la validazione automatica delle regole generate?

La validazione automatica e il gate critico prima che una regola entri nel repository, poiché un LLM può produrre YAML sintatticamente valido ma semanticamente errato.

Fonti e approfondimenti

Prova Questi Strumenti

Strumenti gratuiti correlati a questo argomento.

Commenti

Caricamento commenti...

Accedi per commentare

Discuti questo articolo nella community

Unisciti al gruppo Telegram per discutere con altri sviluppatori, fare domande e condividere le tue esperienze.

Ti è piaciuto questo articolo?

Esplora altri contenuti sul blog o scopri i miei progetti