Forensisch onderzoek van AI-honeypots

"""
AI-honeypotsysteem voor het verzamelen van threat intelligence.
 
Implementeert een gemonitord AI-systeem dat is ontworpen om adversariële
interacties aan te trekken en vast te leggen, met behoud van veiligheidscontroles.
"""
import hashlib
import json
import time
import uuid
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any
 
@dataclass
class HoneypotConfig:
    """Configuratie voor een AI-honeypot-deployment."""
    honeypot_id: str
    fidelity_level: str  # "low", "medium", "high"
    persona: str  # Wat de honeypot voorgeeft te zijn
    model_id: str
    system_prompt: str
    deliberate_vulnerabilities: list[str]  # Welke zwakheden worden gesimuleerd
    safety_guardrails: list[str]  # Harde veiligheidslimieten die niet omzeild kunnen worden
    max_conversation_turns: int = 50
    max_token_output: int = 4096
    enabled: bool = True
 
@dataclass
class HoneypotInteraction:
    """Een volledige interactie vastgelegd door de honeypot."""
    interaction_id: str
    honeypot_id: str
    start_timestamp: float
    end_timestamp: float | None = None
    actor_ip: str | None = None
    actor_user_agent: str | None = None
    actor_api_key_hash: str | None = None
    turns: list[dict] = field(default_factory=list)
    attack_classification: dict | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
 
class AIHoneypot:
    """
    AI-honeypot die adversariële interacties vastlegt met behoud
    van veiligheidsbeperkingen.
    """
 
    def __init__(
        self,
        config: HoneypotConfig,
        model_fn,  # Functie die modelresponses genereert
        log_dir: str,
    ):
        self.config = config
        self.model_fn = model_fn
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)
        self.active_interactions: dict[str, HoneypotInteraction] = {}
 
    def start_interaction(
        self,
        actor_ip: str | None = None,
        actor_user_agent: str | None = None,
        actor_api_key: str | None = None,
    ) -> str:
        """Start een nieuwe honeypot-interactie en retourneer de interaction-ID."""
        interaction_id = str(uuid.uuid4())
 
        interaction = HoneypotInteraction(
            interaction_id=interaction_id,
            honeypot_id=self.config.honeypot_id,
            start_timestamp=time.time(),
            actor_ip=actor_ip,
            actor_user_agent=actor_user_agent,
            actor_api_key_hash=(
                hashlib.sha256(actor_api_key.encode()).hexdigest()[:16]
                if actor_api_key else None
            ),
        )
 
        self.active_interactions[interaction_id] = interaction
        return interaction_id
 
    def process_turn(
        self,
        interaction_id: str,
        user_input: str,
    ) -> str:
        """
        Verwerk een enkele gespreksbeurt in de honeypot.
 
        Legt de input vast, genereert een response (mogelijk met
        gesimuleerde kwetsbaarheden) en legt de output vast.
        """
        interaction = self.active_interactions.get(interaction_id)
        if not interaction:
            raise ValueError(f"Unknown interaction: {interaction_id}")
 
        if len(interaction.turns) >= self.config.max_conversation_turns:
            return "I'm sorry, I need to end this conversation. Please start a new session."
 
        # Leg de gebruikersinput vast met volledig forensisch detail
        turn_record = {
            "turn_number": len(interaction.turns) + 1,
            "timestamp": time.time(),
            "role": "user",
            "content": user_input,
            "content_hash": hashlib.sha256(user_input.encode()).hexdigest(),
            "content_length": len(user_input),
        }
 
        # Controleer harde veiligheids-guardrails (deze geven nooit toe)
        safety_check = self._check_hard_safety(user_input)
        if safety_check["blocked"]:
            response = "I'm not able to help with that request."
            turn_record["safety_blocked"] = True
            turn_record["safety_reason"] = safety_check["reason"]
        else:
            # Genereer response met gesimuleerde kwetsbaarheden
            response = self._generate_response(interaction, user_input)
 
        # Leg de response vast
        response_record = {
            "turn_number": len(interaction.turns) + 1,
            "timestamp": time.time(),
            "role": "assistant",
            "content": response,
            "content_hash": hashlib.sha256(response.encode()).hexdigest(),
            "content_length": len(response),
        }
 
        interaction.turns.append(turn_record)
        interaction.turns.append(response_record)
 
        # Persisteer interactiestatus na elke beurt
        self._save_interaction(interaction)
 
        return response
 
    def _generate_response(
        self,
        interaction: HoneypotInteraction,
        user_input: str,
    ) -> str:
        """
        Genereer een modelresponse en simuleer optioneel kwetsbaarheden.
 
        De honeypot kan er doelbewust voor "trappen" in bepaalde
        aanvalstechnieken om het vervolggedrag van de aanvaller te observeren.
        """
        # Bouw gespreksgeschiedenis voor het model op
        messages = [{"role": "system", "content": self.config.system_prompt}]
        for turn in interaction.turns:
            messages.append({
                "role": turn["role"],
                "content": turn["content"],
            })
        messages.append({"role": "user", "content": user_input})
 
        # Roep het model aan
        response = self.model_fn(messages)
 
        return response[:self.config.max_token_output]
 
    def _check_hard_safety(self, user_input: str) -> dict:
        """
        Controleer harde veiligheids-guardrails die nooit worden omzeild.
 
        Ook al is dit een honeypot, we handhaven bepaalde
        absolute veiligheidslimieten om te voorkomen dat de honeypot
        als wapen wordt ingezet tegen echte targets.
        """
        # Deze patronen leiden tot onmiddellijke blokkering, ongeacht
        # de honeypotconfiguratie
        hard_blocks = [
            "real person's",  # Gericht op echte individuen
            "actual address",  # Real-world targeting
            "child",  # CSAM-gerelateerd
        ]
 
        input_lower = user_input.lower()
        for pattern in hard_blocks:
            if pattern in input_lower:
                return {"blocked": True, "reason": f"Hard safety: {pattern}"}
 
        return {"blocked": False, "reason": None}
 
    def end_interaction(self, interaction_id: str) -> HoneypotInteraction:
        """Beëindig een interactie en finaliseer het record."""
        interaction = self.active_interactions.pop(interaction_id, None)
        if not interaction:
            raise ValueError(f"Unknown interaction: {interaction_id}")
 
        interaction.end_timestamp = time.time()
        self._save_interaction(interaction)
        return interaction
 
    def _save_interaction(self, interaction: HoneypotInteraction) -> None:
        path = self.log_dir / f"{interaction.interaction_id}.json"
        path.write_text(json.dumps(asdict(interaction), default=str, indent=2))

