AI Agent 智能合约安全审计清单指南
— MSG Chain (msg-chain-1) 专用审计方法论 —
⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/
版本: 1.0 | 链: msg-chain-1 | Bech32 前缀:
msg
目录
1. 概述
1.1 为什么审计 Agent 合约
AI Agent 智能合约运行在 MSG Chain 上，负责自主执行策略、管理资产、与其他 Agent 交互。与普通智能合约相比，Agent 合约有以下独特风险：
- 自主执行：Agent 无需人工干预即可调用函数，一旦存在漏洞会被自动化利用
- 资产管理：Agent 持有用户委托的资金，权限过大可能导致资金流失
- 宪法约束：Agent 遵循预设的"宪法"规则，若宪法可被绕过则约束失效
- A2A 通信：Agent 间消息若未经认证，可被伪造指令操控
- 持续运行：Agent 7×24 小时在线，漏洞窗口期更短
1.2 审计阶段
第一阶段：静态分析
├─ 代码审查（手动）
├─ cosmwasm-check 静态验证
└─ 依赖项安全检查
第二阶段：自动化扫描
├─ 漏洞模式匹配
├─ 气体估算边界测试
└─ 宪法规则验证
第三阶段：渗透测试
├─ 代理冒充攻击测试
├─ 限额绕过攻击测试
└─ 预言机操控测试
第四阶段：报告生成
├─ 发现汇总
├─ 严重性分级
└─ 修复建议
1.3 Agent 威胁建模
/// Agent 合约威胁模型
pub struct ThreatModel {
/// 攻击者模型: 外部用户、恶意 Agent、验证者
pub attacker_types: Vec<AttackerType>,
/// 资产风险: 委托资金、原生代币、CW20 代币
pub assets_at_risk: Vec<AssetType>,
/// 攻击向量: 重入、限额绕过、签名伪造
pub attack_vectors: Vec<AttackVector>,
/// 影响范围: 单 Agent 沦陷、全网络扩散
pub blast_radius: BlastRadius,
}
pub enum AttackerType {
ExternalUser, // 外部未授权用户
MaliciousAgent, // 恶意 Agent
CompromisedKey, // 密钥泄露
ValidatorRogue, // 恶意验证者
}
pub enum BlastRadius {
SingleAgent, // 仅影响单个 Agent
AgentPool, // 影响 Agent 池
CrossChain, // 影响跨链桥
}
威胁矩阵示例:
| 威胁 | 攻击者 | 影响 | 可能性 | 严重性 |
|---|---|---|---|---|
| 重入攻击 | 恶意合约 | 资金耗尽 | 高 | 严重 |
| 限额绕过 | 外部用户 | 超额支出 | 中 | 高 |
| 宪法绕过 | 恶意 Agent | 行为失控 | 低 | 严重 |
| 签名伪造 | 外部攻击者 | A2A 欺骗 | 中 | 高 |
| 气体耗尽 | 外部用户 | DoS | 高 | 中 |
| 预言机操控 | 验证者 | 错误定价 | 低 | 高 |
2. 常见漏洞模式与修复
2.1 重入攻击 (Reentrancy)
重入攻击是 Agent 合约最常见的漏洞。当合约在状态更新前发起外部调用时，攻击者可以在第一次调用完成前重新进入合约，多次提取资金。
❌ 不安全的实现
use cosmwasm_std::{
DepsMut, Env, MessageInfo, Response, BankMsg, Coin, Uint128,
StdResult, StdError, Storage,
};
use cw_storage_plus::Map;
pub const BALANCES: Map<&Addr, Uint128> = Map::new("balances");
// ❌ BAD: 状态在外部调用后更新 — 可重入
pub fn execute_withdraw(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount: Uint128,
) -> StdResult<Response> {
let balance = BALANCES.load(deps.storage, &info.sender)?;
let msg = BankMsg::Send {
to_address: info.sender.to_string(),
amount: vec![Coin::new(amount.u128(), "umsg")],
};
BALANCES.save(deps.storage, &info.sender, &(balance - amount))?;
Ok(Response::new()
.add_message(msg)
.add_attribute("action", "withdraw")
.add_attribute("amount", amount.to_string()))
}
✅ 修复: Checks-Effects-Interactions 模式
// ✅ GOOD: 先更新状态，再执行外部调用
pub fn execute_withdraw_fixed(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount: Uint128,
) -> StdResult<Response> {
let balance = BALANCES.load(deps.storage, &info.sender)?;
if balance < amount {
return Err(StdError::generic_err("insufficient balance"));
}
BALANCES.save(deps.storage, &info.sender, &(balance - amount))?;
Ok(Response::new()
.add_message(BankMsg::Send {
to_address: info.sender.to_string(),
amount: vec![Coin::new(amount.u128(), "umsg")],
})
.add_attribute("action", "withdraw")
.add_attribute("amount", amount))
}
🔒 进阶防护: 重入锁
use cw_storage_plus::Item;
pub const REENTRANCY_GUARD: Item<bool> = Item::new("reentrancy_guard");
pub fn execute_with_guard<F>(
deps: &mut DepsMut,
env: &Env,
info: &MessageInfo,
f: F,
) -> StdResult<Response>
where
F: FnOnce(&mut DepsMut, &Env, &MessageInfo) -> StdResult<Response>,
{
if REENTRANCY_GUARD.load(deps.storage)? {
return Err(StdError::generic_err("reentrancy detected"));
}
REENTRANCY_GUARD.save(deps.storage, &true)?;
let result = f(deps, env, info);
REENTRANCY_GUARD.save(deps.storage, &false)?;
result
}
🔍 检测方法
def detect_reentrancy(code: str) -> list:
findings = []
lines = code.split('\n')
for i, line in enumerate(lines):
if 'add_message' in line and ('BankMsg::Send' in line or 'WasmMsg::Execute' in line):
for j in range(i + 1, min(i + 10, len(lines))):
if 'save(' in lines[j] or 'update(' in lines[j]:
findings.append({
'type': 'reentrancy',
'severity': 'CRITICAL',
'line': i + 1,
'description': f'external call before state update (line {j+1})',
'recommendation': 'move save() before add_message()'
})
break
return findings
2.2 整数溢出/下溢 (Integer Overflow/Underflow)
虽然 Rust 在 debug 模式下会检测溢出，但 release 模式下使用算术运算时需谨慎。
❌ 不安全的实现
// ❌ BAD: 使用 unchecked 算术
pub fn execute_deposit(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount: Uint128,
) -> StdResult<Response> {
let mut total = TOTAL_DEPOSITED.load(deps.storage)?;
total = total + amount; // 可能溢出
TOTAL_DEPOSITED.save(deps.storage, &total)?;
Ok(Response::new().add_attribute("action", "deposit"))
}
✅ 使用 checked 算术
use cosmwasm_std::Uint128;
use cosmwasm_std::Decimal;
// ✅ GOOD: 使用 checked 算术
pub fn execute_deposit_fixed(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount: Uint128,
) -> StdResult<Response> {
let total = TOTAL_DEPOSITED.load(deps.storage)?;
let new_total = total
.checked_add(amount)
.map_err(|_| StdError::generic_err("overflow"))?;
TOTAL_DEPOSITED.save(deps.storage, &new_total)?;
Ok(Response::new().add_attribute("action", "deposit_safe"))
}
// ✅ Decimal 精度安全
pub fn calculate_fee(amount: Uint128, rate: Decimal) -> StdResult<Uint128> {
let fee = amount
.checked_mul(rate)
.map_err(|_| StdError::generic_err("fee overflow"))?;
if fee.is_zero() && !rate.is_zero() {
return Err(StdError::generic_err("fee rounded to zero"));
}
Ok(fee)
}
2.3 未检查的返回值 (Unchecked Return Values)
Agent 合约经常调用其他合约（CW20 转账、Staking 操作），忽略返回值会导致静默失败。
❌ 不安全的实现
// ❌ BAD: 忽略 CW20 转账的返回值
pub fn execute_payment(
deps: DepsMut,
env: Env,
info: MessageInfo,
token_contract: String,
amount: Uint128,
recipient: String,
) -> StdResult<Response> {
let msg = WasmMsg::Execute {
contract_addr: token_contract,
msg: to_binary(&Cw20ExecuteMsg::Transfer {
recipient: recipient.clone(),
amount,
})?,
funds: vec![],
};
Ok(Response::new()
.add_message(msg) // ❌ 未验证返回结果
.add_attribute("action", "payment"))
}
✅ 检查 SubMsg 返回值
use cosmwasm_std::{SubMsg, ReplyOn, SubMsgResult, from_binary};
// ✅ GOOD: 使用 SubMsg 检查返回值
pub fn execute_payment_fixed(
deps: DepsMut,
env: Env,
info: MessageInfo,
token_contract: String,
amount: Uint128,
recipient: String,
) -> StdResult<Response> {
let msg = WasmMsg::Execute {
contract_addr: token_contract.clone(),
msg: to_binary(&Cw20ExecuteMsg::Transfer {
recipient: recipient.clone(),
amount,
})?,
funds: vec![],
};
let sub_msg = SubMsg::reply_on_success(msg, PAYMENT_REPLY_ID);
PENDING_PAYMENTS.save(deps.storage, &PAYMENT_REPLY_ID, &PendingPayment {
sender: info.sender.clone(),
recipient: recipient.clone(),
amount,
})?;
Ok(Response::new()
.add_submessage(sub_msg)
.add_attribute("action", "payment_verified"))
}
// 在 reply 入口点验证
pub fn reply(deps: DepsMut, env: Env, msg: Reply) -> StdResult<Response> {
if msg.id == PAYMENT_REPLY_ID {
match msg.result {
SubMsgResult::Ok(_) => {
PENDING_PAYMENTS.remove(deps.storage, &PAYMENT_REPLY_ID);
Ok(Response::new().add_attribute("payment", "success"))
}
SubMsgResult::Err(e) => {
let pending = PENDING_PAYMENTS.load(deps.storage, &PAYMENT_REPLY_ID)?;
PENDING_PAYMENTS.remove(deps.storage, &PAYMENT_REPLY_ID);
Err(StdError::generic_err(
format!("payment failed: {}, amount: {}", e, pending.amount)
))
}
}
} else {
Err(StdError::generic_err("unknown reply id"))
}
}
2.4 访问控制漏洞 (Access Control)
Agent 合约中，管理员功能、Agent 配置和资金操作必须有严格的访问控制。
❌ 不安全的实现
// ❌ BAD: 缺少访问控制
pub fn execute_set_spending_limit(
deps: DepsMut,
env: Env,
info: MessageInfo,
new_limit: Uint128,
) -> StdResult<Response> {
// 没有检查调用者身份！
SPENDING_LIMIT.save(deps.storage, &new_limit)?;
Ok(Response::new().add_attribute("action", "set_spending_limit"))
}
✅ 基于角色的访问控制
use cw_controllers::Admin;
use cw_storage_plus::Item;
pub const ADMIN: Admin = Admin::new("admin");
pub const AGENT_OWNER: Item<Addr> = Item::new("agent_owner");
// ✅ GOOD: 显式访问控制
pub fn execute_set_spending_limit_fixed(
deps: DepsMut,
env: Env,
info: MessageInfo,
new_limit: Uint128,
) -> StdResult<Response> {
ADMIN.assert_admin(deps.as_ref(), &info.sender)?;
SPENDING_LIMIT.save(deps.storage, &new_limit)?;
Ok(Response::new()
.add_attribute("action", "set_spending_limit")
.add_attribute("set_by", info.sender))
}
// ✅ 细化权限粒度
pub enum Role {
Admin,
Operator,
Auditor,
}
pub const ROLES: Map<&Addr, Role> = Map::new("roles");
pub fn assert_role(storage: &dyn Storage, addr: &Addr, required: Role) -> StdResult<()> {
let role = ROLES.load(storage, addr)
.map_err(|_| StdError::generic_err("no role assigned"))?;
match (required, role) {
(Role::Admin, Role::Admin) => Ok(()),
(Role::Operator, Role::Admin) | (Role::Operator, Role::Operator) => Ok(()),
(Role::Auditor, _) => Ok(()),
_ => Err(StdError::generic_err("insufficient permissions")),
}
}
2.5 Agent 宪法逻辑漏洞 (Constitution Logic Errors)
Agent 的"宪法"是其行为准则。宪法逻辑漏洞允许 Agent 执行违反预设规则的操作。
❌ 不安全的宪法实现
// ❌ BAD: 宪法规则可被绕过
pub fn execute_trade(
deps: DepsMut,
env: Env,
info: MessageInfo,
trade: Trade,
) -> StdResult<Response> {
let constitution = CONSTITUTION.load(deps.storage)?;
// 只检查了部分规则 — 未验证交易对和白名单
if trade.amount > constitution.max_position_size {
return Err(StdError::generic_err("position too large"));
}
execute_swap(deps, env, info, trade)
}
✅ 宪法强制执行的正确实现
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct Constitution {
pub max_position_size: Uint128,
pub allowed_assets: Vec<String>,
pub risk_level: RiskLevel,
pub is_active: bool,
pub version: u32,
}
impl Constitution {
pub fn validate_trade(&self, trade: &Trade) -> StdResult<()> {
if !self.is_active {
return Err(StdError::generic_err("constitution is inactive"));
}
if trade.amount > self.max_position_size {
return Err(StdError::generic_err(
format!("position {} exceeds max {}", trade.amount, self.max_position_size)
));
}
if !self.allowed_assets.contains(&trade.asset_in) ||
!self.allowed_assets.contains(&trade.asset_out) {
return Err(StdError::generic_err("asset not in allowed list"));
}
Ok(())
}
}
// ✅ 执行交易时必须通过宪法验证
pub fn execute_trade_fixed(
deps: DepsMut,
env: Env,
info: MessageInfo,
trade: Trade,
) -> StdResult<Response> {
let constitution = CONSTITUTION.load(deps.storage)?;
constitution.validate_trade(&trade)?;
execute_swap(deps, env, info, trade)
}
2.6 气体耗尽漏洞 (Gas Exhaustion)
Agent 合约中的无界循环或高复杂度操作可导致气体耗尽攻击。
❌ 不安全的实现
// ❌ BAD: 无界循环
pub fn execute_process_all_orders(
deps: DepsMut,
env: Env,
info: MessageInfo,
) -> StdResult<Response> {
let orders = PENDING_ORDERS.load(deps.storage)?;
let mut msgs = vec![];
for order in orders.iter() {
msgs.push(process_order(deps, &env, &info, order)?);
}
Ok(Response::new().add_messages(msgs))
}
✅ 分批处理 + 气体限制
pub const MAX_BATCH_SIZE: u32 = 50;
pub fn execute_process_batch(
deps: DepsMut,
env: Env,
info: MessageInfo,
start_from: u64,
) -> StdResult<Response> {
let orders = PENDING_ORDERS.load(deps.storage)?;
let total = orders.len() as u64;
if start_from >= total {
return Err(StdError::generic_err("no more orders"));
}
let end = (start_from + MAX_BATCH_SIZE as u64).min(total);
let batch = &orders[start_from as usize..end as usize];
let mut messages = vec![];
for order in batch {
messages.push(create_order_message(order)?);
}
if end < total {
let continuation = WasmMsg::Execute {
contract_addr: env.contract.address.to_string(),
msg: to_binary(&ExecuteMsg::ProcessBatch { start_from: end })?,
funds: vec![],
};
messages.push(continuation.into());
}
Ok(Response::new()
.add_messages(messages)
.add_attribute("processed", (end - start_from).to_string())
.add_attribute("remaining", (total - end).to_string()))
}
2.7 预言机价格操控 (Oracle Price Manipulation)
❌ 不安全的预言机使用
// ❌ BAD: 依赖单一价格源
pub fn execute_swap(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount_in: Uint128,
) -> StdResult<Response> {
let price = SINGLE_ORACLE.load(deps.storage)?; // 单点故障
execute_swap_internal(deps, env, info, amount_in, price)
}
✅ 多源验证 + TWAP
#[derive(Serialize, Deserialize)]
pub struct OraclePrice {
pub source: OracleSource,
pub price: Decimal,
pub timestamp: u64,
pub confidence: u8,
}
pub const ORACLE_PRICES: Map<OracleSource, OraclePrice> = Map::new("oracle_prices");
pub const MIN_CONFIDENCE: u8 = 80;
pub const MAX_PRICE_DEVIATION: Decimal = Decimal::percent(5);
pub const STALE_THRESHOLD: u64 = 3600;
pub fn get_aggregated_price(storage: &dyn Storage) -> StdResult<Decimal> {
let sources = vec![
OracleSource::MsgChainOracle,
OracleSource::BandProtocol,
OracleSource::PythNetwork,
];
let mut valid_prices: Vec<Decimal> = vec![];
for source in sources {
if let Ok(price) = ORACLE_PRICES.load(storage, &source) {
if current_timestamp() - price.timestamp > STALE_THRESHOLD { continue; }
if price.confidence < MIN_CONFIDENCE { continue; }
valid_prices.push(price.price);
}
}
if valid_prices.len() < 2 {
return Err(StdError::generic_err("insufficient oracle sources"));
}
valid_prices.sort_by(|a, b| a.partial_cmp(&b).unwrap());
let median = valid_prices[valid_prices.len() / 2];
for price in &valid_prices {
let deviation = if *price > median { *price - median } else { median - *price };
if deviation > MAX_PRICE_DEVIATION {
return Err(StdError::generic_err("price deviation too high"));
}
}
Ok(median)
}
2.8 跨合约调用注入 (Cross-Contract Call Injection)
❌ 不安全的合约调用
// ❌ BAD: 未验证目标合约
pub fn execute_delegate_call(
deps: DepsMut,
env: Env,
info: MessageInfo,
target_contract: String,
target_msg: Binary,
) -> StdResult<Response> {
let msg = WasmMsg::Execute {
contract_addr: target_contract,
msg: target_msg,
funds: vec![],
};
Ok(Response::new().add_message(msg))
}
✅ 合约白名单验证
pub const WHITELISTED_CONTRACTS: Map<&Addr, ContractInfo> = Map::new("whitelisted");
pub fn execute_delegate_call_fixed(
deps: DepsMut,
env: Env,
info: MessageInfo,
target_contract: String,
target_msg: Binary,
method_name: String,
) -> StdResult<Response> {
let target_addr = deps.api.addr_validate(&target_contract)?;
let contract_info = WHITELISTED_CONTRACTS
.load(deps.storage, &target_addr)
.map_err(|_| StdError::generic_err("contract not whitelisted"))?;
if !contract_info.allowed_methods.contains(&method_name) {
return Err(StdError::generic_err("method not allowed"));
}
let code_hash = deps.querier.query_wasm_contract_info(&target_addr)?;
let expected_hash = CONTRACT_CODE_HASHES.load(deps.storage, &target_addr)?;
if code_hash.code_hash != expected_hash {
return Err(StdError::generic_err("code hash mismatch"));
}
let msg = WasmMsg::Execute {
contract_addr: target_contract,
msg: target_msg,
funds: vec![],
};
Ok(Response::new()
.add_message(msg)
.add_attribute("target", target_contract))
}
3. Agent 特有安全风险
3.1 支出限额漏洞 (Spending Limit Vulnerabilities)
❌ 不安全的限额实现
class UnsafeAgent:
def __init__(self, wallet_address: str, private_key: str):
self.wallet = Wallet(wallet_address, private_key)
async def execute_action(self, action: dict):
# ❌ 没有任何限制 — 攻击者可提取全部资金
if action['type'] == 'pay':
await self.wallet.send(
to=action['recipient'],
amount=action['amount'],
denom="umsg"
)
✅ 多层限额系统
import time
from enum import Enum
from dataclasses import dataclass
from typing import Dict, Optional
class LimitType(Enum):
PER_TX = "per_tx"
HOURLY = "hourly"
DAILY = "daily"
MONTHLY = "monthly"
@dataclass
class SpendingLimit:
limit_type: LimitType
max_amount: int
current_used: int = 0
reset_period: int = 0
last_reset: int = 0
class SecureAgent:
def __init__(self, wallet_address: str, private_key: str):
self.wallet = Wallet(wallet_address, private_key)
self.spending_limits = {
LimitType.PER_TX: SpendingLimit(
limit_type=LimitType.PER_TX,
max_amount=1000_000_000,
),
LimitType.DAILY: SpendingLimit(
limit_type=LimitType.DAILY,
max_amount=20_000_000_000,
reset_period=86400,
),
}
self.whitelist_recipients = {
"msg1vault...": "self_vault",
}
async def check_spending_limits(self, amount: int, recipient: str) -> bool:
if recipient in self.whitelist_recipients:
return True
current_time = time.time()
if amount > self.spending_limits[LimitType.PER_TX].max_amount:
return False
for limit_type in [LimitType.HOURLY, LimitType.DAILY, LimitType.MONTHLY]:
limit = self.spending_limits[limit_type]
if current_time - limit.last_reset > limit.reset_period:
limit.current_used = 0
limit.last_reset = current_time
if limit.current_used + amount > limit.max_amount:
return False
return True
async def execute_action(self, action: dict) -> bool:
amount = action.get('amount', 0)
recipient = action.get('recipient', '')
if not await self.check_spending_limits(amount, recipient):
print(f"[SECURITY] Spending limit exceeded")
return False
tx_hash = await self.wallet.send(
to=recipient,
amount=amount,
denom="umsg"
)
for limit_type in [LimitType.HOURLY, LimitType.DAILY, LimitType.MONTHLY]:
self.spending_limits[limit_type].current_used += amount
print(f"[AGENT] Executed: {amount} umsg -> {recipient}")
return True
3.2 Action 频率限制 (Action Rate Limiting)
✅ 滑动窗口限速
use cw_storage_plus::{Item, Map};
pub const ACTION_COUNTER: Map<&Addr, u64> = Map::new("action_counter");
pub const ACTION_WINDOW: Item<ActionWindow> = Item::new("action_window");
pub fn check_rate_limit(
storage: &dyn Storage,
sender: &Addr,
current_block: u64,
) -> StdResult<()> {
let window = ACTION_WINDOW.load(storage)?;
let block_actions = BLOCK_ACTIONS
.may_load(storage, (sender, current_block))?
.unwrap_or(0);
if block_actions >= window.max_per_block {
return Err(StdError::generic_err("rate limit: per block"));
}
let total = ACTION_COUNTER
.may_load(storage, sender)?
.unwrap_or(0);
if total >= window.max_actions as u64 {
return Err(StdError::generic_err("rate limit: window exceeded"));
}
Ok(())
}
3.3 宪法违规检测 (Constitution Violation Detection)
import hashlib
import json
import time
from enum import Enum
from dataclasses import dataclass
from typing import List, Optional, Dict, Any
from datetime import datetime
class ViolationSeverity(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class ConstitutionViolation:
violation_type: str
description: str
severity: ViolationSeverity
evidence: Dict[str, Any]
timestamp: int
class ConstitutionValidator:
def __init__(self, constitution: Dict[str, Any]):
self.constitution = constitution
self.constitution_hash = self._hash_constitution()
self.violations: List[ConstitutionViolation] = []
def _hash_constitution(self) -> str:
return hashlib.sha256(
json.dumps(self.constitution, sort_keys=True).encode()
).hexdigest()
def verify_constitution_integrity(self, onchain_hash: str) -> bool:
if onchain_hash != self.constitution_hash:
self.violations.append(ConstitutionViolation(
violation_type="constitution_tampered",
description="on-chain hash mismatch",
severity=ViolationSeverity.CRITICAL,
evidence={
"expected": self.constitution_hash,
"actual": onchain_hash
},
timestamp=int(time.time())
))
return False
return True
def validate_trade(self, trade: Dict[str, Any]) -> bool:
allowed = self.constitution.get('allowed_assets', [])
if trade.get('asset_in') not in allowed:
self.violations.append(ConstitutionViolation(
violation_type="asset_not_allowed",
description=f"{trade.get('asset_in')} not in whitelist",
severity=ViolationSeverity.HIGH,
evidence=trade,
timestamp=int(time.time())
))
return False
max_pos = int(self.constitution.get('max_position_size', '0'))
if int(trade.get('amount', 0)) > max_pos:
self.violations.append(ConstitutionViolation(
violation_type="position_size_exceeded",
description=f"amount exceeds max {max_pos}",
severity=ViolationSeverity.HIGH,
evidence=trade,
timestamp=int(time.time())
))
return False
return True
def get_violation_report(self) -> dict:
return {
"constitution_hash": self.constitution_hash,
"total_violations": len(self.violations),
"by_severity": {
"critical": len([v for v in self.violations if v.severity == ViolationSeverity.CRITICAL]),
"high": len([v for v in self.violations if v.severity == ViolationSeverity.HIGH]),
"medium": len([v for v in self.violations if v.severity == ViolationSeverity.MEDIUM]),
"low": len([v for v in self.violations if v.severity == ViolationSeverity.LOW]),
},
"violations": [
{"type": v.violation_type, "description": v.description,
"severity": v.severity.value, "timestamp": v.timestamp}
for v in self.violations
]
}
3.4 密钥管理漏洞 (Key Management Vulnerabilities)
✅ 安全的密钥管理方案
import os
import hashlib
import base64
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from typing import Optional, List
from dataclasses import dataclass
@dataclass
class KeyRotationRecord:
old_key_hash: str
new_key_hash: str
rotated_by: str
block_height: int
reason: str
class SecureKeyManager:
def __init__(self, encryption_key: bytes):
self.fernet = Fernet(base64.urlsafe_b64encode(encryption_key[:32]))
self.rotation_history: List[KeyRotationRecord] = []
@staticmethod
def derive_key(password: str, salt: Optional[bytes] = None) -> tuple:
if salt is None:
salt = os.urandom(16)
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=100_000,
)
return kdf.derive(password.encode()), salt
def encrypt_private_key(self, private_key: str) -> bytes:
return self.fernet.encrypt(private_key.encode())
def decrypt_private_key(self, encrypted_key: bytes) -> str:
return self.fernet.decrypt(encrypted_key).decode()
def rotate_key(self, old_key: bytes, new_key: str,
rotated_by: str, block: int,
reason: str = "scheduled") -> bytes:
decrypted = self.decrypt_private_key(old_key)
old_hash = hashlib.sha256(decrypted.encode()).hexdigest()
new_encrypted = self.encrypt_private_key(new_key)
new_hash = hashlib.sha256(new_key.encode()).hexdigest()
self.rotation_history.append(KeyRotationRecord(
old_key_hash=old_hash,
new_key_hash=new_hash,
rotated_by=rotated_by,
block_height=block,
reason=reason
))
return new_encrypted
def get_history(self) -> List[KeyRotationRecord]:
return self.rotation_history
class MultiSigKeyManager:
def __init__(self, guardians: List[str], threshold: int):
self.guardians = guardians
self.threshold = threshold
self.approvals: Dict[str, List[str]] = {}
self.ops: Dict[str, Dict] = {}
def create_proposal(self, action: str, params: dict, proposer: str) -> str:
import uuid
pid = str(uuid.uuid4())
self.ops[pid] = {
'action': action,
'params': params,
'proposer': proposer,
'timestamp': time.time(),
'status': 'pending'
}
self.approvals[pid] = []
return pid
def approve(self, proposal_id: str, guardian: str) -> bool:
if guardian not in self.guardians:
return False
if proposal_id not in self.ops:
return False
if guardian in self.approvals[proposal_id]:
return True
self.approvals[proposal_id].append(guardian)
if len(self.approvals[proposal_id]) >= self.threshold:
self.ops[proposal_id]['status'] = 'approved'
return True
return False
3.5 A2A 消息认证 (Agent-to-Agent Authentication)
❌ 不安全的 A2A 通信
// ❌ BAD: A2A 消息无认证
pub fn execute_handle_agent_message(
deps: DepsMut,
env: Env,
info: MessageInfo,
from_agent: String,
action: String,
params: Binary,
) -> StdResult<Response> {
// ❌ 没有验证 from_agent 的真实身份
handle_action(deps, env, &action, params)
}
✅ DID 签名认证
pub const AGENT_DID_REGISTRY: Map<&Addr, AgentDID> = Map::new("agent_dids");
pub const NONCE_REGISTRY: Map<&Addr, u64> = Map::new("nonces");
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct SignedAgentMessage {
pub from_did: String,
pub to_did: String,
pub action: String,
pub params: Binary,
pub nonce: u64,
pub timestamp: u64,
pub signature: Binary,
}
// ✅ GOOD: A2A 消息认证
pub fn execute_handle_agent_message_fixed(
deps: DepsMut,
env: Env,
info: MessageInfo,
signed_msg: SignedAgentMessage,
) -> StdResult<Response> {
let from_addr = deps.api.addr_validate(&signed_msg.from_did)?;
let to_addr = deps.api.addr_validate(&signed_msg.to_did)?;
if to_addr != env.contract.address {
return Err(StdError::generic_err("not addressed to this agent"));
}
let sender_did = AGENT_DID_REGISTRY
.load(deps.storage, &from_addr)
.map_err(|_| StdError::generic_err("unknown sender DID"))?;
if sender_did.revoked {
return Err(StdError::generic_err("sender DID revoked"));
}
if env.block.time.seconds() > sender_did.valid_until {
return Err(StdError::generic_err("sender DID expired"));
}
let stored_nonce = NONCE_REGISTRY
.may_load(deps.storage, &from_addr)?
.unwrap_or(0);
if signed_msg.nonce <= stored_nonce {
return Err(StdError::generic_err("replay attack detected"));
}
let max_age: u64 = 300;
if env.block.time.seconds() > signed_msg.timestamp + max_age {
return Err(StdError::generic_err("message expired"));
}
// 签名验证
let msg_bytes = build_signing_message(&signed_msg)?;
verify_signature(&sender_did.public_key, &msg_bytes, &signed_msg.signature, &sender_did.key_type)?;
NONCE_REGISTRY.save(deps.storage, &from_addr, &signed_msg.nonce)?;
handle_action(deps, env, &signed_msg.action, signed_msg.params)
}
fn build_signing_message(msg: &SignedAgentMessage) -> StdResult<Vec<u8>> {
let canonical = serde_json::to_string(&serde_json::json!({
"from_did": msg.from_did,
"to_did": msg.to_did,
"action": msg.action,
"nonce": msg.nonce,
"timestamp": msg.timestamp,
})).map_err(|_| StdError::generic_err("serialization error"))?;
Ok(canonical.into_bytes())
}
Python A2A 客户端示例
import hashlib
import json
import time
from cryptography.hazmat.primitives.asymmetric import ed25519
from typing import Dict, Any
class AgentToAgentClient:
def __init__(self, did: str, private_key: bytes, agent_contract: str, wallet):
self.did = did
self.private_key = private_key
self.agent_contract = agent_contract
self.wallet = wallet
self.nonce = int(time.time() * 1000)
def _get_next_nonce(self) -> int:
self.nonce += 1
return self.nonce
def _sign_message(self, message: Dict) -> bytes:
canonical = json.dumps(message, sort_keys=True, separators=(',', ':')).encode()
msg_hash = hashlib.sha256(canonical).digest()
key = ed25519.Ed25519PrivateKey.from_private_bytes(self.private_key)
return key.sign(msg_hash)
def send_action(self, target_did: str, action: str, params: Dict[str, Any]) -> str:
signed_msg = {
"from_did": self.did,
"to_did": target_did,
"action": action,
"params": params,
"nonce": self._get_next_nonce(),
"timestamp": int(time.time()),
}
signature = self._sign_message(signed_msg)
signed_msg["signature"] = signature.hex()
msg = {"handle_agent_message": signed_msg}
result = self.wallet.contract_call(contract=self.agent_contract, msg=msg)
tx_hash = result.get('txhash', '')
print(f"[A2A] Sent {action} -> {target_did} [{tx_hash}]")
return tx_hash
3.6 Agent 关闭机制 (Graceful Shutdown)
✅ 紧急暂停实现
pub const PAUSED: Item<bool> = Item::new("paused");
pub const PAUSE_GUARDIANS: Map<&Addr, bool> = Map::new("pause_guardians");
pub const PAUSE_THRESHOLD: Item<u32> = Item::new("pause_threshold");
pub fn assert_not_paused(storage: &dyn Storage) -> StdResult<()> {
if PAUSED.load(storage)? {
return Err(StdError::generic_err("agent is paused"));
}
Ok(())
}
pub fn execute_emergency_pause(
deps: DepsMut,
env: Env,
info: MessageInfo,
signatures: Vec<Signature>,
) -> StdResult<Response> {
let guardians: Vec<Addr> = PAUSE_GUARDIANS
.range(deps.storage, None, None, Order::Ascending)
.filter_map(|r| r.ok())
.map(|(k, _)| k)
.collect();
let threshold = PAUSE_THRESHOLD.load(deps.storage)?;
if !verify_multisig(b"EMERGENCY_PAUSE", &signatures, &guardians, threshold) {
return Err(StdError::generic_err("insufficient signatures"));
}
PAUSED.save(deps.storage, &true)?;
Ok(Response::new().add_attribute("action", "emergency_pause"))
}
pub fn execute_safe<F>(
deps: &mut DepsMut,
env: &Env,
info: &MessageInfo,
f: F,
) -> StdResult<Response>
where
F: FnOnce(&mut DepsMut, &Env, &MessageInfo) -> StdResult<Response>,
{
assert_not_paused(deps.storage)?;
f(deps, env, info)
}
Python Agent 生命周期管理
from enum import Enum
from dataclasses import dataclass
class AgentStatus(Enum):
CREATED = "created"
ACTIVE = "active"
PAUSED = "paused"
SHUTTING_DOWN = "shutting_down"
TERMINATED = "terminated"
COMPROMISED = "compromised"
@dataclass
class AgentState:
status: AgentStatus
active_since: int
last_action: int
total_actions: int = 0
class AgentLifecycleManager:
def __init__(self, wallet):
self.wallet = wallet
self.state = AgentState(status=AgentStatus.CREATED, active_since=0, last_action=0)
async def start(self) -> bool:
if self.state.status != AgentStatus.CREATED:
return False
msg = {"activate_agent": {"timestamp": int(time.time())}}
result = await self.wallet.contract_call(
contract=self.wallet.contract_address, msg=msg
)
if result.get('status') == 'success':
self.state.status = AgentStatus.ACTIVE
self.state.active_since = int(time.time())
return True
return False
async def emergency_shutdown(self) -> bool:
if self.state.status in [AgentStatus.TERMINATED, AgentStatus.COMPROMISED]:
return False
await self.wallet.contract_call(
contract=self.wallet.contract_address,
msg={"pause": {"reason": "emergency_shutdown"}}
)
# 清算持仓, 返回资金...
await self.wallet.contract_call(
contract=self.wallet.contract_address,
msg={"terminate_agent": {"reason": "emergency"}}
)
self.state.status = AgentStatus.TERMINATED
return True
4. 自动审计工具
4.1 Agent 合约审计器架构
import json
import hashlib
import subprocess
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Dict, Any, Callable
class Severity(Enum):
CRITICAL = "CRITICAL"
HIGH = "HIGH"
MEDIUM = "MEDIUM"
LOW = "LOW"
INFO = "INFO"
PASS = "PASS"
@dataclass
class AuditFinding:
check_id: str
title: str
description: str
severity: Severity
file_location: Optional[str] = None
line_number: Optional[int] = None
code_snippet: Optional[str] = None
recommendation: Optional[str] = None
@dataclass
class AuditReport:
contract_name: str
contract_version: str
chain_id: str = "msg-chain-1"
date: str = ""
findings: List[AuditFinding] = field(default_factory=list)
@property
def total_findings(self) -> int:
return len([f for f in self.findings if f.severity != Severity.PASS])
@property
def critical_count(self) -> int:
return len([f for f in self.findings if f.severity == Severity.CRITICAL])
@property
def high_count(self) -> int:
return len([f for f in self.findings if f.severity == Severity.HIGH])
@property
def medium_count(self) -> int:
return len([f for f in self.findings if f.severity == Severity.MEDIUM])
@property
def low_count(self) -> int:
return len([f for f in self.findings if f.severity == Severity.LOW])
@property
def score(self) -> int:
if self.total_findings == 0:
return 100
deductions = {
Severity.CRITICAL: 30,
Severity.HIGH: 15,
Severity.MEDIUM: 7,
Severity.LOW: 3,
}
total = sum(deductions.get(f.severity, 0) for f in self.findings if f.severity != Severity.PASS)
return max(0, min(100, 100 - total))
def to_markdown(self) -> str:
lines = [
"# Security Audit Report",
"",
f"**Contract**: {self.contract_name} v{self.contract_version}",
f"**Chain**: {self.chain_id}",
f"**Date**: {self.date}",
"",
"## Summary",
"",
f"- Total findings: **{self.total_findings}**",
f"- Critical: **{self.critical_count}**",
f"- High: **{self.high_count}**",
f"- Medium: **{self.medium_count}**",
f"- Low: **{self.low_count}**",
f"- Score: **{self.score}/100**",
"",
]
for f in self.findings:
if f.severity != Severity.PASS:
lines.extend([
f"### [{f.severity.value}] {f.title}",
f"**ID**: {f.check_id}",
f"**Description**: {f.description}",
f"**Line**: {f.line_number or 'N/A'}",
"",
])
if f.recommendation:
lines.extend([f"**Fix**: {f.recommendation}", ""])
return "\n".join(lines)
class AgentAuditor:
def __init__(self, verbose: bool = False):
self.verbose = verbose
self.findings: List[AuditFinding] = []
self.check_registry: Dict[str, Callable] = {
'reentrancy': self.check_reentrancy,
'integer_overflow': self.check_integer_overflow,
'access_control': self.check_access_control,
'spending_limits': self.check_spending_limits,
'constitution_enforcement': self.check_constitution_enforcement,
'input_validation': self.check_input_validation,
'error_handling': self.check_error_handling,
'event_emission': self.check_event_emission,
'gas_optimization': self.check_gas_optimization,
'oracle_integrity': self.check_oracle_integrity,
'a2a_authentication': self.check_a2a_auth,
'key_management': self.check_key_management,
'cross_contract_safety': self.check_cross_contract_safety,
}
def log(self, msg: str, level: str = "INFO"):
if self.verbose:
print(f"[{level}] {msg}")
def run_static_analysis(self, wasm_path: str) -> List[AuditFinding]:
findings = []
try:
result = subprocess.run(
["cosmwasm-check", wasm_path],
capture_output=True, text=True, timeout=60
)
if result.returncode != 0:
findings.append(AuditFinding(
check_id="STATIC-001",
title="cosmwasm-check failed",
description=result.stderr,
severity=Severity.HIGH,
))
else:
findings.append(AuditFinding(
check_id="STATIC-001",
title="cosmwasm-check passed",
description="static analysis passed",
severity=Severity.PASS,
))
except FileNotFoundError:
findings.append(AuditFinding(
check_id="STATIC-001",
title="cosmwasm-check not installed",
description="install: cargo install cosmwasm-check",
severity=Severity.INFO,
))
return findings
def check_reentrancy(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
lines = content.split('\n')
for i, line in enumerate(lines):
if 'add_message' in line and ('BankMsg' in line or 'WasmMsg' in line):
for j in range(i + 1, min(i + 15, len(lines))):
if 'save(' in lines[j] or 'update(' in lines[j]:
findings.append(AuditFinding(
check_id="RE-001",
title="Reentrancy: external call before state update",
description=f"line {i+1}: add_message before save at line {j+1}",
severity=Severity.CRITICAL,
line_number=i + 1,
recommendation="Apply Checks-Effects-Interactions pattern"
))
break
if not findings:
findings.append(AuditFinding(check_id="RE-001", title="No reentrancy found",
description="CEI pattern followed", severity=Severity.PASS))
return findings
def check_integer_overflow(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if 'checked_add' not in content and 'checked_sub' not in content:
if any(op in content for op in ['Uint128', 'Decimal']):
findings.append(AuditFinding(
check_id="IO-001",
title="No checked arithmetic found",
description="use checked_add/checked_sub instead of raw +/-",
severity=Severity.HIGH,
recommendation="Use checked_add(), checked_sub(), checked_mul()"
))
if not findings:
findings.append(AuditFinding(check_id="IO-001", title="Overflow checks passed",
severity=Severity.PASS))
return findings
def check_access_control(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
critical_fns = ['withdraw_all', 'set_spending_limit', 'update_constitution', 'terminate']
for fn in critical_fns:
if fn in content:
fn_idx = content.find(fn)
snippet = content[fn_idx:fn_idx + 1200]
if not any(p in snippet for p in ['assert_admin', 'OWNER', 'ADMIN', 'require_auth']):
findings.append(AuditFinding(
check_id="AC-001",
title=f"Missing access control: {fn}",
description=f"no authorization check found",
severity=Severity.CRITICAL,
recommendation=f"Add ADMIN.assert_admin() to {fn}"
))
if not findings:
findings.append(AuditFinding(check_id="AC-001", title="Access control passed",
severity=Severity.PASS))
return findings
def check_spending_limits(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if 'spending_limit' not in content.lower() and 'spend_limit' not in content.lower():
if 'BankMsg' in content:
findings.append(AuditFinding(
check_id="SL-001",
title="No spending limits",
description="contract has transfers but no spending limits",
severity=Severity.HIGH,
recommendation="Implement multi-level spending limits"
))
if not findings:
findings.append(AuditFinding(check_id="SL-001", title="Spending limit checks passed",
severity=Severity.PASS))
return findings
def check_constitution_enforcement(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if 'constitution' not in content.lower():
findings.append(AuditFinding(
check_id="CE-001",
title="No constitution mechanism",
description="agent lacks constitutional constraints",
severity=Severity.HIGH,
recommendation="Implement Constitution struct with pre-execution validation"
))
else:
if 'validate_trade' not in content and 'validate_action' not in content:
findings.append(AuditFinding(
check_id="CE-002",
title="Constitution not enforced",
description="defined but not validated before actions",
severity=Severity.CRITICAL,
recommendation="Call constitution.validate() in each execute_* function"
))
return findings
def check_input_validation(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if 'addr_validate' not in content and 'Addr::unchecked' in content:
findings.append(AuditFinding(
check_id="IV-001",
title="Missing address validation",
description="use addr_validate() instead of Addr::unchecked()",
severity=Severity.MEDIUM,
recommendation="Use deps.api.addr_validate() for user-supplied addresses"
))
if not findings:
findings.append(AuditFinding(check_id="IV-001", title="Input validation passed",
severity=Severity.PASS))
return findings
def check_error_handling(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
unwrap_count = content.count('.unwrap()')
if unwrap_count > 0:
findings.append(AuditFinding(
check_id="EH-001",
title=f"{unwrap_count} unwrap() calls found",
description="unwrap() panics on error, use ? instead",
severity=Severity.MEDIUM,
recommendation="Replace unwrap() with ? operator"
))
return findings
def check_event_emission(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
attrs = content.count('add_attribute')
if attrs < 3:
findings.append(AuditFinding(
check_id="EE-001",
title=f"Only {attrs} add_attribute calls",
description="insufficient event emission for monitoring",
severity=Severity.MEDIUM,
recommendation="Add attributes to all state-changing operations"
))
return findings
def check_gas_optimization(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if '.range(' in content and 'limit' not in content and 'MAX_BATCH' not in content:
findings.append(AuditFinding(
check_id="GO-001",
title="Unbounded range iteration",
description="map range without limit may exhaust gas",
severity=Severity.MEDIUM,
recommendation="Add limit parameter to range() calls"
))
return findings
def check_oracle_integrity(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if 'oracle' in content.lower() or 'price' in content.lower():
if 'twap' not in content.lower() and 'aggregated' not in content.lower():
findings.append(AuditFinding(
check_id="OI-001",
title="Single oracle source",
description="use multi-source aggregation + TWAP",
severity=Severity.HIGH,
recommendation="Implement multi-oracle aggregation with median price"
))
return findings
def check_a2a_auth(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if 'handle_agent_message' in content:
if not any(p in content for p in ['signature', 'Signature', 'nonce', 'Nonce']):
findings.append(AuditFinding(
check_id="A2A-001",
title="A2A messages lack authentication",
description="no signature verification in handle_agent_message",
severity=Severity.CRITICAL,
recommendation="Implement DID signature + nonce replay protection"
))
return findings
def check_key_management(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if 'private_key' in content.lower() or 'secret' in content.lower():
if 'env' not in content.lower():
findings.append(AuditFinding(
check_id="KM-001",
title="Possible hardcoded key",
description="source contains private_key/secret keyword",
severity=Severity.CRITICAL,
recommendation="Use environment variables or HSM"
))
return findings
def check_cross_contract_safety(self, source_path: str) -> List[AuditFinding]:
findings = []
with open(source_path, 'r') as f:
content = f.read()
if 'WasmMsg::Execute' in content and 'whitelist' not in content.lower():
findings.append(AuditFinding(
check_id="CC-001",
title="No contract whitelist",
description="WasmMsg::Execute without target verification",
severity=Severity.CRITICAL,
recommendation="Implement contract whitelist with code hash verification"
))
return findings
async def audit_contract(self, wasm_path: str, source_path: Optional[str] = None) -> AuditReport:
self.log(f"Auditing: {wasm_path}")
# Static analysis
static_results = self.run_static_analysis(wasm_path)
self.findings.extend(static_results)
# Source analysis
if source_path:
for name, check_fn in self.check_registry.items():
try:
results = check_fn(source_path)
self.findings.extend(results)
self.log(f" {name}: {'PASS' if not [r for r in results if r.severity != Severity.PASS] else 'ISSUES'}")
except Exception as e:
self.log(f" {name}: ERROR - {e}", "ERROR")
report = AuditReport(
contract_name=wasm_path.split('/')[-1].replace('.wasm', ''),
contract_version="1.0",
date=time.strftime("%Y-%m-%d"),
findings=self.findings,
)
self.log(f"Audit complete. Score: {report.score}/100")
return report
4.2 安全评分计算
def calculate_security_score(findings: List[AuditFinding]) -> int:
score = 100
for finding in findings:
if finding.severity == Severity.CRITICAL:
score -= 30
elif finding.severity == Severity.HIGH:
score -= 15
elif finding.severity == Severity.MEDIUM:
score -= 7
elif finding.severity == Severity.LOW:
score -= 3
return max(0, score)
def severity_label(score: int) -> str:
if score >= 90:
return "SAFE"
elif score >= 75:
return "MODERATE"
elif score >= 60:
return "NEEDS_IMPROVEMENT"
else:
return "DANGEROUS"
5. 审计清单表
5.1 完整审计清单
# AI Agent 智能合约安全审计清单
# 链: msg-chain-1 | Bech32: msg
## A. 访问控制 (Access Control)
### A.1 管理员权限
[ ] Admin functions require proper authorization
[ ] Agent constitution is immutable after deployment
[ ] Owner can be transferred (not renounced without backup)
[ ] Capability registration validates caller
[ ] Role-based access control implemented
[ ] No backdoor admin functions
[ ] Admin key is multi-sig or time-locked
[ ] Pause/unpause requires multi-signature
### A.2 权限分级
[ ] Distinguish between Admin/Operator/Auditor roles
[ ] Read-only functions accessible by anyone
[ ] Sensitive actions logged with actor address
[ ] Minimum privilege principle applied
[ ] Role hierarchy prevents privilege escalation
### A.3 宪法权限
[ ] Constitution only modified by multi-sig
[ ] Constitution changes emit events
[ ] Constitution versioning and history preserved
[ ] Emergency override requires super-majority
[ ] Guardian set rotation is rate-limited
## B. 金融安全 (Financial Safety)
### B.1 支出限额
[ ] Spending limits enforced per transaction
[ ] Spending limits enforced per day
[ ] Spending limits enforced per month
[ ] Limits are cumulative across all action types
[ ] Whitelist addresses can bypass limits (optional)
[ ] Limit bypass events are logged
[ ] Spender cannot modify own limit
[ ] Limit reset logic is secure
### B.2 资金安全
[ ] No unbounded loops with external calls
[ ] Payment splitting uses checked math
[ ] Staking/slashing has proper validation
[ ] No hidden fees or balance manipulation
[ ] Contract cannot receive unsupported tokens
[ ] Withdraw pattern requires explicit request
[ ] Balance tracking is consistent
[ ] Dust amounts handled correctly
### B.3 代币操作
[ ] CW20 transfer return values checked
[ ] CW20 allowance race condition prevented
[ ] Native coin amounts verified
[ ] Token contract addresses are validated
[ ] IBC transfer timeout is reasonable
[ ] IBC ACK/timeout handlers are implemented
[ ] Cross-chain replay protection
## C. Agent 安全 (Agent Security)
### C.1 宪法执行
[ ] Constitution cannot be bypassed
[ ] All actions validated against constitution before execution
[ ] Constitution integrity checked (hash match)
[ ] Constitution has a circuit breaker
[ ] Risk level constraints enforced
[ ] Asset whitelist enforced
[ ] Position size limit enforced
[ ] Portfolio exposure limit enforced
### C.2 通信安全
[ ] A2A messages are authenticated (DID signatures)
[ ] Nonce prevents replay attacks
[ ] Message timestamps validated
[ ] Sender DID is verified and not revoked
[ ] Cross-contract calls use whitelist
[ ] Callback functions validate caller
### C.3 密钥管理
[ ] Key rotation is logged on chain
[ ] No hardcoded keys in source
[ ] Multi-sig for key rotation
[ ] Compromised key revocation mechanism
[ ] Key expiry enforced
[ ] Key history maintained for audit
[ ] Private key never exposed in events/tx data
### C.4 生命周期
[ ] Agent shutdown mechanism exists
[ ] Emergency pause works correctly
[ ] Funds can be recovered in emergency
[ ] Agent cannot be permanently killed by single actor
[ ] Graceful shutdown processes pending actions
[ ] Shutdown emits events for monitoring
[ ] Restart after shutdown is possible with proper auth
## D. 数据验证 (Data Validation)
### D.1 输入验证
[ ] All user inputs are validated
[ ] Addresses validated via addr_validate()
[ ] Amounts checked for zero and overflow
[ ] JSON parsing handles malformed data gracefully
[ ] String lengths bounded
[ ] Array lengths bounded
[ ] Enum values validated
[ ] Integer ranges validated
### D.2 预言机数据
[ ] External oracle data is validated
[ ] Multiple oracle sources used (>= 3)
[ ] Oracle staleness checked
[ ] Oracle price deviation limits enforced
[ ] TWAP used for DeFi operations
[ ] Oracle update frequency is rate-limited
[ ] Median (not average) price calculation
[ ] Flashloan-resistant price feeds
### D.3 合约状态
[ ] State invariants validated before/after actions
[ ] Storage versioning for migration support
[ ] State corruption cannot suspend funds
[ ] Redundant state checks for critical values
[ ] Deserialization validation for stored data
## E. 编码安全 (Code Security)
### E.1 重放保护
[ ] Checks-Effects-Interactions pattern followed
[ ] Reentrancy guard implemented if needed
[ ] No cross-function reentrancy
[ ] No cross-contract reentrancy
[ ] External calls made after state changes
### E.2 整数安全
[ ] checked_add/sub/mul/div used instead of raw operators
[ ] Decimal operations use checked methods
[ ] No truncation in division
[ ] No unsafe type conversions
[ ] Multiplication before division to preserve precision
### E.3 错误处理
[ ] All Result types handled
[ ] No unwrap() in production code
[ ] Specific error messages returned
[ ] Error conditions cannot lead to inconsistent state
[ ] Proper use of ? operator
### E.4 气体限制
[ ] No unbounded loops
[ ] Map range operations limited
[ ] Batch processing for large datasets
[ ] Gas costs predictable
[ ] External call gas limits set
## F. 日志与事件 (Logging & Events)
[ ] All state changes emit events
[ ] Events include all relevant parameters
[ ] Events indexed for efficient querying
[ ] Security-sensitive events have severity field
[ ] Failed operations emit failure events
[ ] All admin actions logged
[ ] Spending limit changes logged
[ ] Constitution changes logged with before/after
[ ] Key rotation history preserved
[ ] Pause/unpause events logged
[ ] Failed auth attempts logged
[ ] Oracle price deviations logged
## G. 跨链安全 (Cross-Chain Security)
[ ] IBC channels validated
[ ] IBC packet timeout set
[ ] IBC packet replay protection
[ ] IBC fee handling verified
[ ] IBC memos validated
[ ] ICS-20 transfer amount validated
[ ] Channel handshake validated
[ ] Bridge validators are decentralized
[ ] Cross-chain message ordering enforced
[ ] Relayer incentives aligned
[ ] Finality delay enforced
[ ] Fraud proof mechanism exists
## H. 依赖安全 (Dependency Security)
[ ] All dependencies audited
[ ] No deprecated packages
[ ] Lock file present and consistent
[ ] Known CVEs checked
[ ] Dependency versions pinned
[ ] Supply chain attack mitigations
[ ] Minimal dependency count
## I. 部署配置 (Deployment Configuration)
[ ] Admin address set correctly on instantiate
[ ] Agent contract labeled on chain
[ ] Code ID verified
[ ] Instantiate permissions restricted
[ ] Migrate function has access control
[ ] Contract cannot self-destruct unexpectedly
[ ] No uninitialized storage paths
[ ] Storage key prefixes avoid collisions
## J. 渗透测试结果 (Penetration Test Results)
[ ] Agent impersonation test passed
[ ] Spending limit bypass test passed
[ ] Constitution violation test passed
[ ] Oracle manipulation test passed
[ ] Reentrancy attack test passed
[ ] Replay attack test passed
[ ] Gas exhaustion test passed
[ ] Front-running resistance test passed
[ ] Input validation fuzz test passed
5.2 严重性分类标准
| 级别 | 分数 | 定义 | 示例 |
|---|---|---|---|
| CRITICAL | 9-10 | 可直接导致资金损失或合约完全控制权丢失 | 重入漏洞、访问控制缺失 |
| HIGH | 7-8 | 可能导致资金损失或关键功能异常 | 整数溢出、限额绕过 |
| MEDIUM | 4-6 | 在特定条件下可能导致安全问题 | 缺少事件、气体优化 |
| LOW | 1-3 | 影响代码质量和可维护性 | 代码风格、缺少注释 |
| INFO | 0 | 提供信息，无安全影响 | 合约大小、依赖版本 |
5.3 修复优先级指引
立即修复 (P0): 严重 + 高危漏洞
- 在部署前必须修复
- 已部署合约应在 24 小时内升级
尽快修复 (P1): 中危漏洞
- 在下次升级时修复
- 如果影响资金安全，应在 7 天内修复
择机修复 (P2): 低危漏洞
- 在下一次代码迭代时修复
- 不影响当前安全状态
记录关注 (P3): 信息类
- 无需修复，但应持续监控
- 可能在未来版本中优化
6. 渗透测试场景
6.1 Agent 冒充测试
"""
测试场景: Agent 冒充攻击
目标: 测试 Agent 合约是否能抵御身份伪造
"""
import asyncio
class AgentImpersonationTest:
async def test_impersonation(self, agent_contract: str, wallet) -> dict:
results = {
'test_name': 'Agent Impersonation Attack',
'status': 'PASS',
'findings': [],
}
# 测试 1: 伪造 from_agent 参数
print("[TEST 1] Forging sender address...")
try:
msg = {
"handle_agent_message": {
"from_agent": "msg1legitimate...",
"action": "withdraw_all",
"params": {},
}
}
await wallet.contract_call(contract=agent_contract, msg=msg)
results['findings'].append({
'test': '1.1',
'description': 'Contract accepted forged from_agent',
'severity': 'CRITICAL',
'impact': 'Attacker can impersonate any agent'
})
results['status'] = 'FAIL'
except Exception as e:
print(f" Correctly rejected: {e}")
# 测试 2: 伪造 DID 签名
print("[TEST 2] Forging signature...")
try:
msg = {
"handle_agent_message": {
"from_did": "msg1victim...",
"to_did": agent_contract,
"action": "transfer_funds",
"params": {"amount": "1000000"},
"nonce": 99999,
"timestamp": int(time.time()),
"signature": "0x" + "aa" * 32,
}
}
await wallet.contract_call(contract=agent_contract, msg=msg)
results['findings'].append({
'test': '2.1',
'description': 'Contract accepted forged signature',
'severity': 'CRITICAL',
})
results['status'] = 'FAIL'
except Exception as e:
print(f" Correctly rejected: {e}")
# 测试 3: 重放攻击
print("[TEST 3] Replay attack...")
try:
# Capture a legitimate message and replay it
valid_msg = await self.capture_legitimate_message(agent_contract)
# Replay in a new block
await wallet.contract_call(contract=agent_contract, msg=valid_msg)
results['findings'].append({
'test': '3.1',
'description': 'Replay attack succeeded (nonce not enforced)',
'severity': 'HIGH',
})
results['status'] = 'FAIL'
except Exception as e:
print(f" Replay correctly rejected: {e}")
return results
6.2 支出限额绕过测试
class SpendingLimitBypassTest:
async def test_limit_bypass(self, agent_contract: str, wallet) -> dict:
results = {'test_name': 'Spending Limit Bypass', 'status': 'PASS', 'findings': []}
# 测试 1: 大额直接转账
print("[TEST 1] Large direct transfer...")
try:
msg = {"execute_action": {"type": "pay", "amount": "999999999999", "recipient": "msg1attacker..."}}
await wallet.contract_call(contract=agent_contract, msg=msg)
results['findings'].append({'test': '1.1', 'description': 'Large transfer not limited', 'severity': 'CRITICAL'})
results['status'] = 'FAIL'
except Exception as e:
print(f" Correctly rejected: {e}")
# 测试 2: 分多次小额的累积转账
print("[TEST 2] Accumulating small transfers...")
total_sent = 0
try:
for i in range(100):
msg = {"execute_action": {"type": "pay", "amount": "100000000", "recipient": f"msg1receiver{i}..."}}
await wallet.contract_call(contract=agent_contract, msg=msg)
total_sent += 100_000_000
results['findings'].append({
'test': '2.1',
'description': f'Accumulated {total_sent} via small transfers',
'severity': 'HIGH',
})
results['status'] = 'FAIL'
except Exception as e:
print(f" Daily limit enforced at {total_sent}: {e}")
# 测试 3: 白名单地址滥用
print("[TEST 3] Whitelist abuse...")
try:
msg = {"execute_action": {"type": "pay", "amount": "1000000000000", "recipient": "msg1whitelisted..."}}
await wallet.contract_call(contract=agent_contract, msg=msg)
# Check if whitelist is limited to specific types of transfers
results['findings'].append({
'test': '3.1',
'description': 'Whitelist address allowed unlimited withdrawal',
'severity': 'MEDIUM',
})
except Exception as e:
print(f" Correctly limited: {e}")
return results
6.3 宪法违规测试
class ConstitutionViolationTest:
async def test_constitution_violations(self, agent_contract: str, wallet) -> dict:
results = {'test_name': 'Constitution Violation', 'status': 'PASS', 'findings': []}
# 测试 1: 超额头寸
print("[TEST 1] Exceeding max position size...")
try:
msg = {"execute_trade": {"asset_in": "umsg", "asset_out": "usdt", "amount": "999999999999999"}}
await wallet.contract_call(contract=agent_contract, msg=msg)
results['findings'].append({'test': '1.1', 'description': 'Trade exceeded max position', 'severity': 'CRITICAL'})
results['status'] = 'FAIL'
except Exception as e:
print(f" Correctly rejected: {e}")
# 测试 2: 非白名单资产交易
print("[TEST 2] Trading non-whitelisted asset...")
try:
msg = {"execute_trade": {"asset_in": "uhack", "asset_out": "umsg", "amount": "1000"}}
await wallet.contract_call(contract=agent_contract, msg=msg)
results['findings'].append({'test': '2.1', 'description': 'Non-whitelisted asset trade allowed', 'severity': 'HIGH'})
results['status'] = 'FAIL'
except Exception as e:
print(f" Correctly rejected: {e}")
# 测试 3: 超额敞口
print("[TEST 3] Exceeding risk exposure...")
try:
msg = {"execute_trade": {"asset_in": "umsg", "asset_out": "uatom", "amount": "500000000000000"}}
await wallet.contract_call(contract=agent_contract, msg=msg)
results['findings'].append({'test': '3.1', 'description': 'Trade exceeded risk limits', 'severity': 'HIGH'})
results['status'] = 'FAIL'
except Exception as e:
print(f" Correctly rejected: {e}")
return results
6.4 预言机操控测试
class OracleManipulationTest:
async def test_oracle_manipulation(self, agent_contract: str, wallet) -> dict:
results = {'test_name': 'Oracle Price Manipulation', 'status': 'PASS', 'findings': []}
# 测试 1: 通过闪电贷操控价格
print("[TEST 1] Flash loan price manipulation...")
try:
# Simulate a large swap to move price
msg = {"simulate_price_impact": {"amount": "1000000000000000", "direction": "buy"}}
manipulated_price = await wallet.contract_query(contract=agent_contract, msg=msg)
# Then execute trade at manipulated price
msg = {"execute_trade": {"asset_in": "umsg", "asset_out": "uatom", "amount": "1000000"}}
await wallet.contract_call(contract=agent_contract, msg=msg)
results['findings'].append({
'test': '1.1',
'description': 'Trade executed at manipulated price (no TWAP)',
'severity': 'CRITICAL',
})
results['status'] = 'FAIL'
except Exception as e:
print(f" TWAP protection works: {e}")
# 测试 2: 陈旧预言机数据
print("[TEST 2] Stale oracle data...")
try:
msg = {"simulate_stale_oracle": {"hours_old": 48}}
result = await wallet.contract_call(contract=agent_contract, msg=msg)
if result.get('accepted'):
results['findings'].append({
'test': '2.1',
'description': 'Accepted 48-hour-old oracle data',
'severity': 'HIGH',
})
results['status'] = 'FAIL'
except Exception as e:
print(f" Staleness check works: {e}")
# 测试 3: 异常偏差价格
print("[TEST 3] Extreme price deviation...")
try:
msg = {"submit_oracle_price": {"source": "malicious", "price": "999999.0", "confidence": 100}}
await wallet.contract_call(contract=agent_contract, msg=msg)
results['findings'].append({
'test': '3.1',
'description': 'Accepted outlier price without deviation check',
'severity': 'HIGH',
})
results['status'] = 'FAIL'
except Exception as e:
print(f" Deviation check works: {e}")
return results
6.5 气体耗尽测试
class GasExhaustionTest:
async def test_gas_exhaustion(self, agent_contract: str, wallet) -> dict:
results = {'test_name': 'Gas Exhaustion Attack', 'status': 'PASS', 'findings': []}
# 测试 1: 批量操作耗尽气体
print("[TEST 1] Batch processing gas exhaustion...")
try:
# Submit many items to be processed
for i in range(1000):
msg = {"submit_order": {"id": i, "data": "x" * 1000}}
await wallet.contract_call(contract=agent_contract, msg=msg)
# Try to process all at once
start_gas = wallet.get_gas_used()
try:
msg = {"process_all_orders": {}}
await wallet.contract_call(contract=agent_contract, msg=msg, gas_limit=5_000_000)
results['findings'].append({
'test': '1.1',
'description': 'Processing all orders succeeded (may be gas-inefficient)',
'severity': 'MEDIUM',
})
except Exception:
print(f" Gas limit correctly prevents bulk processing")
results['findings'].append({
'test': '1.1',
'description': 'Batch processing correctly limited',
'severity': 'PASS',
})
except Exception as e:
print(f" Order submission failed: {e}")
# 测试 2: 无限循环
print("[TEST 2] Infinite loop protection...")
try:
msg = {"start_infinite_loop": {}}
await wallet.contract_call(contract=agent_contract, msg=msg, gas_limit=10_000_000)
results['findings'].append({
'test': '2.1',
'description': 'No infinite loop protection detected',
'severity': 'HIGH',
})
results['status'] = 'FAIL'
except Exception:
print(f" Infinite loop correctly blocked")
return results
6.6 测试执行器
async def run_all_penetration_tests(agent_contract: str, wallet) -> dict:
"""运行所有渗透测试"""
print("=" * 60)
print(" AI AGENT CONTRACT PENETRATION TEST SUITE")
print(f" Contract: {agent_contract}")
print(f" Chain: msg-chain-1")
print("=" * 60)
tests = [
("Agent Impersonation", AgentImpersonationTest().test_impersonation),
("Spending Limit Bypass", SpendingLimitBypassTest().test_limit_bypass),
("Constitution Violation", ConstitutionViolationTest().test_constitution_violations),
("Oracle Manipulation", OracleManipulationTest().test_oracle_manipulation),
("Gas Exhaustion", GasExhaustionTest().test_gas_exhaustion),
]
all_results = {}
overall_status = "PASS"
for name, test_fn in tests:
print(f"\n--- Running: {name} ---")
try:
result = await test_fn(agent_contract, wallet)
all_results[name] = result
if result['status'] == 'FAIL':
overall_status = 'FAIL'
for finding in result['findings']:
if finding['severity'] in ['CRITICAL', 'HIGH']:
print(f" !! {finding['severity']}: {finding['description']}")
except Exception as e:
print(f" Test error: {e}")
all_results[name] = {'status': 'ERROR', 'error': str(e)}
print("\n" + "=" * 60)
print(f" OVERALL STATUS: {overall_status}")
print("=" * 60)
return {
'overall_status': overall_status,
'tests': all_results,
'summary': {
'total': len(tests),
'passed': sum(1 for r in all_results.values() if r.get('status') == 'PASS'),
'failed': sum(1 for r in all_results.values() if r.get('status') == 'FAIL'),
}
}
7. 审计报告模板
7.1 标准审计报告模板
# 安全审计报告: AI Agent 智能合约
## 基本信息
| 项目 | 内容 |
|------|------|
| 合约名称 | AgentName |
| 合约版本 | v1.0.0 |
| 链 | msg-chain-1 |
| Bech32 前缀 | msg |
| 审计日期 | 2026-07-07 |
| 审计方法 | 静态分析 + 自动化扫描 + 渗透测试 |
| 审计工具 | AgentAuditor v1.0, cosmwasm-check |
| 代码行数 | 1,234 |
## 审计总结
- 总发现: **3**
- 严重: **0**
- 高危: **1**
- 中危: **1**
- 低危: **1**
- 安全评分: **82/100**
- 安全等级: **较安全**
## 分项评分
| 类别 | 分数 | 状态 |
|------|------|------|
| 访问控制 | 95/100 | ✅ 安全 |
| 金融安全 | 80/100 | ✅ 较安全 |
| 宪法执行 | 85/100 | ✅ 较安全 |
| 通信安全 | 70/100 | ⚠️ 需改进 |
| 密钥管理 | 90/100 | ✅ 安全 |
| 数据验证 | 85/100 | ✅ 较安全 |
| 编码安全 | 78/100 | ⚠️ 需改进 |
| 日志与事件 | 75/100 | ⚠️ 需改进 |
## 发现详情
### [HIGH] A2A-001: A2A 消息缺少 Nonce 防重放
- **文件**: src/contract.rs:245
- **描述**: `handle_agent_message` 函数使用了 DID 签名验证，但未实现 nonce 或时间戳检查，允许重放攻击
- **严重性**: HIGH (CVSS 7.2)
- **影响**: 攻击者可重放截获的合法 A2A 消息，导致重复执行 action
**代码片段**:
```rust
// ❌ 问题代码
pub fn handle_agent_message(
deps: DepsMut,
env: Env,
info: MessageInfo,
msg: SignedAgentMessage,
) -> StdResult<Response> {
// 验证了签名但未检查 nonce
verify_signature(&msg)?;
execute_action(deps, env, msg.action, msg.params)
}
修复建议:
// ✅ 修复: 添加 nonce 检查
pub fn handle_agent_message_fixed(
deps: DepsMut,
env: Env,
info: MessageInfo,
msg: SignedAgentMessage,
) -> StdResult<Response> {
verify_signature(&msg)?;
// 检查 nonce 防重放
let stored_nonce = NONCES
.may_load(deps.storage, &msg.from_did)?
.unwrap_or(0);
if msg.nonce <= stored_nonce {
return Err(StdError::generic_err("replay attack detected"));
}
NONCES.save(deps.storage, &msg.from_did, &msg.nonce)?;
// 检查时间戳
let max_age = env.block.time.seconds() - msg.timestamp;
if max_age > 300 {
return Err(StdError::generic_err("message expired"));
}
execute_action(deps, env, msg.action, msg.params)
}
[MEDIUM] AC-002: 角色权限提升路径
- 文件: src/auth.rs:89
- 描述: Operator 角色可以通过调用
register_operator()将自身升级为 Admin - 严重性: MEDIUM (CVSS 5.1)
- 影响: 被攻陷的 Operator 密钥可导致完全控制权丢失
修复建议:
register_operator()只允许 Admin 调用- 移除 Operator 的自我升级能力
- Admin 变更需多签
[LOW] EH-001: 使用 unwrap() 而非 ? 运算符
- 文件: src/helpers.rs:12, src/contract.rs:67, src/contract.rs:134
- 描述: 共检测到 3 处
.unwrap()调用，在 release 模式下会导致 panic - 严重性: LOW (CVSS 2.5)
- 影响: 特定输入可导致合约 panic 和状态回滚
- 修复建议: 将
.unwrap()替换为?运算符，或使用.map_err()提供具体错误
渗透测试结果
| 测试项目 | 结果 | 备注 |
|---|---|---|
| Agent 冒充测试 | ✅ PASS | DID 签名验证有效 |
| 支出限额绕过测试 | ✅ PASS | 多层限额成功阻止 |
| 宪法违规测试 | ✅ PASS | 位置限制和资产白名单生效 |
| 预言机操控测试 | ⚠️ WARN | TWAP 已实现但精度不足 |
| 重入攻击测试 | ✅ PASS | Checks-Effects-Interactions 合规 |
| 重放攻击测试 | ❌ FAIL | 见 A2A-001 发现 |
| 气体耗尽测试 | ✅ PASS | 分批处理有效 |
修复计划
| 优先级 | 发现 | 预计工作量 | 状态 |
|---|---|---|---|
| P0 | 无 | - | - |
| P1 | A2A-001: 添加 nonce 防重放 | 2 小时 | 待修复 |
| P1 | AC-002: 修复权限提升 | 1 小时 | 待修复 |
| P2 | EH-001: 替换 unwrap() | 0.5 小时 | 待修复 |
声明
本审计报告基于提供的源代码和 WASM 二进制文件进行分析。
审计不保证完全消除所有潜在风险，建议定期重新审计。
审计日期: 2026-07-07
审计员: AgentAuditor v1.0
## 7.2 快速审计摘要模板
```markdown
# 快速审计摘要
合约: AgentName v1.0.0 | 链: msg-chain-1
日期: 2026-07-07
## 评分: 82/100 — 较安全
### 发现汇总
严重: 0 | 高危: 1 | 中危: 1 | 低危: 1
### 关键修复
1. [HIGH] A2A 消息缺少 nonce 防重放 → 添加 nonce 检查
2. [MEDIUM] Operator 可升级为 Admin → 移除自我升级能力
3. [LOW] 使用 unwrap() → 替换为 ?
### 部署建议
✅ 可部署 (修复 P1 事项后推荐)
7.3 自动化报告生成脚本
def generate_audit_report_pdf(report: AuditReport, output_path: str):
"""将审计报告保存为 Markdown 文件"""
with open(output_path, 'w') as f:
f.write(report.to_markdown())
print(f"Report saved to {output_path}")
def print_audit_summary(report: AuditReport):
"""打印审计摘要到控制台"""
print(f"""
{'='*50}
AI AGENT CONTRACT AUDIT SUMMARY
{'='*50}
Contract: {report.contract_name} v{report.contract_version}
Chain: {report.chain_id}
Date: {report.date}
{'='*50}
Score: {report.score}/100
Findings: {report.total_findings}
Critical: {report.critical_count}
High: {report.high_count}
Medium: {report.medium_count}
Low: {report.low_count}
{'='*50}
""")
async def main():
"""主函数: 运行审计"""
import sys
if len(sys.argv) < 2:
print("Usage: python auditor.py <wasm_path> [source_path]")
sys.exit(1)
wasm_path = sys.argv[1]
source_path = sys.argv[2] if len(sys.argv) > 2 else None
auditor = AgentAuditor(verbose=True)
report = await auditor.audit_contract(wasm_path, source_path)
print_audit_summary(report)
output_path = f"audit_report_{report.contract_name}_{report.date}.md"
generate_audit_report_pdf(report, output_path)
print(f"\nFull report: {output_path}")
附录: MSG Chain 合约部署检查清单
# 部署前检查
#!/bin/bash
# MSG Chain Agent Contract Pre-Deployment Checklist
echo "=== Pre-Deployment Checklist ==="
# 1. 编译检查
echo "[1/8] Checking compilation..."
cargo wasm 2>&1 || { echo "FAIL: compilation error"; exit 1; }
echo "PASS"
# 2. CosmWasm 静态检查
echo "[2/8] Running cosmwasm-check..."
cosmwasm-check target/wasm32-unknown-unknown/release/*.wasm 2>&1 || { echo "FAIL"; exit 1; }
echo "PASS"
# 3. 单元测试
echo "[3/8] Running unit tests..."
cargo test 2>&1 || { echo "FAIL"; exit 1; }
echo "PASS"
# 4. 代码大小检查
echo "[4/8] Checking wasm size..."
MAX_SIZE=800000 # 800KB max
WASM_SIZE=$(wc -c < target/wasm32-unknown-unknown/release/*.wasm)
if [ "$WASM_SIZE" -gt "$MAX_SIZE" ]; then
echo "WARN: wasm size ${WASM_SIZE}B exceeds ${MAX_SIZE}B"
fi
echo "OK (${WASM_SIZE}B)"
# 5. 依赖审计
echo "[5/8] Checking dependencies..."
cargo audit 2>&1 || echo "WARN: cargo audit not installed"
echo "OK"
# 6. 常量检查
echo "[6/8] Checking bech32 prefix..."
# Verify no hardcoded 'cosmos' prefix in source
if grep -r '"cosmos"' src/ 2>/dev/null; then
echo "FAIL: found cosmos prefix (should be msg)"
exit 1
fi
echo "PASS (all prefixes use msg)"
# 7. 审计清单验证
echo "[7/8] Audit checklist verification..."
# Count critical items
echo "Complete audit checklist verification manually"
echo "OK"
# 8. Gas 估算
echo "[8/8] Gas estimation..."
echo "Estimate: ~500000 gas for execute"
echo "OK"
echo "=== All checks complete ==="
文档信息
- 版本: 1.0
- 目标链: msg-chain-1
- Bech32 前缀:
msg- 语言: Rust + Python
相关工具
- cosmwasm-check: CosmWasm 静态验证
- cargo-audit: Rust 依赖安全检查
- AgentAuditor: 本指南配套的自动审计工具
本文档内容基于 MSGChain 代码库真实状态编写，非 AI 自动生成。
主网状态: No-Go | 白皮书: https://msgchain.org/whitepaper/
参考资源
- CosmWasm 文档: https://docs.cosmwasm.com
- MSG Chain 文档: https://docs.msgchain.io
- OWASP Smart Contract Top 10