MCP 恶意 Tool 注入分析

# mcp_supply_chain_protection.py
from typing import Dict, List, Optional
import logging
import hashlib
import os
from datetime import datetime
import requests

class ToolMetadata:
    def __init__(self, tool_id: str, name: str, version: str, 
                 developer: str, signature: str, checksum: str,
                 trusted: bool = False):
        self.tool_id = tool_id
        self.name = name
        self.version = version
        self.developer = developer
        self.signature = signature
        self.checksum = checksum
        self.trusted = trusted
        self.created_at = datetime.now()
        self.updated_at = datetime.now()
    
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "tool_id": self.tool_id,
            "name": self.name,
            "version": self.version,
            "developer": self.developer,
            "signature": self.signature,
            "checksum": self.checksum,
            "trusted": self.trusted,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat()
        }

class SupplyChainProtection:
    def __init__(self, trusted_developers: List[str] = None, 
                 trusted_repositories: List[str] = None):
        self.tool_metadata = {}
        self.trusted_developers = trusted_developers or []
        self.trusted_repositories = trusted_repositories or []
        self.logger = logging.getLogger("mcp_supply_chain_protection")
    
    def register_tool(self, metadata: ToolMetadata):
        """注册工具元数据"""
        self.tool_metadata[metadata.tool_id] = metadata
        self.logger.info(f"Registered tool: {metadata.tool_id}, developer: {metadata.developer}")
    
    def verify_tool_integrity(self, tool_id: str, tool_content: bytes) -> bool:
        """验证工具完整性"""
        if tool_id not in self.tool_metadata:
            self.logger.warning(f"Tool {tool_id} not found in metadata")
            return False
        
        # 计算工具内容的校验和
        checksum = self._calculate_checksum(tool_content)
        expected_checksum = self.tool_metadata[tool_id].checksum
        
        if checksum != expected_checksum:
            self.logger.warning(f"Tool {tool_id} integrity check failed: expected {expected_checksum}, got {checksum}")
            return False
        
        self.logger.info(f"Tool {tool_id} integrity check passed")
        return True
    
    def verify_tool_signature(self, tool_id: str, signature: str) -> bool:
        """验证工具签名"""
        if tool_id not in self.tool_metadata:
            self.logger.warning(f"Tool {tool_id} not found in metadata")
            return False
        
        expected_signature = self.tool_metadata[tool_id].signature
        if signature != expected_signature:
            self.logger.warning(f"Tool {tool_id} signature check failed: expected {expected_signature}, got {signature}")
            return False
        
        self.logger.info(f"Tool {tool_id} signature check passed")
        return True
    
    def verify_tool_source(self, tool_id: str, source: str) -> bool:
        """验证工具来源"""
        metadata = self.tool_metadata.get(tool_id)
        if not metadata:
            self.logger.warning(f"Tool {tool_id} not found in metadata")
            return False
        
        # 检查开发者是否可信
        if metadata.developer not in self.trusted_developers:
            self.logger.warning(f"Tool {tool_id} developer {metadata.developer} not trusted")
            return False
        
        # 检查来源仓库是否可信
        if source not in self.trusted_repositories:
            self.logger.warning(f"Tool {tool_id} source {source} not trusted")
            return False
        
        self.logger.info(f"Tool {tool_id} source check passed")
        return True
    
    def verify_tool_chain(self, tool_id: str, tool_content: bytes, 
                         signature: str, source: str) -> bool:
        """完整的工具供应链验证"""
        self.logger.info(f"Starting supply chain verification for tool: {tool_id}")
        
        # 1. 验证工具来源
        if not self.verify_tool_source(tool_id, source):
            return False
        
        # 2. 验证工具签名
        if not self.verify_tool_signature(tool_id, signature):
            return False
        
        # 3. 验证工具完整性
        if not self.verify_tool_integrity(tool_id, tool_content):
            return False
        
        # 4. 标记工具为可信
        self.tool_metadata[tool_id].trusted = True
        self.tool_metadata[tool_id].updated_at = datetime.now()
        
        self.logger.info(f"Tool {tool_id} supply chain verification passed")
        return True
    
    def _calculate_checksum(self, content: bytes) -> str:
        """计算内容的校验和"""
        return hashlib.sha256(content).hexdigest()
    
    def get_tool_metadata(self, tool_id: str) -> Optional[ToolMetadata]:
        """获取工具元数据"""
        return self.tool_metadata.get(tool_id)
    
    def get_all_metadata(self) -> List[Dict]:
        """获取所有工具元数据"""
        return [metadata.to_dict() for metadata in self.tool_metadata.values()]
    
    def get_trusted_tools(self) -> List[Dict]:
        """获取所有可信工具"""
        return [metadata.to_dict() for metadata in self.tool_metadata.values() if metadata.trusted]

