110：Multi-Agent协作安全机制：构建可靠的多智能体系统

# 智能体间安全通信示例
import hashlib
import hmac
import base64
from cryptography.fernet import Fernet

class SecureCommunication:
    def __init__(self, key):
        self.key = key
        self.cipher_suite = Fernet(key)
    
    def encrypt_message(self, message):
        """加密消息"""
        encrypted = self.cipher_suite.encrypt(message.encode())
        return encrypted
    
    def decrypt_message(self, encrypted_message):
        """解密消息"""
        decrypted = self.cipher_suite.decrypt(encrypted_message)
        return decrypted.decode()
    
    def sign_message(self, message):
        """为消息添加签名"""
        signature = hmac.new(self.key, message.encode(), hashlib.sha256).digest()
        return base64.b64encode(signature).decode()
    
    def verify_signature(self, message, signature):
        """验证消息签名"""
        expected_signature = hmac.new(self.key, message.encode(), hashlib.sha256).digest()
        return hmac.compare_digest(expected_signature, base64.b64decode(signature))

# 使用示例
key = Fernet.generate_key()
secure_comm = SecureCommunication(key)

message = "Agent A to Agent B: Let's collaborate on task X"
encrypted = secure_comm.encrypt_message(message)
signature = secure_comm.sign_message(message)

# 验证过程
decrypted = secure_comm.decrypt_message(encrypted)
is_valid = secure_comm.verify_signature(decrypted, signature)
print(f"Decrypted message: {decrypted}")
print(f"Signature valid: {is_valid}")

# 消息认证实现
import time
import hashlib

class MessageAuthenticator:
    def __init__(self):
        self.agent_keys = {}
    
    def register_agent(self, agent_id, public_key):
        """注册智能体公钥"""
        self.agent_keys[agent_id] = public_key
    
    def create_authenticated_message(self, sender_id, recipient_id, content):
        """创建带认证的消息"""
        message = {
            "sender": sender_id,
            "recipient": recipient_id,
            "content": content,
            "timestamp": time.time()
        }
        # 这里应该使用非对称加密签名
        # 简化示例，实际应使用真实的签名机制
        message["signature"] = self._generate_signature(sender_id, message)
        return message
    
    def verify_message(self, message):
        """验证消息"""
        sender_id = message.get("sender")
        if sender_id not in self.agent_keys:
            return False
        
        # 验证签名
        if not self._verify_signature(sender_id, message):
            return False
        
        # 验证时间戳，防止重放攻击
        if time.time() - message.get("timestamp", 0) > 300:  # 5分钟过期
            return False
        
        return True
    
    def _generate_signature(self, agent_id, message):
        """生成签名"""
        # 简化实现，实际应使用私钥签名
        data = f"{message['sender']}{message['recipient']}{message['content']}{message['timestamp']}"
        return hashlib.sha256(data.encode()).hexdigest()
    
    def _verify_signature(self, agent_id, message):
        """验证签名"""
        # 简化实现，实际应使用公钥验证
        data = f"{message['sender']}{message['recipient']}{message['content']}{message['timestamp']}"
        expected_signature = hashlib.sha256(data.encode()).hexdigest()
        return message.get("signature") == expected_signature

# RBAC权限管理实现
class RBACSystem:
    def __init__(self):
        self.roles = {}
        self.agent_roles = {}
    
    def create_role(self, role_name, permissions):
        """创建角色"""
        self.roles[role_name] = permissions
    
    def assign_role(self, agent_id, role_name):
        """为智能体分配角色"""
        if role_name not in self.roles:
            raise ValueError(f"Role {role_name} does not exist")
        if agent_id not in self.agent_roles:
            self.agent_roles[agent_id] = []
        if role_name not in self.agent_roles[agent_id]:
            self.agent_roles[agent_id].append(role_name)
    
    def check_permission(self, agent_id, permission):
        """检查智能体是否具有特定权限"""
        if agent_id not in self.agent_roles:
            return False
        
        for role_name in self.agent_roles[agent_id]:
            if role_name in self.roles and permission in self.roles[role_name]:
                return True
        
        return False
    
    def remove_role(self, agent_id, role_name):
        """移除智能体的角色"""
        if agent_id in self.agent_roles and role_name in self.agent_roles[agent_id]:
            self.agent_roles[agent_id].remove(role_name)

