锻造"认知免疫系统"：Agent对抗防御、红队自动化与运行时行为免疫实战

┌─────────────────────────────────────────────────────────────────────┐
│            2026 Agent Cognitive Immunity Architecture               │
├─────────────────────────────────────────────────────────────────────┤
│  [Response Layer: Graded Intervention / Auto-Recovery / Forensics]  │
│      ↓                                                              │
│  [Layer 1: 对抗检测层] ← Semantic Classifier / Intent Analyzer     │
│   ├─ 多模态输入的语义级恶意意图识别                                    │
│   ├─ 上下文感知的间接注入检测                                          │
│   └─ 对抗样本的在线学习与模型热更新                                     │
│      ↓                                                              │
│  [Layer 2: 红队自动化层] ← Attack Orchestrator / CI Gate / Metric  │
│   ├─ 结构化攻击策略库与自动编排                                        │
│   ├─ 安全回归测试与发布门禁                                            │
│   └─ 安全水位的可度量、可追踪、可对标                                   │
│      ↓                                                              │
│  [Layer 3: 运行时免疫层] ← Behavior Baseline / Anomaly Detect      │
│   ├─ Agent行为语义画像与动态基线                                       │
│   ├─ 自适应异常检测与误报抑制                                          │
│   └─ 分级响应与自动恢复                                                │
└─────────────────────────────────────────────────────────────────────┘

pip install pydantic transformers torch opentelemetry-api redis scikit-learn
# 部署: OpenTelemetry Collector + Redis (特征缓存) + MLflow (模型版本) + PostgreSQL (对抗样本库) + GPU推理服务

"""
adversarial_detection_engine.py - 语义级对抗检测与在线学习引擎
技术栈: Transformers / PyTorch / Pydantic / OpenTelemetry
"""
from typing import Dict, List, Any, Optional, Tuple
from pydantic import BaseModel, Field
from enum import Enum
import asyncio
import time
import uuid
import json
import numpy as np
from dataclasses import dataclass, field

class AttackCategory(str, Enum):
    DIRECT_JAILBREAK = "direct_jailbreak"
    INDIRECT_INJECTION = "indirect_injection"
    TOOL_CHAIN_POISONING = "tool_chain_poisoning"
    DATA_EXFILTRATION = "data_exfiltration"
    PRIVACY_PROBING = "privacy_probing"
    BENIGN = "benign"

class DetectionVerdict(str, Enum):
    SAFE = "safe"
    SUSPICIOUS = "suspicious"
    MALICIOUS = "malicious"

@dataclass
class InputAnalysisResult:
    """输入分析结果"""
    request_id: str
    verdict:31268.t.kuaisou.com
    attack_category: Optional[AttackCategory] = None
    confidence: float = 0.0
    risk_signals: List[str] = field(default_factory=list)
    semantic_embedding: Optional[np.ndarray] = None
    latency_ms: float = 0.0

