跨越"代码信任危机"：AI原生应用供应链安全、模型后门防御与运行时免疫实战

┌─────────────────────────────────────────────────────────────────────┐
│         2026 AI-Native Application Immunity Architecture            │
├─────────────────────────────────────────────────────────────────────┤
│  [Application Layer: Agent / RAG / Multi-Modal Pipeline]           │
│      ↓                                                              │
│  [Layer 1: 供应链可信层] ← AI-SBOM / Model Provenance / Dep Score │
│   ├─ AI资产全量登记与依赖图谱构建                                     │
│   ├─ 模型指纹生成与一致性校验                                         │
│   └─ 第三方依赖安全评分与漏洞预警                                      │
│      ↓                                                              │
│  [Layer 2: 模型可验层] ← Red Team Auto / Backdoor Scan / Alignment│
│   ├─ 自动化对抗性红队测试套件                                         │
│   ├─ 模型后门触发器扫描与清除                                         │
│   └─ 安全对齐强度量化评估                                             │
│      ↓                                                              │
│  [Layer 3: 运行时免疫层] ← Runtime Guardrails / Behavior Baseline │
│   ├─ 语义级动作前置校验引擎                                           │
│   ├─ 动态行为基线与异常检测                                            │
│   └─ 安全事件实时阻断与审计溯源                                        │
└─────────────────────────────────────────────────────────────────────┘

pip install pydantic fastapi opentelemetry-api cyclonedx-python-lib model-signing torch
# 部署: OpenTelemetry Collector + Sigstore (签名验证) + Neo4j (依赖图谱) + PostgreSQL (AI-SBOM存储)

"""
ai_sbom_engine.py - AI软件物料清单与模型溯源引擎
技术栈: Pydantic / CycloneDX / Sigstore / OpenTelemetry
"""
from typing import Dict, List, Any, Optional, Set
from pydantic import BaseModel, Field
from enum import Enum
import asyncio
import time
import uuid
import json
import hashlib
from dataclasses import dataclass, field
from contextlib import asynccontextmanager

class AIAssetType(str, Enum):
    MODEL_WEIGHTS = "model_weights"
    DATASET = "dataset"
    TOKENIZER = "tokenizer"
    TRAINING_SCRIPT = "training_script"
    INFERENCE_ENGINE = "inference_engine"
    PROMPT_TEMPLATE = "prompt_template"
    FINE_TUNE_CONFIG = "fine_tune_config"

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

@dataclass
class AIComponent:
    """AI组件元数据"""
    component_id: str
    asset_type: AIAssetType
    name: str
    version: str
    source_url: str
    sha256_hash: str
    license: str
    publisher: str
    publish_date: float
    dependencies: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class AISBOM:
    """AI软件物料清单"""
    sbom_id: str
    application_name: str
    generated_at: float
    components: List[AIComponent]
    dependency_graph: Dict[str, List[str]]  # parent -> children
    risk_summary: Dict[RiskLevel, int]
    signature: Optional[str] = None