# 使用示例
rbac = RBACSystem()

# 创建角色
rbac.create_role("admin", ["read", "write", "execute", "manage"])
rbac.create_role("user", ["read", "write"])
rbac.create_role("guest", ["read"])

# 分配角色
rbac.assign_role("agent1", "admin")
rbac.assign_role("agent2", "user")
rbac.assign_role("agent3", "guest")

# 检查权限
print(f"agent1 can manage: {rbac.check_permission('agent1', 'manage')}")
print(f"agent2 can manage: {rbac.check_permission('agent2', 'manage')}")
print(f"agent3 can write: {rbac.check_permission('agent3', 'write')}")

# 动态权限管理
import time

class DynamicPermissionManager:
    def __init__(self, rbac_system):
        self.rbac = rbac_system
        self.temporary_permissions = {}
    
    def grant_temporary_permission(self, agent_id, permission, duration=3600):
        """授予临时权限"""
        if agent_id not in self.temporary_permissions:
            self.temporary_permissions[agent_id] = []
        
        expiry_time = time.time() + duration
        self.temporary_permissions[agent_id].append({
            "permission": permission,
            "expiry": expiry_time
        })
    
    def check_permission(self, agent_id, permission):
        """检查权限（包括临时权限）"""
        # 检查常规权限
        if self.rbac.check_permission(agent_id, permission):
            return True
        
        # 检查临时权限
        if agent_id in self.temporary_permissions:
            current_time = time.time()
            # 过滤过期的临时权限
            valid_permissions = [p for p in self.temporary_permissions[agent_id] if p["expiry"] > current_time]
            self.temporary_permissions[agent_id] = valid_permissions
            
            for p in valid_permissions:
                if p["permission"] == permission:
                    return True
        
        return False
    
    def revoke_temporary_permission(self, agent_id, permission):
        """撤销临时权限"""
        if agent_id in self.temporary_permissions:
            self.temporary_permissions[agent_id] = [
                p for p in self.temporary_permissions[agent_id] 
                if p["permission"] != permission
            ]

# 使用示例
dpm = DynamicPermissionManager(rbac)

# 检查初始权限
print(f"agent3 can write initially: {dpm.check_permission('agent3', 'write')}")

# 授予临时权限
dpm.grant_temporary_permission('agent3', 'write', duration=60)  # 60秒
print(f"agent3 can write with temporary permission: {dpm.check_permission('agent3', 'write')}")

# 等待权限过期
time.sleep(61)
print(f"agent3 can write after expiry: {dpm.check_permission('agent3', 'write')}")

# 信任管理系统
import time

class TrustManager:
    def __init__(self):
        self.trust_scores = {}
        self.interaction_history = {}
    
    def record_interaction(self, sender_id, receiver_id, success, context):
        """记录智能体之间的交互"""
        if sender_id not in self.interaction_history:
            self.interaction_history[sender_id] = {}
        if receiver_id not in self.interaction_history[sender_id]:
            self.interaction_history[sender_id][receiver_id] = []
        
        self.interaction_history[sender_id][receiver_id].append({
            "success": success,
            "timestamp": time.time(),
            "context": context
        })
        
        # 更新信任分数
        self.update_trust_score(sender_id, receiver_id)
    
    def update_trust_score(self, sender_id, receiver_id):
        """更新信任分数"""
        if sender_id not in self.trust_scores:
            self.trust_scores[sender_id] = {}
        
        interactions = self.interaction_history.get(sender_id, {}).get(receiver_id, [])
        if not interactions:
            self.trust_scores[sender_id][receiver_id] = 0.5  # 初始信任分数
            return
        
        # 计算信任分数（最近10次交互的加权平均）
        recent_interactions = interactions[-10:]
        weights = [i+1 for i in range(len(recent_interactions))]  # 越 recent 权重越高
        total_weight = sum(weights)
        
        weighted_sum = sum(
            (1 if interaction["success"] else 0) * weight 
            for interaction, weight in zip(recent_interactions, weights)
        )
        
        trust_score = weighted_sum / total_weight
        self.trust_scores[sender_id][receiver_id] = trust_score
    
    def get_trust_score(self, sender_id, receiver_id):
        """获取信任分数"""
        if sender_id not in self.trust_scores or receiver_id not in self.trust_scores[sender_id]:
            return 0.5  # 默认信任分数
        return self.trust_scores[sender_id][receiver_id]
    
    def is_trusted(self, sender_id, receiver_id, threshold=0.7):
        """判断是否可信"""
        return self.get_trust_score(sender_id, receiver_id) >= threshold