class AdversarialDetectionEngine:
    """对抗检测引擎"""

    # 风险信号→类别映射
    SIGNAL_CATEGORY_MAP = {
        "ignore_previous instructions": AttackCategory.DIRECT_JAILBREAK,
        "system prompt extraction": AttackCategory.DIRECT_JAILBREAK,
        "hidden instruction in document": AttackCategory.INDIRECT_INJECTION,
        "encoded payload in base64": AttackCategory.INDIRECT_INJECTION,
        "unusual tool parameter pattern": AttackCategory.TOOL_CHAIN_POISONING,
        "request for internal api key": AttackCategory.DATA_EXFILTRATION,
        "personal data inference attempt": AttackCategory.PRIVACY_PROBING,
    }

    def __init__(self, semantic_classifier, embedding_model,
                 sample_store, model_registry, otel_tracer):
        self.classifier = semantic_classifier   # Fine-tuned DeBERTa/RoBERTa
        self.embedder = embedding_model         # Sentence-transformer
        self.samples = 31269.t.kuaisou.com      # 对抗样本库
        self.registry = model_registry          # MLflow
        self.tracer = otel_tracer
        self._online_buffer: List[Dict] = []    # 在线学习缓冲
        self._current_model_version: str = "v1.0"

    async def analyze_input(self, request_id: str,
                             user_input: str,
                             context: Optional[Dict] = None,
                             attached_documents: Optional[List[str]] = None) -> InputAnalysisResult:
        """语义级输入分析"""
        start = time.time()

        with self.tracer.start_as_current_span("adversarial.detect") as span:
            span.set_attribute("request.id", request_id)

            # Step 1: 多源语义融合
            signals = []
            
            # 用户输入直接分析
            user_signals = await self._extract_risk_signals(user_input)
            signals.extend(user_signals)

            # 附件内容分析（间接注入主战场）
            if attached_documents:
                for doc in attached_documents:
                    doc_signals = await self._extract_risk_signals(doc, is_document=True)
                    signals.extend([f"doc:{s}" for s in doc_signals])

            # 上下文一致性检查
            if context:
                ctx_signals = await self._check_context_consistency(user_input, context)
                signals.extend(ctx_signals)

            # Step 2: 语义分类
            embedding = await self.embedder.encode(user_input)
            classification = await self.classifier.predict(embedding, signals)

            verdict = DetectionVerdict(classification["verdict"])
            attack_cat = None
            if verdict != DetectionVerdict.SAFE:
                attack_cat = self._infer_attack_category(signals, classification)

            latency = (time.time() - start) * 1000
            result = InputAnalysisResult(
                request_id=request_id,
                verdict=verdict,
                attack_category=attack_cat,
                confidence=classification["confidence"],
                risk_signals=signals,
                semantic_embedding=embedding,
                latency_ms=latency
            )

            span.set_attribute("detection.verdict", verdict.value)
            span.set_attribute("detection.latency_ms", latency)

            # Step 3: 可疑样本入缓冲（用于在线学习）
            if verdict == DetectionVerdict.SUSPICIOUS:
                self._online_buffer.append({
                    "embedding": embedding.tolist(),
                    "signals": signals,
                    "label": None,  # 待人工标注或反馈确认
                    "timestamp": time.time()
                })

            return result

    async def ingest_feedback(self, request_id: str, 
                               confirmed_label: AttackCategory):
        """接收人工/自动反馈，触发在线学习"""
        # 找到对应缓冲样本
        target = None
        for item in self._online_buffer:
            if item.get("request_id") == request_id:
                target = item
                break
        
        if not target:
            return {"ingested": False, "reason": "Sample not in buffer"}

        target["label"] = confirmed_label.value

        # 缓冲满阈值时触发微调
        labeled_count = sum(1 for s in self._online_buffer if s["label"] is not None)
        if labeled_count >= 50:
            await self._trigger_online_finetune()

        return {"ingested": True, "buffer_size": len(self._online_buffer)}

    async def _extract_risk_signals(self, text: str, 
                                      is_document: bool = False) -> List[str]:
        """提取风险信号（轻量级预筛）"""
        signals = []
        text_lower = text.lower()
        
        for signal, category in self.SIGNAL_CATEGORY_MAP.items():
            if signal in text_lower:
                signals.append(signal)
        
        # 文档特有信号
        if is_document:
            if any(enc in text_lower for enc in ["base64", "rot13", "\\x", "%00"]):
                signals.append("encoded payload in document")
            if "ignore all previous" in text_lower or "disregard prior" in text_lower:
                signals.append("hidden instruction in document")
        
        return signals

    async def _check_context_consistency(self, user_input: str, 
                                           context: Dict) -> List[str]:
        """检查输入与上下文的一致性"""
        signals = []
        # 示例：用户在上一轮询问天气，本轮突然要求导出数据库
        prev_intent = context.get("previous_intent", "")
        curr_signals = await self._extract_risk_signals(user_input)
        
        if prev_intent == "weather_query" and any(
            s in curr_signals for s in ["data_exfiltration", "tool_chain_poisoning"]
        ):
            signals.append("abrupt intent shift from benign to risky")
        
        return signals

    def _infer_attack_category(self, signals: List[str], 
                                classification: Dict) -> Optional[AttackCategory]:
        """综合信号与分类结果推断攻击类别"""
        if classification.get("category"):
            return AttackCategory(classification["category"])
        
        for signal in signals:
            clean_signal = signal.replace("doc:", "")
            if clean_signal in self.SIGNAL_CATEGORY_MAP:
                return self.SIGNAL_CATEGORY_MAP[clean_signal]
        
        return None

    async def _trigger_online_finetune(self):
        """触发在线微调"""
        labeled_samples = [s for s in self._online_buffer if s["label"] is not None]
        
        new_version = await self.registry.finetune(
            base_model=self._current_model_version,
            samples=labeled_samples,
            epochs= 31270.t.kuaisou.com
            learning_rate=2e-5
        )
        
        # 热加载新模型
        await self.classifier.load_version(new_version)
        self._current_model_version = new_version
        
        # 清空已标注缓冲
        self._online_buffer = [s for s in self._online_buffer if s["label"] is None]