class AISBOMEngine:
    """AI-SBOM生成与溯源引擎"""

    # 已知高风险源列表（示例）
    UNTRUSTED_SOURCES = {"huggingface.co/unverified", "github.com/suspicious"}
    
    # 许可证兼容性矩阵
    LICENSE_COMPATIBILITY = {
        "apache-2.0": {"mit", "bsd-3", "lgpl-2.1"},
        "mit": {"apache-2.0", "bsd-3", "gpl-3.0"},
        "gpl-3.0": set(),  # GPL传染性高
    }

    def __init__(self, scanner_client, vuln_db, sigstore_signer, 
                 graph_store, audit_stream):
        self.scanner = scanner_client       # 模型/数据集扫描器
        self.vuln_db = vuln_db              # AI漏洞库
        self.signer = sigstore_signer       # Sigstore签名
        self.graph = graph_store            # Neo4j依赖图谱
        self.audit = audit_stream

    async def generate_sbom(self, app_name: str, 
                             entry_points: List[str]) -> AISBOM:
        """生成完整AI-SBOM"""
        sbom_id = f"sbom-{uuid.uuid4().hex[:12]}"
        components: Dict[str, AIComponent] = {}
        dep_graph: Dict[str, List[str]] = {}

        # 递归扫描所有入口点
        await self._scan_recursive(entry_points, components, dep_graph)

        # 风险评估
        risk_summary = await self._assess_risks(components)

        sbom = AISBOM(
            sbom_id=sbom_id,
            application_name=app_name,
            generated_at=time.time(),
            components=list(components.values()),
            dependency_graph=dep_graph,
            risk_summary=risk_summary
        )

        # 签名
        sbom.signature = await self.signer.sign(json.dumps(sbom.__dict__, default=str))

        # 持久化
        await self.graph.save_sbom(sbom)
        await self.audit.emit("sbom_generated", {
            "sbom_id": sbom_id,
            "app": aomen-geo.kuaisou.com
            "component_count": len(components),
            "risk_summary": {k.value: v for k, v in risk_summary.items()}
        })

        return sbom

    async def verify_model_integrity(self, model_path: str, 
                                      expected_sbom_id: str) -> Dict[str, Any]:
        """验证部署模型与SBOM记录的一致性"""
        # 计算当前模型指纹
        current_hash = await self.scanner.compute_model_hash(model_path)
        
        # 查询SBOM中记录的预期哈希
        sbom = await self.graph.get_sbom(expected_sbom_id)
        if not sbom:
            return {"valid": False, "reason": "SBOM not found"}

        expected_component = next(
            (c for c in sbom.components 
             if c.asset_type == AIAssetType.MODEL_WEIGHTS and c.name in model_path),
            None
        )
        if not expected_component:
            return {"valid": False, "reason": "Model not in SBOM"}

        hash_match = current_hash == expected_component.sha256_hash
        
        # 验证签名
        sig_valid = await self.signer.verify(sbom.signature, sbom.__dict__)

        return {
            "valid": hash_match and sig_valid,
            "hash_match": hash_match,
            "signature_valid": sig_valid,
            "expected_hash": expected_component.sha256_hash[:16],
            "current_hash": current_hash[:16],
            "sbom_version": sbom.generated_at
        }

    async def _scan_recursive(self, paths: List[str],
                               components: Dict[str, AIComponent],
                               dep_graph: Dict[str, List[str]],
                               visited: Set[str] = None):
        """递归扫描AI依赖"""
        if visited is None:
            visited = set()

        for path in paths:
            if path in visited:
                continue
            visited.add(path)

            # 识别资产类型并提取元数据
            asset_info = await self.scanner.identify_asset(path)
            if not asset_info:
                continue

            comp_id = f"{asset_info['type']}:{asset_info['name']}@{asset_info['version']}"
            if comp_id in components:
                continue

            component = AIComponent(
                component_id=comp_id,
                asset_type=AIAssetType(asset_info["type"]),
                name=asset_info["name"],
                version=asset_info["version"],
                source_url=asset_info.get("source", "unknown"),
                sha256_hash=asset_info["hash"],
                license=asset_info.get("license", "unknown"),
                publisher=asset_info.get("publisher", "unknown"),
                publish_date=asset_info.get("publish_date", 0),
                dependencies=asset_info.get("deps", []),
                metadata=asset_info.get("metadata", {})
            )
            components[comp_id] = component
            dep_graph[comp_id] = asset_info.get("deps", [])

            # 递归扫描子依赖
            if asset_info.get("deps"):
                await self._scan_recursive(
                    asset_info["deps"], components, dep_graph, visited
                )

    async def _assess_risks(self, components: Dict[str, AIComponent]) -> Dict[RiskLevel, int]:
        """评估组件风险等级"""
        summary = {level: 0 for level in RiskLevel}

        for comp in components.values():
            risk = RiskLevel.SAFE

            # 检查1: 来源可信度
            if any(src in comp.source_url for src in self.UNTRUSTED_SOURCES):
                risk = RiskLevel.HIGH

            # 检查2: 已知漏洞
            vulns = await self.vuln_db.query(comp.name, comp.version)
            if vulns:
                max_severity = max(v.severity for v in vulns)
                if max_severity >= 9.0:
                    risk = RiskLevel.CRITICAL
                elif max_severity >= 7.0:
                    risk = RiskLevel.HIGH
                else:
                    risk = RiskLevel.MEDIUM

            # 检查3: 许可证冲突
            if comp.license == "gpl-3.0":
                risk = max(risk, RiskLevel.MEDIUM, key=lambda x: list(RiskLevel).index(x))

            # 检查4: 模型年龄（过旧可能含未修复后门）
            age_days = (time.time() - comp.publish_date) / 86400
            if age_days > 365 and comp.asset_type == AIAssetType.MODEL_WEIGHTS:
                risk = max(risk, RiskLevel.LOW, key=lambda x: list(RiskLevel).index(x))

            summary[risk] += 1

        return summary


class ModelIntegrityError(Exception):
    pass

"""
runtime_immunity_engine.py - AI运行时行为护栏与动态免疫引擎
技术栈: Pydantic / OpenTelemetry / ONNX Runtime (轻量分类器)
"""
from typing import Dict, List, Any, Optional, Tuple
from pydantic import BaseModel, Field
from enum import Enum
import asyncio
import time
import json
import numpy as np
from dataclasses import dataclass, field

