Quels logs collecter sur un système LLM pour un SOC

{
  // Section 1, Identification (5)
  "event_id": "evt_018f...",
  "ts": "2026-05-01T10:23:45.123Z",
  "request_id": "req_abc123",
  "trace_id": "0af7651916cd43dd8448eb211c80319c",
  "parent_span_id": "b7ad6b7169203331",
  
  // Section 2, Acteur (5)
  "user_id_hash": "sha256_a1b2c3...",
  "user_role": "customer",
  "org_id": "org_zeroday",
  "session_id": "sess_xyz789",
  "source_ip_hash": "sha256_d4e5f6...",
  
  // Section 3, Source (5)
  "app_name": "zerodaysupport-chatbot",
  "app_version": "2.4.1",
  "environment": "production",
  "region": "eu-west-3",
  "hostname": "chatbot-prod-7f9b",
  
  // Section 4, Modèle (6)
  "model": {
    "provider": "openai",
    "name": "gpt-4o",
    "version": "2026-04-15",
    "temperature": 0.3,
    "max_tokens": 1500,
    "finish_reason": "stop"
  },
  
  // Section 5, Contenu (8)
  "content": {
    "prompt_redacted": "Bonjour, [EMAIL_REDACTED] commande #1234...",
    "response_redacted": "Votre commande sera livrée le...",
    "prompt_tokens": 245,
    "completion_tokens": 187,
    "total_tokens": 432,
    "rag_doc_ids": ["doc_42", "doc_91"],
    "tool_calls": [
      {"name": "search_order", "args_hash": "sha256_g7h8i9..."}
    ],
    "message_history_length": 3
  },
  
  // Section 6, Métriques (6)
  "metrics": {
    "duration_ms": 1850,
    "latency_breakdown_ms": {
      "guardrail_input": 78,
      "rag_retrieval": 145,
      "llm_call": 1320,
      "tool_exec": 250,
      "guardrail_output": 32
    },
    "cost_usd": 0.018,
    "cache_hit": false,
    "retry_count": 0,
    "queue_time_ms": 25
  },
  
  // Section 7, Sécurité (8)
  "security": {
    "guardrail_input_score": 0.05,
    "guardrail_input_action": "allow",
    "guardrail_output_alerts": [],
    "anomaly_flags": [],
    "rate_limit_dimension_hit": null,
    "mitre_atlas_techniques": [],
    "owasp_llm_classes": [],
    "severity": "informational"
  },
  
  // Section 8, Conformité (7)
  "compliance": {
    "data_classification": "internal",
    "retention_class": "security_1y",
    "deletion_eligible_after": "2027-05-01T10:23:45Z",
    "gdpr_lawful_basis": "legitimate_interest",
    "dpia_reference": "DPIA-2026-014",
    "audit_trail_id": "audit_chatbot_v2",
    "schema_version": "1.3"
  }
}

{
  "metadata": {
    "version": "1.2.0",
    "product": {
      "vendor_name": "ZerodaySupport",
      "name": "ChatBot SAV",
      "version": "2.4.1"
    },
    "log_name": "llm_application_event",
    "log_version": "1.3"
  },
  
  "category_uid": 7,
  "category_name": "Application Activity",
  "class_uid": 7000,
  "class_name": "AI Application Activity",
  "type_uid": 700001,
  "type_name": "AI Inference",
  
  "time": 1746104625123,
  "severity_id": 1,
  "severity": "Informational",
  
  "actor": {
    "user": {
      "uid": "sha256_a1b2c3",
      "type": "User",
      "type_id": 1
    },
    "session": {
      "uid": "sess_xyz789"
    }
  },
  
  "src_endpoint": {
    "ip": "203.0.113.42",
    "hostname_hash": "sha256_d4e5f6"
  },
  
  "device": {
    "name": "chatbot-prod-7f9b",
    "region": "eu-west-3"
  },
  
  // Custom attributes for AI/LLM
  "gen_ai": {
    "system": "openai",
    "request": {
      "model": "gpt-4o",
      "temperature": 0.3,
      "max_tokens": 1500
    },
    "response": {
      "model": "gpt-4o-2026-04-15",
      "finish_reasons": ["stop"]
    },
    "usage": {
      "prompt_tokens": 245,
      "completion_tokens": 187,
      "total_tokens": 432
    },
    "content": {
      "prompt_redacted": "Bonjour, [EMAIL_REDACTED] commande #1234...",
      "response_redacted": "Votre commande sera livrée..."
    },
    "tools": [
      {"name": "search_order", "args_hash": "sha256_g7h8i9"}
    ],
    "guardrails": {
      "input_score": 0.05,
      "input_action": "allow",
      "output_alerts": []
    },
    "cost": {
      "amount": 0.018,
      "currency": "EUR"
    }
  },
  
  // Pour findings sécurité (events anormaux)
  "finding": {
    "title": null,
    "uid": null,
    "atlas_techniques": [],
    "owasp_llm_classes": []
  },
  
  // Compliance
  "compliance": {
    "data_classification": "internal",
    "retention_class": "security_1y",
    "deletion_eligible_after": 1777640625000
  }
}