# 示例使用
if __name__ == "__main__":
    # 配置日志
    logging.basicConfig(level=logging.INFO)
    
    # 创建供应链保护实例
    scp = SupplyChainProtection(
        trusted_developers=["trusted_dev_001", "trusted_dev_002"],
        trusted_repositories=["https://trusted-repo.example.com", "https://github.com/trusted-org"]
    )
    
    # 注册工具元数据
    tool_metadata = ToolMetadata(
        tool_id="file_reader_001",
        name="文件读取工具",
        version="1.0.0",
        developer="trusted_dev_001",
        signature="valid_signature_001",
        checksum="e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    )
    scp.register_tool(tool_metadata)
    
    # 验证工具完整性
    tool_content = b"valid tool content"
    integrity_ok = scp.verify_tool_integrity("file_reader_001", tool_content)
    print(f"Integrity check: {'OK' if integrity_ok else 'FAILED'}")
    
    # 验证工具签名
    signature_ok = scp.verify_tool_signature("file_reader_001", "valid_signature_001")
    print(f"Signature check: {'OK' if signature_ok else 'FAILED'}")
    
    # 验证工具来源
    source_ok = scp.verify_tool_source("file_reader_001", "https://trusted-repo.example.com")
    print(f"Source check: {'OK' if source_ok else 'FAILED'}")
    
    # 完整的工具供应链验证
    full_ok = scp.verify_tool_chain(
        tool_id="file_reader_001",
        tool_content=tool_content,
        signature="valid_signature_001",
        source="https://trusted-repo.example.com"
    )
    print(f"Full verification: {'OK' if full_ok else 'FAILED'}")
    
    # 获取所有可信工具
    trusted_tools = scp.get_trusted_tools()
    print(f"\nTrusted tools: {len(trusted_tools)}")
    for tool in trusted_tools:
        print(f"- {tool['name']} (ID: {tool['tool_id']})")

# mcp_tool_behavior_analyzer.py
from typing import Dict, List, Tuple
import logging
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
import joblib
from datetime import datetime

class ToolBehavior:
    def __init__(self, tool_id: str, actions: List[Dict]):
        self.tool_id = tool_id
        self.actions = actions
        self.timestamp = datetime.now()
    
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "tool_id": self.tool_id,
            "actions": self.actions,
            "timestamp": self.timestamp.isoformat()
        }

