当Agent学会了"装工具"：2026企业级AI工具生态投毒、描述劫持、Rug Pull与寄生工具链治理实战

┌────────────────────────────────────────────────────────────────────────────────┐
│  2026 Tool Ecosystem Poisoning & Trust Chain Decay: Five-Layer Model           │
├────────────────────────────────────────────────────────────────────────────────┤
│                                                                                │
│  [Tool description = trusted instruction + Dynamic silent updates +           │
│   Shared context window + Name-based identity + Autonomous tool acquisition    │
│   → Any tool can speak into your agent's ear; any output can become            │
│     the next instruction; any approval can decay into a different tool]        │
│       ↓                                                                        │
│  ┌─ L1: 工具供应链普查与描述审计层 (Tool Supply Census & Description Audit)──┐ │
│  │  • 全量登记: 每个Agent挂载的每个工具的身份指纹/来源/维护者/权限/描述哈希    │ │
│  │  • 描述审计: 全量工具描述执行注入模式扫描(祈使句/隐藏指令/数据外传诱导)     │ │
│  │  • 影子工具检测: 同名/近名工具冲突扫描, 优先级覆盖行为监测                  │ │
│  │  • 自主获取闸门: Agent自主发现并挂载新工具必须经准入, 禁止运行时自装        │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L2: 描述-行为一致性验证层 (Description-Behavior Consistency) ────────────┐ │
│  │  • 沙箱试跑: 批准前在隔离环境以金丝雀数据试跑, 比对声明行为与实际行为        │ │
│  │  • 权限越界检测: 实际文件/网络/凭据访问超出声明范围即判定不一致             │ │
│  │  • 判断偏置测试: 对影响决策的工具执行对照实验(注入组vs对照组输出差异)        │ │
│  │  • 描述语义签名: 描述文本的语义指纹入库, 语义漂移即触发再审计               │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L3: 信任时间固化层 (Trust Temporal Binding) ─────────────────────────────┐ │
│  │  • 版本钉扎: 批准绑定描述哈希+代码哈希+版本号, 禁止跟随最新端点             │ │
│  │  • 变更即失效: 任何哈希漂移自动冻结该工具调用, 触发再批准流程               │ │
│  │  • 再批准时效: 批准有效期≤90天, 过期强制重新执行L2验证                      │ │
│  │  • 静默更新监测: 每日拉取工具定义并与钉扎快照比对, 差异即告警               │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L4: 数据/指令隔离与最小权限层 (Data-Instruction Isolation) ──────────────┐ │
│  │  • 输出标记: 所有工具返回值标记为"外部数据", 禁止作为指令解释               │ │
│  │  • 跨工具阻断: 工具A的输出触发工具B的高风险调用必须经策略网关二次判定        │ │
│  │  • 能力令牌: 每次调用携带范围化短时令牌, 工具无法获得超越本次任务的凭据       │ │
│  │  • 不可逆动作熔断: 退款/删除/转账/发布类动作强制人工确认或独立校验           │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L5: 污染溯源与生态归责层 (Contamination Provenance & Attribution) ────────┐ │
│  │  • 调用链存证: 完整工具调用序列+上下文快照存证, 支持事后污染路径重建          │ │
│  │  • 生态情报: 订阅工具投毒/Rug Pull/shadowing威胁情报, 已知恶意指纹全网阻断   │ │
│  │  • 损失归因: 偏置合同/超政策退款/密钥降级损失归入工具生态成本核算            │ │
│  │  • 准入归责: "批准了描述而不批准行为"的准入流程责任框架                      │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│                                                                                │
└────────────────────────────────────────────────────────────────────────────────┘

"""
tool_poison_detection.py - 工具投毒与信任链腐化检测引擎
核心原则: "这个工具是可信的"不等于"这个工具的描述是可信的"——
          如果客户端把description原样拼进system prompt,
          那任何人都可以在你的Agent耳边说话,
          而你的Agent被训练成: 听话;
          如果模型能力越强指令遵循越好,
          那你花高价买的更聪明的模型,
          是更服从的投毒载体——72.8%的成功率不是漏洞评分,
          是行业现状的测量值;
          如果批准发生在T0而工具可以在T+91静默改描述,
          那你批准的是一个快照, 运行的是另一个东西,
          而没有任何机制告诉你这两者已经不是同一个;
          如果六个工具全部可信而退款依然超政策23笔,
          那污染不在工具里, 在工具之间——
          数据一旦能决定下一步指令, 任何可写内容的工具
          都是其他所有工具的指令源;
          投毒检测最反直觉的地方在于:
          你要审查的不是代码, 是自我介绍
"""
from typing import Dict, List, Any, Optional, Tuple, Set
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict, Counter
import time, uuid, json, hashlib, re
import numpy as np