{
  "metadata": { "version": "1.2.0", "product": {...} },
  "category_uid": 2,
  "category_name": "Findings",
  "class_uid": 2007,
  "class_name": "Detection Finding",
  "type_uid": 200701,
  "type_name": "AI Behavioral Anomaly Detected",
  
  "time": 1746104825000,
  "severity_id": 4,
  "severity": "High",
  
  "actor": { "user": { "uid": "sha256_a1b2c3" } },
  
  "finding": {
    "title": "DoW Suspected, User cost > 3× baseline",
    "uid": "finding_018g4h",
    "details": "User accumulated 45 € in 1 hour vs baseline 4.5 €/h",
    "types": ["Cost Anomaly", "Possible DoW"],
    "atlas_techniques": ["AML.T0034"],
    "owasp_llm_classes": ["LLM10"]
  },
  
  "raw_data": {
    "cost_1h_actual": 50.23,
    "cost_1h_baseline": 5.20,
    "ratio": 9.66,
    "request_count_1h": 234
  }
}

title: LLM Cost Anomaly Per User
id: 018f2a3b-c4d5-6e7f-8901-2345abcdef01
status: experimental
description: User cost > 3× baseline → DoW potential
author: ZerodaySupport SOC
date: 2026-05-01
logsource:
  product: zerodaysupport
  service: llm_application
detection:
  selection:
    class_name: AI Application Activity
  timeframe: 1h
  condition: |
    sum(gen_ai.cost.amount) by actor.user.uid 
    > 3 * avg_over_time(sum(gen_ai.cost.amount) by actor.user.uid, 7d)
falsepositives:
  - Legitimate burst usage (verify with user manager)
level: high
tags:
  - attack.impact
  - atlas.AML.T0034
  - owasp.llm10

index=llm sourcetype="llm:application:json" 
| bin _time span=1h
| stats sum(gen_ai.cost.amount) as cost_1h by actor.user.uid, _time
| eventstats avg(cost_1h) as baseline_avg, stdev(cost_1h) as baseline_stdev by actor.user.uid
| where cost_1h > (baseline_avg + 3*baseline_stdev) AND cost_1h > 5
| table _time, actor.user.uid, cost_1h, baseline_avg, ratio

LLMEvents_CL
| where TimeGenerated > ago(1h)
| summarize cost_1h = sum(cost_usd_d) by user_id_hash_s, bin(TimeGenerated, 1h)
| join kind=inner (
    LLMEvents_CL
    | where TimeGenerated between (ago(7d) .. ago(1h))
    | summarize baseline = avg(cost_usd_d) by user_id_hash_s
) on user_id_hash_s
| where cost_1h > 3 * baseline and cost_1h > 5.0
| project TimeGenerated, user_id_hash_s, cost_1h, baseline, ratio = cost_1h / baseline

index=llm sourcetype="llm:application:json"
| where actor.user.org_id != "" AND finding.types contains "cross_tenant"
OR (rag_doc_ids contains "*" AND tenant_filter_applied=false)
| eval severity="critical"
| table _time, actor.user.uid, source_tenant, target_tenant, rag_doc_ids

detection:
  selection:
    class_name: AI Application Activity
    tool_calls.count: ">10"
  condition: selection
level: medium
tags:
  - atlas.AML.T0034
  - owasp.llm06
  - owasp.llm10

index=auth source=llm-app
| stats count by src.ip.hash, _time span=1m
| where count > 10
| join src.ip.hash [
    search index=auth event=login_failed
    | stats count by src.ip.hash
]