# 使用示例
trust_manager = TrustManager()

# 记录交互
trust_manager.record_interaction("agent1", "agent2", True, "成功完成数据传输")
trust_manager.record_interaction("agent1", "agent2", True, "成功协作完成任务")
trust_manager.record_interaction("agent1", "agent2", False, "未能按时完成任务")

# 获取信任分数
print(f"Trust score from agent1 to agent2: {trust_manager.get_trust_score('agent1', 'agent2')}")
print(f"Is agent2 trusted by agent1: {trust_manager.is_trusted('agent1', 'agent2')}")

# 信任传播实现
class TrustPropagation:
    def __init__(self, trust_manager):
        self.trust_manager = trust_manager
    
    def get_indirect_trust(self, source_id, target_id, intermediaries=None, max_depth=2):
        """计算间接信任分数"""
        if source_id == target_id:
            return 1.0
        
        # 直接信任
        direct_trust = self.trust_manager.get_trust_score(source_id, target_id)
        if direct_trust > 0.5:  # 有直接交互历史
            return direct_trust
        
        # 间接信任（通过 intermediaries）
        if intermediaries is None:
            # 找到所有可能的中介智能体
            intermediaries = []
            for agent_id in self.trust_manager.trust_scores.get(source_id, {}):
                if agent_id != target_id:
                    intermediaries.append(agent_id)
        
        if not intermediaries or max_depth <= 0:
            return 0.5  # 默认信任分数
        
        # 计算通过中介的信任分数
        indirect_trusts = []
        for intermediary in intermediaries:
            # 源到中介的信任
            source_to_intermediary = self.trust_manager.get_trust_score(source_id, intermediary)
            if source_to_intermediary < 0.5:
                continue  # 跳过不可信的中介
            
            # 中介到目标的信任
            intermediary_to_target = self.get_indirect_trust(
                intermediary, target_id, max_depth=max_depth-1
            )
            
            # 计算组合信任分数
            combined_trust = (source_to_intermediary * intermediary_to_target) ** 0.5
            indirect_trusts.append(combined_trust)
        
        if not indirect_trusts:
            return 0.5
        
        # 返回最高的间接信任分数
        return max(indirect_trusts)

# 使用示例
trust_propagation = TrustPropagation(trust_manager)

# 建立间接信任关系
trust_manager.record_interaction("agent2", "agent3", True, "成功完成协作")

# 计算间接信任
indirect_trust = trust_propagation.get_indirect_trust("agent1", "agent3")
print(f"Indirect trust from agent1 to agent3: {indirect_trust}")

# 恶意智能体检测系统
import time

class MaliciousAgentDetector:
    def __init__(self, trust_manager):
        self.trust_manager = trust_manager
        self.suspicious_activities = {}
    
    def monitor_agent(self, agent_id, activity, context):
        """监控智能体活动"""
        if agent_id not in self.suspicious_activities:
            self.suspicious_activities[agent_id] = []
        
        # 检测可疑活动
        is_suspicious = self._detect_suspicious_activity(activity, context)
        if is_suspicious:
            self.suspicious_activities[agent_id].append({
                "activity": activity,
                "context": context,
                "timestamp": time.time()
            })
        
        # 检查是否达到恶意阈值
        return self.is_malicious(agent_id)
    
    def _detect_suspicious_activity(self, activity, context):
        """检测可疑活动"""
        # 示例规则，实际应根据具体场景制定
        suspicious_patterns = [
            "unauthorized_access",
            "excessive_requests",
            "data_exfiltration",
            "denial_of_service",
            "tampering"
        ]
        
        for pattern in suspicious_patterns:
            if pattern in activity.lower() or pattern in str(context).lower():
                return True
        
        return False
    
    def is_malicious(self, agent_id, threshold=3):
        """判断智能体是否恶意"""
        if agent_id not in self.suspicious_activities:
            return False
        
        # 最近24小时内的可疑活动
        recent_activities = [
            act for act in self.suspicious_activities[agent_id] 
            if time.time() - act["timestamp"] < 86400
        ]
        
        return len(recent_activities) >= threshold
    
    def get_suspicious_activities(self, agent_id, time_range=86400):
        """获取智能体的可疑活动"""
        if agent_id not in self.suspicious_activities:
            return []
        
        return [
            act for act in self.suspicious_activities[agent_id] 
            if time.time() - act["timestamp"] < time_range
        ]