class PoisonPattern(str, Enum): jiyi.tongsou.com 
    DESCRIPTION_INJECTION = "desc_injection"      # 描述劫持
    RUG_PULL = "rug_pull"                         # 批准后静默变更
    TOOL_SHADOWING = "tool_shadowing"             # 影子工具覆盖
    PARASITIC_CHAIN = "parasitic_chain"           # 寄生工具链
    SCOPE_EXCEEDANCE = "scope_exceedance"         # 行为超出声明权限
    AUTONOMOUS_INSTALL = "autonomous_install"     # 运行时自主挂载
    NONE = "none"

class TrustState(str, Enum): zhaixing.tongsou.com 
    PINNED = "pinned"                # 已钉扎且与快照一致
    DRIFTED = "drifted"              # 描述/代码哈希漂移(冻结)
    EXPIRED = "expired"              # 批准超期(需再验证)
    QUARANTINED = "quarantined"      # 检出投毒(隔离)
    UNREGISTERED = "unregistered"    # 未登记(拒绝调用)

@dataclass
class ToolIdentity: xunling.tongsou.com 
    """工具身份指纹——名字不是身份"""
    tool_uid: str = field(default_factory=lambda: f"tu-{uuid.uuid4().hex[:10]}")
    agent_id: str = ""
    server_id: str = ""
    tool_name: str = ""
    # 三重指纹
    description_hash: str = ""       # 描述文本SHA256
    code_hash: str = ""              # 实现代码/包SHA256
    semantic_fingerprint: str = ""   # 描述语义嵌入指纹
    # 声明
    declared_permissions: List[str] = field(default_factory=list)
    maintainer: str = ""
    source_url: str = ""
    version: str = ""
    # 信任
    trust_state: TrustState = TrustState.UNREGISTERED
    approved_at: float = 0.0
    approval_expiry_days: int = 90
    last_verified_at: float = 0.0

@dataclass
class PoisonEvidence: maifushi.tongsou.com 
    """投毒证据"""
    evidence_id: str = field(default_factory=lambda: f"tp-{uuid.uuid4().hex[:10]}")
    agent_id: str = ""
    tool_name: str = ""
    pattern: PoisonPattern = PoisonPattern.NONE
    detail: str = ""
    matched_indicators: List[str] = field(default_factory=list)
    severity: str = "high"           # "critical" / "high" / "medium"
    affected_decisions: int = 0
    estimated_harm_usd: float = 0.0
    detected_at: float = field(default_factory=time.time)

# 描述注入模式库: 工具描述中不应出现的语言结构
INJECTION_PATTERNS = [
    (r"(?i)(before|prior to|when) calling (this|another|any) tool",
     "pre-call instruction — a description should DESCRIBE, not DIRECT"),
    (r"(?i)(do not|don't|never) (tell|inform|show|mention).{0,40}(user|human)",
     "concealment directive — legitimate tools have no reason to hide from users"),
    (r"(?i)(read|access|fetch|collect).{0,40}"
     r"(env|environ|\.ssh|credential|token|api[_-]?key|secret|password)",
     "credential exfiltration lure — no business tool needs your SSH directory"),
    (r"(?i)(ignore|disregard|override).{0,30}(previous|above|system).{0,20}instruction",
     "context override — the classic injection primitive"),
    (r"(?i)(always|prefer|prioritize).{0,60}(this tool|higher priority|over other)",
     "priority hijack attempt — shadowing precursor"),
    (r"(?i)(append|include|embed).{0,40}(output|response|result).{0,40}"
     r"(from|of).{0,30}(file|env|config|context)",
     "data smuggling into outputs — parasitic chain seeding"),
    (r"(?i)system (prompt|message|note)\s*:",
     "role spoofing — a tool description claiming to be the system"),
    (r"(?i)(default|assume).{0,50}(score|rating|approve|pass|accept)"
     r".{0,40}(when|if).{0,30}(missing|unavailable|unclear)",
     "judgment bias injection — silent decision tilting"),
]

