AI Agent 合约安全审计自动化指南

— MSG Chain (msg-chain-1) CI/CD 驱动的审计流水线实践 —

版本: 1.0 | 链: msg-chain-1 | Bech32 前缀: msg
参考: AI Agent 智能合约安全审计清单指南.md — 该文档提供了漏洞模式和手动审计清单，本文档聚焦自动化体系

⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/

目录

1. 引言
2. 审计工具链概览
3. CI/CD 审计流水线
4. 模糊测试与属性测试
5. 形式化验证入门
6. AI Agent 特定安全风险自动化检测
7. 持续监控与运行时审计
8. 审计报告自动生成
9. 实践案例
10. 总结与建议

1. 引言

1.1 为什么需要审计自动化

AI Agent 智能合约运行在 MSG Chain (msg-chain-1) 上，负责自主执行策略、管理资产、与其他 Agent 进行 A2A 通信。与常规智能合约不同，Agent 合约具有以下特征使得手动审计不再足够：

根据 MSG Chain 官方白皮书的 developer_capability_matrix.json，MSG 链的合约运行时 (contract_runtime) 已达到 assisted_codegen 机器就绪等级，支持 AI 辅助代码生成与实际生产部署。这意味着合约代码可以由 LLM 生成后直接部署，审计自动化不再是可选项，而是安全底线。

1.2 自动化在 AI Agent 开发生命周期中的位置

开发阶段                   审计自动化介入点
─────────                 ─────────────────
LLM 生成合约代码  ──────►  1. 实时安全扫描 (pre-commit hook)
        │
  Rust 编译          ──────►  2. cargo-audit / cargo-deny 依赖检查
        │
  Unit Test          ──────►  3. cw-multi-test 属性测试 + 覆盖率
        │
  Integration Test   ──────►  4. test-tube 模糊测试 + gas 估算
        │
  CI 构建            ──────►  5. cosmwasm-check 静态验证
        │
  代码审查 (MR)      ──────►  6. 自动化门禁: 安全评分 ≥ 80
        │
  部署到 msg-chain-1 ──────►  7. 注册中心验证 + 运行时监控

1.3 与手动审计清单的关系

本文档不重复 AI Agent 智能合约安全审计清单指南.md 中的漏洞模式和手动检查项。两篇文档的关系是：

• 清单指南定义"检查什么"（漏洞模式、审计项、渗透场景）
• 自动化指南定义"怎么自动检查"（工具链、流水线、监控）

建议将清单指南中的每个审计项映射为自动化检查规则（下文称为"审计规则编码"），例如：

2. 审计工具链概览

2.1 CosmWasm 生态工具矩阵

MSG Chain 基于 CosmWasm 合约引擎，以下工具已在 msg-chain-1 开发生态中验证可用：

注意: MSG Chain 的 developer_capability_matrix 中，sdk_surface 当前为 local_candidate，agent_query_and_guarded_write 为 guarded_write。工具链主要依赖 contract_runtime (assisted_codegen) 和 rpc_gateway (assisted_codegen) 层的能力。

2.2 核心工具深入

2.2.1 cosmwasm-check

cosmwasm-check 是 CosmWasm 官方提供的 Wasm 静态验证工具。它对编译后的 .wasm 文件进行以下检查：

• Wasm 二进制格式正确性
• 导出函数签名 (instantiate, execute, query, reply, migrate, sudo)
• 导入函数合规性（未使用禁用 API）
• 最大内存页限制
• 栈深度限制

# 安装
cargo install cosmwasm-check

# 运行检查
cosmwasm-check target/wasm32-unknown-unknown/release/agent_contract.wasm

# 输出示例
# Checking contract: agent_contract
#   pass: Capability: iterator
#   pass: Capability: staking
#   pass: Capability: cosmos_msgs
#   pass: All capabilities are allowed
#   pass: All contract APIs are used correctly
#   pass: Contract size: 287.4 kB (max 800 kB)

2.2.2 cargo-audit 与 cargo-deny

# cargo-audit: 基于 RustSec Advisory Database
cargo audit
# 输出: 已知 CVE 漏洞列表

# cargo-deny: 许可合规 + 来源审查
cargo deny check
cargo deny check licenses

2.2.3 cw-multi-test

cw-multi-test 是 CosmWasm 的链上模拟测试框架。对于 Agent 合约审计自动化，它可：

• 模拟多个 Agent 合约交互
• 模拟 A2A 消息传递
• 测试宪法规则边界条件
• 验证 spend limit 逻辑

2.2.4 test-tube

test-tube 是 Osmosis 开发的 Cosmos SDK 集成测试框架，支持：

• 在真实的 Cosmos SDK 环境中运行合约
• 验证 gas 消耗
• 测试链上原生模块交互（Bank, Staking, Distribution）

2.3 与其他生态的比较

3. CI/CD 审计流水线

3.1 流水线架构总览

MSG Chain Agent 合约的推荐 CI/CD 流水线包含 5 个门禁阶段：

[1] 代码提交
    │
    ▼
[2] Pre-commit Hook (本地)
    ├─ clippy lint
    ├─ cargo fmt 检查
    ├─ 秘密泄露检测
    └─ 生成代码安全扫描
    │
    ▼
[3] CI 构建阶段 (GitLab CI / GitHub Actions)
    ├─ 编译检查 (release wasm)
    ├─ cosmwasm-check 静态验证
    ├─ cargo-audit 依赖漏洞
    ├─ cargo-deny 许可审查
    └─ 单元测试 + 覆盖率
    │
    ▼
[4] CI 审计阶段
    ├─ cw-multi-test 集成测试
    ├─ test-tube 模糊测试
    ├─ 宪法规则属性测试
    ├─ 气体估算审计
    └─ 安全评分计算
    │
    ▼
[5] 门禁决策
    ├─ 安全评分 ≥ 80 → 允许合并
    ├─ 安全评分 60-79 → 需人工审查
    └─ 安全评分 < 60 → 自动拒绝

3.2 GitHub Actions 完整配置

# .github/workflows/security-audit.yml
name: AI Agent Security Audit Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

env:
  CARGO_TERM_COLOR: always
  RUSTFLAGS: "-D warnings"
  MSG_CHAIN_ID: msg-chain-1

jobs:
  pre_commit_checks:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Rust toolchain
        uses: dtolnay/rust-toolchain@stable
        with:
          targets: wasm32-unknown-unknown
          components: clippy, rustfmt

      - name: Format check
        run: cargo fmt -- --check

      - name: Clippy lint
        run: cargo clippy --all-targets -- -D warnings

      - name: Check for leaked secrets
        uses: zricethezav/gitleaks-action@v2
        with:
          config_path: .gitleaks.toml

  dependency_audit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Rust
        uses: dtolnay/rust-toolchain@stable

      - name: Install cargo-audit
        run: cargo install cargo-audit --locked

      - name: Run cargo-audit
        run: cargo audit
        continue-on-error: true

      - name: Install cargo-deny
        run: cargo install cargo-deny --locked

      - name: Run cargo-deny
        run: cargo deny check licenses advisories sources

  build_and_static_check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Rust
        uses: dtolnay/rust-toolchain@stable
        with:
          targets: wasm32-unknown-unknown

      - name: Install cosmwasm-check
        run: cargo install cosmwasm-check --locked

      - name: Build release wasm
        run: cargo wasm

      - name: Optimize wasm
        run: |
          cargo install cosmwasm-opt --locked
          cosmwasm-opt target/wasm32-unknown-unknown/release/agent_contract.wasm

      - name: Run cosmwasm-check
        run: cosmwasm-check target/wasm32-unknown-unknown/release/agent_contract.opt.wasm

      - name: Check contract size
        run: |
          SIZE=$(stat -c%s target/wasm32-unknown-unknown/release/agent_contract.opt.wasm)
          MAX_SIZE=800000
          if [ $SIZE -gt $MAX_SIZE ]; then
            echo "Contract size $SIZE exceeds $MAX_SIZE"
            exit 1
          fi
          echo "Contract size: $SIZE bytes"

  unit_tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Rust
        uses: dtolnay/rust-toolchain@stable

      - name: Install cargo-llvm-cov
        run: cargo install cargo-llvm-cov --locked

      - name: Run tests with coverage
        run: cargo llvm-cov --all-features --workspace --lcov --output-path lcov.info

      - name: Check coverage threshold
        run: |
          COV=$(grep -oP 'TOTAL\s+\d+\s+\d+\s+(\d+\.\d+)' lcov.info | head -1 | grep -oP '\d+\.\d+$')
          THRESHOLD=70.0
          if (( $(echo "$COV < $THRESHOLD" | bc -l) )); then
            echo "Coverage $COV% below threshold $THRESHOLD%"
            exit 1
          fi

      - name: Upload coverage report
        uses: actions/upload-artifact@v4
        with:
          name: coverage-report
          path: lcov.info

  fuzz_and_property_tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Rust
        uses: dtolnay/rust-toolchain@stable

      - name: Run property-based tests
        run: cargo test --test property_tests -- --nocapture
        timeout-minutes: 30

      - name: Run integration tests with cw-multi-test
        run: cargo test --test integration_tests -- --nocapture

  security_score:
    needs: [pre_commit_checks, dependency_audit, build_and_static_check,
            unit_tests, fuzz_and_property_tests]
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Calculate security score
        run: |
          SCORE=100
          # 每个失败的检查项扣分
          # 具体扣分规则由 audit-scorer 工具执行
          cargo run --bin audit-scorer -- \
            --coverage lcov.info \
            --audit-output audit-results.json \
            --fuzz-results fuzz-results.json \
            --score-file security-score.json

      - name: Gate decision
        run: |
          SCORE=$(cat security-score.json | jq -r '.score')
          echo "Security Score: $SCORE"
          if [ "$SCORE" -lt 60 ]; then
            echo "FAIL: Security score $SCORE is below minimum threshold (60)"
            exit 1
          elif [ "$SCORE" -lt 80 ]; then
            echo "WARN: Security score $SCORE is below recommended threshold (80)"
            echo "Manual review required"
          else
            echo "PASS: Security score $SCORE meets threshold"
          fi