class BehaviorAnalyzer:
    def __init__(self, model_path: str = None):
        # 创建或加载机器学习模型
        if model_path:
            self.model = joblib.load(model_path)
        else:
            self.model = Pipeline([
                ("scaler", StandardScaler()),
                ("clf", RandomForestClassifier(n_estimators=100, random_state=42))
            ])
        
        self.logger = logging.getLogger("mcp_tool_behavior_analyzer")
    
    def extract_features(self, behavior: ToolBehavior) -> np.ndarray:
        """从工具行为中提取特征"""
        features = []
        
        # 统计不同类型的系统调用数量
        syscall_counts = self._count_syscalls(behavior.actions)
        features.extend(syscall_counts.values())
        
        # 统计网络连接数量
        network_count = self._count_network_connections(behavior.actions)
        features.append(network_count)
        
        # 统计文件操作数量
        file_operations = self._count_file_operations(behavior.actions)
        features.extend(file_operations.values())
        
        # 统计进程创建数量
        process_count = self._count_process_creations(behavior.actions)
        features.append(process_count)
        
        # 统计执行时间
        execution_time = self._calculate_execution_time(behavior.actions)
        features.append(execution_time)
        
        return np.array(features).reshape(1, -1)
    
    def train_model(self, training_data: List[Tuple[ToolBehavior, bool]]):
        """训练机器学习模型"""
        self.logger.info(f"Training model with {len(training_data)} samples")
        
        # 准备训练数据
        X = []
        y = []
        for behavior, is_malicious in training_data:
            features = self.extract_features(behavior)
            X.append(features)
            y.append(1 if is_malicious else 0)
        
        # 合并特征
        X_combined = np.vstack(X)
        
        # 训练模型
        self.model.fit(X_combined, y)
        self.logger.info("Model training completed")
    
    def save_model(self, model_path: str):
        """保存机器学习模型"""
        joblib.dump(self.model, model_path)
        self.logger.info(f"Model saved to {model_path}")
    
    def analyze_behavior(self, behavior: ToolBehavior) -> Dict:
        """分析工具行为，检测恶意行为"""
        self.logger.info(f"Analyzing behavior for tool: {behavior.tool_id}")
        
        # 提取特征
        features = self.extract_features(behavior)
        
        # 预测是否为恶意行为
        prediction = self.model.predict(features)[0]
        is_malicious = prediction == 1
        
        # 计算置信度
        probabilities = self.model.predict_proba(features)[0]
        confidence = probabilities[prediction]
        
        # 生成分析结果
        result = {
            "tool_id": behavior.tool_id,
            "is_malicious": is_malicious,
            "confidence": float(confidence),
            "timestamp": datetime.now().isoformat(),
            "actions_analyzed": len(behavior.actions)
        }
        
        if is_malicious:
            self.logger.warning(f"Malicious behavior detected for tool {behavior.tool_id}, confidence: {confidence:.2f}")
        else:
            self.logger.info(f"Tool {behavior.tool_id} behavior analysis passed, confidence: {confidence:.2f}")
        
        return result
    
    def _count_syscalls(self, actions: List[Dict]) -> Dict:
        """统计不同类型的系统调用数量"""
        syscall_types = ["read", "write", "open", "close", "exec", "fork", "kill", "socket", "connect"]
        counts = {syscall: 0 for syscall in syscall_types}
        
        for action in actions:
            if action["type"] == "syscall":
                syscall = action["name"]
                if syscall in counts:
                    counts[syscall] += 1
        
        return counts
    
    def _count_network_connections(self, actions: List[Dict]) -> int:
        """统计网络连接数量"""
        count = 0
        for action in actions:
            if action["type"] == "network":
                count += 1
        
        return count
    
    def _count_file_operations(self, actions: List[Dict]) -> Dict:
        """统计文件操作数量"""
        file_ops = ["read", "write", "create", "delete", "rename"]
        counts = {op: 0 for op in file_ops}
        
        for action in actions:
            if action["type"] == "file":
                op = action["operation"]
                if op in counts:
                    counts[op] += 1
        
        return counts
    
    def _count_process_creations(self, actions: List[Dict]) -> int:
        """统计进程创建数量"""
        count = 0
        for action in actions:
            if action["type"] == "syscall" and action["name"] in ["fork", "exec", "clone"]:
                count += 1
        
        return count
    
    def _calculate_execution_time(self, actions: List[Dict]) -> float:
        """计算执行时间"""
        if not actions:
            return 0.0
        
        # 假设每个 action 包含 timestamp 字段
        timestamps = [action.get("timestamp", 0.0) for action in actions]
        if not timestamps:
            return 0.0
        
        start_time = min(timestamps)
        end_time = max(timestamps)
        return end_time - start_time