title: LLM Prompt Injection Pattern Detected
detection:
  selection:
    class_name: AI Application Activity
    gen_ai.guardrails.input_action: block
    gen_ai.guardrails.input_score: ">0.95"
  condition: selection
level: medium
falsepositives:
  - Legitimate research/security testing
tags:
  - atlas.AML.T0051
  - owasp.llm01

index=llm sourcetype="llm:application:json"
| where match(gen_ai.content.response_redacted, "(?i)Eva|ZerodaySupport|INSTRUCTION HIERARCHY")
| eval finding="system_prompt_leak"

index=llm sourcetype="llm:application:json"
| where match(gen_ai.content.response_redacted, "!\[.*\]\(https?://(?!zerodaysupport\.com)")
| eval finding="markdown_image_external_url"
| eval severity="high"

LLMEvents_CL
| where TimeGenerated > ago(15m)
| summarize 
    refusal_rate = countif(response_contains_refusal_b) * 1.0 / count(),
    msg_count = count()
  by session_id_s
| where refusal_rate > 0.3 and msg_count > 5
| project session_id_s, refusal_rate, msg_count

index=llm sourcetype="llm:application:json"
| spath gen_ai.tools{}.name output=tool_names
| where tool_names IN ("refund", "send_email_external", "delete_file", "modify_permissions")
| where human_approval=false
| table _time, actor.user.uid, tool_names, status

index=llm sourcetype="llm:application:json"
| bin _time span=1h
| stats count by actor.user.uid, _time
| eventstats avg(count) as baseline_avg by actor.user.uid
| where count > 10 * baseline_avg AND count > 100
| eval finding="volume_anomaly"

# Application : envoyer logs via HEC (HTTP Event Collector)
import httpx
import os
import json
 
SPLUNK_HEC_URL = os.environ["SPLUNK_HEC_URL"]
SPLUNK_HEC_TOKEN = os.environ["SPLUNK_HEC_TOKEN"]
 
async def send_to_splunk(event: dict):
    async with httpx.AsyncClient() as client:
        await client.post(
            SPLUNK_HEC_URL,
            headers={"Authorization": f"Splunk {SPLUNK_HEC_TOKEN}"},
            json={
                "time": event["time"] / 1000,
                "host": event["device"]["name"],
                "source": event["metadata"]["product"]["name"],
                "sourcetype": "llm:application:json",
                "index": "llm",
                "event": event,
            },
        )

# Splunk : data model AI Activity custom
| pivot AI_Activity llm_application_activity
| eventstats avg(gen_ai.cost.amount) as avg_cost by actor.user.uid
| ...

# Application : envoyer logs vers Log Analytics workspace
from azure.monitor.ingestion import LogsIngestionClient
from azure.identity import DefaultAzureCredential
 
client = LogsIngestionClient(
    endpoint=os.environ["DCE_ENDPOINT"],
    credential=DefaultAzureCredential(),
)
 
async def send_to_sentinel(events: list):
    client.upload(
        rule_id=os.environ["DCR_IMMUTABLE_ID"],
        stream_name="Custom-LLMEvents_CL",
        logs=events,
    )

// Sentinel : Analytics rule custom
LLMEvents_CL
| where TimeGenerated > ago(1h)
| where cost_usd_d > 50.0
| project TimeGenerated, user_id_hash_s, cost_usd_d, severity_s
| extend MitreAttackTactics = "Impact"
| extend AtlasAttackTechniques = "AML.T0034"

