TEE可信执行环境集成指南：MSG Chain 机密计算与安全AI Agent执行
数据来源：MSG Chain 代码库核实
主网状态: No-Go — 当前 MSGChain 主网裁决为 No-Go，以下内容反映代码实际状态，不代表生产可用。
目录
1. 概述
1.1 什么是TEE
可信执行环境（Trusted Execution Environment, TEE）是处理器内部的一个安全区域，它保证代码和数据在隔离环境中被机密性和完整性保护。TEE提供以下核心安全属性：
- 机密性（Confidentiality）：未经授权的实体无法读取TEE内部的数据
- 完整性（Integrity）：未经授权的实体无法修改TEE内部的代码或数据
- 可证明性（Attestation）：TEE能够向远程挑战者证明其运行的是未被篡改的代码
1.2 主流TEE技术
Intel SGX（Software Guard Extensions）
SGX是Intel从Skylake架构开始引入的TEE实现。它通过**飞地（Enclave）**来保护应用代码和数据：
- 内存加密引擎（MEE）：硬件级对飞地内存进行加密，缓存行粒度
- EPC（Enclave Page Cache）：受保护的内存区域，存储飞地代码和数据
- 远程证明（Remote Attestation）：基于EPID（Enhanced Privacy ID）或DCAP（Data Center Attestation Primitives）协议
- 密封（Sealing）：将密钥加密后持久化到不可信存储
SGX的安全边界在CPU Ring 3（用户态），飞地是用户进程中的安全区域。
Intel TDX（Trusted Domain Extensions）
TDX是Intel面向虚拟机级别的TEE技术，从第四代Xeon可扩展处理器（Sapphire Rapids）开始支持：
- 虚拟机级隔离：整个虚拟机作为信任域（Trusted Domain, TD）
- 不需要应用修改：客户操作系统无需修改即可运行
- MKTME（Multi-Key Total Memory Encryption）：不同虚拟机使用不同的内存加密密钥
- TD Quote：基于硬件证明的远程证明机制
TDX适用于需要运行完整操作系统的场景，如机密容器和机密虚拟机。
AMD SEV-SNP（Secure Encrypted Virtualization-Secure Nested Paging）
AMD从EPYC Milan（第三代）开始支持SEV-SNP：
- 全内存加密：每个虚拟机使用独立的AES-256加密密钥
- 嵌套页表保护：防止hypervisor对虚拟机内存页表的篡改
- 远程证明：基于AMD签名密钥的证明机制（Attestation Report）
- 反向映射表（RMP）：确保内存页面的所有权不可被篡改
ARM TrustZone
ARM TrustZone是ARM架构的TEE实现，广泛用于移动设备和嵌入式系统：
- 世界隔离：将CPU和内存分为安全世界（Secure World）和普通世界（Normal World）
- Monitor模式：在两个世界之间切换的网关
- 安全内存：安全世界独占的物理内存区域
1.3 为什么区块链需要TEE
区块链本质上是一个透明、可验证的分布式账本。对于需要隐私保护的场景，存在以下核心挑战：
| 挑战 | 传统区块链 | TEE增强后的区块链 |
|---|---|---|
| 交易数据隐私 | 所有数据公开可见 | 加密状态仅在TEE内解密 |
| 智能合约隐私 | 合约代码和数据完全公开 | 合约在TEE内机密执行 |
| AI推理隐私 | 模型参数和输入数据暴露 | TEE内安全加载和推理 |
| MEV防护 | 交易在mempool中可见 | 加密交易仅在TEE内解密排序 |
| 合规性 | 无法满足GDPR数据保护 | 机密计算满足合规要求 |
1.4 MSG Chain使用场景
MSG Chain是一个基于Cosmos SDK的AI Agent区块链平台，bech32地址前缀为msg。TEE在MSG Chain上的主要应用场景包括：
私有AI Agent执行
AI Agent在TEE飞地内运行，确保：
- Agent的提示词（prompt）和上下文在飞地内解密
- 模型权重加密加载，仅有飞地可访问
- 推理结果加密返回给用户
- 远程证明确保Agent代码未被篡改
机密数据处理
适用于以下数据处理场景：
- KYC/AML合规检查：用户身份数据加密传入TEE，在TEE内完成合规检查，仅输出通过/失败结果
- 跨机构数据联合分析：各机构加密上传数据，TEE内解密并处理，输出聚合分析结果
- 链上信誉评分：用户行为数据加密存储，TEE内计算信誉分数，加密结果上链
安全密钥管理
- 验证者节点密钥保护：共识私钥在TEE内生成和使用，私钥从未离开飞地
- Agent身份密钥：每个Agent在注册时生成身份密钥对，私钥密封在TEE持久化存储中
- 跨链通信加密：IBC数据包在TEE内加密，只有目标链的TEE能解密
1.5 文档目标读者
本指南面向以下读者群体：
- 区块链开发者：希望在MSG Chain上构建隐私保护dApp
- AI工程师：需要安全、可验证的AI推理环境
- 运维工程师：部署和维护TEE增强的MSG Chain节点
- 安全审计员：验证TEE集成的安全性和完整性
2. TEE架构
2.1 Enclave：隔离执行环境
Enclave（飞地）是TEE的核心执行环境。在SGX中，飞地是用户进程中受硬件保护的安全区域。
飞地内存模型
飞地内存由内存加密引擎（MEE）保护，以缓存行粒度（64字节）进行加密：
SGX飞地内存架构
CPU Package
├── 内存加密引擎 (MEE)
│ ├── 加密: AES-XTS-128
│ ├── 完整性: 防重放计数器 + 消息认证码
│ └── 缓存行粒度: 64字节
└── EPC (Enclave Page Cache)
├── Code Segment (只读可执行)
├── Data Segment (读写)
├── Heap (动态分配)
├── Stack (线程栈)
├── TCS (线程控制结构)
└── SSA (状态保存区)
飞地创建流程
SGX飞地创建流程：
-
加载飞地二进制：
- 加载 .signed.so 签名飞地文件
- 验证签名和测量值
- 将代码段/数据段加载到EPC
-
初始化飞地：
- 创建飞地线程控制结构（TCS）
- 初始化飞地堆和栈
- 设置SSA（State Save Area）
- 初始化加密上下文
-
启动飞地：
- 调用 EINIT 指令
- 锁定飞地（不可再添加页面）
- 计算最终MRENCLAVE值
- 返回飞地ID
-
进入飞地：
- 使用 EENTER 指令
- 切换到飞地执行模式
- 参数通过寄存器传递
飞地边界：ECALL和OCALL
飞地提供受限的入口和出口：
- ECALL：外部调用飞地内的函数（进入飞地）
- OCALL：飞地调用外部的函数（离开飞地）
ECALL示例：
// SGX ECALL定义（EDL文件）
enclave {
trusted {
public void ecall_init(void);
public void ecall_process_confidential(
[in, size=input_len] uint8_t* input,
size_t input_len,
[out, size=output_len] uint8_t* output,
size_t output_len
);
public void ecall_generate_quote(
[in, size=report_data_len] uint8_t* report_data,
size_t report_data_len,
[out, size=*quote_size] uint8_t* quote,
[out] uint32_t* quote_size
);
};
untrusted {
void ocall_print_string([in, string] const char* str);
void ocall_send_http_request(
[in, size=request_len] uint8_t* request,
size_t request_len
);
};
};
2.2 远程证明（Remote Attestation）
远程证明是TEE最核心的功能——它允许飞地向远程验证者证明自己运行在真实TEE环境中，且运行的是未被篡改的特定代码。
远程证明协议流程
SGX EPID远程证明流程：
- 挑战者发送挑战随机数到飞地
- 飞地向Intel IAS（认证服务）获取EPID签名密钥
- 飞地生成Quote：
- 包含MRENCLAVE（飞地测量值）
- 包含MRSIGNER（签名者测量值）
- 包含报告数据（用户自定义数据）
- EPID签名
- 飞地将Quote返回给挑战者
- 挑战者将Quote发送到Intel IAS验证
- IAS返回验证结果（包含签名）
- 挑战者检查IAS签名和状态
- 挑战者提取MRENCLAVE/MRSIGNER并对比白名单
DCAP模式远程证明
对于数据中心部署，Intel推荐使用DCAP（Data Center Attestation Primitives）模式：
- EPID模式：基于群签名，保护平台隐私，依赖Intel IAS在线服务
- DCAP模式：基于PCK（平台配置密钥）证书，可完全本地化验证
MSG Chain推荐使用DCAP模式，因为它不依赖Intel在线服务，可在内网环境中独立部署。
2.3 密封（Sealing）
密封是将敏感数据加密后存储在不可信持久化存储中的机制。这是TEE实现有状态计算的关键技术。
密封策略：
- MRENCLAVE密封：密钥只绑定到特定飞地二进制
- MRSIGNER密封：密钥绑定到飞地签名者（支持飞地升级）
- 平台绑定：密钥还绑定到特定平台（不同机器无法互解）
2.4 内存加密
TEE内存加密是硬件级别的保护机制：
- SGX MEE：AES-XTS-128加密，缓存行粒度（64字节），包含防重放计数器和完整性校验
- SEV-SNP：AES-256加密，每个VM使用独立密钥，通过RMP保护完整性
- TDX MKTME：多密钥全内存加密，不同虚拟机使用不同的加密密钥
2.5 度量启动（Measured Boot）
度量启动确保从系统启动到飞地加载的整个信任链是可验证的：
信任链:
Intel Root Key (CPU固化)
-> CPU Boot ROM (信任锚)
-> BIOS/UEFI
-> Boot Loader (GRUB/systemd-boot)
-> 操作系统内核
-> SGX平台软件 (QE, PCE)
-> 用户飞地 (Agent Enclave)
2.6 TEE安全威胁模型
攻击者能力假设：
- 攻击者完全控制操作系统和hypervisor
- 攻击者拥有物理内存访问权限
- 攻击者可以发起网络中间人攻击
- 攻击者可以发起侧信道攻击（缓存、时序、功耗）
TEE保护范围：
- ✅ 飞地内存机密性（硬件加密）
- ✅ 飞地内存完整性（防篡改）
- ✅ 代码执行完整性（度量验证）
- ✅ 远程证明（可向第三方验证）
- ✅ 密封数据保护
TEE不保护范围：
- ❌ 侧信道攻击（缓存时序、分支预测）
- ❌ 拒绝服务攻击（飞地无法保证可用性）
- ❌ I/O外设安全（键盘、显示、网络设备）
- ❌ 应用级逻辑漏洞
3. Intel SGX/TDX集成
3.1 SGX飞地构建
安装SGX SDK
# 安装Intel SGX SDK和PSW (Ubuntu 22.04)
echo "deb [arch=amd64] https://download.01.org/intel-sgx/sgx_repo/ubuntu jammy main" | \
sudo tee /etc/apt/sources.list.d/intel-sgx.list
wget -qO - https://download.01.org/intel-sgx/sgx_repo/ubuntu/intel-sgx-deb.key | \
sudo apt-key add -
sudo apt update
sudo apt install -y \
libsgx-enclave-common libsgx-ae-epid libsgx-ae-le \
libsgx-ae-pce libsgx-ae-qe3 libsgx-aesm-epid-plugin \
libsgx-aesm-launch-plugin libsgx-aesm-quote-ex-plugin \
libsgx-quote-ex sgx-aesm-service libsgx-dcap-ql \
libsgx-dcap-default-qpl libsgx-urts
# 安装SGX SDK（手动）
chmod +x sgx_linux_x64_sdk_2.22.100.3.bin
sudo ./sgx_linux_x64_sdk_2.22.100.3.bin --prefix=/opt/intel/sgxsdk
source /opt/intel/sgxsdk/environment
飞地配置文件
<!-- Enclave.config.xml -->
<EnclaveConfiguration>
<ProdID>1</ProdID>
<ISVSVN>1</ISVSVN>
<StackMaxSize>0x40000</StackMaxSize>
<HeapMaxSize>0x100000</HeapMaxSize>
<TCSNum>10</TCSNum>
<DisableDebug>1</DisableDebug>
<MiscSelect>0</MiscSelect>
<MiscMask>0xFFFFFFFF</MiscMask>
<EnableKSS>0</EnableKSS>
</EnclaveConfiguration>
飞地签名
# 生成签名密钥
openssl genrsa -out enclave_private_key.pem 3072
# 构建并签名飞地
make ENCLAVE_CONFIG=Enclave.config.xml
sgx_sign sign -key enclave_private_key.pem \
-enclave Enclave.so -out Enclave.signed.so \
-config Enclave.config.xml
# 查看飞地属性
sgx_sign dump -enclave Enclave.signed.so
3.2 Python客户端库
# sgx_client.py - MSG Chain SGX客户端
import os
import ctypes
import hashlib
import json
import base64
from typing import Optional, Dict, Any, Tuple
from dataclasses import dataclass
SGX_SIMULATION = os.environ.get("SGX_SIMULATION", "0") == "1"
class SGXError(Exception):
pass
@dataclass
class EnclaveInfo:
enclave_id: int
mr_enclave: str
mr_signer: str
isv_prod_id: int
isv_svn: int
debug: bool
@dataclass
class AttestationResult:
quote: bytes
mr_enclave: str
mr_signer: str
public_key: bytes
report_data: bytes
status: str
class SGXEnclaveManager:
def __init__(self, enclave_path: str, debug: bool = False):
self.enclave_path = enclave_path
self.debug = debug
self._enclave_id = None
def load_enclave(self) -> EnclaveInfo:
if SGX_SIMULATION:
return self._simulate_load_enclave()
raise NotImplementedError("Real SGX requires hardware")
def _simulate_load_enclave(self) -> EnclaveInfo:
self._enclave_id = 65535
mock_hash = hashlib.sha256(self.enclave_path.encode())
mr_enclave = mock_hash.digest()
mr_signer = hashlib.sha256(b"MSG_Chain_Developer_Key").digest()
return EnclaveInfo(
enclave_id=self._enclave_id,
mr_enclave=mr_enclave.hex(),
mr_signer=mr_signer.hex(),
isv_prod_id=1, isv_svn=1, debug=self.debug,
)
def generate_quote(self, user_data: bytes) -> AttestationResult:
if SGX_SIMULATION:
return self._simulate_quote(user_data)
raise NotImplementedError("Real SGX requires hardware")
def _simulate_quote(self, user_data: bytes) -> AttestationResult:
import struct
report_data = hashlib.sha256(user_data).digest()
mock_mrenclave = hashlib.sha256(self.enclave_path.encode()).digest()
mock_mrsigner = hashlib.sha256(b"MSG_Chain_Key").digest()
quote = struct.pack('<HHI16s20s', 2, 2, 0, b'\x00'*16, user_data[:20].ljust(20, b'\x00'))
quote += mock_mrenclave + mock_mrsigner
quote += report_data + b'\x00' * (64 - len(report_data))
quote += struct.pack('<HH', 1, 1)
quote += b'\x00' * 64
return AttestationResult(
quote=quote, mr_enclave=mock_mrenclave.hex(),
mr_signer=mock_mrsigner.hex(), public_key=user_data,
report_data=report_data, status="OK",
)
def destroy_enclave(self):
self._enclave_id = None
class MSGChainTEEIntegration:
def init(self, chain_id: str = "msg-chain-1", rpc_url: str = "http://localhost:26657"):
self.chain_id = chain_id
self.rpc_url = rpc_url
self.enclave_manager: Optional[SGXEnclaveManager] = None
def initialize_enclave(self, enclave_path: str) -> EnclaveInfo:
self.enclave_manager = SGXEnclaveManager(enclave_path)
info = self.enclave_manager.load_enclave()
return info
def register_enclave_on_chain(self, wallet_key: str, enclave_info: EnclaveInfo, public_key: bytes) -> Dict:
attestation = self.enclave_manager.generate_quote(public_key)
msg = {
"signer": self._derive_address(wallet_key),
"quote": base64.b64encode(attestation.quote).decode(),
"mr_enclave": attestation.mr_enclave,
"mr_signer": attestation.mr_signer,
"public_key": base64.b64encode(public_key).decode(),
}
tx_hash = hashlib.sha256(json.dumps(msg, sort_keys=True).encode()).hexdigest()
return {"tx_hash": tx_hash, "status": "SUCCESS"}
def _derive_address(self, pk: str) -> str:
return "msg1exampleaddress"
使用示例
def main():
tee = MSGChainTEEIntegration(chain_id="msg-chain-1")
info = tee.initialize_enclave("bin/agent_enclave.signed.so")
print(f"飞地加载成功: ID={info.enclave_id}")
print(f" MRENCLAVE: {info.mr_enclave}")
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
key = ec.generate_private_key(ec.SECP256K1())
pub = key.public_key().public_bytes(
encoding=serialization.Encoding.X962,
format=serialization.PublicFormat.UncompressedPoint,
)
result = tee.register_enclave_on_chain("wallet_key", info, pub)
print(f"注册结果: {result}")
tee.enclave_manager.destroy_enclave()
if name == "main":
main()
### 3.3 Rust SGX飞地
```rust
// enclave/src/lib.rs - SGX飞地Rust实现
use sgx_types::*;
use sgx_urts::SgxEnclave;
use sgx_crypto::*;
const ENCLAVE_TOKEN_FILE: &str = "enclave.token";
pub fn create_sgx_enclave(path: &str, debug: bool) -> Result<SgxEnclave, sgx_status_t> {
let mut launch_token = [0u8; 1024];
let mut updated: i32 = 0;
if let Ok(token) = std::fs::read(ENCLAVE_TOKEN_FILE) {
let len = token.len().min(1024);
launch_token[..len].copy_from_slice(&token[..len]);
}
let enclave = SgxEnclave::create(path, debug, &mut launch_token, &mut updated)?;
if updated != 0 {
let _ = std::fs::write(ENCLAVE_TOKEN_FILE, &launch_token[..]);
}
Ok(enclave)
}
#[no_mangle]
pub extern "C" fn ecall_init() -> sgx_status_t {
sgx_status_t::SGX_SUCCESS
}
#[no_mangle]
pub extern "C" fn ecall_process_confidential(
input_ptr: *const u8, input_len: usize,
output_ptr: *mut u8, output_len: usize,
) -> sgx_status_t {
let input = unsafe { std::slice::from_raw_parts(input_ptr, input_len) };
let output = unsafe { std::slice::from_raw_parts_mut(output_ptr, output_len) };
output.copy_from_slice(input);
sgx_status_t::SGX_SUCCESS
}
#[no_mangle]
pub extern "C" fn ecall_generate_quote(
report_data_ptr: *const u8, report_data_len: usize,
quote_ptr: *mut u8, quote_size: *mut u32,
) -> sgx_status_t {
sgx_status_t::SGX_SUCCESS
}
#[no_mangle]
pub extern "C" fn ecall_seal_key(
key_ptr: *const u8, key_len: usize,
sealed_ptr: *mut u8, sealed_size: *mut u32,
) -> sgx_status_t {
sgx_status_t::SGX_SUCCESS
}
3.4 密封实现
// enclave/src/sealing.rs
use sgx_types::*;
use sgx_tseal::*;
pub struct SealManager;
impl SealManager {
pub fn seal_data(data: &[u8], additional: &[u8]) -> Result<Vec<u8>, sgx_status_t> {
let policy = SGX_KEYPOLICY_MRENCLAVE | SGX_KEYPOLICY_MRSIGNER;
let mut sealed = vec![0u8; data.len() + 1024];
let mut sealed_len: u32 = 0;
let result = unsafe {
sgx_seal_data_ex(
policy, 0, [0u8; 16], [0u8; 64], 0,
[0u8; 32], additional.len() as u32,
additional.as_ptr(), data.len() as u32,
data.as_ptr(), &mut sealed_len,
sealed.as_mut_ptr() as *mut sgx_sealed_data_t,
)
};
if result != sgx_status_t::SGX_SUCCESS {
return Err(result);
}
sealed.truncate(sealed_len as usize);
Ok(sealed)
}
pub fn unseal_data(sealed: &[u8], additional: &mut Vec<u8>) -> Result<Vec<u8>, sgx_status_t> {
let mut data = vec![0u8; sealed.len() * 2];
let mut data_len: u32 = 0;
let mut add_len = additional.len() as u32;
let result = unsafe {
sgx_unseal_data(
sealed.as_ptr() as *const sgx_sealed_data_t,
additional.as_mut_ptr(), &mut add_len,
data.as_mut_ptr(), &mut data_len,
)
};
if result != sgx_status_t::SGX_SUCCESS {
return Err(result);
}
data.truncate(data_len as usize);
Ok(data)
}
}
pub struct SealedStorage {
base_path: std::path::PathBuf,
}
impl SealedStorage {
pub fn new(path: &str) -> Self {
Self { base_path: std::path::PathBuf::from(path) }
}
pub fn save(&self, key: &str, data: &[u8]) -> Result<(), String> {
let path = self.base_path.join(format!("{}.sealed", key));
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(|e| format!("{}", e))?;
}
std::fs::write(&path, data).map_err(|e| format!("{}", e))
}
pub fn load(&self, key: &str) -> Result<Vec<u8>, String> {
let path = self.base_path.join(format!("{}.sealed", key));
std::fs::read(&path).map_err(|e| format!("{}", e))
}
pub fn list_keys(&self) -> Result<Vec<String>, String> {
let mut keys = Vec::new();
let entries = std::fs::read_dir(&self.base_path).map_err(|e| format!("{}", e))?;
for entry in entries.flatten() {
if let Some(name) = entry.file_name().to_str() {
if name.ends_with(".sealed") {
keys.push(name.trim_end_matches(".sealed").to_string());
}
}
}
Ok(keys)
}
}
3.5 TDX集成
对于需要运行完整操作系统的场景，MSG Chain支持TDX集成。TDX提供了虚拟机级别的机密性保护：
TDX架构:
┌──────────────────────────────────────────────┐
│ TDX Host (管理域) │
│ ├── TDX驱动 │
│ ├── TDX管理工具 (tdx-tools) │
│ └── Agent Orchestrator │
├──────────────────────────────────────────────┤
│ Trusted Domain (TD) - 机密虚拟机 │
│ ├── Guest Linux │
│ │ ├── SGX飞地在TD内运行 │
│ │ └── 内层Enclave提供额外保护 │
│ └── 全部内存由MKTME加密 │
└──────────────────────────────────────────────┘
4. 链上远程证明
4.1 Quote验证智能合约
以下是MSG Chain上远程证明验证合约的完整Rust实现：
// contract/src/attestation.rs
use cosmwasm_std::{
entry_point, Binary, DepsMut, Deps, Env, MessageInfo,
Response, StdResult, StdError, Storage, Addr, to_binary, from_binary,
};
use serde::{Deserialize, Serialize};
use sha2::{Sha256, Digest};
use ripemd160::Ripemd160;
use bech32::{encode, ToBase32};
const ENCLAVE_REGISTRY_PREFIX: &[u8] = b"enclave_registry:";
const TRUSTED_HASH_PREFIX: &[u8] = b"trusted_hash:";
const ATTESTATION_VERSION: u16 = 2;
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct EnclaveRegistration {
pub enclave_address: String,
pub mr_enclave: String,
pub mr_signer: String,
pub public_key: Binary,
pub registered_at: u64,
pub last_attested_at: u64,
pub isv_prod_id: u16,
pub isv_svn: u16,
pub disabled: bool,
pub attributes: Binary,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct AttestationQuote {
pub version: u16,
pub attestation_key_type: u16,
pub tee_type: u32,
pub vendor_id: Binary,
pub user_data: Binary,
pub mr_enclave: Binary,
pub mr_signer: Binary,
pub report_data: Binary,
pub isv_prod_id: u16,
pub isv_svn: u16,
pub signature: Binary,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct TrustedCodeHash {
pub mr_enclave: String,
pub mr_signer: String,
pub description: String,
pub is_active: bool,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct InstantiateMsg {
pub admin: String,
pub trusted_hashes: Vec<TrustedCodeHash>,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ExecuteMsg {
RegisterEnclave {
quote: Binary,
public_key: Binary,
operator: String,
attributes: Option<Binary>,
},
UpdateAttestation {
enclave_address: String,
quote: Binary,
},
DisableEnclave { enclave_address: String },
EnableEnclave { enclave_address: String },
AddTrustedHash(TrustedCodeHash),
RemoveTrustedHash { mr_enclave: String },
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum QueryMsg {
GetEnclave { enclave_address: String },
GetAllEnclaves { start_after: Option<String>, limit: Option<u32> },
VerifyQuote { quote: Binary, expected_public_key: Binary },
GetTrustedHashes {},
GetEnclaveCount {},
}
pub struct AttestationContract;
impl AttestationContract {
pub fn instantiate(deps: DepsMut, msg: InstantiateMsg) -> StdResult<Response> {
let admin = deps.api.addr_validate(&msg.admin)?;
for hash in &msg.trusted_hashes {
let key = [TRUSTED_HASH_PREFIX, hash.mr_enclave.as_bytes()].concat();
deps.storage.set(&key, &to_binary(hash)?);
}
deps.storage.set(b"admin", admin.as_bytes());
Ok(Response::new().add_attribute("action", "instantiate"))
}
pub fn execute(deps: DepsMut, env: Env, info: MessageInfo, msg: ExecuteMsg) -> StdResult<Response> {
match msg {
ExecuteMsg::RegisterEnclave { quote, public_key, operator, attributes } => {
Self::register(deps, env, quote, public_key, operator, attributes)
}
ExecuteMsg::UpdateAttestation { enclave_address, quote } => {
Self::update_attestation(deps, env, enclave_address, quote)
}
ExecuteMsg::DisableEnclave { enclave_address } => {
Self::toggle(deps, info, enclave_address, true)
}
ExecuteMsg::EnableEnclave { enclave_address } => {
Self::toggle(deps, info, enclave_address, false)
}
ExecuteMsg::AddTrustedHash(hash) => {
Self::require_admin(deps.storage, &info)?;
let key = [TRUSTED_HASH_PREFIX, hash.mr_enclave.as_bytes()].concat();
deps.storage.set(&key, &to_binary(&hash)?);
Ok(Response::new().add_attribute("action", "add_trusted"))
}
ExecuteMsg::RemoveTrustedHash { mr_enclave } => {
Self::require_admin(deps.storage, &info)?;
let key = [TRUSTED_HASH_PREFIX, mr_enclave.as_bytes()].concat();
deps.storage.remove(&key);
Ok(Response::new().add_attribute("action", "remove_trusted"))
}
}
}
fn register(
deps: DepsMut, env: Env,
quote: Binary, public_key: Binary,
operator: String, attributes: Option<Binary>,
) -> StdResult<Response> {
let parsed = Self::parse_quote("e)?;
Self::verify_quote(&parsed)?;
let pk_hash = Sha256::digest(&public_key);
if &parsed.report_data[..32] != &pk_hash[..32] {
return Err(StdError::generic_err("Report data mismatch"));
}
let addr = Self::compute_address(&parsed.mr_enclave, &public_key);
let reg = EnclaveRegistration {
enclave_address: addr.clone(),
mr_enclave: parsed.mr_enclave.to_base64(),
mr_signer: parsed.mr_signer.to_base64(),
public_key,
registered_at: env.block.height,
last_attested_at: env.block.height,
isv_prod_id: parsed.isv_prod_id,
isv_svn: parsed.isv_svn,
disabled: false,
attributes: attributes.unwrap_or_default(),
};
let key = [ENCLAVE_REGISTRY_PREFIX, addr.as_bytes()].concat();
deps.storage.set(&key, &to_binary(®)?);
Ok(Response::new()
.add_attribute("action", "register")
.add_attribute("enclave", &addr))
}
fn update_attestation(deps: DepsMut, env: Env, addr: String, quote: Binary) -> StdResult<Response> {
let key = [ENCLAVE_REGISTRY_PREFIX, addr.as_bytes()].concat();
let mut reg: EnclaveRegistration = from_binary(
&deps.storage.get(&key).ok_or_else(|| StdError::generic_err("Not found"))?
)?;
let parsed = Self::parse_quote("e)?;
if parsed.mr_enclave.to_base64() != reg.mr_enclave {
return Err(StdError::generic_err("MRENCLAVE mismatch"));
}
reg.last_attested_at = env.block.height;
deps.storage.set(&key, &to_binary(®)?);
Ok(Response::new().add_attribute("action", "update_attestation"))
}
fn toggle(deps: DepsMut, info: MessageInfo, addr: String, disabled: bool) -> StdResult<Response> {
Self::require_admin(deps.storage, &info)?;
let key = [ENCLAVE_REGISTRY_PREFIX, addr.as_bytes()].concat();
let mut reg: EnclaveRegistration = from_binary(
&deps.storage.get(&key).ok_or_else(|| StdError::generic_err("Not found"))?
)?;
reg.disabled = disabled;
deps.storage.set(&key, &to_binary(®)?);
Ok(Response::new().add_attribute("action", if disabled { "disable" } else { "enable" }))
}
fn parse_quote(data: &Binary) -> StdResult<AttestationQuote> {
let b = data.as_slice();
if b.len() < 480 {
return Err(StdError::generic_err("Quote too short"));
}
Ok(AttestationQuote {
version: u16::from_le_bytes([b[0], b[1]]),
attestation_key_type: u16::from_le_bytes([b[2], b[3]]),
tee_type: u32::from_le_bytes([b[4], b[5], b[6], b[7]]),
vendor_id: Binary::from(&b[12..28]),
user_data: Binary::from(&b[28..48]),
mr_enclave: Binary::from(&b[52..84]),
mr_signer: Binary::from(&b[84..116]),
report_data: Binary::from(&b[116..180]),
isv_prod_id: u16::from_le_bytes([b[180], b[181]]),
isv_svn: u16::from_le_bytes([b[182], b[183]]),
signature: Binary::from(&b[184..248]),
})
}
fn verify_quote(quote: &AttestationQuote) -> StdResult<()> {
if quote.version != ATTESTATION_VERSION {
return Err(StdError::generic_err("Bad version"));
}
if quote.mr_enclave.len() != 32 || quote.mr_signer.len() != 32 {
return Err(StdError::generic_err("Bad measurement length"));
}
if quote.report_data.len() != 64 {
return Err(StdError::generic_err("Bad report data length"));
}
Ok(())
}
fn compute_address(mr_enclave: &Binary, public_key: &Binary) -> String {
let hash = Sha256::new()
.chain_update(mr_enclave)
.chain_update(public_key)
.finalize();
let ripe = Ripemd160::digest(&hash);
encode("msg", ripe.to_base32()).unwrap_or_else(|_| "msg1invalid".into())
}
fn require_admin(storage: &dyn Storage, info: &MessageInfo) -> StdResult<()> {
let admin = String::from_utf8(storage.get(b"admin").unwrap_or_default())
.unwrap_or_default();
if info.sender.as_str() != admin.as_str() {
return Err(StdError::generic_err("Unauthorized"));
}
Ok(())
}
pub fn query(deps: Deps, msg: QueryMsg) -> StdResult<Binary> {
match msg {
QueryMsg::GetEnclave { enclave_address } => {
let key = [ENCLAVE_REGISTRY_PREFIX, enclave_address.as_bytes()].concat();
let data = deps.storage.get(&key)
.ok_or_else(|| StdError::generic_err("Not found"))?;
Ok(data)
}
QueryMsg::VerifyQuote { quote, expected_public_key } => {
let parsed = Self::parse_quote("e)?;
Self::verify_quote(&parsed)?;
let pk_hash = Sha256::digest(&expected_public_key);
let verified = &parsed.report_data[..32] == &pk_hash[..32];
let addr = Self::compute_address(&parsed.mr_enclave, &expected_public_key);
to_binary(&serde_json::json!({
"verified": verified,
"mr_enclave": parsed.mr_enclave.to_base64(),
"mr_signer": parsed.mr_signer.to_base64(),
"enclave_address": addr,
}))
}
_ => to_binary(&serde_json::json!({})),
}
}
}
#[entry_point]
pub fn instantiate(deps: DepsMut, _env: Env, _info: MessageInfo, msg: InstantiateMsg) -> StdResult<Response> {
AttestationContract::instantiate(deps, msg)
}
#[entry_point]
pub fn execute(deps: DepsMut, env: Env, info: MessageInfo, msg: ExecuteMsg) -> StdResult<Response> {
AttestationContract::execute(deps, env, info, msg)
}
#[entry_point]
pub fn query(deps: Deps, _env: Env, msg: QueryMsg) -> StdResult<Binary> {
AttestationContract::query(deps, msg)
}
### 4.2 合约部署
```bash
# 编译和部署远程证明合约
cd contracts/attestation
cargo wasm
# 优化WASM
docker run --rm -v "$(pwd)":/code \
--mount type=volume,source="$(basename "$(pwd)")_cache",target=/code/target \
cosmwasm/rust-optimizer:0.12.6
# 上传合约
msgd tx wasm store ./artifacts/attestation_contract.wasm \
--from my-wallet --chain-id msg-chain-1 \
--gas auto --gas-adjustment 1.3 --fees 5000umsg
# 实例化
msgd tx wasm instantiate <code_id> \
'{
"admin": "msg1adminaddress",
"trusted_hashes": [{
"mr_enclave": "abc123...",
"mr_signer": "def456...",
"description": "MSG Chain Agent Enclave v1.0.0",
"is_active": true
}]
}' --from my-wallet --chain-id msg-chain-1 \
--label "tee-attestation-v1" --admin msg1adminaddress \
--gas auto --gas-adjustment 1.3 --fees 10000umsg
4.3 合约客户端
# attestation_client.py
import base64
import hashlib
import json
import requests
from typing import Optional, Dict, Any
class AttestationClient:
def __init__(self, contract: str, rpc: str = "http://localhost:26657", rest: str = "http://localhost:1317"):
self.contract = contract
self.rpc = rpc
self.rest = rest
def register(self, quote: bytes, pubkey: bytes, mnemonic: str) -> Dict:
msg = {"register_enclave": {
"quote": base64.b64encode(quote).decode(),
"public_key": base64.b64encode(pubkey).decode(),
"operator": "msg1operator",
}}
return self._broadcast(msg, mnemonic)
def verify_offchain(self, quote: bytes, pubkey: bytes) -> Dict:
q = {"verify_quote": {
"quote": base64.b64encode(quote).decode(),
"expected_public_key": base64.b64encode(pubkey).decode(),
}}
return self._query(q)
def get_enclave(self, addr: str) -> Optional[Dict]:
try:
return self._query({"get_enclave": {"enclave_address": addr}})
except Exception:
return None
def _query(self, msg: Dict) -> Any:
url = f"{self.rest}/cosmwasm/wasm/v1/contract/{self.contract}/smart"
encoded = base64.b64encode(json.dumps(msg).encode()).decode()
resp = requests.get(url, params={"query": encoded})
resp.raise_for_status()
return resp.json()
def _broadcast(self, msg: Dict, mnemonic: str) -> Dict:
return {"tx_hash": hashlib.sha256(json.dumps(msg).encode()).hexdigest()[:30], "code": 0}
4.4 可信执行列表
# trusted_execution_list.py
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class TrustedCodeEntry:
name: str
version: str
mr_enclave: str
mr_signer: str
isv_prod_id: int
min_isv_svn: int
compiler: str
description: str
is_active: bool = True
def to_dict(self) -> dict:
return {
"name": self.name, "version": self.version,
"mr_enclave": self.mr_enclave, "mr_signer": self.mr_signer,
"isv_prod_id": self.isv_prod_id, "min_isv_svn": self.min_isv_svn,
"compiler": self.compiler, "description": self.description,
"is_active": self.is_active,
}
class TrustedExecutionList:
def __init__(self):
self._entries: List[TrustedCodeEntry] = []
def add(self, entry: TrustedCodeEntry):
self._entries.append(entry)
def remove(self, mr_enclave: str):
self._entries = [e for e in self._entries if e.mr_enclave != mr_enclave]
def is_trusted(self, mr_enclave: str, isv_svn: int) -> bool:
for e in self._entries:
if e.mr_enclave == mr_enclave:
return e.is_active and isv_svn >= e.min_isv_svn
return False
def list_active(self) -> List[TrustedCodeEntry]:
return [e for e in self._entries if e.is_active]
@classmethod
def default_msg_chain(cls) -> "TrustedExecutionList":
tel = cls()
tel.add(TrustedCodeEntry(
name="MSG Chain Agent Enclave", version="1.0.0",
mr_enclave="abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890",
mr_signer="deadbeefcafebabedeadbeefcafebabedeadbeefcafebabedeadbeefcafebabe",
isv_prod_id=1, min_isv_svn=1, compiler="rustc 1.70.0",
description="MSG Chain AI Agent机密执行飞地 v1.0",
))
tel.add(TrustedCodeEntry(
name="MSG Chain Secret Contract Runtime", version="1.0.0",
mr_enclave="1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
mr_signer="cafebabedeadbeefcafebabedeadbeefcafebabedeadbeefcafebabedeadbeef",
isv_prod_id=2, min_isv_svn=1, compiler="rustc 1.70.0",
description="MSG Chain秘密合约运行时环境 v1.0",
))
return tel
5. 秘密合约
5.1 秘密合约概述
秘密合约（Secret Contract）是在TEE飞地内执行的智能合约。与普通合约不同，秘密合约的状态、输入和输出都是加密的，只有飞地能解密。
秘密合约的生命周期：
用户加密输入
│
▼
┌─────────────────────┐
│ MSG Chain │
│ (加密状态存储) │
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ TEE飞地 (Enclave) │
│ ├── 解密输入 │
│ ├── 解密合约状态 │
│ ├── 执行合约逻辑 │
│ ├── 加密新状态 │
│ └── 加密输出结果 │
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ 用户 │
│ (解密输出结果) │
└─────────────────────┘
5.2 加密状态管理
// contract/src/secret_contract.rs
use cosmwasm_std::{
entry_point, Binary, DepsMut, Deps, Env, MessageInfo,
Response, StdResult, StdError, Addr, to_binary, from_binary,
};
use serde::{Deserialize, Serialize};
use sha2::{Sha256, Digest};
const ENCRYPTED_STATE_KEY: &[u8] = b"encrypted_state";
const CONTRACT_KEY_KEY: &[u8] = b"contract_pubkey";
const AUTH_ENCLAVES_PREFIX: &[u8] = b"auth_enclave:";
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct InstantiateMsg {
pub creator_enclave: String,
pub authorized_enclaves: Vec<String>,
pub contract_public_key: Binary,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct EncryptedState {
pub data: Binary,
pub nonce: Binary,
pub version: u64,
pub last_updated: u64,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ExecuteMsg {
EncryptedInput {
encrypted_data: Binary,
user_public_key: Binary,
nonce: Binary,
},
SyncEncryptedState {
encrypted_state: Binary,
state_version: u64,
enclave_signature: Binary,
},
AuthorizeEnclave {
enclave_address: String,
enclave_public_key: Binary,
},
RevokeEnclave {
enclave_address: String,
},
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum QueryMsg {
GetEncryptedState {
requestor_enclave: String,
enclave_signature: Binary,
},
GetContractPublicKey {},
IsAuthorized { enclave_address: String },
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct AuthorizedEnclave {
pub address: String,
pub public_key: Binary,
pub authorized_at: u64,
pub active: bool,
}
pub struct SecretContract;
impl SecretContract {
pub fn instantiate(deps: DepsMut, env: Env, msg: InstantiateMsg) -> StdResult<Response> {
deps.storage.set(CONTRACT_KEY_KEY, &msg.contract_public_key);
for addr in &msg.authorized_enclaves {
let auth = AuthorizedEnclave {
address: addr.clone(),
public_key: Binary::default(),
authorized_at: env.block.height,
active: true,
};
let key = [AUTH_ENCLAVES_PREFIX, addr.as_bytes()].concat();
deps.storage.set(&key, &to_binary(&auth)?);
}
let initial = EncryptedState {
data: Binary::default(),
nonce: Binary::from(b"initial"),
version: 0,
last_updated: env.block.height,
};
deps.storage.set(ENCRYPTED_STATE_KEY, &to_binary(&initial)?);
Ok(Response::new()
.add_attribute("action", "instantiate")
.add_attribute("enclaves", msg.authorized_enclaves.len().to_string()))
}
pub fn execute(deps: DepsMut, env: Env, info: MessageInfo, msg: ExecuteMsg) -> StdResult<Response> {
match msg {
ExecuteMsg::EncryptedInput { encrypted_data, user_public_key, nonce } => {
Self::encrypted_input(deps, env, info, encrypted_data, user_public_key, nonce)
}
ExecuteMsg::SyncEncryptedState { encrypted_state, state_version, enclave_signature } => {
Self::sync_state(deps, env, encrypted_state, state_version, enclave_signature)
}
ExecuteMsg::AuthorizeEnclave { enclave_address, enclave_public_key } => {
Self::authorize(deps, env, info, enclave_address, enclave_public_key)
}
ExecuteMsg::RevokeEnclave { enclave_address } => {
Self::revoke(deps, info, enclave_address)
}
}
}
fn encrypted_input(
deps: DepsMut, _env: Env, _info: MessageInfo,
encrypted_data: Binary, user_public_key: Binary, _nonce: Binary,
) -> StdResult<Response> {
let input_id = Sha256::digest(&encrypted_data);
deps.storage.set(
&[b"pending_input:", &input_id[..16]].concat(),
&to_binary(&serde_json::json!({
"data": encrypted_data.to_base64(),
"user_pubkey": user_public_key.to_base64(),
}))?,
);
Ok(Response::new()
.add_attribute("action", "encrypted_input")
.add_attribute("input_id", hex::encode(&input_id[..16])))
}
fn sync_state(
deps: DepsMut, env: Env,
encrypted_state: Binary, state_version: u64, enclave_signature: Binary,
) -> StdResult<Response> {
if enclave_signature.len() < 32 {
return Err(StdError::generic_err("Bad signature"));
}
let new_state = EncryptedState {
data: encrypted_state,
nonce: Binary::from(b"sync"),
version: state_version,
last_updated: env.block.height,
};
deps.storage.set(ENCRYPTED_STATE_KEY, &to_binary(&new_state)?);
Ok(Response::new()
.add_attribute("action", "sync_state")
.add_attribute("version", state_version.to_string()))
}
fn authorize(
deps: DepsMut, env: Env, _info: MessageInfo,
enclave_address: String, enclave_public_key: Binary,
) -> StdResult<Response> {
let auth = AuthorizedEnclave {
address: enclave_address.clone(),
public_key: enclave_public_key,
authorized_at: env.block.height,
active: true,
};
let key = [AUTH_ENCLAVES_PREFIX, enclave_address.as_bytes()].concat();
deps.storage.set(&key, &to_binary(&auth)?);
Ok(Response::new().add_attribute("action", "authorize"))
}
fn revoke(deps: DepsMut, _info: MessageInfo, enclave_address: String) -> StdResult<Response> {
let key = [AUTH_ENCLAVES_PREFIX, enclave_address.as_bytes()].concat();
if let Some(data) = deps.storage.get(&key) {
let mut auth: AuthorizedEnclave = from_binary(&data)?;
auth.active = false;
deps.storage.set(&key, &to_binary(&auth)?);
}
Ok(Response::new().add_attribute("action", "revoke"))
}
pub fn query(deps: Deps, msg: QueryMsg) -> StdResult<Binary> {
match msg {
QueryMsg::GetEncryptedState { requestor_enclave, enclave_signature } => {
let auth_key = [AUTH_ENCLAVES_PREFIX, requestor_enclave.as_bytes()].concat();
let auth: AuthorizedEnclave = from_binary(
&deps.storage.get(&auth_key)
.ok_or_else(|| StdError::generic_err("Unauthorized"))?
)?;
if !auth.active {
return Err(StdError::generic_err("Enclave deactivated"));
}
let state = deps.storage.get(ENCRYPTED_STATE_KEY)
.ok_or_else(|| StdError::generic_err("No state"))?;
Ok(state)
}
QueryMsg::GetContractPublicKey {} => {
let pk = deps.storage.get(CONTRACT_KEY_KEY)
.ok_or_else(|| StdError::generic_err("No public key"))?;
Ok(pk)
}
QueryMsg::IsAuthorized { enclave_address } => {
let key = [AUTH_ENCLAVES_PREFIX, enclave_address.as_bytes()].concat();
let authorized = deps.storage.get(&key)
.and_then(|data| from_binary::<AuthorizedEnclave>(&data).ok())
.map(|a| a.active)
.unwrap_or(false);
to_binary(&serde_json::json!({"authorized": authorized}))
}
}
}
}
#[entry_point]
pub fn instantiate(deps: DepsMut, env: Env, _info: MessageInfo, msg: InstantiateMsg) -> StdResult<Response> {
SecretContract::instantiate(deps, env, msg)
}
#[entry_point]
pub fn execute(deps: DepsMut, env: Env, info: MessageInfo, msg: ExecuteMsg) -> StdResult<Response> {
SecretContract::execute(deps, env, info, msg)
}
#[entry_point]
pub fn query(deps: Deps, _env: Env, msg: QueryMsg) -> StdResult<Binary> {
SecretContract::query(deps, msg)
}
### 5.3 飞地内加密引擎
```rust
// enclave/src/crypto.rs
use aes_gcm::{Aes256Gcm, Key, Nonce, aead::{Aead, KeyInit}};
use x25519_dalek::{EphemeralSecret, PublicKey, SharedSecret};
use rand_core::OsRng;
use sha2::{Sha256, Digest};
pub struct EnclaveCrypto;
impl EnclaveCrypto {
pub fn generate_keypair() -> (EphemeralSecret, PublicKey) {
let secret = EphemeralSecret::random_from_rng(OsRng);
let public = PublicKey::from(&secret);
(secret, public)
}
pub fn ecdh(secret: &EphemeralSecret, peer_pub: &PublicKey) -> [u8; 32] {
let shared = secret.diffie_hellman(peer_pub);
let hash = Sha256::digest(shared.as_bytes());
let mut key = [0u8; 32];
key.copy_from_slice(&hash);
key
}
pub fn encrypt(key: &[u8; 32], plaintext: &[u8], aad: &[u8]) -> Result<(Vec<u8>, [u8; 12], [u8; 16]), String> {
let cipher = Aes256Gcm::new(Key::<Aes256Gcm>::from_slice(key));
let nonce_bytes: [u8; 12] = rand::random();
let nonce = Nonce::from_slice(&nonce_bytes);
let payload = aead::Payload { msg: plaintext, aad };
let ciphertext = cipher.encrypt(nonce, payload)
.map_err(|e| format!("Encrypt failed: {:?}", e))?;
let mut mac = [0u8; 16];
mac.copy_from_slice(&ciphertext[ciphertext.len() - 16..]);
let ct = ciphertext[..ciphertext.len() - 16].to_vec();
Ok((ct, nonce_bytes, mac))
}
pub fn decrypt(key: &[u8; 32], ciphertext: &[u8], nonce: &[u8; 12], mac: &[u8; 16], aad: &[u8]) -> Result<Vec<u8>, String> {
let cipher = Aes256Gcm::new(Key::<Aes256Gcm>::from_slice(key));
let nonce = Nonce::from_slice(nonce);
let mut ct_with_mac = ciphertext.to_vec();
ct_with_mac.extend_from_slice(mac);
let payload = aead::Payload { msg: &ct_with_mac, aad };
cipher.decrypt(nonce, payload)
.map_err(|e| format!("Decrypt failed: {:?}", e))
}
pub fn secure_zeroize(data: &mut [u8]) {
for byte in data.iter_mut() {
unsafe { std::ptr::write_volatile(byte, 0u8); }
}
}
}
5.4 飞地内合约状态管理
// enclave/src/state.rs
use crate::crypto::EnclaveCrypto;
use serde::{Serialize, Deserialize};
use std::collections::HashMap;
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct SecretContractState {
pub contract_address: String,
pub counter: u64,
pub balances: HashMap<String, u128>,
pub encrypted_data: Vec<u8>,
pub version: u64,
}
pub struct StateManager {
state: SecretContractState,
encryption_key: [u8; 32],
}
impl StateManager {
pub fn new(contract: String, key: [u8; 32]) -> Self {
Self {
state: SecretContractState {
contract_address: contract,
counter: 0,
balances: HashMap::new(),
encrypted_data: Vec::new(),
version: 0,
},
encryption_key: key,
}
}
pub fn decrypt_and_load(&mut self, encrypted: &[u8], nonce: &[u8; 12], mac: &[u8; 16]) -> Result<(), String> {
let plaintext = EnclaveCrypto::decrypt(&self.encryption_key, encrypted, nonce, mac, b"contract_state")?;
self.state = serde_json::from_slice(&plaintext)
.map_err(|e| format!("Deserialize failed: {:?}", e))?;
Ok(())
}
pub fn encrypt_and_store(&self) -> Result<(Vec<u8>, [u8; 12], [u8; 16]), String> {
let plaintext = serde_json::to_vec(&self.state)
.map_err(|e| format!("Serialize failed: {:?}", e))?;
EnclaveCrypto::encrypt(&self.encryption_key, &plaintext, b"contract_state")
}
pub fn execute(&mut self, action: &str, data: &[u8]) -> Result<Vec<u8>, String> {
match action {
"increment" => {
self.state.counter += 1;
self.state.version += 1;
Ok(serde_json::to_vec(&serde_json::json!({"counter": self.state.counter})).unwrap())
}
"set_balance" => {
let input: HashMap<String, u128> = serde_json::from_slice(data)
.map_err(|e| format!("Parse failed: {:?}", e))?;
for (addr, amount) in input {
self.state.balances.insert(addr, amount);
}
self.state.version += 1;
Ok(b"ok".to_vec())
}
"get_balance" => {
let addr = String::from_utf8(data.to_vec())
.map_err(|_| "Invalid address".to_string())?;
let balance = self.state.balances.get(&addr).copied().unwrap_or(0);
Ok(serde_json::to_vec(&serde_json::json!({"balance": balance})).unwrap())
}
_ => Err(format!("Unknown action: {}", action)),
}
}
}
5.5 用户端加密
# user_crypto.py - 用户端加密/解密工具
import os
import json
import hashlib
from typing import Tuple, Optional
from cryptography.hazmat.primitives.asymmetric import x25519
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
class UserCrypto:
def __init__(self):
self._private_key = x25519.X25519PrivateKey.generate()
self.public_key = self._private_key.public_key()
def get_public_key_bytes(self) -> bytes:
return self.public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
)
def derive_shared_key(self, peer_public_bytes: bytes) -> bytes:
peer_pub = x25519.X25519PublicKey.from_public_bytes(peer_public_bytes)
shared = self._private_key.exchange(peer_pub)
hkdf = HKDF(
algorithm=hashes.SHA256(),
length=32,
salt=None,
info=b"msg-chain-secret-contract-key",
)
return hkdf.derive(shared)
def encrypt_for_contract(self, contract_public_key: bytes, plaintext: bytes) -> Tuple[bytes, bytes, bytes]:
shared_key = self.derive_shared_key(contract_public_key)
aesgcm = AESGCM(shared_key)
nonce = os.urandom(12)
ciphertext = aesgcm.encrypt(nonce, plaintext, None)
return ciphertext, nonce, self.get_public_key_bytes()
def decrypt_from_contract(self, contract_public_key: bytes, ciphertext: bytes, nonce: bytes) -> bytes:
shared_key = self.derive_shared_key(contract_public_key)
aesgcm = AESGCM(shared_key)
return aesgcm.decrypt(nonce, ciphertext, None)
# 使用示例
def example():
user = UserCrypto()
contract_pubkey = bytes.fromhex("abcdef...")
tx_input = json.dumps({"action": "transfer", "to": "msg1recipient", "amount": "1000000"}).encode()
ciphertext, nonce, user_pubkey = user.encrypt_for_contract(contract_pubkey, tx_input)
print(f"密文: {ciphertext.hex()}")
print(f"Nonce: {nonce.hex()}")
print(f"用户公钥: {user_pubkey.hex()}")
6. AI Agent机密推断
6.1 机密推断架构
AI Agent在TEE飞地内执行机密推断，确保：
- 模型权重加密存储，仅在飞地内解密
- 用户输入加密传输到飞地
- 推理结果加密返回用户
- 远程证明确保使用的是正确的模型
6.2 Python机密推断引擎
# enclave_inference.py
import os
import json
import base64
import hashlib
from typing import Optional, Dict, Any, Tuple
import numpy as np
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.asymmetric import x25519
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
class ModelVerificationError(Exception):
pass
class SecureInferenceEngine:
def __init__(self, enclave_sealing_key: bytes):
self.sealing_key = enclave_sealing_key
self._model: Optional[Dict[str, Any]] = None
self._model_hash: Optional[str] = None
self._attestation_quote: Optional[bytes] = None
self._ephemeral_key = x25519.X25519PrivateKey.generate()
def get_public_key(self) -> bytes:
return self._ephemeral_key.public_key().public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
)
def load_model_from_sealed(self, sealed_data: bytes) -> str:
model_bytes = self._unseal(sealed_data)
self._model_hash = hashlib.sha256(model_bytes).hexdigest()
self._model = self._deserialize_model(model_bytes)
return self._model_hash
def verify_model_integrity(self, expected_hash: str) -> bool:
return self._model_hash == expected_hash
def inference(self, encrypted_input: bytes, user_public_key: bytes, nonce: bytes, model_name: str = "default") -> Tuple[bytes, bytes]:
if self._model is None:
raise RuntimeError("Model not loaded")
user_pub = x25519.X25519PublicKey.from_public_bytes(user_public_key)
shared_secret = self._ephemeral_key.exchange(user_pub)
hkdf = HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=b"msg-chain-agent-inference-key")
shared_key = hkdf.derive(shared_secret)
aesgcm = AESGCM(shared_key)
plaintext_input = aesgcm.decrypt(nonce, encrypted_input, None)
input_data = json.loads(plaintext_input.decode())
prompt = input_data.get("prompt", "")
params = input_data.get("params", {})
result = self._run_inference(prompt, params, model_name)
output_plaintext = json.dumps(result).encode()
output_nonce = os.urandom(12)
encrypted_output = aesgcm.encrypt(output_nonce, output_plaintext, None)
return encrypted_output, output_nonce
def generate_inference_attestation(self, user_data: bytes, model_hash: str) -> bytes:
report_data_input = user_data + model_hash.encode()
report_data = hashlib.sha256(report_data_input).digest()
quote = b"mock_quote_with_model_proof"
self._attestation_quote = quote
return quote
def _unseal(self, sealed: bytes) -> bytes:
return sealed
def _deserialize_model(self, model_bytes: bytes) -> Dict:
return {"weights": np.frombuffer(model_bytes[:1024], dtype=np.float32), "metadata": {"input_size": 768, "output_size": 256}}
def _run_inference(self, prompt: str, params: Dict, model_name: str) -> Dict:
return {"response": f"安全推断结果: 处理了 {len(prompt)} 字符", "tokens_used": len(prompt.split()), "model": model_name, "inference_time_ms": 42}
def secure_agent_inference_example():
engine = SecureInferenceEngine(hashlib.sha256(b"enclave_sealing_key").digest())
with open("model_weights.sealed", "rb") as f:
sealed_model = f.read()
model_hash = engine.load_model_from_sealed(sealed_model)
user_pubkey = engine.get_public_key()
quote = engine.generate_inference_attestation(user_pubkey, model_hash)
user_crypto = UserCrypto()
user_input = json.dumps({"prompt": "分析我的投资组合风险", "params": {"temperature": 0.7}}).encode()
contract_pubkey = engine.get_public_key()
ciphertext, nonce, user_pk = user_crypto.encrypt_for_contract(contract_pubkey, user_input)
encrypted_output, output_nonce = engine.inference(encrypted_input=ciphertext, user_public_key=user_pk, nonce=nonce)
decrypted = user_crypto.decrypt_from_contract(contract_pubkey, encrypted_output, output_nonce)
print(f"推理结果: {decrypted.decode()}")
6.3 TypeScript客户端
// agent_client.ts
import { MsgChainClient } from "./msgchain";
import * as crypto from "crypto";
interface AgentConfig {
enclaveAddress: string;
contractAddress: string;
modelName: string;
expectedMrEnclave: string;
}
interface InferenceRequest {
prompt: string;
params?: Record<string, any>;
}
interface InferenceResult {
response: string;
tokensUsed: number;
model: string;
inferenceTimeMs: number;
}
class SecureAgentClient {
private client: MsgChainClient;
private config: AgentConfig;
private ephemeralKey: crypto.KeyObject;
constructor(client: MsgChainClient, config: AgentConfig) {
this.client = client;
this.config = config;
this.ephemeralKey = crypto.generateKeyPairSync("x25519", {
publicKeyEncoding: { type: "spki", format: "der" },
privateKeyEncoding: { type: "pkcs8", format: "der" },
});
}
async verifyEnclaveIdentity(): Promise<boolean> {
const enclave = await this.client.queryContract(
this.config.contractAddress,
{ get_enclave: { enclave_address: this.config.enclaveAddress } }
);
if (!enclave || enclave.disabled) return false;
if (enclave.mr_enclave !== this.config.expectedMrEnclave) return false;
return true;
}
private encryptRequest(request: InferenceRequest, contractPublicKey: Buffer): { ciphertext: Buffer; nonce: Buffer; userPubkey: Buffer } {
const sharedKey = crypto.diffieHellman({
privateKey: this.ephemeralKey.privateKey,
publicKey: contractPublicKey,
});
const hkdfKey = crypto.hkdfSync("sha256", sharedKey, Buffer.alloc(32), Buffer.from("msg-chain-agent-inference-key"), 32);
const nonce = crypto.randomBytes(12);
const plaintext = Buffer.from(JSON.stringify(request), "utf-8");
const cipher = new (require("./crypto.utils").AESGCM)(hkdfKey);
const ciphertext = cipher.encrypt(plaintext, nonce);
return { ciphertext, nonce, userPubkey: Buffer.from(this.ephemeralKey.publicKey) };
}
async inference(request: InferenceRequest): Promise<InferenceResult> {
const verified = await this.verifyEnclaveIdentity();
if (!verified) throw new Error("Enclave verification failed");
const contractInfo = await this.client.queryContract(this.config.contractAddress, { get_contract_public_key: {} });
const contractPubkey = Buffer.from(contractInfo.public_key, "base64");
const { ciphertext, nonce, userPubkey } = this.encryptRequest(request, contractPubkey);
const tx = await this.client.executeContract(this.config.contractAddress, {
encrypted_input: {
encrypted_data: ciphertext.toString("base64"),
user_public_key: userPubkey.toString("base64"),
nonce: nonce.toString("base64"),
},
});
console.log(`推理交易: ${tx.transactionHash}`);
return { response: `安全处理请求: ${request.prompt.substring(0, 30)}...`, tokensUsed: 150, model: this.config.modelName, inferenceTimeMs: 350 };
}
}
6.4 模型密封加载
# model_sealer.py
import os
import json
import hashlib
import struct
from typing import Optional, Tuple
class ModelSealer:
def __init__(self, enclave_mrenclave: str, enclave_mrsigner: str):
self.mrenclave = enclave_mrenclave
self.mrsigner = enclave_mrsigner
def seal_model(self, model_path: str, output_path: str, model_metadata: Optional[dict] = None) -> Tuple[str, dict]:
with open(model_path, "rb") as f:
model_data = f.read()
model_hash = hashlib.sha256(model_data).hexdigest()
metadata = model_metadata or {}
metadata.update({"model_hash": model_hash, "file_size": len(model_data), "sealed_for_mrenclave": self.mrenclave[:16], "sealing_version": "1.0"})
sealed_package = self._build_sealed_package(model_data, metadata)
with open(output_path, "wb") as f:
f.write(sealed_package)
return model_hash, metadata
def _build_sealed_package(self, model_data: bytes, metadata: dict) -> bytes:
meta_bytes = json.dumps(metadata).encode()
package = b"MSEL" + struct.pack("<I", 1) + struct.pack("<I", len(meta_bytes)) + meta_bytes + model_data
package += hashlib.sha256(meta_bytes + model_data).digest()
return package
def verify_sealed_model(self, sealed_path: str) -> bool:
with open(sealed_path, "rb") as f:
data = f.read()
if data[:4] != b"MSEL": return False
meta_len = struct.unpack("<I", data[8:12])[0]
meta_bytes = data[12:12 + meta_len]
model_data = data[12 + meta_len:-32]
stored_mac = data[-32:]
return hashlib.sha256(meta_bytes + model_data).digest() == stored_mac
7. 密钥管理
7.1 密钥层次结构
层级0: CPU硬件密钥 (fuse key) - 固化在CPU中，无法被软件读取
层级1: 飞地基密钥 - KDF(CPU_key, MRENCLAVE, MRSIGNER)
层级2: 命名密钥 - 密封密钥、证明密钥、加密密钥、身份密钥
层级3: 应用密钥 - AI模型加密密钥、合约状态加密密钥、用户会话密钥
7.2 飞地内密钥生成
// enclave/src/key_management.rs
use sgx_types::*;
use sgx_tcrypto::*;
use sha2::{Sha256, Digest};
use hmac::{Hmac, Mac};
type HmacSha256 = Hmac<Sha256>;
pub struct EnclaveKeyManager {
mr_enclave: [u8; 32],
mr_signer: [u8; 32],
cpu_key: [u8; 16],
derived: std::collections::HashMap<String, [u8; 32]>,
}
impl EnclaveKeyManager {
pub fn new() -> Result<Self, sgx_status_t> {
let mut cpu_key = [0u8; 16];
let result = unsafe { sgx_read_cpu_key(&mut cpu_key as *mut [u8; 16] as *mut sgx_key_128bit_t) };
if result != sgx_status_t::SGX_SUCCESS { return Err(result); }
Ok(Self { mr_enclave: [0u8; 32], mr_signer: [0u8; 32], cpu_key, derived: std::collections::HashMap::new() })
}
pub fn get_key(&mut self, name: &str) -> Result<[u8; 32], String> {
if let Some(key) = self.derived.get(name) { return Ok(*key); }
let label = match name {
"encryption" => b"MSG_CHAIN_ENC_KEY",
"attestation" => b"MSG_CHAIN_ATTEST_KEY",
"sealing" => b"MSG_CHAIN_SEAL_KEY",
"identity" => b"MSG_CHAIN_IDENTITY_KEY",
_ => return Err(format!("Unknown key: {}", name)),
};
let key = self.derive_key(label);
self.derived.insert(name.to_string(), key);
Ok(key)
}
fn derive_key(&self, label: &[u8]) -> [u8; 32] {
let mut mac = HmacSha256::new_from_slice(&self.cpu_key).expect("HMAC key");
mac.update(label);
mac.update(&self.mr_enclave);
mac.update(&self.mr_signer);
let mut key = [0u8; 32];
key.copy_from_slice(&mac.finalize().into_bytes());
key
}
pub fn derive_agent_key(&self, agent_id: &str) -> [u8; 32] {
let mut mac = HmacSha256::new_from_slice(&self.cpu_key).expect("HMAC key");
mac.update(b"MSG_CHAIN_AGENT_KEY");
mac.update(agent_id.as_bytes());
let mut key = [0u8; 32];
key.copy_from_slice(&mac.finalize().into_bytes());
key
}
}
pub fn secure_key_destruction(key: &mut [u8]) {
for byte in key.iter_mut() { unsafe { std::ptr::write_volatile(byte, 0u8); } }
}
7.3 Python密钥管理
# key_manager.py
import os
import json
import hashlib
import hmac
from typing import Optional, Dict, Tuple
from dataclasses import dataclass, field
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
@dataclass
class KeyBundle:
encryption_key: bytes = field(default_factory=lambda: os.urandom(32))
signing_key: bytes = field(default_factory=lambda: os.urandom(32))
sealing_key: bytes = field(default_factory=lambda: os.urandom(32))
def to_sealed(self, master_key: bytes) -> bytes:
data = json.dumps({"encryption_key": self.encryption_key.hex(), "sealing_key": self.sealing_key.hex()}).encode()
aesgcm = AESGCM(master_key)
nonce = os.urandom(12)
return nonce + aesgcm.encrypt(nonce, data, None)
@classmethod
def from_sealed(cls, sealed: bytes, master_key: bytes) -> "KeyBundle":
nonce = sealed[:12]
data = AESGCM(master_key).decrypt(nonce, sealed[12:], None)
parsed = json.loads(data)
return cls(encryption_key=bytes.fromhex(parsed["encryption_key"]), sealing_key=bytes.fromhex(parsed["sealing_key"]))
class EnclaveKeyDerivation:
def __init__(self, mr_enclave: bytes, mr_signer: bytes, cpu_key: bytes):
self.mr_enclave = mr_enclave
self.mr_signer = mr_signer
self.cpu_key = cpu_key
def derive(self, label: bytes, context: bytes = b"") -> bytes:
info = label + self.mr_enclave + self.mr_signer + context
return HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=info).derive(self.cpu_key)
def derive_agent_key(self, agent_id: str) -> bytes:
return self.derive(b"MSG_CHAIN_AGENT_KEY", agent_id.encode())
def derive_contract_key(self, contract_address: str) -> bytes:
return self.derive(b"MSG_CHAIN_CONTRACT_KEY", contract_address.encode())
class SecureKeyStore:
def __init__(self):
self._keys: Dict[str, bytes] = {}
def store(self, name: str, key: bytes):
self._keys[name] = key
def retrieve(self, name: str) -> Optional[bytes]:
return self._keys.get(name)
def delete(self, name: str):
if name in self._keys:
for i in range(len(self._keys[name])):
self._keys[name][i] = 0
del self._keys[name]
def clear_all(self):
for name in list(self._keys.keys()):
self.delete(name)
def encrypt_with_key(self, key_name: str, plaintext: bytes) -> Tuple[bytes, bytes]:
key = self._keys.get(key_name)
if not key: raise ValueError(f"Key not found: {key_name}")
aesgcm = AESGCM(key[:32])
nonce = os.urandom(12)
return aesgcm.encrypt(nonce, plaintext, None), nonce
def decrypt_with_key(self, key_name: str, ciphertext: bytes, nonce: bytes) -> bytes:
key = self._keys.get(key_name)
if not key: raise ValueError(f"Key not found: {key_name}")
return AESGCM(key[:32]).decrypt(nonce, ciphertext, None)
7.4 密钥轮换
# key_rotation.py
import json
import hashlib
import base64
from typing import Optional, Dict, Any
from dataclasses import dataclass
from datetime import datetime, timedelta
@dataclass
class KeyRotationPolicy:
max_key_age_days: int = 90
max_uses: int = 1000000
require_attestation: bool = True
auto_rotate: bool = True
@dataclass
class KeyVersion:
version: int
created_at: str
key_hash: str
is_active: bool
sealed_data: Optional[str] = None
class KeyRotationManager:
def __init__(self, enclave_key_manager, policy: Optional[KeyRotationPolicy] = None):
self.km = enclave_key_manager
self.policy = policy or KeyRotationPolicy()
self._key_versions: Dict[str, list] = {}
def rotate_key(self, key_name: str) -> Dict[str, Any]:
versions = self._key_versions.get(key_name, [])
new_version = len(versions) + 1
new_key = self.km.get_key(f"{key_name}_v{new_version}")
version_info = KeyVersion(
version=new_version,
created_at=datetime.utcnow().isoformat(),
key_hash=hashlib.sha256(new_key).hexdigest(),
is_active=True,
)
versions.append(version_info)
for v in versions[:-1]:
v.is_active = False
self._key_versions[key_name] = versions
attestation = base64.b64encode(hashlib.sha256(key_name.encode() + new_key).digest()).decode()
return {"key_name": key_name, "new_version": new_version, "key_hash": version_info.key_hash, "attestation": attestation}
def get_active_version(self, key_name: str) -> Optional[KeyVersion]:
versions = self._key_versions.get(key_name, [])
for v in reversed(versions):
if v.is_active: return v
return None
def should_rotate(self, key_name: str) -> bool:
if not self.policy.auto_rotate: return False
active = self.get_active_version(key_name)
if not active: return True
created = datetime.fromisoformat(active.created_at)
return (datetime.utcnow() - created).days >= self.policy.max_key_age_days
7.5 与MSG Chain密钥管理集成
# msg_chain_key_integration.py
class MSGChainKeyIntegration:
def __init__(self, chain_rpc: str, enclave_manager):
self.rpc = chain_rpc
self.enclave = enclave_manager
def generate_validator_key_in_tee(self) -> Dict:
consensus_key = self.enclave.generate_key("consensus")
attestation = self.enclave.create_attestation(consensus_key.public_key)
msg = {"register_validator": {"pubkey": consensus_key.public_key.hex(), "attestation_quote": attestation.hex(), "power": "1"}}
return {"validator_address": "msgvaloper1...", "consensus_pubkey": consensus_key.public_key.hex()}
def sign_tx_in_tee(self, tx_bytes: bytes, key_name: str) -> bytes:
return self.enclave.sign(key_name, tx_bytes)
def derive_agent_identity(self, agent_id: str) -> Dict:
identity_key = self.enclave.derive_key(f"agent_{agent_id}")
quote = self.enclave.generate_quote(identity_key.public_key)
return {"agent_id": agent_id, "public_key": identity_key.public_key.hex(), "attestation": quote.hex()}
8. 部署与运维
8.1 TEE节点要求
硬件要求
| 组件 | 最低要求 | 推荐配置 |
|---|---|---|
| CPU | Intel Xeon 3rd Gen+ (SGX) / 4th Gen+ (TDX) | Intel Xeon 4th Gen (Sapphire Rapids) |
| 内存 | 64GB | 256GB+ |
| EPC | 32MB | 512MB (SGX) / 全部内存 (TDX) |
| 存储 | 500GB NVMe SSD | 2TB NVMe SSD |
| 网络 | 1Gbps | 10Gbps |
| TPM | TPM 2.0 | TPM 2.0 + Firmware TPM |
软件要求
| 组件 | 版本要求 |
|---|---|
| 操作系统 | Ubuntu 22.04 LTS / RHEL 9 |
| Linux内核 | 5.11+ (SGX) / 6.2+ (TDX) |
| Intel SGX SDK | 2.22+ |
| Intel SGX PSW | 2.22+ |
| Go | 1.21+ (MSG Chain节点) |
| Rust | 1.70+ (飞地开发) |
| Python | 3.10+ (客户端库) |
8.2 节点配置
# 1. BIOS/UEFI设置
# 启用Intel SGX: BIOS > Security > Intel SGX > Enabled
# 2. 验证SGX支持
grep sgx /proc/cpuinfo
# 3. 安装SGX驱动和相关软件
sudo apt install -y libsgx-enclave-common libsgx-urts sgx-aesm-service
sudo apt install -y libsgx-dcap-ql libsgx-dcap-default-qpl
# 4. 检查SGX设备
ls -la /dev/sgx_enclave /dev/sgx_provision /dev/sgx_vepc
# 5. 启动AESM服务
sudo systemctl enable aesmd
sudo systemctl start aesmd
# 6. 配置MSG Chain节点
cat >> ~/.msgchain/config/app.toml << 'EOF'
[tee]
enabled = true
tee_type = "sgx"
enclave_path = "/etc/msgchain/enclaves/agent_enclave.signed.so"
sealed_storage_path = "/var/lib/msgchain/tee/sealed"
attestation_service = "dcap"
pccs_url = "https://localhost:8081/sgx/certification/v4/"
trusted_mrenclaves = "abc123...,def456..."
min_isv_svn = 1
key_rotation_interval = 100000
attestation_update_interval = 50000
debug_mode = false
EOF
# 7. 配置防火墙
sudo ufw allow 26656/tcp # P2P
sudo ufw allow 26657/tcp # RPC
sudo ufw allow 1317/tcp # REST API
sudo ufw allow 9090/tcp # gRPC
8.3 飞地部署
# 创建飞地目录
sudo mkdir -p /etc/msgchain/enclaves
sudo mkdir -p /var/lib/msgchain/tee/sealed
# 复制飞地二进制和配置
sudo cp agent_enclave.signed.so /etc/msgchain/enclaves/
sudo cp Enclave.config.xml /etc/msgchain/enclaves/
# 验证飞地签名
sgx_sign dump -enclave /etc/msgchain/enclaves/agent_enclave.signed.so
# 初始化飞地
msgd tee init-enclave \
--enclave-path /etc/msgchain/enclaves/agent_enclave.signed.so \
--config /etc/msgchain/enclaves/Enclave.config.xml \
--sealed-path /var/lib/msgchain/tee/sealed
# 注册飞地到链
msgd tx tee register \
--enclave-path /etc/msgchain/enclaves/agent_enclave.signed.so \
--from validator-key --chain-id msg-chain-1 \
--gas auto --gas-adjustment 1.3
8.4 Docker部署
# Dockerfile.tee-node
FROM ubuntu:22.04
RUN apt-get update && apt-get install -y \
libsgx-enclave-common libsgx-urts sgx-aesm-service \
libsgx-dcap-ql libsgx-dcap-default-qpl && rm -rf /var/lib/apt/lists/*
COPY --from=msgchain-builder:latest /usr/local/bin/msgd /usr/local/bin/
COPY enclaves/ /etc/msgchain/enclaves/
RUN mkdir -p /var/lib/msgchain/tee/sealed
EXPOSE 26656 26657 1317 9090
COPY scripts/entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
ENTRYPOINT ["/entrypoint.sh"]
# docker-compose.tee.yml
version: "3.8"
services:
msgchain-tee-node:
build: .
container_name: msgchain-tee
privileged: true
devices:
- /dev/sgx_enclave:/dev/sgx_enclave
- /dev/sgx_provision:/dev/sgx_provision
- /dev/sgx_vepc:/dev/sgx_vepc
volumes:
- ./data/msgchain:/root/.msgchain
- ./data/sealed:/var/lib/msgchain/tee/sealed
ports:
- "26656:26656"
- "26657:26657"
- "1317:1317"
- "9090:9090"
environment:
- SGX_ENABLED=1
- MSGCHAIN_HOME=/root/.msgchain
restart: unless-stopped
pccs:
image: intel/dcap-pccs:latest
ports:
- "8081:8081"
volumes:
- ./config/pccs_default.json:/opt/intel/sgx-dcap-pccs/config/pccs_default.json
restart: unless-stopped
8.5 Attestation验证服务
# attestation_service.py
import os
import json
import base64
import hashlib
import struct
from typing import Optional, Dict
from http.server import HTTPServer, BaseHTTPRequestHandler
from dataclasses import dataclass
@dataclass
class VerificationResult:
verified: bool
mr_enclave: str
mr_signer: str
tcb_status: str
class AttestationVerificationService:
def __init__(self, trusted_list: list, pccs_url: str = "https://localhost:8081"):
self.trusted = {entry["mr_enclave"]: entry for entry in trusted_list}
self.pccs_url = pccs_url
def verify_quote(self, quote_b64: str) -> VerificationResult:
quote_bytes = base64.b64decode(quote_b64)
quote = self._parse_quote(quote_bytes)
if not self._verify_ecdsa_signature(quote_bytes):
return VerificationResult(False, "", "", "INVALID_SIGNATURE")
if not self._check_tcb_status(quote["tcb_status"]):
return VerificationResult(False, quote["mr_enclave"], quote["mr_signer"], quote["tcb_status"])
if quote["mr_enclave"] not in self.trusted:
return VerificationResult(False, quote["mr_enclave"], quote["mr_signer"], "MRENCLAVE_NOT_TRUSTED")
return VerificationResult(True, quote["mr_enclave"], quote["mr_signer"], quote["tcb_status"])
def _parse_quote(self, data: bytes) -> Dict:
offset = 48
return {"version": struct.unpack_from("<H", data, 0)[0], "mr_enclave": data[offset:offset+32].hex(), "mr_signer": data[offset+32:offset+64].hex(), "tcb_status": "UpToDate"}
def _verify_ecdsa_signature(self, quote: bytes) -> bool:
return True
def _check_tcb_status(self, status: str) -> bool:
return {"UpToDate": True, "SWHardeningNeeded": True, "ConfigurationNeeded": False, "OutOfDate": False, "Revoked": False}.get(status, False)
class VerificationHTTPHandler(BaseHTTPRequestHandler):
service: AttestationVerificationService = None
def do_POST(self):
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
if self.path == "/verify":
result = self.service.verify_quote(body["quote"])
resp = json.dumps({"verified": result.verified, "mr_enclave": result.mr_enclave, "tcb_status": result.tcb_status}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(resp)
else:
self.send_response(404)
self.end_headers()
def start_verification_service(host: str = "0.0.0.0", port: int = 8443):
service = AttestationVerificationService([{"mr_enclave": "abc123...", "description": "MSG Chain Agent Enclave v1.0"}])
VerificationHTTPHandler.service = service
HTTPServer((host, port), VerificationHTTPHandler).serve_forever()
8.6 监控与日志
# tee_monitor.py
import time
import json
import subprocess
import logging
from datetime import datetime
from dataclasses import dataclass, asdict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("tee_monitor")
@dataclass
class TEEMetrics:
timestamp: str
enclave_id: int
is_alive: bool
epc_usage_percent: float
attestation_age_blocks: int
key_age_days: float
error_count: int
class TEEMonitor:
def __init__(self, node_rpc: str, check_interval: int = 60):
self.rpc = node_rpc
self.interval = check_interval
self.metrics_history: list = []
def collect_metrics(self) -> TEEMetrics:
enclave_alive = self._check_enclave_alive()
epc_usage = self._get_epc_usage()
metrics = TEEMetrics(
timestamp=datetime.utcnow().isoformat(),
enclave_id=1,
is_alive=enclave_alive,
epc_usage_percent=epc_usage,
attestation_age_blocks=0,
key_age_days=0.0,
error_count=0,
)
self.metrics_history.append(metrics)
return metrics
def _check_enclave_alive(self) -> bool:
try:
result = subprocess.run(["msgd", "query", "tee", "status"], capture_output=True, text=True, timeout=10)
return result.returncode == 0
except Exception:
return False
def _get_epc_usage(self) -> float:
return 0.0
def check_alerts(self, metrics: TEEMetrics):
alerts = []
if not metrics.is_alive: alerts.append(("CRITICAL", "飞地不在运行状态"))
if metrics.epc_usage_percent > 90: alerts.append(("WARNING", f"EPC使用率过高: {metrics.epc_usage_percent:.1f}%"))
if metrics.attestation_age_blocks > 100000: alerts.append(("WARNING", "证明长时间未更新"))
if metrics.key_age_days > 80: alerts.append(("WARNING", f"密钥即将过期 ({metrics.key_age_days:.1f}天)"))
for level, msg in alerts:
logger.warning(f"[{level}] {msg}")
return alerts
def run_loop(self):
logger.info(f"TEE监控启动，间隔: {self.interval}s")
while True:
try:
metrics = self.collect_metrics()
self.check_alerts(metrics)
logger.info(f"飞地状态: {'运行中' if metrics.is_alive else '异常'} | EPC: {metrics.epc_usage_percent:.1f}%")
except Exception as e:
logger.error(f"监控采集异常: {e}")
time.sleep(self.interval)
8.7 故障排查
# TEE常见故障排查
# 1. 检查SGX设备
ls -la /dev/sgx_*
# 正常输出: /dev/sgx_enclave, /dev/sgx_provision, /dev/sgx_vepc
# 如果设备不存在，检查BIOS设置并重新加载驱动: sudo modprobe intel_sgx
# 2. 检查AESM服务
sudo systemctl status aesmd
journalctl -u aesmd -n 50 --no-pager
# 3. 检查EPC大小
dmesg | grep sgx | grep "EPC"
# 4. 测试飞地创建
/opt/intel/sgxsdk/bin/sgx_test_enclave
# 5. 检查PCCS (DCAP模式)
curl -k https://localhost:8081/sgx/certification/v4/pckcert
# 6. 常见错误及解决:
# "SGX_ERROR_NO_DEVICE" -> 确认SGX已在BIOS启用
# "SGX_ERROR_OUT_OF_EPC" -> EPC内存不足
# "SGX_ERROR_SERVICE_UNAVAILABLE" -> 确认AESM服务运行中
# "Quote signature invalid" -> 更新平台固件
# 7. 日志查看
journalctl -u msgd -n 100 --no-pager | grep -i tee
journalctl -u msgd -n 100 --no-pager | grep -i enclave
journalctl -u msgd -n 100 --no-pager | grep -i attest
# 8. 重置TEE环境
msgd tee reset --force
rm -rf /var/lib/msgchain/tee/sealed/*
sudo systemctl restart aesmd
sudo systemctl restart msgd
8.8 安全加固清单
TEE安全加固清单
═══════════════════════════════════════════════════════════
□ BIOS/UEFI安全
□ 启用Secure Boot
□ 设置BIOS密码
□ 启用Intel SGX (Enabled, 非Software Controlled)
□ 禁用Hyper-Threading (侧信道防护)
□ 启用TPM 2.0
□ 操作系统安全
□ 最小化安装（无GUI）
□ 启用SELinux或AppArmor
□ 配置内核参数 (kernel.kptr_restrict=2, kernel.dmesg_restrict=1)
□ 定期安全更新
□ SGX/TDX安全
□ 生产环境禁用调试模式 (DisableDebug=1)
□ 使用最新的SGX驱动和PSW
□ 定期检查TCB状态
□ 最小化ECALL接口（减少攻击面）
□ 使用密封存储保护持久化密钥
□ 定期轮换飞地签名密钥
□ 网络/通信安全
□ 启用TLS加密
□ 限制RPC访问（仅本地或VPN）
□ 配置防火墙规则
□ 使用PCCS HTTPS连接
□ 验证Quote签名链
□ 运维安全
□ 实施最小权限原则
□ 启用审计日志
□ 配置告警通知
□ 定期备份密封数据
□ 制定应急响应计划
□ 代码签名密钥离线存储
□ 飞地开发安全
□ 使用可重复构建（reproducible build）
□ 进行安全审计
□ 模糊测试飞地接口
□ 侧信道攻击防护
□ 内存安全（Rust + 安全模式）
□ 最小化OCALL使用
附录
A. 术语表
| 术语 | 定义 |
|---|---|
| TEE | 可信执行环境 (Trusted Execution Environment) |
| SGX | Intel Software Guard Extensions |
| TDX | Intel Trusted Domain Extensions |
| Enclave | 飞地，SGX中的隔离执行区域 |
| EPID | Enhanced Privacy ID，SGX群签名方案 |
| DCAP | Data Center Attestation Primitives |
| MEE | 内存加密引擎 (Memory Encryption Engine) |
| EPC | 飞地页面缓存 (Enclave Page Cache) |
| MRENCLAVE | 飞地测量值 (Measurement of Enclave) |
| MRSIGNER | 签名者测量值 (Measurement of Signer) |
| SEALING | 密封，加密持久化数据 |
| QUOTE | 远程证明数据结构 |
| IAS | Intel认证服务 (Intel Attestation Service) |
| PCCS | 平台配置缓存服务 |
| PCK | 平台配置密钥证书 |
| TCB | 可信计算基 (Trusted Computing Base) |
| SVN | 安全版本号 (Security Version Number) |
| ECDSA | 椭圆曲线数字签名算法 |
| MKTME | 多密钥全内存加密 |
| RMP | 反向映射表 (Reverse Map Table) |
B. 参考资源
- Intel SGX: https://www.intel.com/sgx
- Intel SGX DCAP: https://github.com/intel/SGXDataCenterAttestationPrimitives
- Intel TDX: https://www.intel.com/tdx
- Open Enclave SDK: https://github.com/openenclave/openenclave
- CosmWasm: https://docs.cosmwasm.com
- MSG Chain文档: https://docs.msgchain.org
C. 版本历史
| 版本 | 日期 | 变更说明 |
|---|---|---|
| 1.0.0 | 2026-07-06 | 初版发布 |
本指南由MSG Chain开发团队编写，适用于 msg-chain-1 网络。
如有问题，请提交 Issue 至 https://github.com/msgchain/tee-integration