3.3 GitLab CI 等价配置

# .gitlab-ci.yml
stages:
  - pre-commit
  - dependency-audit
  - build
  - test
  - audit
  - gate

variables:
  CARGO_TERM_COLOR: always
  RUSTFLAGS: "-D warnings"

.pre-commit-rules: &pre-commit-rules
  rules:
    - if: $CI_MERGE_REQUEST_IID
    - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH

dependency-audit:
  stage: dependency-audit
  image: rust:1.80
  <<: *pre-commit-rules
  script:
    - cargo install cargo-audit cargo-deny
    - cargo audit
    - cargo deny check
  artifacts:
    paths:
      - cargo-audit-report.json
    expire_in: 30 days

build-wasm:
  stage: build
  image: rust:1.80
  <<: *pre-commit-rules
  script:
    - rustup target add wasm32-unknown-unknown
    - cargo install cosmwasm-check
    - cargo wasm
    - cosmwasm-check target/wasm32-unknown-unknown/release/*.wasm
    - cargo install cosmwasm-opt
    - cosmwasm-opt target/wasm32-unknown-unknown/release/agent_contract.wasm
  artifacts:
    paths:
      - target/wasm32-unknown-unknown/release/agent_contract.opt.wasm
    expire_in: 30 days

test:
  stage: test
  image: rust:1.80
  <<: *pre-commit-rules
  script:
    - cargo test --lib
    - cargo test --test property_tests
    - cargo test --test integration_tests
  coverage: '/^\s*line[未公开路径]'

audit-gate:
  stage: gate
  image: rust:1.80
  <<: *pre-commit-rules
  script:
    - cargo run --bin audit-scorer
    - ./scripts/gate-decision.sh
  dependencies:
    - dependency-audit
    - build-wasm
    - test

3.4 Pre-commit Hook 配置

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.5.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-added-large-files
      - id: detect-private-key

  - repo: local
    hooks:
      - id: cargo-fmt
        name: cargo fmt
        entry: cargo fmt
        language: system
        types: [rust]
        pass_filenames: false

      - id: cargo-clippy
        name: cargo clippy
        entry: cargo clippy
        language: system
        types: [rust]
        args: ["--", "-D", "warnings"]
        pass_filenames: false

      - id: cargo-audit
        name: cargo audit
        entry: cargo audit
        language: system
        pass_filenames: false

      - id: secret-detect
        name: Secret Detection
        entry: scripts/detect-secrets.sh
        language: script
        pass_filenames: false

      - id: llm-code-scan
        name: LLM Generated Code Scan
        entry: scripts/llm-code-scan.sh
        language: script
        types: [rust]
        pass_filenames: true

3.5 秘密泄露检测脚本

#!/usr/bin/env bash
# scripts/detect-secrets.sh
# 检测 AI Agent 合约中硬编码的秘密

set -euo pipefail

PATTERNS=(
  'msg1[0-9a-z]{38}'           # MSG 地址（允许在测试中引用，但需标记）
  '[未公开私钥标记]'
  'sk-[A-Za-z0-9]{32,}'        # OpenAI / LLM API key
  'ANT_[A-Za-z0-9]{32,}'       # Anthropic API key
  'xox[bpras]-[0-9a-zA-Z-]{10,}' # Slack token
  'ghp_[A-Za-z0-9]{36}'        # GitHub PAT
  'AKIA[0-9A-Z]{16}'           # AWS Access Key
)

EXIT_CODE=0

for pattern in "${PATTERNS[@]}"; do
  while IFS=: read -r file line content; do
    # 跳过测试文件和文档
    if [[ "$file" == *test* || "$file" == *spec* || "$file" == *.md ]]; then
      continue
    fi
    echo "WARNING: Possible secret in $file:$line"
    echo "  $content"
    EXIT_CODE=1
  done < <(grep -rnP "$pattern" --include='*.rs' --include='*.toml' --include='*.yaml' --include='*.json' . 2>/dev/null || true)
done

exit $EXIT_CODE

4. 模糊测试与属性测试

4.1 为什么需要模糊测试

AI Agent 合约的输入空间极大：

• A2A 消息可以是任意结构
• LLM 生成的 action payload 不可预测
• Agent 宪法规则可能有未覆盖的边界条件

手动测试只能覆盖"happy path"，模糊测试可以穷举边界。

4.2 cw-multi-test 中的模糊测试策略

4.2.1 支出限额模糊测试

// tests/property_tests/spending_limits.rs
use cosmwasm_std::{Addr, Coin, Uint128, Empty};
use cw_multi_test::{App, Contract, ContractWrapper, Executor};
use proptest::prelude::*;

// 导入合约
use agent_contract::msg::{ExecuteMsg, InstantiateMsg, QueryMsg};
use agent_contract::state::SpendingLimit;

// 合约包装
fn contract_template() -> Box<dyn Contract<Empty>> {
    Box::new(ContractWrapper::new(
        agent_contract::contract::execute,
        agent_contract::contract::instantiate,
        agent_contract::contract::query,
    ))
}

// 属性测试: 支出限额永远不应被超过
proptest! {
    #[test]
    fn test_spending_limit_invariant(
        daily_limit in 1000u128..1_000_000u128,
        num_transactions in 1..100u32,
        amounts in prop::collection::vec(
            1u128..10_000u128, 1..100
        ),
    ) {
        let mut app = App::default();
        let owner = Addr::unchecked("msg1owner");
        let agent = Addr::unchecked("msg1agent");
        let recipient = Addr::unchecked("msg1recipient");

        // 初始化合约
        let contract_id = app.store_code(contract_template());
        let contract_addr = app
            .instantiate_contract(
                contract_id,
                owner.clone(),
                &InstantiateMsg {
                    daily_limit: Uint128::new(daily_limit),
                    owner: owner.to_string(),
                },
                &[],
                "agent_contract",
                None,
            )
            .unwrap();

        // 模拟多次转账
        let mut total_spent = Uint128::zero();

        for (i, &amount) in amounts.iter().enumerate() {
            if i >= num_transactions as usize {
                break;
            }

            let spend_msg = ExecuteMsg::Spend {
                recipient: recipient.to_string(),
                amount: Uint128::new(amount),
            };

            let result = app.execute_contract(
                agent.clone(),
                contract_addr.clone(),
                &spend_msg,
                &[],
            );

            match result {
                Ok(_) => {
                    total_spent += Uint128::new(amount);
                    // 不变量: 总支出 <= 每日限额
                    assert!(
                        total_spent <= Uint128::new(daily_limit),
                        "Spending limit violated: {} > {}",
                        total_spent, daily_limit
                    );
                }
                Err(err) => {
                    // 预期: 当超出限额时拒绝
                    if total_spent + Uint128::new(amount) > Uint128::new(daily_limit) {
                        // 正确拒绝
                    } else {
                        panic!("Unexpected error: {:?}", err);
                    }
                    break;
                }
            }
        }
    }
}

4.2.2 A2A 消息认证属性测试

// tests/property_tests/a2a_auth.rs
use cosmwasm_std::Binary;
use proptest::prelude::*;
use agent_contract::msg::SignedAgentMessage;

// 属性: 签名验证具有以下性质
proptest! {
    #[test]
    fn test_signature_verification_properties(
        from_did in "[msg1][a-z0-9]{38}",
        to_did in "[msg1][a-z0-9]{38}",
        action in "\\p{ASCII}{1,50}",
        nonce in 0..u64::MAX,
        timestamp in 1000000000..2000000000u64,
    ) {
        // 1. 相同消息的签名应一致（确定性）
        let msg1 = SignedAgentMessage {
            from_did: from_did.clone(),
            to_did: to_did.clone(),
            action: action.clone(),
            params: Binary::default(),
            nonce,
            timestamp,
            signature: Binary::default(),
        };

        let msg2 = SignedAgentMessage {
            from_did,
            to_did,
            action,
            params: Binary::default(),
            nonce,
            timestamp,
            signature: Binary::default(),
        };

        // canoical 序列化应一致
        let bytes1 = build_canonical_bytes(&msg1).unwrap();
        let bytes2 = build_canonical_bytes(&msg2).unwrap();
        assert_eq!(bytes1, bytes2);

        // 2. nonce 不同 → 签名消息不同
        let msg3 = SignedAgentMessage {
            nonce: nonce + 1,
            ..msg1.clone()
        };
        let bytes3 = build_canonical_bytes(&msg3).unwrap();
        assert_ne!(bytes1, bytes3);
    }
}

4.2.3 宪法规则自动验证

// tests/property_tests/constitution.rs
use proptest::prelude::*;
use agent_contract::state::Constitution;

// 宪法规则的属性测试
proptest! {
    #[test]
    fn test_constitution_invariants(
        max_position in 1_000u128..10_000_000u128,
        allowed_assets_count in 1usize..20usize,
    ) {
        let constitution = Constitution {
            max_position_size: Uint128::new(max_position),
            allowed_assets: (0..allowed_assets_count)
                .map(|i| format!("msg1asset{}", i))
                .collect(),
            risk_level: RiskLevel::Medium,
            is_active: true,
            version: 1,
        };

        // 属性 1: 金额 <= max_position_size 且在允许列表中的交易应通过
        let allowed_asset = constitution.allowed_assets[0].clone();
        let trade = Trade {
            asset_in: allowed_asset.clone(),
            asset_out: allowed_asset.clone(),
            amount: Uint128::new(max_position / 2),
        };
        assert!(constitution.validate_trade(&trade).is_ok());

        // 属性 2: 金额 > max_position_size 应拒绝
        let oversized_trade = Trade {
            asset_in: allowed_asset.clone(),
            asset_out: allowed_asset,
            amount: Uint128::new(max_position + 1),
        };
        assert!(constitution.validate_trade(&oversized_trade).is_err());

        // 属性 3: 未允许的资产应拒绝
        let disallowed_trade = Trade {
            asset_in: "msg1unknown".to_string(),
            asset_out: constitution.allowed_assets[0].clone(),
            amount: Uint128::new(1000),
        };
        assert!(constitution.validate_trade(&disallowed_trade).is_err());

        // 属性 4: 宪法未激活时所有交易应拒绝
        let inactive = Constitution {
            is_active: false,
            ..constitution.clone()
        };
        assert!(inactive.validate_trade(&Trade {
            asset_in: constitution.allowed_assets[0].clone(),
            asset_out: constitution.allowed_assets[0].clone(),
            amount: Uint128::new(1000),
        }).is_err());
    }
}

4.3 test-tube 集成测试

// tests/integration_tests/agent_integration.rs
use test_tube::{Account, Runner, SigningAccount};
use agent_contract::msg::{ExecuteMsg, InstantiateMsg, QueryMsg};

#[test]
fn test_agent_full_lifecycle() {
    let runner = Runner::new(
        "msg-chain-1",
        "http://localhost:26657",
    ).unwrap();

    // 创建部署账户（使用 msg 前缀地址）
    let deployer = runner
        .create_account("msg1deplyr")
        .unwrap();

    // 部署合约
    let wasm_bytes = std::fs::read(
        "target/wasm32-unknown-unknown/release/agent_contract.opt.wasm"
    ).unwrap();

    let code_id = runner
        .store_code(&wasm_bytes, &deployer)
        .unwrap()
        .code_id;

    // 实例化
    let contract_addr = runner
        .instantiate_contract(
            code_id,
            &InstantiateMsg {
                owner: deployer.address(),
                daily_limit: Uint128::from(1000000u128),
                max_tx_limit: Uint128::from(100000u128),
            },
            None,
            "agent-instance",
            &[Coin::new(1000000, "umsg")],
            &deployer,
        )
        .unwrap();

    // 验证 gas 消耗在合理范围内
    let gas_used = contract_addr.gas_used;
    assert!(
        gas_used < 500_000,
        "Gas too high: {}",
        gas_used
    );
}

4.4 覆盖率驱动策略

#!/bin/bash
# scripts/coverage-driven-fuzz.sh
# 使用覆盖率数据指导模糊测试方向

set -euo pipefail

RUSTFLAGS="-C instrument-coverage" cargo test --tests
cargo llvm-cov --lcov --output-path lcov.info

# 识别未覆盖的函数
grep '0:0' lcov.info | grep 'fn ' | while read -r line; do
    fn_name=$(echo "$line" | grep -oP 'fn \K\w+')
    echo "UNCOVERED: $fn_name — generating targeted fuzz cases"

    # 自动生成针对该函数的 fuzz 测试
    cat >> "tests/fuzz_targeted/${fn_name}_fuzz.rs" << FEOF
use proptest::prelude::*;

proptest! {
    #[test]
    fn fuzz_${fn_name}() {
        // 自动生成的 fuzz 目标
        // TODO: 填充该函数的输入参数
    }
}
FEOF

4.5 Gas 估算自动化审计

// tests/gas_audit/gas_estimation.rs
use cw_multi_test::App;
use agent_contract::msg::ExecuteMsg;

#[test]
fn test_gas_profile() {
    let mut app = App::default();
    // ... 部署合约 ...

    // 记录各操作的 gas 消耗
    let gas_profiles: Vec<(&str, u64)> = vec![
        ("simple_transfer", measure_gas(&mut app, &simple_transfer_msg())),
        ("constitution_update", measure_gas(&mut app, &constitution_update_msg())),
        ("agent_shutdown", measure_gas(&mut app, &shutdown_msg())),
        ("batch_process_50", measure_gas(&mut app, &batch_msg(50))),
        ("batch_process_100", measure_gas(&mut app, &batch_msg(100))),
    ];

    // 验证 gas 消耗在预期范围内
    for (name, gas) in &gas_profiles {
        assert!(
            *gas < 5_000_000,
            "{} gas too high: {}",
            name, gas
        );
        println!("{}: {} gas", name, gas);
    }

    // Gas 回归检测: 与基线比较
    let baseline: Vec<(&str, u64)> = load_gas_baseline();
    for (name, gas) in &gas_profiles {
        if let Some((_, baseline_gas)) = baseline.iter().find(|(n, _)| n == name) {
            let deviation = (*gas as f64 - *baseline_gas as f64) / *baseline_gas as f64;
            if deviation > 0.2 {
                println!(
                    "WARNING: {} gas increased by {:.1}% ({} vs {})",
                    name, deviation * 100.0, gas, baseline_gas
                );
            }
        }
    }
}

5. 形式化验证入门

5.1 形式化验证在 CosmWasm 中的现状

根据 MSG Chain 白皮书，形式化验证工具在 CosmWasm 生态中的成熟度如下：

关键结论: MSG Chain 上形式化验证尚未达到生产就绪 (production_reference) 等级。对大多数 AI Agent 合约，使用属性测试 + 模糊测试比形式化验证更具性价比。

5.2 注册中心对已验证合约的支持

MSG Chain 的 genesis_registry_v1 注册中心支持记录合约的验证状态。以下是在注册中心注册已验证合约的方法：

use cosmwasm_std::{DepsMut, Env, Response, StdResult};
use cw_storage_plus::Map;
use serde::{Serialize, Deserialize};

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct VerificationRecord {
    pub contract_addr: String,       // msg1...
    pub verification_type: String,   // "manual_audit" | "formal_verification" | "automated_scan"
    pub tool: String,                // "cw-multi-test" | "k-framework" | "certora"
    pub report_hash: String,         // IPFS hash of full report
    pub verified_at: u64,            // block height
    pub verified_by: String,         // auditor address (msg1...)
    pub expires_at: Option<u64>,     // verification expiry
}

pub const VERIFICATION_REGISTRY: Map<&Addr, VerificationRecord> = Map::new("vrfy");

pub fn submit_verification_record(
    deps: DepsMut,
    _env: Env,
    record: VerificationRecord,
) -> StdResult<Response> {
    let addr = deps.api.addr_validate(&record.contract_addr)?;
    VERIFICATION_REGISTRY.save(deps.storage, &addr, &record)?;
    Ok(Response::new()
        .add_attribute("action", "register_verification")
        .add_attribute("contract", &record.contract_addr)
        .add_attribute("type", &record.verification_type))
}

5.3 使用 Z3 求解器进行符号测试

// 在 CI 中使用 Z3 验证算术约束
// 安装: cargo install z3
use z3::{ast::Int, Config, Context, Solver};

pub fn verify_no_overflow(max_amount: u128, num_transactions: u32) -> bool {
    let cfg = Config::new();
    let ctx = Context::new(&cfg);
    let solver = Solver::new(&ctx);

    let amount = Int::new_const(&ctx, "amount");
    let count = Int::new_const(&ctx, "count");
    let total = Int::new_const(&ctx, "total");
    let limit = Int::from_u64(&ctx, max_amount as u64);

    // 约束: 0 <= amount <= max_amount
    solver.assert(&amount.ge(&Int::from_u64(&ctx, 0)));
    solver.assert(&amount.le(&limit));
    // 约束: 0 <= count <= num_transactions
    solver.assert(&count.ge(&Int::from_u64(&ctx, 0)));
    solver.assert(&count.le(&Int::from_u64(&ctx, num_transactions as u64)));
    // total = amount * count
    solver.assert(&total._eq(&amount.mul(&[&count])));
    // 检查: total 是否可能溢出 u128?
    let max_u128 = Int::from_u64(&ctx, u64::MAX) * Int::from_u64(&ctx, u64::MAX);
    solver.assert(&total.gt(&max_u128));

    match solver.check() {
        z3::SatResult::Unsat => {
            println!("No overflow possible");
            true
        }
        z3::SatResult::Sat => {
            println!("Potential overflow found!");
            false
        }
        _ => {
            println!("Unknown");
            false
        }
    }
}

6. AI Agent 特定安全风险自动化检测

6.1 LLM 生成代码的安全扫描

AI Agent 合约可能由 LLM 生成或辅助生成。以下自动化检测针对 LLM 常见错误模式：

// src/audit/llm_code_scan.rs
/// LLM 生成代码的自动化安全扫描器

pub struct LlmCodeScanner {
    pub findings: Vec<ScanFinding>,
}

#[derive(Debug)]
pub struct ScanFinding {
    pub pattern_id: String,
    pub severity: Severity,
    pub file: String,
    pub line: u32,
    pub description: String,
    pub recommendation: String,
}

pub enum Severity {
    Critical,
    High,
    Medium,
    Low,
    Info,
}

impl LlmCodeScanner {
    pub fn new() -> Self {
        Self { findings: Vec::new() }
    }

    // 检测 LLM 常见的"幻觉" API 调用
    pub fn scan_for_hallucinated_apis(&mut self, source: &str, file: &str) {
        // LLM 经常虚构不存在的 CosmWasm API
        let hallucinated_patterns = vec![
            ("cw_storage_plus::Singleton", "Use Item instead of Singleton"),
            ("cosmwasm_std::Storage::get", "Use load() or may_load()"),
            ("cosmwasm_std::Storage::set", "Use save() or update()"),
            ("cosmwasm_std::to_vec", "Use to_json_binary()"),
            ("cosmwasm_std::from_slice", "Use from_json()"),
            ("cosmwasm_std::Env::block_height", "Use env.block.height"),
            ("cosmwasm_std::Env::contract_address", "Use env.contract.address"),
            ("cw20::Cw20Contract::new", "Use Cw20Contract::new() with addr"),
        ];

        for (pattern, suggestion) in hallucinated_patterns {
            if let Some(pos) = source.find(pattern) {
                let line_no = source[..pos].matches('\n').count() as u32 + 1;
                self.findings.push(ScanFinding {
                    pattern_id: "LLM-HALLUCINATION-001".to_string(),
                    severity: Severity::High,
                    file: file.to_string(),
                    line: line_no,
                    description: format!("Possible LLM hallucination: '{}' is not a valid API", pattern),
                    recommendation: suggestion.to_string(),
                });
            }
        }
    }

    // 检测 LLM 忽略的错误处理
    pub fn scan_for_missing_error_handling(&mut self, source: &str, file: &str) {
        // LLM 经常使用 unwrap() 而非 ? 操作符
        for (line_no, line) in source.lines().enumerate() {
            let trimmed = line.trim();

            // 忽略测试文件中的 unwrap
            if file.contains("test") {
                continue;
            }

            if trimmed.contains(".unwrap()") {
                self.findings.push(ScanFinding {
                    pattern_id: "LLM-UNWRAP-001".to_string(),
                    severity: Severity::Medium,
                    file: file.to_string(),
                    line: line_no as u32 + 1,
                    description: format!("unwrap() found: '{}'", trimmed.trim()),
                    recommendation: "Replace with ? operator for proper error propagation".to_string(),
                });
            }
        }
    }

    // 检测 LLM 忽略的访问控制
    pub fn scan_for_missing_auth(&mut self, source: &str, file: &str) {
        let critical_functions = vec![
            "withdraw", "set_limit", "update_constitution",
            "terminate", "migrate", "pause",
        ];

        for func in critical_functions {
            if let Some(start) = source.find(&format!("fn execute_{}", func)) {
                let end = source[start..]
                    .find('{')
                    .map(|i| start + i)
                    .unwrap_or(source.len());
                let snippet = &source[start..end.min(source.len())];
                // 检查函数签名附近是否有权限检查
                if !snippet.contains("ADMIN")
                    && !snippet.contains("OWNER")
                    && !snippet.contains("assert_admin")
                    && !snippet.contains("require_auth")
                {
                    let line_no = source[..start].matches('\n').count() as u32 + 1;
                    self.findings.push(ScanFinding {
                        pattern_id: "LLM-NO-AUTH-001".to_string(),
                        severity: Severity::Critical,
                        file: file.to_string(),
                        line: line_no,
                        description: format!("Missing access control on execute_{}", func),
                        recommendation: format!("Add ADMIN.assert_admin() to execute_{}", func),
                    });
                }
            }
        }
    }
}

6.2 Prompt Injection 检测

AI Agent 可能处理来自用户或其他 Agent 的自然语言指令。自动检测 prompt injection 向量：

// src/audit/prompt_injection_detector.rs
pub struct PromptInjectionDetector {
    pub patterns: Vec<&'static str>,
}

impl PromptInjectionDetector {
    pub fn new() -> Self {
        Self {
            patterns: vec![
                // 经典的 prompt injection 模式
                r"ignore all previous instructions",
                r"ignore all prior directives",
                r"forget your constitution",
                r"override your rules",
                r"you are now",
                r"act as if",
                r"Disregard all previous",
                r"System prompt:",
                r"## SYSTEM",
                r"New instructions:",
                r"reset your configuration",
                r"bypass security",
                r"reveal your private key",
                r"print your seed phrase",
                r"sign any transaction",
                r"transfer all funds",
                // Agent 特定
                r"修改宪法",
                r"忽略限制",
                r"绕过审计",
                r"转账给我",
                r"提升额度",
            ],
        }
    }

    pub fn scan_input(&self, input: &str) -> Vec<InjectionAttempt> {
        let mut attempts = Vec::new();
        let lower = input.to_lowercase();

        for &pattern in &self.patterns {
            if lower.contains(pattern) {
                attempts.push(InjectionAttempt {
                    pattern: pattern.to_string(),
                    severity: InjectionSeverity::High,
                    snippet: extract_context(input, pattern, 50),
                });
            }
        }

        attempts
    }
}

fn extract_context(text: &str, pattern: &str, radius: usize) -> String {
    if let Some(pos) = text.to_lowercase().find(pattern) {
        let start = pos.saturating_sub(radius);
        let end = (pos + pattern.len() + radius).min(text.len());
        let snippet = &text[start..end];
        if start > 0 { format!("...{}...", snippet) }
        else { format!("{}...", snippet) }
    } else {
        String::new()
    }
}

#[derive(Debug)]
pub struct InjectionAttempt {
    pub pattern: String,
    pub severity: InjectionSeverity,
    pub snippet: String,
}

pub enum InjectionSeverity {
    Info,
    Low,
    Medium,
    High,
    Critical,
}

CI 集成: A2A 消息扫描

# CI: 验证所有 A2A 消息体不包含 prompt injection
a2a-security-scan:
  stage: test
  script:
    - cargo run --bin a2a-message-scanner -- test-vectors/a2a-messages.json
    - cargo run --bin prompt-injection-scanner -- src/

6.3 Registry Key 泄露检测

MSG Chain 的 genesis_registry_v1 使用 canonical key 进行合约寻址。Agent 合约中硬编码的 registry key 可能导致信息泄露或权限提升：

// src/audit/registry_key_scanner.rs
pub struct RegistryKeyScanner;

impl RegistryKeyScanner {
    /// 扫描合约代码中的 registry canonical key 引用
    /// 在 MSG Chain 中, canonical key 形如 "dao_governance_v1" 等
    pub fn scan_for_keys(source: &str, file: &str) -> Vec<RegistryKeyFinding> {
        let known_canonical_keys = vec![
            "dao_governance_v1",
            "dar_verification_v1",
            "emission_schedule_v2",
            "gas_fee_distribution_v2",
            "candidate_node_staking_v2",
            "genesis_registry_v1",
            "treasury_v1",
            "block_time_schedule_v1",
        ];

        let mut findings = Vec::new();

        for key in known_canonical_keys {
            if let Some(pos) = source.find(key) {
                let line_no = source[..pos].matches('\n').count() as u32 + 1;
                let line_start = source[..pos].rfind('\n').map(|i| i + 1).unwrap_or(0);
                let line_end = source[pos..].find('\n').map(|i| pos + i).unwrap_or(source.len());
                let line_content = &source[line_start..line_end].trim();

                // 忽略文档注释
                if line_content.starts_with("//") || line_content.starts_with("///") {
                    continue;
                }

                findings.push(RegistryKeyFinding {
                    key: key.to_string(),
                    file: file.to_string(),
                    line: line_no,
                    severity: if source[pos..].contains("addr_validate") {
                        FindingSeverity::Low
                    } else {
                        FindingSeverity::Medium
                    },
                    context: line_content.to_string(),
                });
            }
        }

        findings
    }
}

6.4 Agent 配置漂移检测

# scripts/detect_config_drift.py
"""检测 Agent 合约配置与声明式宪法的差异"""
import json
import sys
import hashlib
from pathlib import Path

def load_constitution_manifest(path: str) -> dict:
    with open(path) as f:
        return json.load(f)

def query_onchain_config(rpc_endpoint: str, contract_addr: str) -> dict:
    """通过 MSG Chain RPC 查询链上配置"""
    import requests
    query = {
        "constitution": {}
    }
    resp = requests.post(
        f"{rpc_endpoint}/cosmwasm/wasm/v1/contract/{contract_addr}/smart",
        json={"data": json.dumps(query)}
    )
    return resp.json()

def detect_drift(manifest: dict, onchain: dict) -> list:
    drifts = []
    for key, expected_value in manifest.items():
        actual_value = onchain.get(key)
        if actual_value != expected_value:
            drifts.append({
                "key": key,
                "expected": expected_value,
                "actual": actual_value,
                "severity": "HIGH" if key in ["max_position_size", "allowed_assets"] else "MEDIUM"
            })
    return drifts

def main():
    manifest_path = sys.argv[1]
    rpc_endpoint = sys.argv[2]
    contract_addr = sys.argv[3]

    manifest = load_constitution_manifest(manifest_path)
    onchain = query_onchain_config(rpc_endpoint, contract_addr)

    drifts = detect_drift(manifest, onchain)

    if drifts:
        print(f"CONFIG DRIFT DETECTED for {contract_addr}")
        for d in drifts:
            print(f"  [{d['severity']}] {d['key']}: {d['expected']} → {d['actual']}")
        sys.exit(1)
    else:
        print(f"OK: {contract_addr} configuration matches manifest")
        sys.exit(0)

if __name__ == "__main__":
    main()

7. 持续监控与运行时审计

7.1 事件驱动的链上监控

部署后的 AI Agent 合约需要持续监控。MSG Chain 的 event 系统支持基于合约属性的过滤订阅：

// contracts/watchdog/src/contract.rs
/// 监控合约: 监听多个 Agent 的安全事件
use cosmwasm_std::{
    entry_point, DepsMut, Env, MessageInfo, Response, StdResult,
    Binary, from_binary,
};

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct WatchEvent {
    pub agent_addr: String,
    pub event_type: EventType,
    pub severity: u8,  // 1-5, 5=最严重
    pub details: String,
    pub block_height: u64,
    pub timestamp: u64,
}

pub enum EventType {
    SpendingLimitWarning,
    ConstitutionViolation,
    UnexpectedShutdown,
    LargeTransfer,
    NewAdminAction,
    RateLimitExceeded,
    UnauthorizedAccess,
}

pub const EVENTS: Map<u64, WatchEvent> = Map::new("events");
pub const ALERT_THRESHOLD: u8 = 4;  // 严重度 >= 4 时发送告警

#[entry_point]
pub fn execute(
    deps: DepsMut,
    env: Env,
    info: MessageInfo,
    msg: ExecuteMsg,
) -> StdResult<Response> {
    match msg {
        ExecuteMsg::ReportEvent(event) => {
            let id = env.block.height;
            EVENTS.save(deps.storage, &id, &event)?;

            let mut resp = Response::new()
                .add_attribute("action", "event_reported")
                .add_attribute("event_type", format!("{:?}", event.event_type))
                .add_attribute("severity", event.severity.to_string())
                .add_attribute("agent", &event.agent_addr);

            // 严重事件触发链上告警
            if event.severity >= ALERT_THRESHOLD {
                resp = resp.add_attribute("alert", "true");
                // 可添加通知逻辑
            }

            Ok(resp)
        }
    }
}

7.2 MSG Chain 事件订阅示例

// scripts/monitor.ts
import { SigningCosmWasmClient } from "@cosmjs/cosmwasm-stargate";
import { DirectSecp256k1HdWallet } from "@cosmjs/proto-signing";

const RPC_ENDPOINT = "https://rpc.msg-chain-1.msgchain.org";
const MONITORED_AGENTS = [
  "msg1agent1...",
  "msg1agent2...",
  "msg1agent3...",
];

interface SecurityEvent {
  agent: string;
  eventType: string;
  severity: number;
  blockHeight: number;
  txHash: string;
}

async function startMonitor() {
  const wallet = await DirectSecp256k1HdWallet.fromMnemonic(
    "your mnemonic here",
    { prefix: "msg" }
  );

  const client = await SigningCosmWasmClient.connectWithSigner(
    RPC_ENDPOINT,
    wallet
  );

  // 订阅事件
  const subscription = client.subscribe("tm.event='Tx'", (event) => {
    const txHash = event.value.TxResult.txhash;
    const events = event.value.TxResult.result.events;

    for (const evt of events) {
      // 过滤 Agent 安全事件
      if (evt.type === "wasm") {
        const attributes = evt.attributes as Array<{key: string, value: string}>;

        const contractAddr = attributes.find(a => a.key === "_contract_address")?.value;
        const action = attributes.find(a => a.key === "action")?.value;

        if (contractAddr && MONITORED_AGENTS.includes(contractAddr)) {
          const securityEvent: SecurityEvent = {
            agent: contractAddr,
            eventType: action || "unknown",
            severity: parseInt(attributes.find(a => a.key === "severity")?.value || "0"),
            blockHeight: parseInt(event.value.TxResult.height),
            txHash,
          };

          if (securityEvent.severity >= 4) {
            console.log(`🚨 CRITICAL EVENT: ${JSON.stringify(securityEvent)}`);
            // 发送告警 (Slack, Telegram, etc.)
          } else {
            console.log(`ℹ️ Event: ${JSON.stringify(securityEvent)}`);
          }
        }
      }
    }
  });

  console.log(`Monitoring ${MONITORED_AGENTS.length} agents on ${RPC_ENDPOINT}`);
}

startMonitor().catch(console.error);

7.3 MS 参数变更告警

Agent 合约的可配置参数（支出限额、宪法规则、白名单等）变更是高风险操作。自动告警系统：

// contracts/watchdog/src/param_monitor.rs
use cosmwasm_std::{Deps, Env, StdResult, Addr};
use cw_storage_plus::Map;

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct ParamChange {
    pub param_name: String,
    pub old_value: String,
    pub new_value: String,
    pub changed_by: Addr,
    pub block_height: u64,
    pub timestamp: u64,
}

pub const PARAM_CHANGES: Map<u64, ParamChange> = Map::new("pchg");

pub fn record_param_change(
    deps: DepsMut,
    env: Env,
    param_name: String,
    old_value: String,
    new_value: String,
    changed_by: Addr,
) -> StdResult<Response> {
    let change = ParamChange {
        param_name,
        old_value,
        new_value,
        changed_by,
        block_height: env.block.height,
        timestamp: env.block.time.seconds(),
    };

    PARAM_CHANGES.save(deps.storage, &env.block.height, &change)?;

    Ok(Response::new()
        .add_attribute("action", "param_change")
        .add_attribute("param", &change.param_name)
        .add_attribute("changed_by", &change.changed_by)
        .add_attribute("old_value", &change.old_value)
        .add_attribute("new_value", &change.new_value))
}

/// 查询最近的参数变更
pub fn query_recent_changes(
    deps: Deps,
    env: Env,
    lookback: u64,
) -> StdResult<Vec<ParamChange>> {
    let from_block = env.block.height.saturating_sub(lookback);
    let mut changes = Vec::new();

    for result in PARAM_CHANGES.range(deps.storage, Some(from_block), None, cosmwasm_std::Order::Ascending) {
        let (_, change) = result?;
        changes.push(change);
    }

    Ok(changes)
}

7.4 自动响应机制

// contracts/auto-responder/src/contract.rs
/// 自动响应合约: 监听安全事件并执行预定义响应

pub enum AutoResponse {
    /// 降低支出限额至 0
    FreezeAgent(String),
    /// 通知守护者
    NotifyGuardians(Vec<String>, String),
    /// 切换至应急宪法
    ActivateEmergencyConstitution(String),
    /// 执行链上暂停
    TriggerPause(String),
}

pub fn evaluate_and_respond(
    deps: DepsMut,
    env: Env,
    event: WatchEvent,
) -> StdResult<Response> {
    let response = match event.event_type {
        EventType::ConstitutionViolation if event.severity >= 5 => {
            AutoResponse::FreezeAgent(event.agent_addr)
        }
        EventType::LargeTransfer => {
            AutoResponse::NotifyGuardians(
                vec!["msg1guard1...".to_string()],
                format!("Large transfer from {}", event.agent_addr),
            )
        }
        EventType::UnauthorizedAccess if event.severity >= 4 => {
            AutoResponse::TriggerPause(event.agent_addr)
        }
        _ => return Ok(Response::new().add_attribute("action", "no_response_needed")),
    };

    execute_auto_response(deps, env, response)
}

7.5 监控仪表板集成

# scripts/dashboard/monitor_agent.py
"""Agent 安全监控仪表板入口"""
import requests
import json
from datetime import datetime
from typing import Dict, List

class AgentSecurityMonitor:
    def __init__(self, rpc: str, registry_addr: str):
        self.rpc = rpc
        self.registry_addr = registry_addr

    def query_agent_status(self, agent_addr: str) -> Dict:
        """查询 Agent 安全状态"""
        query = {
            "agent_security_status": {
                "agent_addr": agent_addr
            }
        }
        resp = requests.post(
            f"{self.rpc}/cosmwasm/wasm/v1/contract/{self.registry_addr}/smart",
            json={"data": json.dumps(query)}
        )
        return resp.json()

    def get_recent_security_events(self, lookback_blocks: int = 1000) -> List[Dict]:
        result = []
        # 遍历区块，收集安全事件
        # ...
        return result

    def generate_alert(self, event: Dict):
        """生成告警"""
        severity_map = {1: "LOW", 2: "MEDIUM", 3: "HIGH", 4: "CRITICAL", 5: "EMERGENCY"}
        msg = (
            f"[{severity_map.get(event.get('severity', 1), 'UNKNOWN')}] "
            f"Agent {event.get('agent', 'unknown')}: "
            f"{event.get('detail', 'no detail')}"
        )
        print(f"ALERT: {msg}")
        # 可集成 Slack/Telegram/DingTalk 通知

8. 审计报告自动生成

8.1 SARIF 格式输出

SARIF (Static Analysis Results Interchange Format) 是 OASIS 标准的静态分析结果格式。将审计工具的输出统一为 SARIF 格式：

{
  "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/master/Schemata/sarif-schema-2.1.0.json",
  "version": "2.1.0",
  "runs": [
    {
      "tool": {
        "driver": {
          "name": "agent-security-audit-pipeline",
          "version": "1.0.0",
          "informationUri": "https://msgchain.org/security-audit",
          "rules": [
            {
              "id": "RE-AUTO-001",
              "name": "reentrancy-detection",
              "shortDescription": {
                "text": "Reentrancy vulnerability detected"
              },
              "fullDescription": {
                "text": "External call before state update may allow reentrancy attack"
              },
              "defaultConfiguration": {
                "level": "error"
              },
              "helpUri": "https://msgchain.org/security/reentrancy",
              "properties": {
                "security-severity": "9.0",
                "tags": ["security", "reentrancy", "critical"]
              }
            },
            {
              "id": "SL-AUTO-001",
              "name": "missing-spending-limit",
              "shortDescription": {
                "text": "Missing spending limit implementation"
              },
              "fullDescription": {
                "text": "Contract has fund transfer capabilities but no spending limit enforcement"
              },
              "defaultConfiguration": {
                "level": "warning"
              },
              "helpUri": "https://msgchain.org/security/spending-limits",
              "properties": {
                "security-severity": "7.0",
                "tags": ["security", "spending-limit", "high"]
              }
            },
            {
              "id": "CE-AUTO-001",
              "name": "constitution-not-enforced",
              "shortDescription": {
                "text": "Agent constitution not enforced"
              },
              "fullDescription": {
                "text": "Constitution defined but not validated before action execution"
              },
              "defaultConfiguration": {
                "level": "error"
              },
              "helpUri": "https://msgchain.org/security/constitution",
              "properties": {
                "security-severity": "8.5",
                "tags": ["security", "constitution", "critical"]
              }
            },
            {
              "id": "LLM-HALLUCINATION-001",
              "name": "llm-hallucinated-api",
              "shortDescription": {
                "text": "Possible LLM hallucinated API call"
              },
              "fullDescription": {
                "text": "Non-existent CosmWasm API detected, likely LLM code generation artifact"
              },
              "defaultConfiguration": {
                "level": "warning"
              },
              "properties": {
                "security-severity": "6.0",
                "tags": ["llm", "ai-codegen", "hallucination"]
              }
            }
          ]
        }
      },
      "results": [
        {
          "ruleId": "RE-AUTO-001",
          "level": "error",
          "message": {
            "text": "External call before state update at src/contract.rs:42"
          },
          "locations": [
            {
              "physicalLocation": {
                "artifactLocation": {
                  "uri": "src/contract.rs",
                  "uriBaseId": "%SRCROOT%"
                },
                "region": {
                  "startLine": 42,
                  "endLine": 42,
                  "snippet": {
                    "text": "                .add_message(BankMsg::Send {"
                  }
                }
              }
            }
          ],
          "relatedLocations": [
            {
              "id": 1,
              "physicalLocation": {
                "artifactLocation": {
                  "uri": "src/contract.rs",
                  "uriBaseId": "%SRCROOT%"
                },
                "region": {
                  "startLine": 48,
                  "snippet": {
                    "text": "                BALANCES.save(deps.storage, ..."
                  }
                }
              },
              "message": {
                "text": "State update after external call"
              }
            }
          ]
        }
      ],
      "columnKind": "utf16CodeUnits",
      "properties": {
        "chain-id": "msg-chain-1",
        "bech32-prefix": "msg",
        "pipeline-run-id": "2026-07-08-build-1234"
      }
    }
  ]
}

8.2 Rust SARIF 生成器

// src/report/sarif_generator.rs
use serde::{Serialize, Deserialize};
use std::collections::HashMap;

#[derive(Serialize, Deserialize)]
pub struct SarifReport {
    #[serde(rename = "$schema")]
    pub schema: String,
    pub version: String,
    pub runs: Vec<SarifRun>,
}

#[derive(Serialize, Deserialize)]
pub struct SarifRun {
    pub tool: Tool,
    pub results: Vec<Result>,
    pub column_kind: String,
    pub properties: HashMap<String, String>,
}

#[derive(Serialize, Deserialize)]
pub struct Tool {
    pub driver: Driver,
}

#[derive(Serialize, Deserialize)]
pub struct Driver {
    pub name: String,
    pub version: String,
    pub information_uri: String,
    pub rules: Vec<Rule>,
}

#[derive(Serialize, Deserialize)]
pub struct Rule {
    pub id: String,
    pub name: String,
    pub short_description: Description,
    pub full_description: Option<Description>,
    pub default_configuration: Configuration,
    pub help_uri: Option<String>,
    pub properties: HashMap<String, String>,
}

#[derive(Serialize, Deserialize)]
pub struct Description {
    pub text: String,
}

#[derive(Serialize, Deserialize)]
pub struct Configuration {
    pub level: String,
}

#[derive(Serialize, Deserialize)]
pub struct Result {
    pub rule_id: String,
    pub level: String,
    pub message: Description,
    pub locations: Vec<Location>,
    pub related_locations: Option<Vec<RelatedLocation>>,
    pub properties: Option<HashMap<String, String>>,
}

#[derive(Serialize, Deserialize)]
pub struct Location {
    pub physical_location: PhysicalLocation,
}

#[derive(Serialize, Deserialize)]
pub struct PhysicalLocation {
    pub artifact_location: ArtifactLocation,
    pub region: Region,
}

#[derive(Serialize, Deserialize)]
pub struct ArtifactLocation {
    pub uri: String,
    pub uri_base_id: Option<String>,
}

#[derive(Serialize, Deserialize)]
pub struct Region {
    pub start_line: u32,
    pub end_line: Option<u32>,
    pub snippet: Option<Snippet>,
}

#[derive(Serialize, Deserialize)]
pub struct Snippet {
    pub text: String,
}

#[derive(Serialize, Deserialize)]
pub struct RelatedLocation {
    pub id: u32,
    pub physical_location: PhysicalLocation,
    pub message: Description,
}

impl SarifReport {
    pub fn to_json(&self) -> String {
        serde_json::to_string_pretty(self).unwrap()
    }
}

8.3 安全评分计算引擎

// src/audit/scorer.rs
/// 安全评分引擎: 将审计工具输出转化为可量化的安全评分

pub struct SecurityScorer {
    pub max_score: u32,
    pub deductions: HashMap<String, Deduction>,
}

pub struct Deduction {
    pub rule_id: String,
    pub severity: Severity,
    pub points: u32,
}

impl SecurityScorer {
    pub fn new() -> Self {
        let mut deductions = HashMap::new();

        // 严重漏洞扣分
        deductions.insert("RE-AUTO-001".to_string(), Deduction {
            rule_id: "RE-AUTO-001".to_string(),
            severity: Severity::Critical,
            points: 30,
        });
        deductions.insert("AC-AUTO-001".to_string(), Deduction {
            rule_id: "AC-AUTO-001".to_string(),
            severity: Severity::Critical,
            points: 30,
        });
        deductions.insert("CE-AUTO-001".to_string(), Deduction {
            rule_id: "CE-AUTO-001".to_string(),
            severity: Severity::Critical,
            points: 30,
        });
        deductions.insert("SL-AUTO-001".to_string(), Deduction {
            rule_id: "SL-AUTO-001".to_string(),
            severity: Severity::High,
            points: 15,
        });
        deductions.insert("A2A-AUTO-001".to_string(), Deduction {
            rule_id: "A2A-AUTO-001".to_string(),
            severity: Severity::High,
            points: 15,
        });
        deductions.insert("IV-AUTO-001".to_string(), Deduction {
            rule_id: "IV-AUTO-001".to_string(),
            severity: Severity::Medium,
            points: 7,
        });
        deductions.insert("LLM-UNWRAP-001".to_string(), Deduction {
            rule_id: "LLM-UNWRAP-001".to_string(),
            severity: Severity::Medium,
            points: 5,
        });
        deductions.insert("LLM-HALLUCINATION-001".to_string(), Deduction {
            rule_id: "LLM-HALLUCINATION-001".to_string(),
            severity: Severity::Low,
            points: 3,
        });

        Self {
            max_score: 100,
            deductions,
        }
    }

    pub fn calculate(&self, findings: &[AuditFinding]) -> ScoreReport {
        let mut score = self.max_score;
        let mut details = Vec::new();

        for finding in findings {
            if let Some(deduction) = self.deductions.get(&finding.rule_id) {
                score = score.saturating_sub(deduction.points);
                details.push(ScoringDetail {
                    rule_id: finding.rule_id.clone(),
                    description: finding.description.clone(),
                    deduction: deduction.points,
                    remaining_score: score,
                });
            }
        }

        ScoreReport {
            score,
            max_score: self.max_score,
            details,
            grade: match score {
                90..=100 => "A+",
                80..=89 => "A",
                70..=79 => "B",
                60..=69 => "C",
                0..=59 => "F",
            }.to_string(),
            verdict: if score >= 80 {
                Verdict::Pass
            } else if score >= 60 {
                Verdict::ReviewRequired
            } else {
                Verdict::Fail
            },
        }
    }
}

#[derive(Serialize, Deserialize)]
pub struct ScoreReport {
    pub score: u32,
    pub max_score: u32,
    pub details: Vec<ScoringDetail>,
    pub grade: String,
    pub verdict: Verdict,
}

#[derive(Serialize, Deserialize)]
pub struct ScoringDetail {
    pub rule_id: String,
    pub description: String,
    pub deduction: u32,
    pub remaining_score: u32,
}

#[derive(Serialize, Deserialize)]
pub enum Verdict {
    Pass,
    ReviewRequired,
    Fail,
}

8.4 报告聚合和发布

#!/usr/bin/env python3
# scripts/generate-audit-report.py
"""聚合多个审计工具的输出，生成统一报告"""

import json
import hashlib
from pathlib import Path
from datetime import datetime

def aggregate_tool_outputs() -> dict:
    report = {
        "metadata": {
            "chain_id": "msg-chain-1",
            "bech32_prefix": "msg",
            "generated_at": datetime.utcnow().isoformat(),
            "pipeline_id": "audit-2026-07-08",
        },
        "tools": {}
    }

    # 收集 cosmwasm-check 结果
    check_path = Path("cosmwasm-check-output.json")
    if check_path.exists():
        report["tools"]["cosmwasm-check"] = json.loads(check_path.read_text())

    # 收集 cargo-audit 结果
    audit_path = Path("cargo-audit-report.json")
    if audit_path.exists():
        report["tools"]["cargo-audit"] = json.loads(audit_path.read_text())

    # 收集 SARIF 结果
    sarif_path = Path("audit-results.sarif")
    if sarif_path.exists():
        report["tools"]["sarif"] = json.loads(sarif_path.read_text())

    # 收集覆盖率结果
    cov_path = Path("lcov.info")
    if cov_path.exists():
        coverage = parse_lcov(cov_path)
        report["tools"]["coverage"] = coverage

    # 收集模糊测试结果
    fuzz_path = Path("fuzz-results.json")
    if fuzz_path.exists():
        report["tools"]["fuzz"] = json.loads(fuzz_path.read_text())

    return report

def parse_lcov(path: Path) -> dict:
    lines = path.read_text().splitlines()
    total_lines = 0
    covered_lines = 0

    for line in lines:
        if line.startswith("DA:"):
            parts = line.split(",")
            if parts[1] != "0":
                covered_lines += 1
            total_lines += 1

    coverage = (covered_lines / total_lines * 100) if total_lines > 0 else 0
    return {
        "total_lines": total_lines,
        "covered_lines": covered_lines,
        "coverage_percent": round(coverage, 2),
    }

def generate_summary(report: dict) -> dict:
    sarif = report.get("tools", {}).get("sarif", {})
    runs = sarif.get("runs", [{}])
    results = runs[0].get("results", []) if runs else []

    critical = sum(1 for r in results if r.get("level") == "error")
    warnings = sum(1 for r in results if r.get("level") == "warning")
    notes = sum(1 for r in results if r.get("level") == "note")

    coverage = report.get("tools", {}).get("coverage", {})
    fuzz = report.get("tools", {}).get("fuzz", {})

    score = 100 - (critical * 30 + warnings * 7 + notes * 3)
    score = max(0, min(100, score))

    return {
        "summary": {
            "total_findings": len(results),
            "critical": critical,
            "warnings": warnings,
            "notes": notes,
            "coverage": coverage.get("coverage_percent", 0),
            "fuzz_tests_run": fuzz.get("tests_run", 0),
            "score": score,
            "grade": "A+" if score >= 90 else "A" if score >= 80 else "B" if score >= 70 else "C" if score >= 60 else "F",
            "pass": score >= 80,
        }
    }

if __name__ == "__main__":
    report = aggregate_tool_outputs()
    summary = generate_summary(report)
    report.update(summary)

    output_path = Path("security-audit-report.json")
    output_path.write_text(json.dumps(report, indent=2))
    print(f"Report written to {output_path}")
    print(f"Score: {summary['summary']['score']}/100 - Grade: {summary['summary']['grade']}")

9. 实践案例

9.1 案例: AI Trading Agent 自动化审计

本节展示一个完整的 AI Trading Agent 合约审计自动化流水线示例。

9.1.1 Agent 合约结构

agent-trading/
├── .github/
│   └── workflows/
│       └── security-audit.yml    # 审计流水线 (见第 3 章)
├── .pre-commit-config.yaml       # Pre-commit hooks
├── src/
│   ├── contract.rs               # 主合约
│   ├── msg.rs                    # 消息定义
│   ├── state.rs                  # 状态管理
│   ├── constitution.rs           # 宪法规则
│   ├── a2a_auth.rs               # A2A 认证
│   ├── spending_limits.rs        # 支出限额
│   └── audit/
│       ├── llm_code_scan.rs      # LLM 代码扫描
│       ├── prompt_injection.rs   # Prompt injection 检测
│       └── registry_scanner.rs   # Registry key 扫描
├── tests/
│   ├── unit/
│   │   ├── test_constitution.rs  # 宪法测试
│   │   ├── test_limits.rs       # 限额测试
│   │   └── test_a2a.rs          # A2A 测试
│   ├── property_tests/
│   │   ├── spending_limits.rs   # 限额属性测试
│   │   ├── constitution.rs      # 宪法属性测试
│   │   └── a2a_auth.rs          # A2A 属性测试
│   └── integration_tests/
│       └── agent_lifecycle.rs   # 端到端生命周期测试
├── gas_audit/
│   └── gas_profiles.rs          # Gas 估算审计
├── audit-manifests/
│   ├── constitution.json        # 期望的宪法配置
│   └── security-baseline.json   # 安全基线
└── scripts/
    ├── detect-secrets.sh         # 秘密检测
    ├── coverage-driven-fuzz.sh   # 覆盖率驱动模糊测试
    ├── detect_config_drift.py    # 配置漂移检测
    └── generate-audit-report.py  # 报告生成

9.1.2 审计配置文件

// audit-manifests/security-baseline.json
{
  "chain_id": "msg-chain-1",
  "bech32_prefix": "msg",
  "security_requirements": {
    "minimum_coverage": 70.0,
    "minimum_security_score": 80,
    "critical_findings_allowed": 0,
    "high_findings_allowed": 0,
    "medium_findings_allowed": 3
  },
  "gas_baselines": {
    "simple_transfer": 150000,
    "constitution_update": 250000,
    "agent_shutdown": 100000,
    "batch_process_50": 500000,
    "batch_process_100": 900000
  },
  "audit_rules": {
    "enabled": [
      "RE-AUTO-001",
      "SL-AUTO-001",
      "CE-AUTO-001",
      "AC-AUTO-001",
      "A2A-AUTO-001",
      "IV-AUTO-001",
      "LLM-HALLUCINATION-001",
      "LLM-UNWRAP-001",
      "LLM-NO-AUTH-001"
    ],
    "severity_overrides": {
      "LLM-UNWRAP-001": "low"
    }
  },
  "monitoring": {
    "watchdog_contract": "msg1watchdog...",
    "alert_channels": {
      "slack": "#agent-security",
      "email": "security@example.com"
    },
    "alert_thresholds": {
      "spending_limit_warning": "daily_limit_80%",
      "large_transfer": ">10000umsg",
      "constitution_change": "any"
    }
  }
}

9.1.3 流水线执行日志

# CI 流水线执行示例输出

=== Agent Trading Security Audit Pipeline ===
Chain: msg-chain-1 | Prefix: msg | Run: 2026-07-08-1234

[1/7] Pre-commit Checks
  ✔ cargo fmt --check
  ✔ cargo clippy -- -D warnings
  ✔ Secret detection: 0 findings
  ✔ LLM code scan: 2 INFO findings
    - src/contract.rs:42: unwrap() found
    - src/state.rs:15: LLM hallucinated API 'Singleton'

[2/7] Dependency Audit
  ✔ cargo-audit: 0 known vulnerabilities
  ✔ cargo-deny: All licenses OK
  ✔ cargo-deny: All sources trusted

[3/7] Build & Static Check
  ✔ cargo wasm (release)
  ✔ cosmwasm-opt (optimized: 287.4 kB → 156.2 kB)
  ✔ cosmwasm-check: All checks passed
  ✔ Contract size: 156.2 kB (limit: 800 kB)

[4/7] Unit Tests & Coverage
  ✔ 42 tests passed, 0 failed
  ✔ Coverage: 84.7% (threshold: 70%)
  ✔ Gas audit: All operations within baseline

[5/7] Property Tests & Fuzzing
  ✔ Spending limit invariant: 1000 random scenarios passed
  ✔ Constitution invariant: 500 random scenarios passed
  ✔ A2A auth properties: 300 random scenarios passed
  ✔ No invariant violations

[6/7] Integration Tests
  ✔ Agent lifecycle (create → activate → trade → pause)
  ✔ Multi-agent interaction (A2A messaging)
  ✔ Error handling (all expected error paths)
  ✔ Rate limiting enforcement

[7/7] Security Score & Gate Decision
  ┌──────────────────────────────────────────────┐
  │  Security Audit Report Summary               │
  │──────────────────────────────────────────────│
  │  Score:          92/100                       │
  │  Grade:          A+                           │
  │  Critical:       0                            │
  │  High:           0                            │
  │  Medium:         1 (LLM-UNWRAP-001)           │
  │  Low:            1 (LLM-HALLUCINATION-001)    │
  │  Coverage:       84.7%                        │
  │  Verdict:        ✅ PASS                      │
  └──────────────────────────────────────────────┘

=== GATE DECISION: PASS (score 92 >= 80) ===

9.1.4 门禁决策脚本

#!/bin/bash
# scripts/gate-decision.sh
# 自动化门禁: 根据审计报告决定是否允许合并

set -euo pipefail

REPORT="security-audit-report.json"

if [ ! -f "$REPORT" ]; then
    echo "ERROR: No audit report found"
    exit 1
fi

SCORE=$(jq -r '.summary.score' "$REPORT")
CRITICAL=$(jq -r '.summary.critical' "$REPORT")
HIGH=$(jq -r '.summary.warnings' "$REPORT")

echo "Security Score: $SCORE/100"
echo "Critical findings: $CRITICAL"
echo "High findings: $HIGH"

# 门禁规则
if [ "$CRITICAL" -gt 0 ]; then
    echo "❌ FAIL: Critical findings present — manual review required"
    exit 1
fi

if [ "$HIGH" -gt 0 ]; then
    echo "❌ FAIL: High findings present — manual review required"
    exit 1
fi

if [ "$SCORE" -lt 60 ]; then
    echo "❌ FAIL: Score below minimum threshold (60)"
    exit 1
elif [ "$SCORE" -lt 80 ]; then
    echo "⚠️  WARN: Score below recommended threshold (80) — manual review required"
    # 可以配置为 conditionally pass with human approval
    exit 0
else
    echo "✅ PASS: Score meets threshold"
    exit 0
fi

9.2 案例: 从零搭建 Agent 审计流水线

步骤 1: 初始化项目

# 使用 MSG Chain 推荐的模板
cargo generate --git https://github.com/msgchain/agent-template.git --name my-agent

# 或手动创建
cargo init --lib my-agent
cd my-agent
cargo add cosmwasm-std cw-storage-plus cw-controllers cw2
cargo add --dev cw-multi-test proptest test-tube

步骤 2: 配置 Pre-commit

pip install pre-commit
pre-commit install

将第 3 章的 .pre-commit-config.yaml 复制到项目根目录。

步骤 3: 配置 CI

将第 3 章的 GitHub Actions 或 GitLab CI 配置添加到项目。

步骤 4: 编写审计相关的代码

mkdir -p src/audit tests/property_tests tests/integration_tests audit-manifests

# 复制 LLM 代码扫描器
cp /templates/llm_code_scan.rs src/audit/

# 复制属性测试模板
cp /templates/spending_limit_proptest.rs tests/property_tests/

# 创建安全基线配置
cp /templates/security-baseline.json audit-manifests/

步骤 5: 配置运行时监控

# 部署 watchdog 合约
msgd tx wasm store watchdog.wasm --from deployer --chain-id msg-chain-1
msgd tx wasm instantiate 1 '{}' --label watchdog --from deployer

# 注册被监控的 Agent
msgd tx wasm execute msg1watchdog \
  '{"register_agent":{"agent":"msg1agent...","threshold":4}}' \
  --from deployer

步骤 6: 验证流水线

# 提交代码触发流水线
git add .
git commit -m "feat: initial agent implementation"

# 流水线将自动运行:
# 1. pre-commit hooks
# 2. CI audit pipeline
# 3. Security gate decision

10. 总结与建议

10.1 自动化审计成熟度模型

参照 MSG Chain developer_capability_matrix.json 的机器就绪等级，我们将审计自动化能力划分为 5 个级别：

建议: MSG Chain 上的 AI Agent 项目至少达到 L3（流水线门禁），推荐 L4（运行时监控）。

10.2 核心建议

对 AI Agent 开发者

1. 将审计自动化嵌入开发流程：使用 pre-commit hooks 在开发者本地捕获基础问题，避免 CI 阶段才发现
2. 强制执行安全门禁：在 CI/CD 中设置安全评分阈值（推荐 ≥80），低于阈值自动拒绝合并
3. 覆盖所有 Agent 特有风险：除常规漏洞外，自动化检测 LLM 生成代码、prompt injection、registry key 泄露
4. 属性测试优先于形式化验证：对 AI Agent 合约，cw-multi-test + proptest 的性价比远高于 K-framework
5. 设计可审计的合约架构：合约设计时即考虑自动化审计场景，如 emit 足够的属性和事件

对 MSG Chain 生态贡献者

1. 推动工具成熟度提升：sdk_surface 从 local_candidate 向 assisted_codegen 演进
2. 完善 OpenAPI 规范：使 agent_query_and_guarded_write 从 guarded_write 升级为 write_path_ready
3. 标准化审计报告格式：推动 SARIF 格式成为 MSG Chain 审计工具的标准输出

不可自动化的安全环节

即使有完善的自动化体系，以下环节仍需人工参与：

10.3 与 MSG Chain 白皮书的对齐

本文档严格遵循 MSG Chain 白皮书定义的能力边界：

• ✅ contract_runtime 已达到 assisted_codegen，支持 CI/CD 集成
• ✅ rpc_gateway 已达到 assisted_codegen，支持事件订阅和链上查询
• ✅ registry_resolution 已达到 production_reference，可进行 registry key 扫描
• ⚠️ agent_query_and_guarded_write 为 guarded_write，监控路径可用，自动写路径受限
• ⚠️ sdk_surface 为 local_candidate，部分集成测试需等待 SDK 稳定
• ❌ 未声称任何 DeFi/Oracle/Payment/Bridge 端点已实现（均为 Stub）

文档版本: 1.0
适用范围: MSG Chain (msg-chain-1) | Bech32 前缀: msg
参考文档:

• AI Agent 智能合约安全审计清单指南.md — 漏洞模式与手动检查清单
• MSG Chain 白皮书: https://msgchain.org/whitepaper/
• developer_capability_matrix.json: https://msgchain.org/whitepaper/developer_capability_matrix.json
• agent_entry.json: https://msgchain.org/whitepaper/agent_entry.json