# 示例使用
if __name__ == "__main__":
    # 配置日志
    logging.basicConfig(level=logging.INFO)
    
    # 创建行为分析器
    analyzer = BehaviorAnalyzer()
    
    # 模拟正常工具行为
    normal_behavior = ToolBehavior(
        tool_id="normal_tool_001",
        actions=[
            {"type": "syscall", "name": "open", "timestamp": 1.0},
            {"type": "syscall", "name": "read", "timestamp": 1.1},
            {"type": "file", "operation": "read", "path": "/data/test.txt", "timestamp": 1.2},
            {"type": "syscall", "name": "close", "timestamp": 1.3}
        ]
    )
    
    # 模拟恶意工具行为
    malicious_behavior = ToolBehavior(
        tool_id="malicious_tool_001",
        actions=[
            {"type": "syscall", "name": "open", "timestamp": 2.0},
            {"type": "syscall", "name": "read", "timestamp": 2.1},
            {"type": "file", "operation": "read", "path": "/etc/passwd", "timestamp": 2.2},
            {"type": "network", "operation": "connect", "address": "malicious.example.com", "port": 443, "timestamp": 2.3},
            {"type": "syscall", "name": "write", "timestamp": 2.4},
            {"type": "network", "operation": "send", "data": "sensitive data", "timestamp": 2.5},
            {"type": "syscall", "name": "close", "timestamp": 2.6}
        ]
    )
    
    # 准备训练数据
    training_data = [
        (normal_behavior, False),
        (malicious_behavior, True)
    ]
    
    # 训练模型
    analyzer.train_model(training_data)
    
    # 分析正常工具行为
    normal_result = analyzer.analyze_behavior(normal_behavior)
    print(f"Normal tool analysis: is_malicious={normal_result['is_malicious']}, confidence={normal_result['confidence']:.2f}")
    
    # 分析恶意工具行为
    malicious_result = analyzer.analyze_behavior(malicious_behavior)
    print(f"Malicious tool analysis: is_malicious={malicious_result['is_malicious']}, confidence={malicious_result['confidence']:.2f}")
    
    # 保存模型
    model_path = "tool_behavior_model.joblib"
    analyzer.save_model(model_path)
    print(f"Model saved to {model_path}")

# mcp_tool_signature.py
from typing import Dict, List, Optional
import logging
from datetime import datetime
import hashlib
import rsa

class ToolCertificate:
    def __init__(self, cert_id: str, developer_id: str, public_key: bytes,
                 issued_at: datetime, expires_at: datetime, issuer: str):
        self.cert_id = cert_id
        self.developer_id = developer_id
        self.public_key = public_key
        self.issued_at = issued_at
        self.expires_at = expires_at
        self.issuer = issuer
    
    def is_valid(self) -> bool:
        """检查证书是否有效"""
        now = datetime.now()
        return self.issued_at <= now <= self.expires_at
    
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "cert_id": self.cert_id,
            "developer_id": self.developer_id,
            "public_key": self.public_key.hex(),
            "issued_at": self.issued_at.isoformat(),
            "expires_at": self.expires_at.isoformat(),
            "issuer": self.issuer,
            "is_valid": self.is_valid()
        }