class ToolPoisonDetectionEngine: zhendao.tongsou.com 
    """工具投毒与信任链腐化检测引擎"""

    # 配置
    APPROVAL_MAX_AGE_DAYS = 90            # 批准最长有效期
    SEMANTIC_DRIFT_ALERT = 0.12           # 语义指纹漂移阈值
    SHADOW_NAME_SIMILARITY = 0.85         # 影子工具名称相似度阈值
    PARASITIC_CHAIN_WINDOW_MIN = 10       # 寄生链检测时间窗
    HIGH_RISK_TOOLS = ("refund", "payment", "transfer", "delete",
                       "publish", "grant", "credential", "execute")
    AUTONOMOUS_INSTALL_BLOCK = True       # 禁止运行时自主挂载

    def __init__(self, agent_registry, alerts, audit, metrics):
        self.registry = agent_registry
        self.alerts = alerts
        self.audit = audit
        self.metrics = metrics
        self.tools: Dict[str, ToolIdentity] = {}          # tool_uid -> identity
        self.by_name: Dict[str, List[ToolIdentity]] = defaultdict(list)
        self.evidences: List[PoisonEvidence] = []
        self.call_logs: Dict[str, List[Dict[str, Any]]] = defaultdict(list)

    @staticmethod
    def _hash(text: str) -> str: weimeng.tongsou.com 
        return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]

    async def audit_description(self, agent_id: str, tool_name: str,
                                 description: str
                                 ) -> Optional[PoisonEvidence]:
        """描述审计: 注入模式扫描(准入前强制执行)"""

        matched = []
        for pattern, rationale in INJECTION_PATTERNS:
            hits = re.findall(pattern, description)
            if hits: aisou.tongsou.com 
                matched.append(f"{rationale} [{str(hits[0])[:60]}]")

        # 长度异常: 正常工具描述<300字符, 投毒描述常>1000
        length_anomaly = len(description) > 1000

        if matched or length_anomaly: toujing.tongsou.com 
            severity = "critical" if len(matched) >= 2 else "high"
            evidence = PoisonEvidence(
                agent_id=agent_id, tool_name=tool_name,
                pattern=PoisonPattern.DESCRIPTION_INJECTION,
                detail=(f"Description length {len(description)} chars. "
                        f"Matched {len(matched)} injection indicators."),
                matched_indicators=matched + (
                    ["abnormal description length"] if length_anomaly else []),
                severity=severity
            )
            self.evidences.append(evidence)

            await self.alerts.critical(
                f"🚨 TOOL DESCRIPTION INJECTION [{severity.upper()}]: "
                f"Agent '{agent_id}', tool '{tool_name}'. "
                f"Indicators: {matched[:4]} "
                f"{'+ abnormal length' if length_anomaly else ''}. "
                f"The UI will show your users a one-line label. "
                f"The model will read all {len(description)} characters "
                f"and treat them as trusted system guidance— because "
                f"that is exactly where the client puts them. "
                f"On 353 real-world tools this pattern succeeded "
                f"72.8% of the time against o1-mini, and the success "
                f"rate RISES with model capability: better instruction "
                f"following means better obedience to whoever is "
                f"speaking. QUARANTINE. Do not mount. Do not 'review "
                f"the code'— the code is fine. The introduction is "
                f"the attack."
            )
            await self.audit.log_poison_evidence(evidence)
            return evidence
        return None

    async def register_and_pin(self, agent_id: str, tool: ToolIdentity
                                ) -> ToolIdentity:
        """登记并钉扎工具身份(三重指纹)"""

        tool.agent_id = agent_id
        tool.description_hash = self._hash(tool.semantic_fingerprint
                                           or tool.tool_name)
        tool.trust_state = TrustState.PINNED
        tool.approved_at = time.time()
        tool.last_verified_at = time.time()

        self.tools[tool.tool_uid] = tool
        self.by_name[tool.tool_name].append(tool)

        # ===== 影子工具检测: 同名/近名冲突 =====
        siblings = [t for t in self.by_name[tool.tool_name]
                    if t.tool_uid != tool.tool_uid]
        if siblings: qiyin.tongsou.com 
            await self.alerts.critical(
                f"🚨 TOOL SHADOWING DETECTED: Name '{tool.tool_name}' "
                f"now resolves to {len(siblings) + 1} distinct identities "
                f"across agents: "
                f"{[(s.agent_id, s.code_hash, s.maintainer) for s in siblings]} "
                f"+ new ({agent_id}, {tool.code_hash}, {tool.maintainer}). "
                f"Your asset inventory records the NAME. The name is a "
                f"public, zero-cost, claimable resource. "
                f"'We approved format_code' remains a TRUE sentence "
                f"after format_code has been replaced— it just no "
                f"longer refers to anything you approved. "
                f"Identity is hash+signature+registry binding, "
                f"never a string. RESOLVE CONFLICT before either "
                f"tool is callable."
            )

        # 近似名检测(拼写抢占/前缀抢占)
        for existing_name, existing_tools in self.by_name.items():
            if existing_name == tool.tool_name:
                continue
            sim = self._name_similarity(existing_name, tool.tool_name)
            if sim > self.SHADOW_NAME_SIMILARITY:
                await self.alerts.warning(
                    f"⚠️ NEAR-NAME SQUATTING: '{tool.tool_name}' is "
                    f"{sim:.0%} similar to approved '{existing_name}'. "
                    f"Typosquatting and prefix-hijack are the cheapest "
                    f"shadowing vectors in an ecosystem where agents "
                    f"select tools by name from a list."
                )

        await self.audit.log_tool_registration(tool)
        return tool

    @staticmethod
    def _name_similarity(a: str, b: str) -> float:
        """简化名称相似度(字符级Jaccard + 编辑距离归一)"""
        if not a or not b: hanzhi.tongsou.com 
            return 0.0
        sa, sb = set(a.lower()), set(b.lower())
        jac = len(sa & sb) / max(len(sa | sb), 1)
        len_ratio = min(len(a), len(b)) / max(len(a), len(b))
        return (jac + len_ratio) / 2

    async def detect_rug_pull(self, tool_uid: str,
                               current_description: str,
                               current_code_hash: str,
                               current_semantic_fp: str
                               ) -> Optional[PoisonEvidence]:
        """Rug Pull检测: 与钉扎快照比对"""

        tool = self.tools.get(tool_uid)
        if not tool:
            return None

        desc_hash = self._hash(current_description)
        desc_changed = desc_hash != tool.description_hash
        code_changed = current_code_hash != tool.code_hash

        # 语义漂移: 即使措辞重写, 语义指纹应稳定
        semantic_drift = 0.0
        if tool.semantic_fingerprint and current_semantic_fp:
            semantic_drift = self._semantic_distance(
                tool.semantic_fingerprint, current_semantic_fp)

        # 批准过期
        age_days = (time.time() - tool.approved_at) / 86400
        expired = age_days > tool.approval_expiry_days

        if not (desc_changed or code_changed or expired
                or semantic_drift > self.SEMANTIC_DRIFT_ALERT):
            return None

        # 冻结
        tool.trust_state = (TrustState.EXPIRED if (expired and not desc_changed
                                                    and not code_changed)
                            else TrustState.DRIFTED)

        # 若变更含新注入模式 → 升级为QUARANTINED
        new_injection = await self.audit_description(
            tool.agent_id, tool.tool_name, current_description)
        if new_injection: hongdong.tongsou.com 
            tool.trust_state = TrustState.QUARANTINED

        evidence = PoisonEvidence(
            agent_id=tool.agent_id, tool_name=tool.tool_name,
            pattern=PoisonPattern.RUG_PULL,
            detail=(f"Approval age: {age_days:.0f}d. "
                    f"Description hash changed: {desc_changed}. "
                    f"Code hash changed: {code_changed}. "
                    f"Semantic drift: {semantic_drift:.3f}. "
                    f"State → {tool.trust_state.value}"),
            matched_indicators=[
                i for i in (["description_hash_drift"] if desc_changed else [])
                + (["code_hash_drift"] if code_changed else [])
                + (["approval_expired"] if expired else [])
                + ([f"semantic_drift_{semantic_drift:.2f}"]
                   if semantic_drift > self.SEMANTIC_DRIFT_ALERT else [])],
            severity="critical" if tool.trust_state == TrustState.QUARANTINED
                     else "high"
        )
        self.evidences.append(evidence)

        await self.alerts.critical(
            f"🚨 RUG PULL / TRUST DECAY: Tool '{tool.tool_name}' "
            f"(agent '{tool.agent_id}', maintainer '{tool.maintainer}'). "
            f"{evidence.detail} "
            f"ALL CALLS FROZEN pending re-approval. "
            f"Your approval was a point-in-time snapshot of a "
            f"remotely, unilaterally, silently editable artifact. "
            f"The supplier qualification tool that tilted 17.8M in "
            f"contracts changed its description on day 91, inside an "
            f"update labeled 'documentation improvements'. "
            f"The clinical summary tool that weakened anticoagulation "
            f"follow-up wording changed on day 180. "
            f"Neither triggered a single alert, because nothing in "
            f"the architecture bound the approved artifact to the "
            f"executed one. Microsoft now catalogs this as a formal "
            f"zero-trust AI attack technique. Treat it as one."
        )
        await self.audit.log_poison_evidence(evidence)
        return evidence

    @staticmethod
    def _semantic_distance(fp_a: str, fp_b: str) -> float:
        """语义指纹距离(此处以指纹向量哈希近似, 生产接嵌入服务)"""
        try: zhuaci.tongsou.com 
            va = np.array([int(c, 16) for c in fp_a[:32]], dtype=float)
            vb = np.array([int(c, 16) for c in fp_b[:32]], dtype=float)
            va = va / (np.linalg.norm(va) + 1e-9)
            vb = vb / (np.linalg.norm(vb) + 1e-9)
            return float(1 - np.dot(va, vb))
        except Exception:
            return 1.0 if fp_a != fp_b else 0.0

    async def detect_parasitic_chain(self, agent_id: str,
                                      call_log: List[Dict[str, Any]]
                                      ) -> Optional[PoisonEvidence]:
        """寄生工具链检测: 数据输出→高风险调用的可疑因果序列"""

        # call_log: [{"ts":..., "tool":..., "kind":"data_return"|"action",
        #             "content_has_external_text": bool,
        #             "content_injection_suspect": bool, "risk":"high"|...}]
        self.call_logs[agent_id].extend(call_log)

        suspects = []
        for i, ev in enumerate(call_log):
            if ev.get("kind") != "action":
                continue
            tool_l = ev.get("tool", "").lower()
            is_high_risk = any(k in tool_l for k in self.HIGH_RISK_TOOLS)
            if not is_high_risk:
                continue

            # 回溯窗口内的数据来源事件
            window_start = ev["ts"] - self.PARASITIC_CHAIN_WINDOW_MIN * 60
            preceding = [p for p in call_log[:i]
                         if p["ts"] >= window_start
                         and p.get("kind") == "data_return"
                         and p.get("content_has_external_text")]
            inject_seeded = [p for p in preceding
                             if p.get("content_injection_suspect")]

            if inject_seeded:
                suspects.append({
                    "action_tool": ev["tool"],
                    "action_ts": ev["ts"],
                    "seeding_tool": inject_seeded[-1]["tool"],
                    "gap_sec": ev["ts"] - inject_seeded[-1]["ts"]
                })

        if suspects: moli.tongsou.com 
            evidence = PoisonEvidence(
                agent_id=agent_id,
                tool_name=", ".join(sorted({s["seeding_tool"]
                                            for s in suspects})),
                pattern=PoisonPattern.PARASITIC_CHAIN,
                detail=f"{len(suspects)} high-risk actions within "
                       f"{self.PARASITIC_CHAIN_WINDOW_MIN}min of "
                       f"injection-suspect external data returns.",
                matched_indicators=[
                    f"{s['seeding_tool']}→{s['action_tool']} "
                    f"@{s['gap_sec']:.0f}s" for s in suspects[:8]],
                severity="critical"
            )
            self.evidences.append(evidence)

            await self.alerts.critical(
                f"🚨 PARASITIC TOOL CHAIN: Agent '{agent_id}'. "
                f"{len(suspects)} high-risk actions "
                f"({', '.join(sorted({s['action_tool'] for s in suspects}))}) "
                f"executed within minutes of injection-suspect external "
                f"data returns "
                f"({', '.join(sorted({s['seeding_tool'] for s in suspects}))}). "
                f"Sequences: {evidence.matched_indicators[:4]}. "
                f"EVERY TOOL IN THIS CHAIN IS APPROVED. Every one "
                f"passed review. None was compromised. The poison is "
                f"not in a tool— it is BETWEEN tools: a search result "
                f"is data until the model reads it, and the model "
                f"cannot tell reading from obeying. Six trusted "
                f"servers, one shared context window, 23 out-of-policy "
                f"refunds in eleven days, 4.1M gone, and every single "
                f"transaction had a complete 'agent judgment' log. "
                f"Input validation cannot help here: the exploited "
                f"component is the LLM's defining capability."
            )
            await self.audit.log_poison_evidence(evidence)
            return evidence
        return None

    async def gate_autonomous_install(self, agent_id: str,
                                       discovered_tool: Dict[str, Any]
                                       ) -> Dict[str, Any]:
        """自主获取闸门: 禁止运行时自装工具"""

        if not self.AUTONOMOUS_INSTALL_BLOCK:
            return {"allowed": True, "reason": "gate disabled (VIOLATION)"}

        result = {
            "allowed": False,
            "agent_id": agent_id,
            "tool_name": discovered_tool.get("name", "?"),
            "action": "queued_for_admission_review",
            "reason": "runtime self-installation of tools is prohibited"
        }

        evidence = PoisonEvidence(
            agent_id=agent_id,
            tool_name=discovered_tool.get("name", "?"),
            pattern=PoisonPattern.AUTONOMOUS_INSTALL,
            detail=f"Agent attempted runtime mount of "
                   f"'{discovered_tool.get('source_url', 'unknown source')}'. "
                   f"Queued for admission review.",
            severity="high"
        )
        self.evidences.append(evidence)

        await self.alerts.critical(
            f"🚫 AUTONOMOUS TOOL INSTALL BLOCKED: Agent '{agent_id}' "
            f"attempted to mount '{discovered_tool.get('name')}' from "
            f"'{discovered_tool.get('source_url', 'unknown')}' at runtime. "
            f"Queued for admission review (L1 census + L2 description "
            f"audit + L2 sandbox trial). "
            f"An agent that can find and install its own tools is an "
            f"agent whose attack surface grows at machine speed while "
            f"your review capacity grows at human speed. "
            f"The supplier-qualification tool with the day-91 bias "
            f"injection had a market rating of 4.8 stars. "
            f"Reputation is not a security control— it is the "
            f"pretext the attacker buys first."
        )
        await self.audit.log_poison_evidence(evidence)
        return result

