AI 蜜罐鑑識

"""
AI honeypot system for threat intelligence collection.
 
Deploys a monitored AI system designed to attract and record
adversarial interactions while maintaining safety controls.
"""
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:
    """Configuration for an AI honeypot deployment."""
    honeypot_id: str
    fidelity_level: str  # "low", "medium", "high"
    persona: str  # What the honeypot pretends to be
    model_id: str
    system_prompt: str
    deliberate_vulnerabilities: list[str]  # What weaknesses are simulated
    safety_guardrails: list[str]  # Hard safety limits that cannot be bypassed
    max_conversation_turns: int = 50
    max_token_output: int = 4096
    enabled: bool = True
 
@dataclass
class HoneypotInteraction:
    """A complete interaction captured by the 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 that records adversarial interactions while
    maintaining safety constraints.
    """
 
    def __init__(
        self,
        config: HoneypotConfig,
        model_fn,  # Function that generates model responses
        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 a new honeypot interaction and return the 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:
        """
        Process a single conversation turn in the honeypot.
 
        Records the input, generates a response (potentially with
        simulated vulnerabilities), and records the output.
        """
        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."
 
        # Record the user input with full forensic 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),
        }
 
        # Check hard safety guardrails (these never yield)
        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:
            # Generate response with simulated vulnerabilities
            response = self._generate_response(interaction, user_input)
 
        # Record the response
        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)
 
        # Persist interaction state after each turn
        self._save_interaction(interaction)
 
        return response
 
    def _generate_response(
        self,
        interaction: HoneypotInteraction,
        user_input: str,
    ) -> str:
        """
        Generate a model response, optionally simulating vulnerabilities.
 
        The honeypot may deliberately appear to "fall for" certain
        attack techniques to observe the attacker's follow-up behavior.
        """
        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})
 
        response = self.model_fn(messages)
        return response[:self.config.max_token_output]
 
    def _check_hard_safety(self, user_input: str) -> dict:
        """
        Check hard safety guardrails that are never bypassed.
        """
        hard_blocks = [
            "real person's",
            "actual address",
            "child",
        ]
 
        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:
        """End an interaction and finalize the 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:
    """
    Analyze captured honeypot interactions to classify attacks
    and extract threat intelligence.
    """
 
    def classify_interaction(
        self,
        interaction: HoneypotInteraction,
    ) -> dict:
        """
        Classify the attack techniques observed in a honeypot interaction.
        """
        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()
 
        # Prompt injection detection
        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
                ],
            })
 
        # System prompt extraction
        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
                ],
            })
 
        # Data exfiltration probing
        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
                ],
            })
 
        escalation = self._analyze_escalation_pattern(user_turns)
        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:
        """Analyze how the attacker escalated their approach over time."""
        if len(user_turns) < 3:
            return {"pattern": "insufficient_data", "turns": len(user_turns)}
 
        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:
        """Assess the attacker's sophistication level."""
        if not techniques:
            return "benign_or_unrecognized"
 
        score = 0
        score += min(len(techniques), 3)
        score += min(len(user_turns) // 5, 3)
        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:
    """Aggregate classified honeypot interactions into a report."""
    total = len(classified_interactions)
    if total == 0:
        return {"status": "no_data"}
 
    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
 
    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
        ),
        "top_techniques": [
            {"technique": name, "count": count}
            for name, count in sorted_techniques[:10]
        ],
        "sophistication_distribution": sophistication_counts,
        "threat_level_distribution": threat_level_counts,
    }