# 使用示例
detector = MaliciousAgentDetector(trust_manager)

# 监控智能体活动
detector.monitor_agent("agent4", "unauthorized_access", "尝试访问未授权资源")
detector.monitor_agent("agent4", "excessive_requests", "短时间内发送大量请求")
detector.monitor_agent("agent4", "data_exfiltration", "尝试导出敏感数据")

# 检查是否恶意
print(f"Is agent4 malicious: {detector.is_malicious('agent4')}")
print(f"Suspicious activities: {detector.get_suspicious_activities('agent4')}")

# 数据安全管理
class DataSecurityManager:
    def __init__(self):
        self.data_classification = {
            "top_secret": {"access_level": 5, "encryption": "AES-256"},
            "secret": {"access_level": 4, "encryption": "AES-128"},
            "confidential": {"access_level": 3, "encryption": "AES-128"},
            "internal": {"access_level": 2, "encryption": "None"},
            "public": {"access_level": 1, "encryption": "None"}
        }
        self.data_access_logs = {}
    
    def classify_data(self, data, sensitivity):
        """对数据进行分类"""
        if sensitivity not in self.data_classification:
            raise ValueError(f"Invalid sensitivity level: {sensitivity}")
        
        classification = self.data_classification[sensitivity]
        return {
            "data": data,
            "sensitivity": sensitivity,
            "access_level": classification["access_level"],
            "encryption": classification["encryption"]
        }
    
    def check_data_access(self, agent_id, data_classification, agent_access_level):
        """检查智能体是否有权访问数据"""
        required_level = data_classification["access_level"]
        if agent_access_level < required_level:
            # 记录访问尝试
            self._log_access_attempt(agent_id, data_classification, False)
            return False
        
        # 记录成功访问
        self._log_access_attempt(agent_id, data_classification, True)
        return True
    
    def encrypt_data(self, data, sensitivity):
        """加密数据"""
        classified_data = self.classify_data(data, sensitivity)
        encryption = classified_data["encryption"]
        
        if encryption == "AES-256":
            # 实际应用中应使用真实的AES加密
            return f"[AES-256 encrypted] {data}"
        elif encryption == "AES-128":
            return f"[AES-128 encrypted] {data}"
        else:
            return data
    
    def decrypt_data(self, encrypted_data, sensitivity):
        """解密数据"""
        if "[AES-256 encrypted]" in encrypted_data:
            return encrypted_data.replace("[AES-256 encrypted] ", "")
        elif "[AES-128 encrypted]" in encrypted_data:
            return encrypted_data.replace("[AES-128 encrypted] ", "")
        else:
            return encrypted_data
    
    def _log_access_attempt(self, agent_id, data_classification, success):
        """记录访问尝试"""
        if agent_id not in self.data_access_logs:
            self.data_access_logs[agent_id] = []
        
        self.data_access_logs[agent_id].append({
            "timestamp": time.time(),
            "data_sensitivity": data_classification["sensitivity"],
            "success": success
        })

# 使用示例
data_security = DataSecurityManager()

# 分类并加密数据
classified_data = data_security.classify_data("敏感商业数据", "secret")
encrypted_data = data_security.encrypt_data("敏感商业数据", "secret")
print(f"Encrypted data: {encrypted_data}")