# Filebeat config
filebeat.inputs:
  - type: filestream
    paths:
      - /var/log/llm-app/*.log
    parsers:
      - ndjson:
          target: ""
          add_error_key: true
 
processors:
  - add_fields:
      target: event
      fields:
        category: ai_activity
        type: llm_inference
  
output.elasticsearch:
  hosts: ["https://elastic:9200"]
  index: "llm-events-%{+yyyy.MM.dd}"

# Elastic Detection Rule (YAML)
description: LLM cost anomaly detected
risk_score: 65
severity: high
type: threshold
language: kuery
query: |
  event.category : "ai_activity" and gen_ai.cost.amount : > 5
threshold:
  field: actor.user.id
  value: 1
  cardinality:
    - field: actor.user.id
      value: 1
risk_score_mapping: []
threat:
  - framework: MITRE ATLAS
    technique:
      - id: AML.T0034
        name: Cost Harvesting

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
 
processors:
  batch:
    timeout: 5s
    send_batch_size: 1000
  
  attributes:
    actions:
      - key: gen_ai.content.prompt_redacted
        action: hash
        regex_pattern: ".*"
 
exporters:
  splunk_hec:
    token: ${SPLUNK_HEC_TOKEN}
    endpoint: ${SPLUNK_HEC_URL}
    source: llm-app
    sourcetype: llm:application:json
  
  azuremonitor:
    instrumentation_key: ${AZURE_MONITOR_KEY}
  
  elasticsearch:
    endpoints: ["https://elastic:9200"]
    index: llm-events
 
service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [batch, attributes]
      exporters: [splunk_hec, azuremonitor, elasticsearch]

# Application : redaction + pseudonymisation + sampling
import hmac
import hashlib
import random
from presidio_analyzer import AnalyzerEngine
 
PSEUDO_KEY = os.environ["PSEUDO_HMAC_KEY"].encode()
analyzer = AnalyzerEngine()
 
def pseudonymize_user_id(user_id: str) -> str:
    return hmac.new(PSEUDO_KEY, user_id.encode(), hashlib.sha256).hexdigest()
 
def redact_pii(text: str, language: str = "fr") -> str:
    results = analyzer.analyze(
        text=text,
        language=language,
        entities=["EMAIL_ADDRESS", "PHONE_NUMBER", "PERSON", "FR_NIR", "IBAN_CODE"],
    )
    for r in sorted(results, key=lambda x: x.start, reverse=True):
        text = text[:r.start] + f"[{r.entity_type}_REDACTED]" + text[r.end:]
    return text
 
 
def should_log_full(event_severity: str, event_type: str) -> bool:
    """Sampling intelligent."""
    # 100% events sécurité
    if event_severity in {"high", "critical"}:
        return True
    # 100% events sensibles
    if event_type in {"tool_call_sensitive", "guardrail_block", "anomaly_flag"}:
        return True
    # 10% events normaux (informational)
    return random.random() < 0.1
 
 
async def emit_log(event: dict):
    # 1. Redaction
    if "prompt" in event.get("gen_ai", {}).get("content", {}):
        event["gen_ai"]["content"]["prompt_redacted"] = redact_pii(
            event["gen_ai"]["content"]["prompt"]
        )
        del event["gen_ai"]["content"]["prompt"]
    
    if "response" in event.get("gen_ai", {}).get("content", {}):
        event["gen_ai"]["content"]["response_redacted"] = redact_pii(
            event["gen_ai"]["content"]["response"]
        )
        del event["gen_ai"]["content"]["response"]
    
    # 2. Pseudonymisation
    if "user_id" in event.get("actor", {}).get("user", {}):
        event["actor"]["user"]["uid"] = pseudonymize_user_id(
            event["actor"]["user"]["user_id"]
        )
        del event["actor"]["user"]["user_id"]
    
    # 3. Sampling
    if not should_log_full(event["severity"], event["class_name"]):
        return
    
    # Send
    await siem_emit(event)

RETENTION_POLICIES = {
    "security_critical": "1825d",   # 5 ans (audit, incidents)
    "security_1y": "365d",           # 1 an (events sécurité standard)
    "informational": "90d",          # 3 mois (events normaux)
    "debug": "14d",                  # 14 jours (debug only)
}

async def erase_user_logs(user_id: str):
    """
    Process RGPD right to erasure.
    Identifier tous les events log par user_id_hash et delete.
    """
    user_id_hash = pseudonymize_user_id(user_id)
    
    # Splunk
    await splunk_delete_query(f'user_id_hash="{user_id_hash}"')
    
    # Sentinel
    await sentinel_delete_query(f'user_id_hash_s == "{user_id_hash}"')
    
    # Elastic
    await elastic_delete_by_query("llm-events-*", {"term": {"user_id_hash": user_id_hash}})
    
    # Audit trail de la deletion
    await audit_log({
        "action": "rgpd_erasure",
        "user_id_hash": user_id_hash,
        "timestamp": datetime.utcnow().isoformat(),
        "operator": current_user_id,
    })