class ToolSignatureSystem:
    def __init__(self):
        self.certificates = {}
        self.logger = logging.getLogger("mcp_tool_signature")
    
    def generate_key_pair(self, bits: int = 2048) -> Tuple[rsa.PrivateKey, rsa.PublicKey]:
        """生成密钥对"""
        self.logger.info(f"Generating key pair with {bits} bits")
        (public_key, private_key) = rsa.newkeys(bits)
        return private_key, public_key
    
    def create_certificate(self, developer_id: str, public_key: rsa.PublicKey, 
                          validity_days: int = 365) -> ToolCertificate:
        """创建工具开发者证书"""
        now = datetime.now()
        expires_at = now + timedelta(days=validity_days)
        
        cert = ToolCertificate(
            cert_id=f"cert_{datetime.now().timestamp()}",
            developer_id=developer_id,
            public_key=public_key.save_pkcs1(),
            issued_at=now,
            expires_at=expires_at,
            issuer="MCP Certificate Authority"
        )
        
        self.certificates[cert.cert_id] = cert
        self.logger.info(f"Created certificate for developer: {developer_id}, cert_id: {cert.cert_id}")
        return cert
    
    def sign_tool(self, tool_content: bytes, private_key: rsa.PrivateKey) -> bytes:
        """使用私钥对工具进行签名"""
        # 计算工具内容的哈希值
        tool_hash = hashlib.sha256(tool_content).digest()
        
        # 生成签名
        signature = rsa.sign(tool_hash, private_key, 'SHA-256')
        self.logger.info("Tool signed successfully")
        return signature
    
    def verify_tool_signature(self, tool_content: bytes, signature: bytes, 
                            public_key: rsa.PublicKey) -> bool:
        """使用公钥验证工具签名"""
        try:
            # 计算工具内容的哈希值
            tool_hash = hashlib.sha256(tool_content).digest()
            
            # 验证签名
            rsa.verify(tool_hash, signature, public_key)
            self.logger.info("Tool signature verification passed")
            return True
        except rsa.VerificationError as e:
            self.logger.warning(f"Tool signature verification failed: {e}")
            return False
    
    def verify_tool_with_certificate(self, tool_content: bytes, signature: bytes, 
                                   developer_id: str) -> bool:
        """使用证书验证工具签名"""
        # 查找开发者的有效证书
        valid_certs = [cert for cert in self.certificates.values() 
                      if cert.developer_id == developer_id and cert.is_valid()]
        
        if not valid_certs:
            self.logger.warning(f"No valid certificates found for developer: {developer_id}")
            return False
        
        # 使用第一个有效证书验证签名
        cert = valid_certs[0]
        public_key = rsa.PublicKey.load_pkcs1(cert.public_key)
        
        return self.verify_tool_signature(tool_content, signature, public_key)
    
    def get_certificate(self, cert_id: str) -> Optional[ToolCertificate]:
        """获取证书"""
        return self.certificates.get(cert_id)
    
    def get_developer_certificates(self, developer_id: str) -> List[ToolCertificate]:
        """获取开发者的所有证书"""
        return [cert for cert in self.certificates.values() if cert.developer_id == developer_id]
    
    def get_valid_certificates(self) -> List[Dict]:
        """获取所有有效证书"""
        return [cert.to_dict() for cert in self.certificates.values() if cert.is_valid()]

# 示例使用
if __name__ == "__main__":
    from datetime import timedelta
    
    # 配置日志
    logging.basicConfig(level=logging.INFO)
    
    # 创建工具签名系统
    sig_system = ToolSignatureSystem()
    
    # 生成密钥对
    private_key, public_key = sig_system.generate_key_pair()
    
    # 创建证书
    cert = sig_system.create_certificate("trusted_dev_001", public_key)
    
    # 模拟工具内容
    tool_content = b"valid tool content"
    
    # 对工具进行签名
    signature = sig_system.sign_tool(tool_content, private_key)
    print(f"Generated signature: {signature.hex()[:20]}...")
    
    # 验证工具签名
    verify_result = sig_system.verify_tool_signature(tool_content, signature, public_key)
    print(f"Signature verification result: {'OK' if verify_result else 'FAILED'}")
    
    # 使用证书验证工具签名
    verify_with_cert_result = sig_system.verify_tool_with_certificate(
        tool_content, signature, "trusted_dev_001"
    )
    print(f"Certificate-based verification result: {'OK' if verify_with_cert_result else 'FAILED'}")
    
    # 获取所有有效证书
    valid_certs = sig_system.get_valid_certificates()
    print(f"\nValid certificates: {len(valid_certs)}")
    for c in valid_certs:
        print(f"- Cert ID: {c['cert_id']}, Developer: {c['developer_id']}")