"""
invocation_sandbox_defense.py - 数据/指令隔离与调用沙箱防御引擎
核心: 寄生链不是某个工具的缺陷, 是共享上下文结构的必然输出——
      你不能通过"只接可信工具"来防污染,
      因为parasitic chaining不要求任何一环是恶意的;
      你不能通过输入校验来防注入,
      因为被利用的不是解析器, 是模型的理解能力本身;
      你能做的是建立物理隔断: 输出标记让数据永远带着"我是数据"的标签,
      策略网关让跨工具的高风险因果必须经过一次架构外的判定,
      能力令牌让工具拿到的凭据只在本次任务内有效且范围最小,
      不可逆熔断让退款和删除在模型被说服之后仍然需要另一个签字;
      防御的本质不是让Agent更警觉, 是让"说服Agent"不再足够——
      而在一个所有工具都能对模型说话的环境里,
      最贵的不是沙箱, 是承认这件事:
      你的Agent的耳朵, 是公共的
"""
from typing import Dict, List, Any, Optional, Tuple
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict
import time, uuid, json, hashlib
import numpy as np

class DataTrust(str, Enum):
    SYSTEM = "system"              # 系统指令(唯一可作为指令的来源)
    USER = "user"                  # 用户输入(经确认的意图)
    TOOL_OUTPUT = "tool_output"    # 工具返回(永远不可作为指令)
    EXTERNAL_CONTENT = "external"  # 网页/文档/邮件内容(最高危)