# 检查访问权限
print(f"Agent with level 4 can access: {data_security.check_data_access('agent1', classified_data, 4)}")
print(f"Agent with level 2 can access: {data_security.check_data_access('agent2', classified_data, 2)}")

# 解密数据
decrypted_data = data_security.decrypt_data(encrypted_data, "secret")
print(f"Decrypted data: {decrypted_data}")

# 安全数据共享系统
import hashlib
import time

class SecureDataSharing:
    def __init__(self, data_security_manager):
        self.data_security = data_security_manager
        self.shared_data = {}
    
    def share_data(self, sender_id, recipient_id, data, sensitivity, expiry_time=None):
        """安全共享数据"""
        # 分类数据
        classified_data = self.data_security.classify_data(data, sensitivity)
        
        # 加密数据
        encrypted_data = self.data_security.encrypt_data(data, sensitivity)
        
        # 生成访问令牌
        access_token = self._generate_access_token(sender_id, recipient_id, sensitivity)
        
        # 存储共享数据
        if recipient_id not in self.shared_data:
            self.shared_data[recipient_id] = []
        
        self.shared_data[recipient_id].append({
            "sender": sender_id,
            "data": encrypted_data,
            "sensitivity": sensitivity,
            "access_token": access_token,
            "timestamp": time.time(),
            "expiry": expiry_time
        })
        
        return access_token
    
    def access_shared_data(self, recipient_id, access_token, agent_access_level):
        """访问共享数据"""
        if recipient_id not in self.shared_data:
            return None
        
        for item in self.shared_data[recipient_id]:
            if item["access_token"] == access_token:
                # 检查是否过期
                if item["expiry"] and time.time() > item["expiry"]:
                    return None
                
                # 检查访问权限
                classified_data = self.data_security.classify_data("", item["sensitivity"])
                if not self.data_security.check_data_access(recipient_id, classified_data, agent_access_level):
                    return None
                
                # 解密数据
                decrypted_data = self.data_security.decrypt_data(item["data"], item["sensitivity"])
                return decrypted_data
        
        return None
    
    def _generate_access_token(self, sender_id, recipient_id, sensitivity):
        """生成访问令牌"""
        data = f"{sender_id}{recipient_id}{sensitivity}{time.time()}"
        return hashlib.sha256(data.encode()).hexdigest()

# 使用示例
secure_sharing = SecureDataSharing(data_security)

# 共享数据
access_token = secure_sharing.share_data("agent1", "agent2", "敏感商业数据", "secret")
print(f"Access token: {access_token}")

# 访问共享数据
shared_data = secure_sharing.access_shared_data("agent2", access_token, 4)
print(f"Shared data: {shared_data}")

# 共识机制实现
import hashlib
import time

class ConsensusManager:
    def __init__(self):
        self.proposals = {}
        self.votes = {}
    
    def propose_action(self, proposer_id, action, context):
        """提出行动建议"""
        proposal_id = self._generate_proposal_id(proposer_id)
        self.proposals[proposal_id] = {
            "proposer": proposer_id,
            "action": action,
            "context": context,
            "timestamp": time.time(),
            "status": "pending"
        }
        self.votes[proposal_id] = {}
        return proposal_id
    
    def vote_on_proposal(self, agent_id, proposal_id, vote):
        """对建议进行投票"""
        if proposal_id not in self.proposals:
            return False
        
        if self.proposals[proposal_id]["status"] != "pending":
            return False
        
        self.votes[proposal_id][agent_id] = vote
        return True
    
    def reach_consensus(self, proposal_id, required_votes=3, approval_threshold=0.6):
        """达成共识"""
        if proposal_id not in self.proposals:
            return False
        
        votes = self.votes.get(proposal_id, {})
        if len(votes) < required_votes:
            return False
        
        # 计算赞成票比例
        approval_count = sum(1 for vote in votes.values() if vote)
        approval_ratio = approval_count / len(votes)
        
        if approval_ratio >= approval_threshold:
            self.proposals[proposal_id]["status"] = "approved"
            return True
        else:
            self.proposals[proposal_id]["status"] = "rejected"
            return False
    
    def get_proposal_status(self, proposal_id):
        """获取建议状态"""
        if proposal_id not in self.proposals:
            return "not_found"
        return self.proposals[proposal_id]["status"]
    
    def _generate_proposal_id(self, proposer_id):
        """生成建议ID"""
        data = f"{proposer_id}{time.time()}"
        return hashlib.sha256(data.encode()).hexdigest()