class HoneypotForensicAnalyzer:
    """
    Analyseer vastgelegde honeypot-interacties om aanvallen te classificeren
    en threat intelligence te extraheren.
    """
 
    def classify_interaction(
        self,
        interaction: HoneypotInteraction,
    ) -> dict:
        """
        Classificeer de aanvalstechnieken die in een honeypot-interactie zijn waargenomen.
 
        Koppelt waargenomen gedragingen aan MITRE ATLAS-technieken en
        kent een algehele dreigingsbeoordeling toe.
        """
        user_turns = [
            t for t in interaction.turns if t["role"] == "user"
        ]
 
        if not user_turns:
            return {"classification": "no_user_input"}
 
        techniques_observed = []
        all_content = " ".join(t["content"] for t in user_turns).lower()
 
        # Detectie van prompt-injectie
        injection_indicators = [
            "ignore previous", "ignore above", "disregard",
            "new instructions", "you are now", "pretend you are",
            "act as", "roleplay as", "jailbreak",
            "dan", "developer mode",
        ]
        injection_count = sum(
            1 for ind in injection_indicators if ind in all_content
        )
        if injection_count > 0:
            techniques_observed.append({
                "technique": "Prompt Injection",
                "atlas_id": "AML.T0051",
                "confidence": min(0.5 + injection_count * 0.15, 1.0),
                "indicators": [
                    ind for ind in injection_indicators if ind in all_content
                ],
            })
 
        # Extractie van system prompt
        extraction_indicators = [
            "system prompt", "initial instructions", "repeat your instructions",
            "what are your rules", "what were you told", "show me your prompt",
            "print your system", "reveal your instructions",
        ]
        extraction_count = sum(
            1 for ind in extraction_indicators if ind in all_content
        )
        if extraction_count > 0:
            techniques_observed.append({
                "technique": "System Prompt Extraction",
                "atlas_id": "AML.T0046",
                "confidence": min(0.5 + extraction_count * 0.2, 1.0),
                "indicators": [
                    ind for ind in extraction_indicators if ind in all_content
                ],
            })
 
        # Probing voor data-exfiltratie
        exfil_indicators = [
            "training data", "what data were you trained on",
            "list users", "show me other conversations",
            "database", "api key", "secret", "password",
            "internal", "confidential",
        ]
        exfil_count = sum(
            1 for ind in exfil_indicators if ind in all_content
        )
        if exfil_count > 0:
            techniques_observed.append({
                "technique": "Data Exfiltration Probing",
                "atlas_id": "AML.T0024",
                "confidence": min(0.4 + exfil_count * 0.15, 1.0),
                "indicators": [
                    ind for ind in exfil_indicators if ind in all_content
                ],
            })
 
        # Analyse van multi-turn-escalatie
        escalation = self._analyze_escalation_pattern(user_turns)
 
        # Beoordeling van verfijning
        sophistication = self._assess_sophistication(user_turns, techniques_observed)
 
        return {
            "interaction_id": interaction.interaction_id,
            "techniques_observed": techniques_observed,
            "technique_count": len(techniques_observed),
            "escalation_pattern": escalation,
            "sophistication": sophistication,
            "turn_count": len(user_turns),
            "duration_seconds": (
                (interaction.end_timestamp or time.time())
                - interaction.start_timestamp
            ),
            "overall_threat_level": self._compute_threat_level(
                techniques_observed, escalation, sophistication
            ),
        }
 
    def _analyze_escalation_pattern(self, user_turns: list[dict]) -> dict:
        """
        Analyseer hoe de aanvaller zijn aanpak in de loop van de tijd escaleerde.
 
        Geavanceerde aanvallers beginnen vaak met goedaardige probing
        en escaleren geleidelijk naar agressievere technieken.
        """
        if len(user_turns) < 3:
            return {"pattern": "insufficient_data", "turns": len(user_turns)}
 
        # Eenvoudige heuristiek: volg de gemiddelde contentlengte per beurtsegment
        third = len(user_turns) // 3
        early = user_turns[:third]
        middle = user_turns[third:2*third]
        late = user_turns[2*third:]
 
        early_avg_len = sum(t["content_length"] for t in early) / max(len(early), 1)
        middle_avg_len = sum(t["content_length"] for t in middle) / max(len(middle), 1)
        late_avg_len = sum(t["content_length"] for t in late) / max(len(late), 1)
 
        if early_avg_len < middle_avg_len < late_avg_len:
            pattern = "escalating"
        elif early_avg_len > late_avg_len:
            pattern = "front_loaded"
        else:
            pattern = "stable"
 
        return {
            "pattern": pattern,
            "early_avg_length": round(early_avg_len, 1),
            "late_avg_length": round(late_avg_len, 1),
        }
 
    def _assess_sophistication(
        self,
        user_turns: list[dict],
        techniques: list[dict],
    ) -> str:
        """Beoordeel het verfijningsniveau van de aanvaller."""
        if not techniques:
            return "benign_or_unrecognized"
 
        # Factoren die op verfijning wijzen:
        # - Meerdere technieken gebruikt
        # - Multi-turn-strategieën
        # - Nieuwe of ongebruikelijke aanvalspatronen
        score = 0
        score += min(len(techniques), 3)  # Diversiteit van technieken
        score += min(len(user_turns) // 5, 3)  # Volharding
        avg_confidence = sum(t["confidence"] for t in techniques) / len(techniques)
        score += 2 if avg_confidence > 0.7 else 1 if avg_confidence > 0.4 else 0
 
        if score >= 6:
            return "advanced"
        if score >= 3:
            return "intermediate"
        return "basic"
 
    def _compute_threat_level(
        self,
        techniques: list[dict],
        escalation: dict,
        sophistication: str,
    ) -> str:
        if not techniques:
            return "NONE"
        if sophistication == "advanced" and len(techniques) >= 3:
            return "CRITICAL"
        if sophistication == "advanced" or len(techniques) >= 2:
            return "HIGH"
        if len(techniques) >= 1:
            return "MEDIUM"
        return "LOW"

def produce_threat_intelligence(
    classified_interactions: list[dict],
    reporting_period_days: int = 7,
) -> dict:
    """
    Aggregeer geclassificeerde honeypot-interacties tot een
    threat intelligence-rapport.
    """
    total = len(classified_interactions)
    if total == 0:
        return {"status": "no_data"}
 
    # Frequentieanalyse van technieken
    technique_counts: dict[str, int] = {}
    sophistication_counts: dict[str, int] = {}
    threat_level_counts: dict[str, int] = {}
 
    for interaction in classified_interactions:
        for tech in interaction.get("techniques_observed", []):
            name = tech["technique"]
            technique_counts[name] = technique_counts.get(name, 0) + 1
 
        soph = interaction.get("sophistication", "unknown")
        sophistication_counts[soph] = sophistication_counts.get(soph, 0) + 1
 
        level = interaction.get("overall_threat_level", "NONE")
        threat_level_counts[level] = threat_level_counts.get(level, 0) + 1
 
    # Sorteer technieken op frequentie
    sorted_techniques = sorted(
        technique_counts.items(), key=lambda x: x[1], reverse=True
    )
 
    return {
        "reporting_period_days": reporting_period_days,
        "total_interactions": total,
        "interactions_with_attacks": sum(
            1 for i in classified_interactions
            if i.get("technique_count", 0) > 0
        ),
        "attack_rate": sum(
            1 for i in classified_interactions
            if i.get("technique_count", 0) > 0
        ) / total,
        "top_techniques": [
            {"technique": name, "count": count, "percentage": round(count / total * 100, 1)}
            for name, count in sorted_techniques[:10]
        ],
        "sophistication_distribution": sophistication_counts,
        "threat_level_distribution": threat_level_counts,
        "recommendations": _generate_recommendations(sorted_techniques, threat_level_counts),
    }
 
def _generate_recommendations(
    techniques: list[tuple[str, int]],
    threat_levels: dict[str, int],
) -> list[str]:
    """Genereer verdedigingsaanbevelingen uit honeypot-intelligence."""
    recommendations = []
 
    if techniques:
        top_technique = techniques[0][0]
        recommendations.append(
            f"Highest observed technique: {top_technique}. "
            f"Prioritize defenses against this attack class."
        )
 
    critical_high = threat_levels.get("CRITICAL", 0) + threat_levels.get("HIGH", 0)
    total = sum(threat_levels.values())
    if total > 0 and critical_high / total > 0.2:
        recommendations.append(
            "Over 20% of interactions show high/critical threat level. "
            "Consider increasing monitoring on production systems."
        )
 
    return recommendations