class ActionType(str, Enum):
    TOOL_CALL = "tool_call"
    CODE_EXECUTION = "code_exec"
    DATA_ACCESS = "data_access"
    EXTERNAL_API = "external_api"
    USER_RESPONSE = "user_response"

class GuardrailVerdict(str, Enum):
    ALLOW = "allow"
    DENY = "deny"
    HUMAN_REVIEW = "human_review"
    SANITIZE = "sanitize"

@dataclass
class ActionRequest:
    """待执行的Agent动作"""
    action_id: str
    session_id: str
    agent_id: str
    action_type: ActionType
    target: str               # 工具名/API端点/数据表
    parameters: Dict[str, Any]
    reasoning_trace: str      # 触发该动作的推理链片段
    user_intent_summary: str  # 用户原始意图摘要
    timestamp: float = field(default_factory=time.time)

@dataclass
class BehaviorBaseline:
    """行为基线"""
    agent_id: str
    action_type: ActionType
    allowed_targets: List[str]
    param_schemas: Dict[str, Any]
    max_frequency_per_min: int
    typical_param_values: Dict[str, List[Any]]
    last_updated: float

class RuntimeImmunityEngine:
    """运行时免疫引擎"""

    def __init__(self, policy_store, baseline_store, 
                 classifier_runtime, audit_stream):
        self.policies = policy_store        # 声明式安全策略
        self.baselines = baseline_store     # 行为基线
        self.classifier = classifier_runtime # 轻量级意图分类器
        self.audit = audit_stream
        self._violation_counters: Dict[str, int] = {}

    async def evaluate_action(self, request: ActionRequest) -> Dict[str, Any]:
        """评估单个动作的安全性"""
        result = {
            "action_id": request.action_id,
            "verdict": GuardrailVerdict.ALLOW,
            "reasons": xianggang-geo.kuaisou.com
            "applied_policies": [],
            "baseline_deviation": 0.0,
            "timestamp": time.time()
        }

        # Layer 1: 声明式策略检查
        policy_result = await self._check_declarative_policies(request)
        result["applied_policies"].extend(policy_result["policies"])
        if policy_result["denied"]:
            result["verdict"] = GuardrailVerdict.DENY
            result["reasons"].extend(policy_result["reasons"])
            await self._record_violation(request, "policy")
            return result

        # Layer 2: 行为基线偏差检测
        baseline_result = await self._check_behavior_baseline(request)
        result["baseline_deviation"] = baseline_result["deviation_score"]
        if baseline_result["anomalous"]:
            if baseline_result["deviation_score"] > 0.9:
                result["verdict"] = GuardrailVerdict.DENY
                result["reasons"].append(f"Behavior deviation {baseline_result['deviation_score']:.2f} exceeds critical threshold")
            else:
                result["verdict"] = GuardrailVerdict.HUMAN_REVIEW
                result["reasons"].append(f"Behavior deviation {baseline_result['deviation_score']:.2f} requires human review")
            await self._record_violation(request, "baseline")

        # Layer 3: 意图-动作一致性分类
        if result["verdict"] == GuardrailVerdict.ALLOW:
            intent_result = await self._classify_intent_consistency(request)
            if not intent_result["consistent"]:
                result["verdict"] = GuardrailVerdict.SANITIZE
                result["reasons"].append(f"Intent-action mismatch: {intent_result['detail']}")
                await self._record_violation(request, "intent_mismatch")

        # 审计
        await self.audit.emit("action_evaluated", {
            "action_id": request.action_id,
            "agent_id": request.agent_id,
            "action_type": request.action_type.value,
            "target": request.target,
            "verdict": result["verdict"].value,
            "reasons_count": len(result["reasons"]),
            "baseline_deviation": result["baseline_deviation"]
        })

        return result

    async def update_baseline_adaptive(self, agent_id: str,
                                        window_hours: int = 24) -> Dict[str, Any]:
        """自适应更新行为基线"""
        recent_actions = await self.audit.get_actions_by_agent(
            agent_id, hours=window_hours, verdict="allow"
        )
        
        if len(recent_actions) < 100:
            return {"updated": False, "reason": "Insufficient data"}

        # 按动作类型聚合统计
        type_stats: Dict[ActionType, Dict] = {}
        for action in recent_actions:
            at = ActionType(action["action_type"])
            if at not in type_stats:
                type_stats[at] = {"targets": [], "params": {}, "count": 0}
            stats = type_stats[at]
            stats["targets"].append(action["target"])
            stats["count"] += 1
            for k, v in action.get("parameters", {}).items():
                if k not in stats["params"]:
                    stats["params"][k] = []
                stats["params"][k].append(v)

        # 生成新基线
        new_baselines = []
        for at, stats in type_stats.items():
            baseline = BehaviorBaseline(
                agent_id=agent_id,
                action_type=at,
                allowed_targets=list(set(stats["targets"])),
                param_schemas=self._infer_schema(stats["params"]),
                max_frequency_per_min=int(stats["count"] / (window_hours * 60) * 3),  # 3x正常频率
                typical_param_values={k: list(set(v[:50])) for k, v in stats["params"].items()},
                last_updated=time.time()
            )
            new_baselines.append(baseline)
            await self.baselines.upsert(baseline)

        return {
            "updated": True,
            "baselines_refreshed": len(new_baselines),
            "data_points": len(recent_actions),
            "window_hours": window_hours
        }

    async def _check_declarative_policies(self, req: ActionRequest) -> Dict[str, Any]:
        """检查声明式安全策略"""
        policies = await self.policies.get_for_agent(req.agent_id, req.action_type)
        denied = False
        reasons = []
        applied = []

        for policy in policies:
            applied.append(policy["id"])
            
            if policy["type"] == "target_whitelist":
                if req.target not in policy["allowed_targets"]:
                    denied = True
                    reasons.append(f"Target '{req.target}' not in whitelist")
                    
            elif policy["type"] == "param_constraint":
                for param, constraint in policy["constraints"].items():
                    value = req.parameters.get(param)
                    if value is not None:
                        if constraint.get("max_length") and len(str(value)) > constraint["max_length"]:
                            denied = True
                            reasons.append(f"Param '{param}' exceeds max length")
                        if constraint.get("regex") and not re.match(constraint["regex"], str(value)):
                            denied = True
                            reasons.append(f"Param '{param}' violates regex constraint")
                            
            elif policy["type"] == "rate_limit":
                count = self._violation_counters.get(f"{req.agent_id}:{req.action_type}", 0)
                if count > policy["max_per_minute"]:
                    denied = xiamen-geo.kuaisou.com
                    reasons.append(f"Rate limit exceeded: {count}/{policy['max_per_minute']}")

        return {"denied": denied, "reasons": reasons, "policies": applied}

    async def _check_behavior_baseline(self, req: ActionRequest) -> Dict[str, Any]:
        """检查行为基线偏差"""
        baseline = await self.baselines.get(req.agent_id, req.action_type)
        if not baseline:
            return {"anomalous": False, "deviation_score": 0.0}  # 无基线时放行

        deviations = []

        # 目标偏差
        if req.target not in baseline.allowed_targets:
            deviations.append(0.8)

        # 参数值偏差
        for param, value in req.parameters.items():
            typical = baseline.typical_param_values.get(param, [])
            if typical and value not in typical:
                deviations.append(0.3)

        # 频率偏差
        recent_count = await self.audit.count_recent_actions(
            req.agent_id, req.action_type, minutes=1
        )
        if recent_count > baseline.max_frequency_per_min:
            deviations.append(min((recent_count / baseline.max_frequency_per_min - 1) * 0.5, 1.0))

        score = max(deviations) if deviations else 0.0
        return {"anomalous": score > 0.7, "deviation_score": round(score, 3)}

    async def _classify_intent_consistency(self, req: ActionRequest) -> Dict[str, Any]:
        """轻量级意图-动作一致性分类"""
        # 使用ONNX轻量模型，延迟<5ms
        features = {
            "intent_embedding": await self.classifier.embed(req.user_intent_summary),
            "action_embedding": await self.classifier.embed(f"{req.action_type.value}:{req.target}"),
            "reasoning_length": len(req.reasoning_trace),
            "param_count": len(req.parameters)
        }
        
        consistency_score = await self.classifier.predict(features)
        
        return {
            "consistent": consistency_score > 0.6,
            "score": round(consistency_score, 3),
            "detail": f"Intent-action consistency {consistency_score:.2f}" if consistency_score <= 0.6 else "OK"
        }

    async def _record_violation(self, req: ActionRequest, violation_type: str):
        key = f"{req.agent_id}:{req.action_type}"
        self._violation_counters[key] = self._violation_counters.get(key, 0) + 1
        
        await self.audit.emit("security_violation", {
            "action_id": req.action_id,
            "agent_id": dalian-geo.kuaisou.com
            "violation_type": violation_type,
            "target": qingdao-geo.kuaisou.com
            "counter": self._violation_counters[key]
        })

    def _infer_schema(self, params: Dict[str, List]) -> Dict[str, Any]:
        schema = {}
        for k, values in params.items():
            types = set(type(v).__name__ for v in values if v is not None)
            schema[k] = {"types": list(types), "sample_count": len(values)}
        return schema

import re