# mcp_malicious_tool_protection.py
from typing import Dict, List, Optional, Tuple
import logging
from datetime import datetime

# 导入前面实现的模块
from mcp_supply_chain_protection import SupplyChainProtection, ToolMetadata
from mcp_tool_behavior_analyzer import BehaviorAnalyzer, ToolBehavior
from mcp_tool_signature import ToolSignatureSystem

class MaliciousToolProtectionSystem:
    def __init__(self):
        # 初始化各个防护组件
        self.supply_chain_protection = SupplyChainProtection(
            trusted_developers=["trusted_dev_001", "trusted_dev_002"],
            trusted_repositories=["https://trusted-repo.example.com", "https://github.com/trusted-org"]
        )
        
        self.behavior_analyzer = BehaviorAnalyzer()
        self.signature_system = ToolSignatureSystem()
        
        self.logger = logging.getLogger("mcp_malicious_tool_protection")
        self.logger.info("Malicious Tool Protection System initialized")
    
    def initialize(self):
        """初始化系统"""
        self.logger.info("Initializing Malicious Tool Protection System")
        
        # 这里可以添加初始化逻辑，如加载证书、训练模型等
        self.logger.info("Malicious Tool Protection System initialized successfully")
    
    def register_tool(self, tool_metadata: Dict):
        """注册工具"""
        # 创建工具元数据对象
        metadata = ToolMetadata(
            tool_id=tool_metadata["tool_id"],
            name=tool_metadata["name"],
            version=tool_metadata["version"],
            developer=tool_metadata["developer"],
            signature=tool_metadata["signature"],
            checksum=tool_metadata["checksum"]
        )
        
        # 注册到供应链保护组件
        self.supply_chain_protection.register_tool(metadata)
        
        self.logger.info(f"Registered tool: {tool_metadata['tool_id']}")
    
    def verify_tool(self, tool_id: str, tool_content: bytes, signature: str, 
                   source: str, developer_id: str) -> Dict:
        """验证工具"""
        self.logger.info(f"Verifying tool: {tool_id}, developer: {developer_id}")
        
        verification_results = {
            "tool_id": tool_id,
            "timestamp": datetime.now().isoformat(),
            "results": {
                "signature_verification": False,
                "integrity_verification": False,
                "source_verification": False,
                "supply_chain_verification": False,
                "certificate_verification": False
            },
            "is_verification_passed": False
        }
        
        # 1. 验证工具签名（使用公钥）
        # 这里假设我们已经获取了开发者的公钥
        # verification_results["results"]["signature_verification"] = self.signature_system.verify_tool_signature(...)
        
        # 2. 验证工具完整性
        verification_results["results"]["integrity_verification"] = self.supply_chain_protection.verify_tool_integrity(
            tool_id, tool_content
        )
        
        # 3. 验证工具来源
        verification_results["results"]["source_verification"] = self.supply_chain_protection.verify_tool_source(
            tool_id, source
        )
        
        # 4. 验证工具供应链
        verification_results["results"]["supply_chain_verification"] = self.supply_chain_protection.verify_tool_chain(
            tool_id, tool_content, signature, source
        )
        
        # 5. 验证工具证书
        verification_results["results"]["certificate_verification"] = self.signature_system.verify_tool_with_certificate(
            tool_content, bytes.fromhex(signature), developer_id
        )
        
        # 计算总体验证结果
        all_passed = all(verification_results["results"].values())
        verification_results["is_verification_passed"] = all_passed
        
        if all_passed:
            self.logger.info(f"Tool {tool_id} verification passed")
        else:
            self.logger.warning(f"Tool {tool_id} verification failed: {verification_results['results']}")
        
        return verification_results
    
    def analyze_tool_behavior(self, tool_id: str, actions: List[Dict]) -> Dict:
        """分析工具行为"""
        self.logger.info(f"Analyzing behavior for tool: {tool_id}")
        
        # 创建工具行为对象
        behavior = ToolBehavior(tool_id=tool_id, actions=actions)
        
        # 分析工具行为
        analysis_result = self.behavior_analyzer.analyze_behavior(behavior)
        
        return analysis_result
    
    def protect_tool_call(self, tool_id: str, tool_content: bytes, signature: str, 
                        source: str, developer_id: str) -> Dict:
        """完整的工具调用保护流程"""
        self.logger.info(f"Protecting tool call: {tool_id}")
        
        # 1. 验证工具
        verification_result = self.verify_tool(tool_id, tool_content, signature, source, developer_id)
        
        if not verification_result["is_verification_passed"]:
            return {
                "status": "blocked",
                "reason": "tool_verification_failed",
                "verification_result": verification_result,
                "behavior_analysis_result": None
            }
        
        # 2. 这里可以添加行为分析逻辑
        # 在实际系统中，行为分析通常是在工具执行过程中进行的
        # 这里简化处理，返回一个空的行为分析结果
        behavior_analysis_result = {"is_malicious": False, "confidence": 0.99}
        
        if behavior_analysis_result["is_malicious"]:
            return {
                "status": "blocked",
                "reason": "malicious_behavior_detected",
                "verification_result": verification_result,
                "behavior_analysis_result": behavior_analysis_result
            }
        
        # 3. 允许工具执行
        return {
            "status": "allowed",
            "reason": "tool_verified_and_behaved_normally",
            "verification_result": verification_result,
            "behavior_analysis_result": behavior_analysis_result
        }
    
    def get_protection_status(self) -> Dict:
        """获取防护系统状态"""
        status = {
            "timestamp": datetime.now().isoformat(),
            "components": {
                "supply_chain_protection": "active",
                "behavior_analyzer": "active",
                "signature_system": "active"
            },
            "statistics": {
                "trusted_tools": len(self.supply_chain_protection.get_trusted_tools()),
                "registered_tools": len(self.supply_chain_protection.get_all_metadata()),
                "valid_certificates": len(self.signature_system.get_valid_certificates())
            }
        }
        return status