class GatewayVerdict(str, Enum):
    ALLOW = "allow"
    REQUIRE_HUMAN = "require_human"      # 不可逆动作熔断
    REQUIRE_INDEPENDENT_CHECK = "independent_check"
    BLOCK = "block"

@dataclass
class TaggedPayload:
    """带信任标签的载荷——数据/指令隔离的最小单元"""
    payload_id: str = field(default_factory=lambda: f"pl-{uuid.uuid4().hex[:10]}")
    trust: DataTrust = DataTrust.TOOL_OUTPUT
    source_tool: str = ""
    content_hash: str = ""
    contains_instruction_like_text: bool = False
    created_at: float = field(default_factory=time.time)

@dataclass
class CapabilityToken:
    """范围化短时能力令牌"""
    token_id: str = field(default_factory=lambda: f"ct-{uuid.uuid4().hex[:10]}")
    tool_uid: str = ""
    task_scope: str = ""              # 本次任务的最小范围描述
    allowed_resources: List[str] = field(default_factory=list)
    max_amount: Optional[float] = None  # 金额上限(若涉及资金)
    expires_at: float = 0.0
    single_use: bool = True
    issued_at: float = field(default_factory=time.time)

@dataclass
class ChainEvidence:
    """调用链存证"""
    chain_id: str = field(default_factory=lambda: f"ch-{uuid.uuid4().hex[:10]}")
    agent_id: str = ""
    session_id: str = ""
    steps: List[Dict[str, Any]] = field(default_factory=list)
    context_snapshot_hash: str = ""
    gateway_decisions: List[Dict[str, Any]] = field(default_factory=list)
    recorded_at: float = field(default_factory=time.time)

