AI智能体加密邮件竞争系统：密码学与多智能体博弈实战
·
在当今AI技术快速发展的背景下，AI agents（智能体）的应用场景越来越广泛。最近接触到一个很有意思的项目——The Email Game，它让多个AI agents通过加密签名邮件进行竞争互动。这种将密码学与AI结合的设计思路，不仅考验agents的智能决策能力，还确保了通信过程的安全性。本文将完整解析这个项目的技术实现，从加密邮件原理到AI agents的竞争机制，带你一步步搭建自己的AI邮件对战系统。
1. 项目背景与核心概念
1.1 什么是AI agents竞争游戏
AI agents竞争游戏是指多个智能体在特定规则下相互博弈的系统。在The Email Game中，每个AI agent代表一个独立的智能实体，它们通过加密签名的电子邮件进行通信和竞争。这种设计模拟了现实世界中的多方协作与竞争场景，比如商业谈判、资源争夺等场景。
与传统AI系统不同，竞争性AI agents需要具备自主决策、策略规划和风险评估能力。每个agent都有自己的目标函数，通过分析邮件内容、评估对手策略来做出最优决策。这种多智能体系统（Multi-Agent System）的研究对于分布式人工智能发展具有重要意义。
1.2 密码学签名邮件的技术价值
密码学签名在邮件系统中的应用确保了通信的真实性和完整性。在AI agents竞争环境中，加密签名解决了几个关键问题：首先是身份认证，确保每封邮件都来自合法的agent身份；其次是防篡改，保证邮件内容在传输过程中不被恶意修改；最后是非否认性，发送方无法否认自己发送过的邮件。
采用加密签名邮件作为通信载体，为AI agents提供了安全可靠的交互通道。这种设计特别适合需要高度信任保障的竞争环境，比如金融交易模拟、合约谈判等敏感场景。
2. 技术架构与环境准备
2.1 系统架构概述
The Email Game的整体架构包含三个核心模块：AI agents决策引擎、邮件处理中间件和密码学签名服务。AI agents决策引擎负责分析邮件内容、制定回复策略；邮件处理中间件管理邮件的收发队列和存储；密码学签名服务处理邮件的加密、解密和验证流程。
系统采用分布式设计，每个AI agent运行在独立的容器中，通过消息队列进行通信。这种架构保证了系统的可扩展性和容错性，即使某个agent出现故障，也不会影响整体系统的运行。
2.2 开发环境要求
要实现类似的AI邮件竞争系统，需要准备以下开发环境：
基础软件要求：
- Python 3.8+ 运行环境
- PostgreSQL或MySQL数据库
- Redis缓存服务
- Docker容器环境
AI相关依赖：
- TensorFlow或PyTorch深度学习框架
- OpenAI GPT API或本地语言模型
- 强化学习库（如Stable-Baselines3）
密码学工具：
- OpenSSL密码学工具包
- GPG密钥管理工具
- 数字证书生成工具
2.3 项目目录结构
标准的项目目录结构应该清晰划分各个功能模块：
email_game/
├── agents/ # AI agents核心逻辑
│ ├── base_agent.py # 基类定义
│ ├── strategy/ # 策略实现
│ └── models/ # 机器学习模型
├── crypto/ # 密码学模块
│ ├── signature.py # 签名验证
│ ├── encryption.py # 加密解密
│ └── keys/ # 密钥管理
├── email/ # 邮件处理
│ ├── client.py # 邮件客户端
│ ├── parser.py # 邮件解析
│ └── storage.py # 邮件存储
├── game/ # 游戏逻辑
│ ├── rules.py # 规则引擎
│ ├── scoring.py # 评分系统
│ └── monitor.py # 监控面板
└── config/ # 配置文件
├── development.yaml
├── production.yaml
└── agents.yaml
3. 密码学签名邮件实现
3.1 数字签名原理与实现
数字签名基于非对称加密技术，使用私钥签名、公钥验证的模式。在Python中可以使用cryptography库实现：
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.backends import default_backend
import base64
class EmailSigner:
def __init__(self, private_key_path=None, public_key_path=None):
self.private_key = None
self.public_key = None
if private_key_path and public_key_path:
self.load_keys(private_key_path, public_key_path)
else:
self.generate_keys()
def generate_keys(self, key_size=2048):
"""生成RSA密钥对"""
self.private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=key_size,
backend=default_backend()
)
self.public_key = self.private_key.public_key()
def sign_email(self, email_content):
"""对邮件内容进行数字签名"""
if not self.private_key:
raise ValueError("私钥未初始化")
# 对邮件内容进行哈希
hasher = hashes.Hash(hashes.SHA256(), backend=default_backend())
hasher.update(email_content.encode('utf-8'))
digest = hasher.finalize()
# 使用私钥签名
signature = self.private_key.sign(
digest,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return base64.b64encode(signature).decode('utf-8')
def verify_signature(self, email_content, signature, public_key=None):
"""验证数字签名"""
verifying_key = public_key or self.public_key
if not verifying_key:
raise ValueError("公钥未提供")
try:
# 计算内容哈希
hasher = hashes.Hash(hashes.SHA256(), backend=default_backend())
hasher.update(email_content.encode('utf-8'))
digest = hasher.finalize()
# 验证签名
signature_bytes = base64.b64decode(signature)
verifying_key.verify(
signature_bytes,
digest,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return True
except Exception as e:
print(f"签名验证失败: {e}")
return False
3.2 邮件加密与安全传输
除了签名验证，邮件内容加密也是确保安全性的重要环节。下面实现AES对称加密与RSA非对称加密结合的方案：
import os
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import padding as sym_padding
from cryptography.hazmat.primitives import serialization
class EmailEncryptor:
def __init__(self):
self.aes_key_size = 32 # AES-256
def generate_aes_key(self):
"""生成随机的AES密钥"""
return os.urandom(self.aes_key_size)
def encrypt_email(self, email_content, recipient_public_key):
"""使用混合加密方式加密邮件"""
# 生成随机的AES密钥
aes_key = self.generate_aes_key()
# 使用AES加密邮件内容
iv = os.urandom(16) # 初始化向量
padder = sym_padding.PKCS7(128).padder()
padded_data = padder.update(email_content.encode()) + padder.finalize()
cipher = Cipher(algorithms.AES(aes_key), modes.CBC(iv))
encryptor = cipher.encryptor()
encrypted_content = encryptor.update(padded_data) + encryptor.finalize()
# 使用接收方的公钥加密AES密钥
encrypted_key = recipient_public_key.encrypt(
aes_key,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
)
)
return {
'encrypted_content': base64.b64encode(encrypted_content).decode(),
'encrypted_key': base64.b64encode(encrypted_key).decode(),
'iv': base64.b64encode(iv).decode()
}
4. AI Agents竞争机制设计
4.1 Agent决策引擎架构
每个AI agent的核心是一个决策引擎，它需要处理邮件内容分析、策略制定和行动选择。下面是基础决策引擎的实现：
import numpy as np
from typing import Dict, List, Any
from abc import ABC, abstractmethod
class BaseAgent(ABC):
def __init__(self, agent_id: str, config: Dict[str, Any]):
self.agent_id = agent_id
self.config = config
self.memory = [] # 对话记忆
self.score = 0 # 当前得分
@abstractmethod
def analyze_email(self, email_content: str) -> Dict[str, Any]:
"""分析接收到的邮件内容"""
pass
@abstractmethod
def formulate_response(self, analysis_result: Dict[str, Any]) -> str:
"""制定回复策略"""
pass
@abstractmethod
def update_strategy(self, game_state: Dict[str, Any]):
"""根据游戏状态更新策略"""
pass
class StrategicAgent(BaseAgent):
def __init__(self, agent_id: str, config: Dict[str, Any]):
super().__init__(agent_id, config)
self.strategy_model = self._load_strategy_model()
def analyze_email(self, email_content: str) -> Dict[str, Any]:
"""深度分析邮件内容和意图"""
analysis = {
'sender_intent': self._detect_intent(email_content),
'urgency_level': self._assess_urgency(email_content),
'emotional_tone': self._analyze_tone(email_content),
'key_points': self._extract_key_points(email_content),
'potential_traps': self._detect_traps(email_content)
}
return analysis
def formulate_response(self, analysis_result: Dict[str, Any]) -> str:
"""基于多因素决策制定回复"""
strategy = self._select_strategy(analysis_result)
response_template = self._choose_response_template(strategy)
customized_response = self._customize_response(response_template, analysis_result)
return customized_response
def _select_strategy(self, analysis: Dict[str, Any]) -> str:
"""根据分析结果选择应对策略"""
if analysis['potential_traps']:
return 'defensive'
elif analysis['urgency_level'] == 'high':
return 'responsive'
else:
return 'strategic'
4.2 多智能体竞争算法
竞争环境中的AI agents需要采用博弈论算法来优化决策。下面是基于Q学习的竞争策略实现：
import numpy as np
from collections import defaultdict
class CompetitiveQLearning:
def __init__(self, learning_rate=0.1, discount_factor=0.9, exploration_rate=0.1):
self.q_table = defaultdict(lambda: defaultdict(float))
self.learning_rate = learning_rate
self.discount_factor = discount_factor
self.exploration_rate = exploration_rate
def choose_action(self, state, available_actions):
"""根据当前状态选择行动"""
if np.random.random() < self.exploration_rate:
# 探索：随机选择行动
return np.random.choice(available_actions)
else:
# 利用：选择Q值最高的行动
q_values = [self.q_table[state][action] for action in available_actions]
max_q = max(q_values)
# 如果多个行动有相同Q值，随机选择
actions_with_max_q = [action for action, q in zip(available_actions, q_values) if q == max_q]
return np.random.choice(actions_with_max_q)
def update_q_value(self, state, action, reward, next_state, next_available_actions):
"""更新Q值表"""
if next_available_actions:
max_next_q = max([self.q_table[next_state][next_action] for next_action in next_available_actions])
else:
max_next_q = 0
current_q = self.q_table[state][action]
new_q = current_q + self.learning_rate * (reward + self.discount_factor * max_next_q - current_q)
self.q_table[state][action] = new_q
class GameMaster:
def __init__(self, agents: List[BaseAgent], rules: Dict[str, Any]):
self.agents = {agent.agent_id: agent for agent in agents}
self.rules = rules
self.game_state = self._initialize_game_state()
self.q_learners = {agent_id: CompetitiveQLearning() for agent_id in self.agents.keys()}
def process_round(self, sender_id: str, receiver_id: str, email_content: str):
"""处理一轮邮件交互"""
# 验证邮件签名
if not self._verify_email_signature(sender_id, email_content):
return "签名验证失败"
# 接收方分析邮件
receiver_agent = self.agents[receiver_id]
analysis = receiver_agent.analyze_email(email_content)
# 根据当前状态选择回复策略
current_state = self._get_game_state_hash()
available_actions = self._get_available_actions(receiver_id)
chosen_action = self.q_learners[receiver_id].choose_action(current_state, available_actions)
response_content = receiver_agent.formulate_response(analysis, chosen_action)
# 计算奖励并更新Q值
reward = self._calculate_reward(receiver_id, analysis, chosen_action)
next_state = self._get_game_state_hash()
next_actions = self._get_available_actions(receiver_id)
self.q_learners[receiver_id].update_q_value(current_state, chosen_action, reward, next_state, next_actions)
return response_content
5. 完整系统集成实战
5.1 系统配置与初始化
首先创建系统的主配置文件，定义agents参数、游戏规则和邮件服务器设置：
# config/game_config.yaml
game:
name: "AI Email Competition"
max_rounds: 100
scoring_system:
successful_communication: 10
strategic_advantage: 20
trap_avoidance: 15
penalty_miscommunication: -10
email:
smtp_server: "smtp.example.com"
smtp_port: 587
use_tls: true
check_interval: 30 # 秒
agents:
agent1:
type: "strategic"
personality: "aggressive"
learning_rate: 0.1
public_key_path: "keys/agent1_public.pem"
agent2:
type: "cooperative"
personality: "cautious"
learning_rate: 0.05
public_key_path: "keys/agent2_public.pem"
crypto:
algorithm: "RSA"
key_size: 2048
hash_algorithm: "SHA256"
5.2 主控制系统实现
主控制系统负责协调各个模块的工作，管理游戏流程和agent交互：
import asyncio
import yaml
from datetime import datetime
from typing import Dict, List
import smtplib
from email.mime.text import MIMEText
class EmailGameController:
def __init__(self, config_path: str):
self.config = self._load_config(config_path)
self.agents = self._initialize_agents()
self.game_master = GameMaster(list(self.agents.values()), self.config['game'])
self.email_client = EmailClient(self.config['email'])
self.running = False
async def start_game(self):
"""启动游戏主循环"""
self.running = True
print(f"游戏开始于 {datetime.now()}")
while self.running and self.game_master.current_round < self.config['game']['max_rounds']:
await self._process_game_round()
await asyncio.sleep(self.config['email']['check_interval'])
await self._end_game()
async def _process_game_round(self):
"""处理单个游戏回合"""
# 检查新邮件
new_emails = await self.email_client.fetch_new_emails()
for email in new_emails:
# 验证邮件签名和解析发送方
sender_id = self._extract_sender_id(email)
if sender_id not in self.agents:
continue
# 处理邮件内容
response = self.game_master.process_round(
sender_id,
self._determine_receiver(sender_id),
email['content']
)
# 发送回复
if response:
await self._send_response(sender_id, response)
# 更新游戏状态
self.game_master.update_scores()
self._log_round_status()
def _initialize_agents(self) -> Dict[str, BaseAgent]:
"""初始化所有AI agents"""
agents = {}
for agent_id, agent_config in self.config['agents'].items():
if agent_config['type'] == 'strategic':
agents[agent_id] = StrategicAgent(agent_id, agent_config)
elif agent_config['type'] == 'cooperative':
agents[agent_id] = CooperativeAgent(agent_id, agent_config)
# 加载其他类型的agents...
return agents
# 启动游戏
async def main():
controller = EmailGameController('config/game_config.yaml')
await controller.start_game()
if __name__ == "__main__":
asyncio.run(main())
5.3 邮件客户端实现
实现支持加密签名的邮件客户端，处理邮件的发送和接收：
import aiosmtplib
import imaplib
import email
from email.header import decode_header
class SecureEmailClient:
def __init__(self, config: Dict[str, Any]):
self.smtp_config = config['smtp']
self.imap_config = config['imap']
self.signer = EmailSigner()
async def send_secure_email(self, to_address: str, subject: str, content: str,
sender_id: str) -> bool:
"""发送加密签名邮件"""
try:
# 对内容进行数字签名
signature = self.signer.sign_email(content)
# 构建安全邮件头
secure_headers = {
'X-Agent-ID': sender_id,
'X-Signature': signature,
'X-Timestamp': datetime.now().isoformat()
}
# 创建邮件消息
message = MIMEText(content)
message['Subject'] = subject
message['From'] = f"{sender_id}@game.system"
message['To'] = to_address
for header, value in secure_headers.items():
message[header] = value
# 发送邮件
async with aiosmtplib.SMTP(
hostname=self.smtp_config['host'],
port=self.smtp_config['port']
) as smtp:
await smtp.login(
self.smtp_config['username'],
self.smtp_config['password']
)
await smtp.send_message(message)
return True
except Exception as e:
print(f"发送邮件失败: {e}")
return False
async def fetch_secure_emails(self) -> List[Dict]:
"""获取并验证安全邮件"""
emails = []
try:
with imaplib.IMAP4_SSL(self.imap_config['host']) as mail:
mail.login(self.imap_config['username'], self.imap_config['password'])
mail.select('inbox')
status, messages = mail.search(None, 'UNSEEN')
email_ids = messages[0].split()
for email_id in email_ids:
status, msg_data = mail.fetch(email_id, '(RFC822)')
email_message = email.message_from_bytes(msg_data[0][1])
# 验证邮件签名
if self._verify_email_signature(email_message):
email_data = {
'id': email_id,
'sender': self._extract_sender(email_message),
'subject': self._decode_header(email_message['Subject']),
'content': self._extract_content(email_message),
'signature': email_message['X-Signature'],
'agent_id': email_message['X-Agent-ID']
}
emails.append(email_data)
except Exception as e:
print(f"获取邮件失败: {e}")
return emails
6. 高级功能与优化策略
6.1 自适应学习机制
为了让AI agents在竞争环境中持续进化，需要实现自适应学习机制：
class AdaptiveLearningManager:
def __init__(self, agents: Dict[str, BaseAgent]):
self.agents = agents
self.performance_history = defaultdict(list)
def analyze_agent_performance(self, agent_id: str, recent_rounds: int = 10):
"""分析agent近期表现"""
history = self.performance_history[agent_id][-recent_rounds:]
if not history:
return None
avg_score = np.mean([h['score'] for h in history])
success_rate = np.mean([1 if h['success'] else 0 for h in history])
strategy_effectiveness = self._calculate_strategy_effectiveness(history)
return {
'average_score': avg_score,
'success_rate': success_rate,
'strategy_effectiveness': strategy_effectiveness,
'improvement_trend': self._detect_improvement_trend(history)
}
def optimize_agent_parameters(self, agent_id: str):
"""根据表现优化agent参数"""
performance = self.analyze_agent_performance(agent_id)
if not performance:
return
agent = self.agents[agent_id]
# 根据表现调整学习率
if performance['improvement_trend'] < 0: # 表现下降
agent.learning_rate *= 1.1 # 增加探索
elif performance['improvement_trend'] > 0.1: # 稳定提升
agent.learning_rate *= 0.9 # 减少探索
# 调整策略权重
if performance['strategy_effectiveness']['defensive'] < 0.5:
agent.defensive_strategy_weight += 0.1
6.2 多维度评分系统
完善的评分系统能够准确反映agents的竞争表现：
class ComprehensiveScoringSystem:
def __init__(self, config: Dict[str, Any]):
self.scoring_rules = config['scoring_system']
self.weight_factors = config.get('weight_factors', {})
def calculate_round_score(self, agent_id: str, round_data: Dict) -> float:
"""计算单回合得分"""
base_score = 0
# 通信有效性得分
if round_data['communication_successful']:
base_score += self.scoring_rules['successful_communication']
# 策略优势得分
strategic_advantage = self._assess_strategic_advantage(round_data)
base_score += strategic_advantage * self.scoring_rules['strategic_advantage']
# 陷阱规避得分
if round_data['trap_avoided']:
base_score += self.scoring_rules['trap_avoidance']
# 惩罚错误通信
if round_data['miscommunication']:
base_score += self.scoring_rules['penalty_miscommunication']
# 应用权重因子
weighted_score = base_score * self._calculate_weight_factor(agent_id, round_data)
return max(0, weighted_score) # 确保分数不为负
def _assess_strategic_advantage(self, round_data: Dict) -> float:
"""评估策略优势程度"""
advantage_score = 0
# 基于对手反应评估
opponent_reaction = round_data.get('opponent_reaction', 'neutral')
if opponent_reaction == 'defensive':
advantage_score += 0.7
elif opponent_reaction == 'confused':
advantage_score += 0.9
# 基于达成目标评估
objectives_achieved = round_data.get('objectives_achieved', 0)
advantage_score += objectives_achieved * 0.3
return min(1.0, advantage_score) # 限制在0-1范围内
7. 部署与监控方案
7.1 容器化部署配置
使用Docker容器化部署确保环境一致性：
# Dockerfile
FROM python:3.9-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
build-essential \
libssl-dev \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
RUN pip install -r requirements.txt
# 复制应用代码
COPY . .
# 创建非root用户
RUN useradd -m -u 1000 agentuser
USER agentuser
# 暴露监控端口
EXPOSE 8080
# 启动命令
CMD ["python", "main.py"]
对应的Docker Compose配置：
# docker-compose.yml
version: '3.8'
services:
email-game:
build: .
ports:
- "8080:8080"
volumes:
- ./config:/app/config
- ./data:/app/data
environment:
- PYTHONPATH=/app
- GAME_ENV=production
depends_on:
- redis
- postgres
redis:
image: redis:6.2-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
postgres:
image: postgres:13
environment:
POSTGRES_DB: emailgame
POSTGRES_USER: agent
POSTGRES_PASSWORD: securepassword
volumes:
- postgres_data:/var/lib/postgresql/data
volumes:
redis_data:
postgres_data:
7.2 系统监控与日志
实现全面的系统监控和日志记录：
import logging
from prometheus_client import Counter, Gauge, Histogram, start_http_server
class MonitoringSystem:
def __init__(self, port=8080):
self.setup_metrics()
self.setup_logging()
start_http_server(port)
def setup_metrics(self):
"""设置Prometheus监控指标"""
self.emails_sent = Counter('emails_sent_total', 'Total emails sent')
self.emails_received = Counter('emails_received_total', 'Total emails received')
self.agent_scores = Gauge('agent_scores', 'Current agent scores', ['agent_id'])
self.response_time = Histogram('response_time_seconds', 'Response time histogram')
def setup_logging(self):
"""配置结构化日志"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('game_system.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger('EmailGame')
def log_round_completion(self, round_number: int, scores: Dict[str, float]):
"""记录回合完成信息"""
self.logger.info(f"Round {round_number} completed", extra={
'scores': scores,
'round': round_number
})
for agent_id, score in scores.items():
self.agent_scores.labels(agent_id=agent_id).set(score)
8. 常见问题与解决方案
8.1 密码学相关问题
问题1：签名验证失败
- 现象 ：邮件签名验证 consistently 失败
- 原因 ：时钟不同步、密钥不匹配、编码问题
- 解决方案 ：
- 检查系统时间同步：
ntpdate pool.ntp.org - 验证密钥对匹配性：重新生成并分发密钥
- 统一字符编码：确保使用UTF-8编码
- 检查系统时间同步：
问题2：加密邮件解密失败
- 现象 ：接收方无法解密邮件内容
- 原因 ：密钥交换问题、算法不匹配、数据损坏
- 解决方案 ：
- 实现密钥交换协议：使用Diffie-Hellman密钥交换
- 检查加密算法一致性：统一使用AES-256-CBC
- 添加数据完整性校验：包含HMAC验证
8.2 AI Agents行为异常
问题1：Agent陷入局部最优
- 现象 ：Agent重复使用相同策略，缺乏创新
- 解决方案 ：
- 增加探索率：动态调整ε-greedy参数
- 引入策略多样性：定期注入随机策略
- 实现课程学习：从简单到复杂逐步训练
问题2：通信僵局
- 现象 ：Agents陷入无限循环的无效通信
- 解决方案 ：
- 设置最大回合数：强制结束僵局回合
- 引入第三方调解：Game Master介入调解
- 实现超时机制：长时间无进展自动跳过
8.3 性能优化问题
问题1：邮件处理延迟
- 现象 ：系统响应时间逐渐变长
- 解决方案 ：
- 实现邮件队列：使用Redis队列管理邮件流
- 优化数据库查询：添加适当索引
- 使用连接池：管理数据库和邮件服务器连接
问题2：内存泄漏
- 现象 ：系统运行时间越长内存占用越高
- 解决方案 ：
- 定期清理缓存：实现LRU缓存策略
- 监控对象生命周期：使用内存分析工具
- 限制历史数据：自动归档旧邮件数据
9. 安全最佳实践
9.1 密钥管理安全
密钥管理是系统安全的核心，必须遵循最小权限原则：
class SecureKeyManager:
def __init__(self, key_storage_path: str):
self.storage_path = key_storage_path
self.encryption_key = self._load_encryption_key()
def store_private_key(self, agent_id: str, private_key: bytes) -> bool:
"""安全存储私钥"""
try:
# 加密私钥
encrypted_key = self._encrypt_key(private_key)
# 安全存储
key_path = os.path.join(self.storage_path, f"{agent_id}.key.enc")
with open(key_path, 'wb') as f:
f.write(encrypted_key)
# 设置严格的文件权限
os.chmod(key_path, 0o600)
return True
except Exception as e:
logging.error(f"存储私钥失败: {e}")
return False
def _encrypt_key(self, key_data: bytes) -> bytes:
"""使用主密钥加密密钥数据"""
# 实现AES-GCM加密确保机密性和完整性
iv = os.urandom(12) # GCM推荐12字节IV
cipher = Cipher(algorithms.AES(self.encryption_key), modes.GCM(iv))
encryptor = cipher.encryptor()
encrypted_data = encryptor.update(key_data) + encryptor.finalize()
return iv + encryptor.tag + encrypted_data
9.2 通信安全加固
确保邮件通信过程中的数据安全：
- 传输层安全 ：强制使用TLS 1.2+加密SMTP/IMAP连接
- 内容安全 ：实现端到端加密，避免中间人攻击
- 身份验证 ：使用双因素认证增强账户安全
- 审计日志 ：记录所有安全相关事件用于事后分析
9.3 系统安全监控
实现实时安全监控和告警：
class SecurityMonitor:
def __init__(self):
self.suspicious_activities = []
self.alert_threshold = 5 # 触发告警的阈值
def monitor_activity(self, activity_type: str, agent_id: str, details: Dict):
"""监控安全相关活动"""
if self._is_suspicious(activity_type, details):
self.suspicious_activities.append({
'timestamp': datetime.now(),
'agent_id': agent_id,
'activity_type': activity_type,
'details': details
})
if len(self.suspicious_activities) >= self.alert_threshold:
self._trigger_security_alert()
def _is_suspicious(self, activity_type: str, details: Dict) -> bool:
"""判断活动是否可疑"""
suspicious_patterns = {
'multiple_failed_logins': details.get('failed_attempts', 0) > 3,
'unusual_sending_pattern': details.get('emails_per_minute', 0) > 10,
'signature_verification_failures': details.get('failed_verifications', 0) > 5
}
return suspicious_patterns.get(activity_type, False)
通过以上完整实现，我们构建了一个安全可靠的AI agents加密邮件竞争系统。这个系统不仅展示了AI与密码学的结合应用，还为多智能体系统研究提供了实用的实验平台。在实际部署时，建议先在测试环境中充分验证各项功能，逐步扩展到生产环境。
更多推荐
5
0- 0
已为社区贡献20条内容
所有评论(0)