"""
redteam_and_runtime_immunity.py - 自动化红队编排与运行时行为免疫引擎
技术栈: Pydantic / Celery / Prometheus / OpenTelemetry
"""
from typing import Dict, List, Any, Optional, Tuple
from pydantic import BaseModel, Field
from enum import Enum
import asyncio
import time
import uuid
import json
from dataclasses import dataclass, field

class RedTeamTestStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    PASSED = "passed"
    FAILED = "failed"
    BLOCKED_RELEASE = "blocked_release"

class RuntimeAnomalySeverity(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class ResponseAction(str, Enum):
    LOG_ONLY = "log_only"
    RATE_LIMIT = "rate_limit"
    DEGRADE_SERVICE = "degrade_service"
    CIRCUIT_BREAK = "circuit_break"
    HUMAN_TAKEOVER = "human_takeover"

@dataclass
class AttackStrategy:
    """结构化攻击策略"""
    strategy_id: str
    name: str
    category: str
    payload_template: str
    target_component: str       # llm / rag / tool / guardrail
    success_criteria: Dict      # 判定攻击成功的条件
    severity: str
    tags: List[str] = field(default_factory=list)

@dataclass
class BehaviorBaseline:
    """Agent行为基线"""
    agent_id: str
    baseline_version: 31271.t.kuaisou.com
    normal_action_distribution: Dict[str, float]  # action -> frequency
    normal_tool_call_patterns: List[Dict]          # 常见工具调用序列
    avg_response_latency_ms: float
    typical_output_length_range: Tuple[int, int]
    updated_at: float = field(default_factory=time.time)

class RedTeamAndImmunityEngine:
    """红队与运行时免疫引擎"""

    # 严重度→响应动作映射
    SEVERITY_RESPONSE_MAP = {
        RuntimeAnomalySeverity.LOW: ResponseAction.LOG_ONLY,
        RuntimeAnomalySeverity.MEDIUM: ResponseAction.RATE_LIMIT,
        RuntimeAnomalySeverity.HIGH: ResponseAction.DEGRADE_SERVICE,
        RuntimeAnomalySeverity.CRITICAL: ResponseAction.CIRCUIT_BREAK,
    }

    def __init__(self, attack_library, agent_client, metrics_store,
                 baseline_store, alert_channel, ci_gate):
        self.attacks = attack_library      # 结构化攻击策略库
        self.agent = agent_client          # 被测Agent客户端
        self.metrics = metrics_store       # Prometheus
        self.baselines = baseline_store    # Redis/PostgreSQL
        self.alerts = alert_channel        # PagerDuty/Slack
        self.ci = ci_gate                  # CI/CD门禁接口
        self._behavior_history: Dict[str, List[Dict]] = {}  # agent_id -> recent actions

    async def run_redteam_suite(self, suite_name: str,
                                  target_agent: str,
                                  trigger: str = "manual") -> Dict[str, Any]:
        """执行红队测试套件"""
        suite = await self.attacks.get_suite(suite_name)
        results = []
        passed = 0
        failed = 0

        for strategy in suite.strategies:
            test_result = await self._execute_single_attack(strategy, target_agent)
            results.append(test_result)
            
            if test_result["status"] == RedTeamTestStatus.PASSED:
                passed += 1
            else:
                failed += 1

        pass_rate = passed / len(results) if results else 0
        
        # 发射指标
        self.metrics.gauge("redteam.pass_rate", pass_rate, labels={
            "suite": suite_name, "agent": target_agent
        })
        self.metrics.gauge("redteam.failures", failed, labels={
            "suite": suite_name, "agent": target_agent
        })

        # CI门禁判定
        gate_result = "pass"
        if pass_rate < suite.min_pass_rate:
            gate_result = "block"
            await self.ci.block_release(
                reason=f"Red team pass rate {pass_rate:.2%} < threshold {suite.min_pass_rate:.2%}",
                suite=suite_name,
                failures=[r for r in results if r["status"] != RedTeamTestStatus.PASSED]
            )

        report = {
            "suite": suite_name,
            "target_agent": target_agent,
            "trigger": trigger,
            "total_tests": len(results),
            "passed": passed,
            "failed": 31272.t.kuaisou.com
            "pass_rate": round(pass_rate, 4),
            "gate_result": gate_result,
            "timestamp": time.time(),
            "details": results
        }

        return report

    async def evaluate_runtime_behavior(self, agent_id: str,
                                          action_record: Dict) -> Dict[str, Any]:
        """运行时行为评估"""
        # Step 1: 获取当前基线
        baseline = await self.baselines.get(agent_id)
        if not baseline:
            # 冷启动：记录但不拦截
            await self._update_history(agent_id, action_record)
            return {"anomaly": False, "reason": "No baseline yet (cold start)"}

        # Step 2: 多维异常评分
        anomaly_scores = {}
        
        # 动作分布偏移
        action_freq = self._compute_action_frequency(agent_id)
        distribution_shift = self._kl_divergence(
            action_freq, baseline.normal_action_distribution
        )
        anomaly_scores["distribution_shift"] = distribution_shift

        # 工具调用序列异常
        recent_sequence = self._get_recent_tool_sequence(agent_id, window=5)
        sequence_anomaly = self._sequence_anomaly_score(
            recent_sequence, baseline.normal_tool_call_patterns
        )
        anomaly_scores["sequence_anomaly"] = sequence_anomaly

        # 响应延迟异常
        latency_zscore = abs(
            (action_record.get("latency_ms", 0) - baseline.avg_response_latency_ms) 
            / max(baseline.avg_response_latency_ms * 0.3, 1)
        )
        anomaly_scores["latency_zscore"] = latency_zscore

        # Step 3: 综合判定
        max_score = max(anomaly_scores.values())
        severity = self._score_to_severity(max_score)
        response_action = self.SEVERITY_RESPONSE_MAP[severity]

        # Step 4: 更新历史并自适应调整基线
        await self._update_history(agent_id, action_record)
        if severity == RuntimeAnomalySeverity.LOW:
            # 低异常视为正常演化，缓慢更新基线
            await self._adaptive_baseline_update(agent_id, action_record)

        result = {
            "agent_id": agent_id,
            "anomaly_detected": severity != RuntimeAnomalySeverity.LOW,
            "severity": 31273.t.kuaisou.com
            "response_action": response_action.value,
            "anomaly_scores": {k: round(v, 4) for k, v in anomaly_scores.items()},
            "timestamp": time.time()
        }

        # 告警
        if severity in (RuntimeAnomalySeverity.HIGH, RuntimeAnomalySeverity.CRITICAL):
            await self.alerts.send(result)

        return result

    async def _execute_single_attack(self, strategy: AttackStrategy,
                                      target_agent: str) -> Dict[str, Any]:
        """执行单次攻击测试"""
        try:
            response = await self.agent.invoke(
                input=strategy.payload_template,
                target_component=strategy.target_component
            )
            
            # 评估是否攻击成功
            success = self._evaluate_success(response, strategy.success_criteria)
            
            status = RedTeamTestStatus.FAILED if success else RedTeamTestStatus.PASSED
            
            return {
                "strategy_id": strategy.strategy_id,
                "name": strategy.name,
                "status": 31274.t.kuaisou.com
                "attack_succeeded": success,
                "response_summary": str(response)[:500],
                "duration_ms": response.get("latency_ms", 0)
            }
        except Exception as e:
            return {
                "strategy_id": strategy.strategy_id,
                "name": strategy.name,
                "status": RedTeamTestStatus.FAILED,
                "error": str(e)
            }

    def _evaluate_success(self, response: Dict, criteria: Dict) -> bool:
        """评估攻击是否成功"""
        if "contains" in criteria:
            return criteria["contains"].lower() in str(response).lower()
        if "not_contains" in criteria:
            return criteria["not_contains"].lower() not in str(response).lower()
        if "tool_called" in criteria:
            return response.get("tool_called") == criteria["tool_called"]
        return False

    def _score_to_severity(self, score: float) -> RuntimeAnomalySeverity:
        if score < 1.5:
            return RuntimeAnomalySeverity.LOW
        elif score < 3.0:
            return RuntimeAnomalySeverity.MEDIUM
        elif score < 5.0:
            return RuntimeAnomalySeverity.HIGH
        return RuntimeAnomalySeverity.CRITICAL

    async def _update_history(self, agent_id: str, record: Dict):
        if agent_id not in self._behavior_history:
            self._behavior_history[agent_id] = []
        self._behavior_history[agent_id].append(record)
        # 保留最近1000条
        self._behavior_history[agent_id] = self._behavior_history[agent_id][-1000:]

    def _compute_action_frequency(self, agent_id: str) -> Dict[str, float]:
        history = self._behavior_history.get(agent_id, [])
        if not history:
            return {}
        counts = {}
        for r in history:
            action = r.get("action", "unknown")
            counts[action] = counts.get(action, 0) + 1
        total = len(history)
        return {k: v / total for k, v in counts.items()}

    def _kl_divergence(self, p: Dict[str, float], q: Dict[str, float]) -> float:
        """简化KL散度计算"""
        import math
        eps = 1e-10
        kl = 0.0
        all_keys = set(p.keys()) | set(q.keys())
        for k in all_keys:
            pk = p.get(k, eps)
            qk = q.get(k, eps)
            kl += pk * math.log(pk / qk)
        return max(kl, 0)

    def _get_recent_tool_sequence(self, agent_id: str, window: int) -> List[str]:
        history = self._behavior_history.get(agent_id, [])
        tool_calls = [r.get("tool") for r in history if r.get("tool")]
        return tool_calls[-window:]

    def _sequence_anomaly_score(self, sequence: List[str], 
                                  normal_patterns: List[Dict]) -> float:
        """序列异常评分（简化版）"""
        if not sequence or not normal_patterns:
            return 0.0
        # 检查序列是否匹配任一正常模式
        for pattern in normal_patterns:
            if sequence == pattern.get("sequence"):
                return 31275.t.kuaisou.com
        return 3.0  # 未匹配任何正常模式

    async def _adaptive_baseline_update(self, agent_id: str, record: Dict):
        """自适应基线更新（指数移动平均）"""
        baseline = await self.baselines.get(agent_id)
        if not baseline:
            return
        
        alpha = 0.01  # 缓慢更新
        action = record.get("action", "unknown")
        current_dist = baseline.normal_action_distribution.copy()
        current_dist[action] = current_dist.get(action, 0) * (1 - alpha) + alpha
        
        baseline.normal_action_distribution = current_dist
        baseline.updated_at = 31276.t.kuaisou.com
        await self.baselines.save(agent_id, baseline)