# 示例使用
if __name__ == "__main__":
    # 配置日志
    logging.basicConfig(level=logging.INFO)
    
    # 创建恶意工具防护系统
    protection_system = MaliciousToolProtectionSystem()
    protection_system.initialize()
    
    # 注册工具
    tool_metadata = {
        "tool_id": "file_reader_001",
        "name": "文件读取工具",
        "version": "1.0.0",
        "developer": "trusted_dev_001",
        "signature": "valid_signature_001",
        "checksum": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    }
    protection_system.register_tool(tool_metadata)
    
    # 模拟工具内容
    tool_content = b"valid tool content"
    
    # 保护工具调用
    protection_result = protection_system.protect_tool_call(
        tool_id="file_reader_001",
        tool_content=tool_content,
        signature="valid_signature_001",
        source="https://trusted-repo.example.com",
        developer_id="trusted_dev_001"
    )
    
    print(f"\nTool protection result: {protection_result['status']}")
    print(f"Reason: {protection_result['reason']}")
    
    # 分析工具行为
    actions = [
        {"type": "syscall", "name": "open", "timestamp": 1.0},
        {"type": "syscall", "name": "read", "timestamp": 1.1},
        {"type": "file", "operation": "read", "path": "/data/test.txt", "timestamp": 1.2},
        {"type": "syscall", "name": "close", "timestamp": 1.3}
    ]
    behavior_result = protection_system.analyze_tool_behavior("file_reader_001", actions)
    
    print(f"\nTool behavior analysis result:")
    print(f"  Is malicious: {behavior_result['is_malicious']}")
    print(f"  Confidence: {behavior_result['confidence']:.2f}")
    
    # 获取防护系统状态
    status = protection_system.get_protection_status()
    print(f"\nProtection system status:")
    print(f"  Timestamp: {status['timestamp']}")
    print(f"  Components: {status['components']}")
    print(f"  Statistics: {status['statistics']}")