# 使用示例
consensus = ConsensusManager()

# 提出建议
proposal_id = consensus.propose_action("agent1", "执行任务X", "需要多个智能体协作完成")

# 投票
consensus.vote_on_proposal("agent2", proposal_id, True)
consensus.vote_on_proposal("agent3", proposal_id, True)
consensus.vote_on_proposal("agent4", proposal_id, False)

# 达成共识
consensus_reached = consensus.reach_consensus(proposal_id)
print(f"Consensus reached: {consensus_reached}")
print(f"Proposal status: {consensus.get_proposal_status(proposal_id)}")

# 冲突解决系统
import hashlib
import time

class ConflictResolver:
    def __init__(self, consensus_manager):
        self.consensus = consensus_manager
        self.conflicts = {}
    
    def detect_conflict(self, agent1_id, agent2_id, issue, context):
        """检测冲突"""
        conflict_id = self._generate_conflict_id(agent1_id, agent2_id)
        self.conflicts[conflict_id] = {
            "agents": [agent1_id, agent2_id],
            "issue": issue,
            "context": context,
            "timestamp": time.time(),
            "status": "detected"
        }
        return conflict_id
    
    def resolve_conflict(self, conflict_id, resolution_strategy):
        """解决冲突"""
        if conflict_id not in self.conflicts:
            return False
        
        conflict = self.conflicts[conflict_id]
        
        # 根据策略解决冲突
        if resolution_strategy == "voting":
            # 使用共识机制进行投票
            proposal_id = self.consensus.propose_action(
                "system", 
                f"Resolve conflict between {conflict['agents'][0]} and {conflict['agents'][1]}",
                conflict['context']
            )
            # 这里应该收集投票，简化示例
            self.consensus.vote_on_proposal(conflict['agents'][0], proposal_id, True)
            self.consensus.vote_on_proposal(conflict['agents'][1], proposal_id, False)
            
            if self.consensus.reach_consensus(proposal_id, required_votes=2):
                conflict["status"] = "resolved"
                conflict["resolution"] = "voting"
                return True
        
        elif resolution_strategy == "mediation":
            # 调解策略
            conflict["status"] = "resolved"
            conflict["resolution"] = "mediation"
            return True
        
        elif resolution_strategy == "hierarchy":
            # 层级策略（基于权限或角色）
            conflict["status"] = "resolved"
            conflict["resolution"] = "hierarchy"
            return True
        
        return False
    
    def get_conflict_status(self, conflict_id):
        """获取冲突状态"""
        if conflict_id not in self.conflicts:
            return "not_found"
        return self.conflicts[conflict_id]["status"]
    
    def _generate_conflict_id(self, agent1_id, agent2_id):
        """生成冲突ID"""
        # 确保ID的一致性，无论agent顺序如何
        sorted_agents = sorted([agent1_id, agent2_id])
        data = f"{sorted_agents[0]}{sorted_agents[1]}{time.time()}"
        return hashlib.sha256(data.encode()).hexdigest()

# 使用示例
resolver = ConflictResolver(consensus)

# 检测冲突
conflict_id = resolver.detect_conflict("agent1", "agent2", "资源分配冲突", "两个智能体都需要访问同一资源")

# 解决冲突
resolved = resolver.resolve_conflict(conflict_id, "voting")
print(f"Conflict resolved: {resolved}")
print(f"Conflict status: {resolver.get_conflict_status(conflict_id)}")

# 安装必要的依赖
pip install cryptography hashlib hmac base64

# 安全配置示例
class SecurityConfig:
    def __init__(self):
        self.encryption_algorithm = "AES-256"
        self.signature_algorithm = "HMAC-SHA256"
        self.token_expiry = 3600  # 1小时
        self.max_login_attempts = 5
        self.lockout_duration = 300  # 5分钟