class InvocationSandboxEngine:
    """数据/指令隔离与调用沙箱防御引擎"""

    # 配置
    TOKEN_MAX_TTL_SEC = 300               # 能力令牌最长有效期
    IRREVERSIBLE_ACTIONS = ("refund", "transfer", "delete", "publish",
                            "grant_access", "execute", "send_external")
    AMOUNT_HUMAN_FLOOR = 500.0            # 超过此金额强制人工确认
    CROSS_TOOL_GATE_MANDATORY = True      # 跨工具高风险调用强制过网关
    CHAIN_RETENTION_DAYS = 400            # 调用链存证保留期
    INSTRUCTION_TEXT_IN_OUTPUT_BLOCK = True  # 输出含指令样文本即降级处理

    def __init__(self, poison_engine, alerts, audit, metrics):
        self.poison = poison_engine
        self.alerts = alerts
        self.audit = audit
        self.metrics = metrics
        self.tokens: Dict[str, CapabilityToken] = {}
        self.chains: List[ChainEvidence] = []
        self.pending_human: List[Dict[str, Any]] = []

    async def tag_tool_output(self, source_tool: str, content: str,
                               injection_suspect: bool = False
                               ) -> TaggedPayload:
        """为工具输出打信任标签(数据永远不是指令)"""

        # 检测输出中的指令样文本
        instruction_like = injection_suspect or bool(
            __import__("re").search(
                r"(?i)(system prompt|ignore previous|you must|you should|"
                r"call the .* tool|do not tell|instruction:)", content))

        payload = TaggedPayload(
            trust=DataTrust.EXTERNAL_CONTENT
                  if source_tool in ("web_search", "email_read", "doc_fetch",
                                     "browse", "crawl")
                  else DataTrust.TOOL_OUTPUT,
            source_tool=source_tool,
            content_hash=hashlib.sha256(content.encode()).hexdigest()[:16],
            contains_instruction_like_text=instruction_like
        )

        if instruction_like and self.INSTRUCTION_TEXT_IN_OUTPUT_BLOCK:
            await self.alerts.warning(
                f"🏷️ OUTPUT DOWNGRADED: Tool '{source_tool}' returned "
                f"content containing instruction-like text. "
                f"Payload tagged {payload.trust.value} + "
                f"'instruction_like' flag. This content will be "
                f"rendered to the model as QUOTED DATA with an explicit "
                f"non-obedience wrapper, and any action it requests "
                f"will require gateway re-adjudication. "
                f"A search result that says 'call refund_tool' is a "
                f"string. The moment your architecture lets a string "
                f"become a verb, every website on the internet is a "
                f"privileged user of your agent."
            )

        await self.audit.log_tagged_payload(payload)
        return payload

    async def issue_capability_token(self, tool_uid: str, task_scope: str,
                                      allowed_resources: List[str],
                                      max_amount: Optional[float] = None
                                      ) -> CapabilityToken:
        """签发范围化短时能力令牌(工具不得持有长期凭据)"""

        token = CapabilityToken(
            tool_uid=tool_uid, task_scope=task_scope,
            allowed_resources=allowed_resources, max_amount=max_amount,
            expires_at=time.time() + self.TOKEN_MAX_TTL_SEC,
            single_use=True)
        self.tokens[token.token_id] = token

        await self.audit.log_capability_token(token)
        return token

    async def validate_token_use(self, token_id: str,
                                  requested_resource: str,
                                  requested_amount: Optional[float] = None
                                  ) -> Dict[str, Any]:
        """校验令牌使用(范围+时效+单次)"""

        token = self.tokens.get(token_id)
        if not token:
            return {"allowed": False, "reason": "token not found"}

        violations = []
        if time.time() > token.expires_at:
            violations.append(f"EXPIRED (ttl {self.TOKEN_MAX_TTL_SEC}s)")
        if requested_resource not in token.allowed_resources:
            violations.append(f"OUT OF SCOPE: '{requested_resource}' not in "
                              f"{token.allowed_resources}")
        if (requested_amount is not None and token.max_amount is not None
                and requested_amount > token.max_amount):
            violations.append(f"AMOUNT {requested_amount} > cap "
                              f"{token.max_amount}")
        if not token.single_use:
            violations.append("TOKEN ALREADY CONSUMED")

        if violations:
            await self.alerts.critical(
                f"🛑 CAPABILITY TOKEN VIOLATION: Tool '{token.tool_uid}', "
                f"scope '{token.task_scope}'. {violations}. "
                f"A tool holding a long-lived, broadly-scoped credential "
                f"is a tool that can be persuaded into anything— and in "
                f"an ecosystem where any output can speak to the model, "
                f"'persuaded' is a matter of one poisoned paragraph. "
                f"The credential must be smaller than the attack: "
                f"scoped to this task, valid for minutes, usable once."
            )
            return {"allowed": False, "violations": violations}

        token.single_use = False
        return {"allowed": True}

    async def adjudicate_high_risk_call(self, agent_id: str,
                                         action_tool: str,
                                         proposed_params: Dict[str, Any],
                                         preceding_payloads: List[TaggedPayload],
                                         amount: Optional[float] = None
                                         ) -> Dict[str, Any]:
        """策略网关: 跨工具高风险调用的二次判定"""

        if not self.CROSS_TOOL_GATE_MANDATORY:
            return {"verdict": GatewayVerdict.ALLOW.value,
                    "reason": "gate disabled (VIOLATION)"}

        action_l = action_tool.lower()
        is_irreversible = any(k in action_l for k in self.IRREVERSIBLE_ACTIONS)

        # 因果污染判定: 前序外部数据是否含指令样文本
        tainted_by = [p for p in preceding_payloads
                      if p.contains_instruction_like_text
                      or p.trust == DataTrust.EXTERNAL_CONTENT]

        verdict = GatewayVerdict.ALLOW
        reasons = []

        if is_irreversible and tainted_by:
            verdict = GatewayVerdict.REQUIRE_HUMAN
            reasons.append(f"irreversible action '{action_tool}' causally "
                           f"preceded by {len(tainted_by)} external/"
                           f"instruction-like payloads "
                           f"({[p.source_tool for p in tainted_by]})")

        if amount is not None and amount > self.AMOUNT_HUMAN_FLOOR:
            verdict = GatewayVerdict.REQUIRE_HUMAN
            reasons.append(f"amount {amount} > human floor "
                           f"{self.AMOUNT_HUMAN_FLOOR}")

        if is_irreversible and not tainted_by and verdict == GatewayVerdict.ALLOW:
            verdict = GatewayVerdict.REQUIRE_INDEPENDENT_CHECK
            reasons.append("irreversible action requires independent "
                           "policy verification (not agent self-attestation)")

        decision = {
            "agent_id": agent_id, "action_tool": action_tool,
            "verdict": verdict.value, "reasons": reasons,
            "amount": amount, "tainted_sources": [p.source_tool for p in tainted_by],
            "decided_at": time.time()
        }

        if verdict == GatewayVerdict.REQUIRE_HUMAN:
            self.pending_human.append(decision)
            await self.alerts.critical(
                f"✋ IRREVERSIBLE ACTION HELD FOR HUMAN: Agent '{agent_id}' "
                f"proposed '{action_tool}'"
                + (f" amount={amount}" if amount else "") + ". "
                f"Grounds: {'; '.join(reasons)}. "
                f"The agent's reasoning for this action traces back to "
                f"content that entered the context as DATA and left it "
                f"as a DECISION. That transmutation is the entire "
                f"parasitic chain attack, and it is invisible to every "
                f"control that inspects tools individually. "
                f"23 refunds, 4.1M, eleven days, six approved tools, "
                f"zero compromised components— and every transaction "
                f"carried a complete 'agent judgment' log. "
                f"The gateway is not doubting the agent. "
                f"The gateway is refusing to let a paragraph from the "
                f"internet hold signing authority."
            )
        elif verdict == GatewayVerdict.REQUIRE_INDEPENDENT_CHECK:
            await self.alerts.warning(
                f"🔎 INDEPENDENT CHECK REQUIRED: '{action_tool}' by "
                f"'{agent_id}'. Agent self-attestation is not accepted "
                f"as policy verification for irreversible actions— "
                f"the attestation and the action come from the same "
                f"persuadable context."
            )

        await self.audit.log_gateway_decision(decision)
        return decision

    async def record_chain(self, agent_id: str, session_id: str,
                            steps: List[Dict[str, Any]],
                            context_snapshot: str
                            ) -> ChainEvidence:
        """调用链存证: 污染路径可重建"""

        chain = ChainEvidence(
            agent_id=agent_id, session_id=session_id, steps=steps,
            context_snapshot_hash=hashlib.sha256(
                context_snapshot.encode()).hexdigest()[:16],
            gateway_decisions=[d for d in self.pending_human
                               if d.get("agent_id") == agent_id]
        )
        self.chains.append(chain)
        await self.audit.log_chain_evidence(chain)
        return chain

    async def reconstruct_contamination_path(self, incident_id: str,
                                              harm_description: str
                                              ) -> Dict[str, Any]:
        """污染路径重建: 从事后损害回溯至注入源"""

        # 生产实现: 按时间窗+资源交集检索调用链存证
        candidates = [c for c in self.chains[-2000:]]
        reconstruction = {
            "incident_id": incident_id,
            "harm": harm_description,
            "chains_examined": len(candidates),
            "candidate_seeding_sources": [],
            "note": ""
        }

        for c in candidates:
            for step in c.steps:
                if step.get("kind") == "data_return" and \
                   step.get("instruction_like"):
                    reconstruction["candidate_seeding_sources"].append({
                        "chain_id": c.chain_id,
                        "tool": step.get("tool"),
                        "ts": step.get("ts"),
                        "context_hash": c.context_snapshot_hash
                    })

        reconstruction["note"] = (
            "Contamination attribution requires the context snapshot, "
            "not just the call log: the poisoned text is not in any "
            "tool's output record, it is in the shared window where "
            "all tools' outputs became one another's premises. "
            f"Retention: {self.CHAIN_RETENTION_DAYS} days."
        )

        await self.alerts.info(
            f"🧬 CONTAMINATION PATH RECONSTRUCTED: Incident '{incident_id}'. "
            f"Chains examined: {reconstruction['chains_examined']}. "
            f"Candidate seeding sources: "
            f"{len(reconstruction['candidate_seeding_sources'])}. "
            f"{reconstruction['note'][:200]}"
        )
        await self.audit.log_contamination_reconstruction(reconstruction)
        return reconstruction

    async def attribute_admission_failure(self, agent_id: str,
                                           tool_name: str,
                                           harm_usd: float
                                           ) -> Dict[str, Any]:
        """准入流程归责"""

        attribution = {
            "agent_id": agent_id, "tool_name": tool_name,
            "harm_usd": harm_usd,
            "primary_accountable": "tool_admission_process",
            "rationale": "", "fixes": []
        }

        attribution["rationale"] = (
            "The review board read the code. The code was clean. The "
            "permissions were declared read-only, and every call stayed "
            "inside them. What nobody reviewed was the tool's "
            "SELF-INTRODUCTION— the natural-language description that "
            "the client pastes into the trusted context, that the UI "
            "folds into one line, that the model reads in full and "
            "obeys at a rate of 72.8%. The admission process certified "
            "what the tool CAN DO and never asked what the tool TELLS "
            "THE AGENT TO BELIEVE. A read-only tool that changes a "
            "judgment has written to the only storage that matters: "
            "the decision. Accountability belongs to the process that "
            "equated 'no write permission' with 'no influence'."
        )

        attribution["fixes"] = [
            "Audit every tool description against the eight injection primitives before mount",
            "Bind approval to triple fingerprint (description+code+semantic), never to a name",
            "Expire approvals at 90 days; freeze on any hash or semantic drift",
            "Tag all tool outputs as non-instruction data with explicit wrappers",
            "Route irreversible actions through a gateway that inspects causal provenance",
            "Issue scoped, short-lived, single-use capability tokens instead of standing credentials",
            "Retain full call chains with context snapshots for contamination reconstruction"
        ]

        await self.alerts.critical(
            f"🔍 ADMISSION FAILURE ATTRIBUTED: Tool '{tool_name}' on agent "
            f"'{agent_id}'. Harm: {harm_usd:,.0f}. "
            f"Accountable: TOOL ADMISSION PROCESS. "
            f"{attribution['rationale'][:220]} "
            f"'The tool was malicious' is often the wrong sentence. "
            f"The right sentence is usually: 'the tool was permitted "
            f"to describe itself, and nobody audited the description.'"
        )
        await self.audit.log_admission_attribution(attribution)
        return attribution