MSG Chain 智能合约安全事件响应指南
数据来源：MSG Chain 代码库核实
主网状态: No-Go — 当前 MSGChain 主网裁决为 No-Go，以下内容反映代码实际状态，不代表生产可用。
目录
1. 概述
1.1 为什么需要事故响应计划
智能合约一旦部署即不可篡改（除非设计升级机制），漏洞可能被瞬间利用。MSG Chain 上的合约事故响应需要秒级检测、分钟级遏制、小时级分析的能力。
截至 2026 年，CosmWasm 生态已因安全事件累计损失超过 20 亿美元。典型攻击窗口期仅 30 秒至 15 分钟 — 在这段时间内，攻击者可提取全部可用流动性。
1.2 MSG Chain 安全假设与风险轮廓
MSG Chain 安全基线
├── 共识层: DAR (Dynamic Authority Rotation)
│ ├── 验证者集: 动态 (最小 4, 无上限)
│ ├── 轮换阈值: 104/96 确认
│ └── 信任假设: > 2/3 验证者诚实
├── 签名层: Dilithium-5 (后量子密码学)
│ ├── 抗量子攻击: 是
│ └── 签名大小: 2426-3366 bytes
├── 存储层: BadgerDB (LSM-Tree)
│ ├── 无 Merkle 证据 (轻客户端需额外机制)
│ └── 高性能但无原生历史查询
├── 合约层: CosmWasm 1.x (wasmd)
│ ├── 沙箱隔离: 有效
│ ├── SubMsg + Reply: 重入风险
│ └── 迁移: 治理控制的升级路径
└── 跨链层: IBC (规划中)
└── 超时 / ACK: 标准 ICS-20 处理
1.3 事故响应生命周期
┌─────────────────────────────────────────────────────────────┐
│ 事故响应生命周期 │
│ │
│ 事前准备 ──→ 攻击检测 ──→ 应急响应 ──→ 资金追回 ──→ 事后复盘 │
│ (Preparation) (Detection) (Containment) (Recovery) (Post) │
│ │
│ ↑ ↓ │
│ └──────── 持续改进 ──────┘ │
└─────────────────────────────────────────────────────────────┘
每个阶段的目标：
| 阶段 | 时间目标 | 核心目标 |
|---|---|---|
| 事前准备 | 持续 | 建立检测、响应、恢复能力 |
| 攻击检测 | 实时 (~秒级) | 尽早发现异常活动 |
| 应急响应 (Triage) | ≤ 5 分钟 | 确认事件、评估严重性、启动响应 |
| 应急响应 (Containment) | ≤ 15 分钟 | 阻止损失扩大 |
| 应急响应 (Investigation) | ≤ 4 小时 | 确定根因和影响范围 |
| 资金追回 | 24h - 2 周 | 追回被盗资金 |
| 事后复盘 | 1 - 2 周 | 改进流程、预防复发 |
1.4 响应团队角色
| 角色 | 职责 | 必备技能 |
|---|---|---|
| 事件指挥官 (IC) | 全局协调、资源调度、决策审批 | 项目管理、技术决策 |
| 安全负责人 (Security Lead) | 技术分析、漏洞定位、修复方案 | CosmWasm 安全审计、Rust |
| 通讯负责人 (Comms Lead) | 内外部沟通、社区公告、媒体关系 | 危机沟通、社区管理 |
| 法务负责人 (Legal Lead) | 监管报告、执法联络、用户通知 | 区块链法规、数据隐私 |
| 合约专家 (Contract SME) | 合约代码审查、补丁开发 | CosmWasm 开发、Rust |
| 基础设施工程师 (Infra) | 节点操作、RPC 端点、链上数据获取 | Cosmos SDK、运维 |
| 追回专家 (Recovery Lead) | 资金追踪、交易所联络、白帽行动 | 链分析、AML/KYC 流程 |
值守要求：
- 核心团队至少 3 人（IC + Security + Comms）保持 24/7 待命
- 每季度进行一次模拟演练
- 角色备份：每个角色至少指定 2 名后备人员
1.5 通信渠道
内部通信 (仅限响应团队)
| 渠道 | 用途 | 安全要求 |
|---|---|---|
| Signal / Telegram 加密群组 | 实时协调 | 端到端加密、阅后即焚可选 |
| 私有 Discord 频道 | 文件共享、日志记录 | 邀请制、多因素认证 |
| 加密视频会议 (如 Zoom E2E) | 危机会议 | 密码保护、等候室 |
| PagerDuty / Opsgenie | 告警推送 | SLA 保证 |
公共通信
| 渠道 | 用途 | 说明 |
|---|---|---|
| MSG Chain 官方论坛 | 正式公告 | 所有公告的权威来源 |
| Twitter / X (@msgchain) | 实时更新 | 简短状态更新 |
| Discord / Telegram 公开群 | 社区问答 | 设 FAQ + 管理员 |
| 项目网站 / 博客 | 事后分析报告 | 透明度报告 |
通信安全规则
┌─────────────────────────────────────────────────────┐
│ 事故通信安全规则 │
├─────────────────────────────────────────────────────┤
│ 1. 所有内部通信假设被监控 → 不讨论未公开细节 │
│ 2. 关键决策需通过 > 2 个渠道确认 │
│ 3. 敏感信息 (私钥、漏洞细节) 永不在公共渠道分享 │
│ 4. 所有对外公告需经 IC + Legal 双重审批 │
│ 5. 使用一次性密码 / 硬件密钥保护敏感通道 │
│ 6. 定期轮换通信密钥 (每次事件后立即轮换) │
└─────────────────────────────────────────────────────┘
1.6 严重等级定义
┌──────────┬─────────────────────────────────┬──────────────────┐
│ 等级 │ 定义 │ 响应时间要求 │
├──────────┼─────────────────────────────────┼──────────────────┤
│ P0 │ 资金正在流失 / 合约被完全控制 │ 立即 (≤5分钟) │
│ Critical │ 攻击仍在进行 │ │
├──────────┼─────────────────────────────────┼──────────────────┤
│ P1 │ 漏洞被发现但未利用 / │ ≤30 分钟 │
│ High │ 资金可被提取但未发生 │ │
├──────────┼─────────────────────────────────┼──────────────────┤
│ P2 │ 非关键功能异常 / │ ≤4 小时 │
│ Medium │ 不影响用户资金安全 │ │
├──────────┼─────────────────────────────────┼──────────────────┤
│ P3 │ 轻微违规 / 配置错误 / │ ≤1 周 │
│ Low │ 建议改进 │ │
└──────────┴─────────────────────────────────┴──────────────────┘
P0 确认条件（任何一条满足即触发）：
- 合约余额正在被非预期地址提取
- 代币被非授权铸造（总供应量异常增加）
- 价格预言机返回值与市场严重偏离 (> 50%)
- 管理权限被非授权地址控制
- 大量用户同时报告资产丢失
2. 事前准备
2.1 安全监控架构
┌─────────────────────────────────────────────────────────────┐
│ 安全监控架构 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 链上数据 │ │ 事件流 │ │交易Mempool│ │ 链下数据 │ │
│ │ (RPC) │ │ (WS) │ │ (Tendermint) (预言机/CEX)│ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │ │
│ ┌────▼──────────────▼─────────────▼─────────────▼──────┐ │
│ │ 监控引擎 (Monitor Engine) │ │
│ │ ├─ 交易分析器 ├─ 事件分析器 │ │
│ │ ├─ 状态差异器 ├─ Mempool 分析器 │ │
│ │ └─ 异常检测 ML └─ 指标聚合器 │ │
│ └───────────────────────┬──────────────────────────────┘ │
│ │ │
│ ┌───────────────────────▼──────────────────────────────┐ │
│ │ 告警路由 (Alert Router) │ │
│ │ ├─ 严重级别分类 ├─ 去重/聚合 │ │
│ │ ├─ 升级逻辑 └─ 静默规则 │ │
│ └──────┬──────────┬──────────┬─────────────────────────┘ │
│ │ │ │ │
│ ┌──────▼──┐ ┌─────▼─────┐ ┌─▼──────────┐ │
│ │ PagerDuty│ │ Telegram │ │ Email/SMS │ │
│ │ (P0/P1) │ │ (实时) │ │ (非紧急) │ │
│ └─────────┘ └───────────┘ └────────────┘ │
└─────────────────────────────────────────────────────────────┘
2.2 链上监控设置
2.2.1 核心监控脚本
#!/usr/bin/env python3
"""
msg_chain_monitor.py — MSG Chain 合约监控守护进程
用途: 实时监控链上事件，检测异常活动并触发告警
"""
import asyncio
import json
import logging
import os
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from decimal import Decimal
from enum import Enum
from typing import Any, Optional
import aiohttp
import httpx
# Logging 配置
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
handlers=[
logging.FileHandler("/var/log/msg_monitor.log"),
logging.StreamHandler(),
],
)
logger = logging.getLogger("msg_monitor")
# ─── 配置 ─────────────────────────────────────────────────────
RPC_ENDPOINT = os.getenv("MSG_RPC", "https://rpc.msgchain.org")
REST_ENDPOINT = os.getenv("MSG_REST", "https://rest.msgchain.org")
WS_ENDPOINT = os.getenv("MSG_WS", "wss://rpc.msgchain.org/websocket")
CONTRACT_ADDRESSES = os.getenv(
"MONITORED_CONTRACTS",
"msg1...contract1,msg1...contract2",
).split(",")
# 告警阈值
THRESHOLDS = {
"large_transfer_usd": Decimal("100000"), # $100k+
"rapid_calls_per_min": 50, # 每分钟 50 次调用
"price_deviation_pct": Decimal("15"), # 15% 价格偏差
"mass_withdrawal_count": 10, # 10 笔同时提现
"unexpected_mint_amount": Decimal("10000"), # 10k+ 非预期铸造
}
# 告警通道
ALERT_WEBHOOKS = {
"telegram": os.getenv("TELEGRAM_BOT_TOKEN", ""),
"pagerduty": os.getenv("PAGERDUTY_KEY", ""),
"slack": os.getenv("SLACK_WEBHOOK_URL", ""),
}
# ─── 数据结构 ──────────────────────────────────────────────────
class Severity(Enum):
INFO = "info"
WARNING = "warning"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class Alert:
severity: Severity
title: str
description: str
tx_hash: Optional[str] = None
contract: Optional[str] = None
affected_users: list[str] = field(default_factory=list)
estimated_loss: Optional[Decimal] = None
timestamp: int = field(default_factory=lambda: int(time.time()))
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class ContractState:
address: str
balance: dict[str, Decimal]
paused: bool = False
last_call_timestamps: list[int] = field(default_factory=list)
call_count_1min: int = 0
# ─── MSG Chain 客户端 ─────────────────────────────────────────
class MsgChainClient:
"""MSG Chain RPC/API 客户端"""
def __init__(self, rpc: str, rest: str):
self.rpc = rpc.rstrip("/")
self.rest = rest.rstrip("/")
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def get_latest_block(self) -> dict[str, Any]:
url = f"{self.rest}/cosmos/base/tendermint/v1beta1/blocks/latest"
async with self.session.get(url) as resp:
return await resp.json()
async def get_contract_balance(self, contract: str) -> list[dict]:
url = f"{self.rest}/cosmos/bank/v1beta1/balances/{contract}"
async with self.session.get(url) as resp:
data = await resp.json()
return data.get("balances", [])
async def get_events(self, event_type: str, height: int) -> list[dict]:
url = f"{self.rest}/cosmos/tx/v1beta1/txs?events={event_type}&pagination.limit=100"
async with self.session.get(url) as resp:
data = await resp.json()
return data.get("tx_responses", [])
async def get_tx(self, tx_hash: str) -> dict:
url = f"{self.rest}/cosmos/tx/v1beta1/txs/{tx_hash}"
async with self.session.get(url) as resp:
return await resp.json()
async def query_contract(self, contract: str, query: dict) -> Any:
url = f"{self.rest}/cosmwasm/wasm/v1/contract/{contract}/smart/{json.dumps(query)}"
async with self.session.get(url) as resp:
data = await resp.json()
return data.get("data")
async def subscribe_events(self, query: str):
import websockets
ws_url = WS_ENDPOINT
async with websockets.connect(ws_url) as ws:
subscribe_msg = {
"jsonrpc": "2.0",
"method": "subscribe",
"params": {"query": query},
"id": 1,
}
await ws.send(json.dumps(subscribe_msg))
async for message in ws:
yield json.loads(message)
# ─── 检测引擎 ──────────────────────────────────────────────────
class DetectionEngine:
"""异常检测引擎"""
def __init__(self, client: MsgChainClient):
self.client = client
self.states: dict[str, ContractState] = {}
self.alert_history: list[Alert] = []
self.price_feeds: dict[str, list[tuple[int, Decimal]]] = {}
async def initialize(self):
for addr in CONTRACT_ADDRESSES:
balances = await self.client.get_contract_balance(addr)
balance_map = {}
for b in balances:
balance_map[b["denom"]] = Decimal(b["amount"])
self.states[addr] = ContractState(address=addr, balance=balance_map)
logger.info(f"Initialized monitoring for {addr}")
async def detect_large_transfer(self, tx: dict, events: list[dict]) -> Optional[Alert]:
for event in events:
if event.get("type") == "transfer":
attributes = {a["key"]: a["value"] for a in event.get("attributes", [])}
amount_str = attributes.get("amount", "0umsg")
if "umsg" in amount_str:
value = Decimal(amount_str.replace("umsg", ""))
if value > THRESHOLDS["large_transfer_usd"]:
return Alert(
severity=Severity.HIGH,
title="大额转账",
description=f"检测到大额转账: {attributes.get('from', '?')} → "
f"{attributes.get('to', '?')}, 金额: {value} umsg",
tx_hash=tx.get("txhash"),
estimated_loss=value,
metadata={"event": event},
)
return None
async def detect_unexpected_mint(self, tx: dict, events: list[dict], contract: str) -> Optional[Alert]:
for event in events:
if event.get("type") == "wasm":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
if attrs.get("action") == "mint":
amount = Decimal(attrs.get("amount", "0"))
if amount > THRESHOLDS["unexpected_mint_amount"]:
return Alert(
severity=Severity.CRITICAL,
title="非预期代币铸造",
description=f"合约 {contract} 铸造 {amount} 代币 (非预期操作)",
tx_hash=tx.get("txhash"),
contract=contract,
estimated_loss=amount,
)
return None
async def detect_rapid_calls(self, contract: str, timestamp: int) -> Optional[Alert]:
state = self.states.get(contract)
if not state:
return None
now = timestamp
state.last_call_timestamps = [t for t in state.last_call_timestamps if now - t < 60]
state.last_call_timestamps.append(now)
state.call_count_1min = len(state.last_call_timestamps)
if state.call_count_1min > THRESHOLDS["rapid_calls_per_min"]:
return Alert(
severity=Severity.WARNING,
title="高频合约调用",
description=f"合约 {contract} 调用频率: {state.call_count_1min}/min (阈值: {THRESHOLDS['rapid_calls_per_min']}/min)",
contract=contract,
metadata={"call_count": state.call_count_1min},
)
return None
async def detect_price_deviation(self, contract: str, current_price: Decimal, timestamp: int) -> Optional[Alert]:
if contract not in self.price_feeds:
self.price_feeds[contract] = []
prices = self.price_feeds[contract]
prices.append((timestamp, current_price))
cutoff = timestamp - 3600
self.price_feeds[contract] = [(t, p) for t, p in prices if t > cutoff]
if len(prices) < 10:
return None
twap_cutoff = timestamp - 300
recent = [(t, p) for t, p in prices if t >= twap_cutoff]
if not recent:
return None
twap = sum(p for _, p in recent) / Decimal(len(recent))
if twap == Decimal("0"):
return None
deviation = abs(current_price - twap) / twap * Decimal("100")
if deviation > THRESHOLDS["price_deviation_pct"]:
return Alert(
severity=Severity.HIGH,
title="价格异常偏离",
description=f"合约 {contract} 价格偏离 TWAP {deviation:.2f}% (当前: {current_price}, TWAP: {twap})",
contract=contract,
metadata={"current_price": str(current_price), "twap": str(twap), "deviation_pct": str(deviation)},
)
return None
async def detect_mass_withdrawal(self, events: list[dict], window_seconds: int = 60) -> Optional[Alert]:
withdrawal_count = 0
total_amount = Decimal("0")
affected = []
for event in events:
if event.get("type") == "wasm":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
if attrs.get("action") in ("withdraw", "withdraw_funds"):
withdrawal_count += 1
amount = Decimal(attrs.get("amount", "0"))
total_amount += amount
affected.append(attrs.get("sender", "unknown"))
if withdrawal_count >= THRESHOLDS["mass_withdrawal_count"]:
return Alert(
severity=Severity.CRITICAL,
title="集中提现",
description=f"{withdrawal_count} 用户在 {window_seconds}s 内提现总计 {total_amount} umsg",
affected_users=affected,
estimated_loss=total_amount,
)
return None
async def detect_contract_pause_bypass(self, events: list[dict]) -> Optional[Alert]:
for event in events:
if event.get("type") == "wasm":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
if attrs.get("action") in ("transfer", "withdraw", "mint") and attrs.get("paused", "false") == "true":
return Alert(
severity=Severity.CRITICAL,
title="暂停绕过尝试",
description=f"合约在暂停状态下执行了 {attrs.get('action')} 操作",
tx_hash=None,
metadata={"event": event},
)
return None
async def detect_admin_role_change(self, events: list[dict], contract: str) -> Optional[Alert]:
for event in events:
if event.get("type") == "wasm":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
if attrs.get("action") in ("update_owner", "transfer_ownership", "add_admin", "remove_admin"):
return Alert(
severity=Severity.HIGH,
title="管理员权限变更",
description=f"合约 {contract} 管理员权限被变更: {attrs.get('action')} → {attrs.get('new_owner', 'N/A')}",
contract=contract,
metadata={"event": event},
)
return None
async def detect_reentrancy_pattern(self, tx: dict) -> Optional[Alert]:
events = tx.get("events", [])
call_stack = []
for event in events:
if event.get("type") == "wasm":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
contract = attrs.get("_contract_address", "")
action = attrs.get("action", "")
call_stack.append((contract, action))
contract_calls = {}
for contract, action in call_stack:
if contract not in contract_calls:
contract_calls[contract] = []
contract_calls[contract].append(action)
for contract, actions in contract_calls.items():
if len(actions) > 3 and len(set(actions)) < 3:
return Alert(
severity=Severity.CRITICAL,
title="重入攻击模式",
description=f"合约 {contract} 在同一交易中被递归调用 {len(actions)} 次",
tx_hash=tx.get("txhash"),
contract=contract,
metadata={"call_count": len(actions), "actions": actions},
)
return None
async def detect_flash_loan_pattern(self, tx: dict) -> Optional[Alert]:
wasm_events = [e for e in tx.get("events", []) if e.get("type") == "wasm"]
actions = []
for e in wasm_events:
attrs = {a["key"]: a["value"] for a in e.get("attributes", [])}
actions.append(attrs.get("action", ""))
if "borrow" in actions and "swap" in actions and "repay" in actions:
return Alert(
severity=Severity.HIGH,
title="闪电贷攻击模式",
description=f"检测到同一交易内含 borrow→swap→repay 模式",
tx_hash=tx.get("txhash"),
metadata={"actions": actions},
)
return None
async def detect_oracle_price_flash(self, events: list[dict]) -> Optional[Alert]:
for event in events:
if event.get("type") == "wasm":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
if attrs.get("action") == "oracle_update":
price = Decimal(attrs.get("price", "0"))
asset = attrs.get("asset", "")
if price == Decimal("0"):
return Alert(
severity=Severity.HIGH,
title="预言机价格归零",
description=f"资产 {asset} 预言机价格被设置为 0",
metadata={"event": event},
)
return None
# ─── 告警路由 ──────────────────────────────────────────────────
class AlertRouter:
def __init__(self, webhooks: dict[str, str]):
self.webhooks = webhooks
self.session: Optional[aiohttp.ClientSession] = None
self.dedupe_cache: set[str] = set()
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
def _dedup_key(self, alert: Alert) -> str:
return f"{alert.severity.value}:{alert.title}:{alert.contract or 'global'}"
def _should_send(self, alert: Alert) -> bool:
key = self._dedup_key(alert)
if key in self.dedupe_cache:
return False
ttl = 60 if alert.severity in (Severity.CRITICAL, Severity.HIGH) else 300
self.dedupe_cache.add(key)
return True
async def route_alert(self, alert: Alert):
if not self._should_send(alert):
return
logger.warning(f"ALERT [{alert.severity.value.upper()}] {alert.title}: {alert.description}")
tasks = []
if alert.severity in (Severity.CRITICAL, Severity.HIGH):
if self.webhooks.get("pagerduty"):
tasks.append(self._send_pagerduty(alert))
if self.webhooks.get("telegram"):
tasks.append(self._send_telegram(alert))
if self.webhooks.get("slack"):
tasks.append(self._send_slack(alert))
await asyncio.gather(*tasks)
async def _send_telegram(self, alert: Alert):
token = self.webhooks.get("telegram")
chat_id = os.getenv("TELEGRAM_CHAT_ID", "")
if not token or not chat_id:
return
emoji_map = {Severity.CRITICAL: "🚨", Severity.HIGH: "🔴", Severity.WARNING: "🟡", Severity.INFO: "ℹ️"}
emoji = emoji_map.get(alert.severity, "")
text = f"{emoji} *[{alert.severity.value.upper()}] {alert.title}*\n\n{alert.description}\n"
if alert.tx_hash:
text += f"\n[交易](https://explorer.msgchain.org/tx/{alert.tx_hash})"
if alert.contract:
text += f"\n合约: `{alert.contract}`"
if alert.estimated_loss:
text += f"\n预估损失: {alert.estimated_loss} umsg"
url = f"https://api.telegram.org/bot{token}/sendMessage"
payload = {"chat_id": chat_id, "text": text, "parse_mode": "Markdown", "disable_web_page_preview": True}
async with self.session.post(url, json=payload) as resp:
if resp.status != 200:
logger.error(f"Telegram send failed: {await resp.text()}")
async def _send_pagerduty(self, alert: Alert):
key = self.webhooks.get("pagerduty")
if not key:
return
severity_map = {Severity.CRITICAL: "critical", Severity.HIGH: "error", Severity.WARNING: "warning", Severity.INFO: "info"}
payload = {
"routing_key": key,
"event_action": "trigger",
"payload": {
"summary": f"[{alert.severity.value.upper()}] {alert.title}",
"severity": severity_map.get(alert.severity, "info"),
"source": "msg-chain-monitor",
"custom_details": {
"description": alert.description,
"tx_hash": alert.tx_hash or "",
"contract": alert.contract or "",
"timestamp": alert.timestamp,
"metadata": alert.metadata,
},
},
"dedup_key": self._dedup_key(alert),
}
url = "https://events.pagerduty.com/v2/enqueue"
async with self.session.post(url, json=payload) as resp:
if resp.status != 202:
logger.error(f"PagerDuty send failed: {await resp.text()}")
async def _send_slack(self, alert: Alert):
url = self.webhooks.get("slack")
if not url:
return
color_map = {Severity.CRITICAL: "#FF0000", Severity.HIGH: "#FF6600", Severity.WARNING: "#FFD700", Severity.INFO: "#36C5F0"}
payload = {
"attachments": [{
"color": color_map.get(alert.severity, "#36C5F0"),
"blocks": [
{"type": "header", "text": {"type": "plain_text", "text": f"[{alert.severity.value.upper()}] {alert.title}"}},
{"type": "section", "text": {"type": "mrkdwn", "text": alert.description}},
],
"ts": alert.timestamp,
}]
}
async with self.session.post(url, json=payload) as resp:
if resp.status != 200:
logger.error(f"Slack send failed: {await resp.text()}")
# ─── 主循环 ──────────────────────────────────────────────────
class MonitorDaemon:
def __init__(self):
self.client = MsgChainClient(RPC_ENDPOINT, REST_ENDPOINT)
self.engine = DetectionEngine(self.client)
self.router = AlertRouter(ALERT_WEBHOOKS)
self.last_checked_height = 0
async def run_block_poller(self, interval: int = 6):
while True:
try:
block = await self.client.get_latest_block()
current_height = int(block["block"]["header"]["height"])
if current_height > self.last_checked_height:
for h in range(self.last_checked_height + 1, current_height + 1):
await self.process_block(h)
self.last_checked_height = current_height
except Exception as e:
logger.error(f"Block poll error: {e}")
await asyncio.sleep(interval)
async def process_block(self, height: int):
events = await self.client.get_events("wasm", height)
for tx in events:
await self.process_transaction(tx)
async def process_transaction(self, tx: dict):
events = tx.get("events", [])
tx_hash = tx.get("txhash", "")
if not tx_hash:
return
alerts = []
detection_tasks = [
self.engine.detect_large_transfer(tx, events),
self.engine.detect_reentrancy_pattern(tx),
self.engine.detect_flash_loan_pattern(tx),
]
results = await asyncio.gather(*detection_tasks)
for result in results:
if result:
alerts.append(result)
for contract in CONTRACT_ADDRESSES:
contract_alerts = await asyncio.gather(
self.engine.detect_unexpected_mint(tx, events, contract),
self.engine.detect_admin_role_change(events, contract),
self.engine.detect_contract_pause_bypass(events),
self.engine.detect_oracle_price_flash(events),
self.engine.detect_mass_withdrawal(events),
)
for alert in contract_alerts:
if alert:
alerts.append(alert)
for alert in alerts:
await self.router.route_alert(alert)
async def run_health_check(self):
while True:
try:
block = await self.client.get_latest_block()
height = block["block"]["header"]["height"]
logger.info(f"Health OK - Block height: {height}")
except Exception as e:
alert = Alert(severity=Severity.HIGH, title="监控系统异常", description=f"无法获取最新区块: {e}")
await self.router.route_alert(alert)
await asyncio.sleep(60)
async def start(self):
logger.info("Starting MSG Chain Monitor Daemon...")
await self.engine.initialize()
tasks = [asyncio.create_task(self.run_block_poller()), asyncio.create_task(self.run_health_check())]
try:
await asyncio.gather(*tasks)
except KeyboardInterrupt:
logger.info("Shutting down...")
finally:
if self.client.session:
await self.client.session.close()
if self.router.session:
await self.router.session.close()
async def main():
daemon = MonitorDaemon()
await daemon.start()
if __name__ == "__main__":
asyncio.run(main())
2.2.2 轻量级健康检查
#!/usr/bin/env python3
"""msg_health_check.py — 快速健康检查脚本"""
import json, os, sys, httpx, asyncio
from datetime import datetime
REST_ENDPOINT = os.getenv("MSG_REST", "https://rest.msgchain.org")
CONTRACTS = os.getenv("MONITORED_CONTRACTS", "").split(",")
async def check_contract_health(contract: str) -> dict:
async with httpx.AsyncClient() as client:
bal_resp = await client.get(f"{REST_ENDPOINT}/cosmos/bank/v1beta1/balances/{contract}")
balance = bal_resp.json().get("balances", [])
info_resp = await client.get(f"{REST_ENDPOINT}/cosmwasm/wasm/v1/contract/{contract}")
info = info_resp.json()
try:
status_resp = await client.get(f"{REST_ENDPOINT}/cosmwasm/wasm/v1/contract/{contract}/smart/" + json.dumps({"status": {}}))
status = status_resp.json()
except:
status = {"error": "no status query"}
return {"contract": contract, "balance": balance, "paused": info.get("contract_info", {}).get("paused", False), "status": status, "checked_at": datetime.utcnow().isoformat()}
async def main():
results = []
for contract in CONTRACTS:
if not contract.strip(): continue
result = await check_contract_health(contract.strip())
results.append(result)
icon = "⚠️" if result.get("paused") else "✅"
print(f"{icon} {result['contract']}: paused={result['paused']}, balance={result['balance']}")
failed = [r for r in results if r.get("paused")]
if failed:
print(f"\n❌ {len(failed)} contract(s) are paused!")
sys.exit(1)
print(f"\n✅ All {len(results)} contracts healthy")
sys.exit(0)
if __name__ == "__main__":
asyncio.run(main())
2.3 告警阈值配置
// alert-thresholds.config.ts
export interface AlertThresholds {
largeTransfer: ThresholdRule;
rapidCalls: ThresholdRule;
priceDeviation: ThresholdRule;
massWithdrawal: ThresholdRule;
unexpectedMint: ThresholdRule;
balanceChange: ThresholdRule;
gasAnomaly: ThresholdRule;
eventAnomaly: ThresholdRule;
}
export interface ThresholdRule {
value: number;
windowSeconds: number;
severity: 'info' | 'warning' | 'high' | 'critical';
cooldownSeconds: number;
enabled: boolean;
}
export const DEFAULT_THRESHOLDS: AlertThresholds = {
largeTransfer: { value: 100_000_000_000, windowSeconds: 60, severity: 'high', cooldownSeconds: 300, enabled: true },
rapidCalls: { value: 50, windowSeconds: 60, severity: 'warning', cooldownSeconds: 120, enabled: true },
priceDeviation: { value: 15, windowSeconds: 300, severity: 'high', cooldownSeconds: 600, enabled: true },
massWithdrawal: { value: 10, windowSeconds: 60, severity: 'critical', cooldownSeconds: 60, enabled: true },
unexpectedMint: { value: 10_000_000_000, windowSeconds: 0, severity: 'critical', cooldownSeconds: 0, enabled: true },
balanceChange: { value: 20, windowSeconds: 300, severity: 'high', cooldownSeconds: 600, enabled: true },
gasAnomaly: { value: 500_000, windowSeconds: 0, severity: 'warning', cooldownSeconds: 60, enabled: true },
eventAnomaly: { value: 100, windowSeconds: 60, severity: 'high', cooldownSeconds: 300, enabled: true },
};
export interface EscalationRule {
severity: string[];
escalateTo: string[];
waitSeconds: number;
channels: string[];
}
export const ESCALATION_RULES: EscalationRule[] = [
{ severity: ['critical'], escalateTo: ['incident_commander', 'security_lead', 'comms_lead'], waitSeconds: 300, channels: ['pagerduty', 'phone', 'telegram'] },
{ severity: ['high'], escalateTo: ['security_lead', 'contract_sme'], waitSeconds: 900, channels: ['pagerduty', 'telegram'] },
];
export interface SilenceRule {
contracts?: string[];
alertPattern?: string;
startCron: string;
endCron: string;
timezone: string;
reason: string;
}
export const SILENCE_RULES: SilenceRule[] = [
{ startCron: '0 2 * * 0', endCron: '0 6 * * 0', timezone: 'UTC', reason: 'Scheduled maintenance window' },
];
2.4 紧急联系人列表
# 紧急联系人清单
## 一级联系人 (P0/P1 事件，15 分钟内必须响应)
| 角色 | 姓名 | 电话 | 备用渠道 | 时区 |
|------|------|------|----------|------|
| 事件指挥官 (IC) | [姓名 1] | [+86 138xxxx] | Telegram @user1 | UTC+8 |
| IC 备份 | [姓名 2] | [+1 415xxxx] | Signal @user2 | UTC-8 |
| 安全负责人 | [姓名 3] | [+86 139xxxx] | Telegram @user3 | UTC+8 |
| 安全备份 | [姓名 4] | [+44 20xxxx] | Discord @user4 | UTC+0 |
| 通讯负责人 | [姓名 5] | [+86 136xxxx] | Telegram @user5 | UTC+8 |
| 通讯备份 | [姓名 6] | [+1 212xxxx] | Signal @user6 | UTC-5 |
## 二级联系人 (P2 事件，4 小时内响应)
| 角色 | 姓名 | 联系方式 | 说明 |
|------|------|----------|------|
| 合约专家 | [姓名 7] | Telegram @user7 | Rust/CosmWasm |
| 合约专家 | [姓名 8] | Signal @user8 | DeFi 安全 |
| 基础设施工程师 | [姓名 9] | Telegram @user9 | Cosmos SDK |
| 法务负责人 | [姓名 10] | Email: legal@example.com | 监管合规 |
## 外部联系人
| 机构 | 联系人 | 联系方式 | 用途 |
|------|--------|----------|------|
| MSG Chain 核心开发 | [Discord] | discord.gg/msgchain | 链级别紧急操作 |
| 交易所 A | [安全团队] | security@exchangeA.com | 资金冻结 |
| 交易所 B | [合规团队] | compliance@exchangeB.com | 资金冻结 |
| 执法机构 | [网络犯罪科] | [当地报警热线] | 报案 |
| 审计公司 | [审计团队] | security@auditfirm.com | 紧急审计 |
| 保险公司 | [理赔团队] | claims@insurance.com | 保险理赔 |
## 联系优先级 (P0 事件)
第 1 梯队 (立即联系):
IC → 安全负责人 → 通讯负责人
第 2 梯队 (5 分钟内):
合约专家 → 基础设施工程师 → 法务负责人
第 3 梯队 (15 分钟内):
交易所联系人 → 外部审计 → 保险公司
第 4 梯队 (1 小时内):
执法机构 → 社区管理员 → 媒体
### 2.5 紧急 DAO 预批准提案
```typescript
// emergency-proposal.ts
import { SigningCosmWasmClient } from "@cosmjs/cosmwasm-stargate";
import { MsgExecuteContract } from "cosmjs-types/cosmwasm/wasm/v1/tx";
export interface EmergencyProposal {
id: string;
title: string;
description: string;
actions: EmergencyAction[];
requiredApprovals: number;
executorAddress: string;
timeLockSeconds: number;
}
export interface EmergencyAction {
contract: string;
message: Record<string, any>;
funds?: Coin[];
}
export const PAUSE_ALL_PROPOSAL: EmergencyProposal = {
id: "EMERGENCY-PAUSE-001",
title: "Emergency Pause All Contracts",
description: "紧急暂停所有非关键操作的合约。仅在确认重大安全事件时使用。",
actions: [
{ contract: "msg1...contract1", message: { emergency_pause: { reason: "Security incident" } } },
{ contract: "msg1...contract2", message: { pause: {} } },
],
requiredApprovals: 3,
executorAddress: "msg1...multisig",
timeLockSeconds: 0,
};
export const RATE_LIMIT_PROPOSAL: EmergencyProposal = {
id: "EMERGENCY-RATELIMIT-001",
title: "Set Emergency Rate Limits",
description: "将所有用户的操作速率限制降低到安全水平。",
actions: [{
contract: "msg1...ratelimit",
message: { set_global_rate_limit: { max_operations_per_minute: 5, max_transfer_amount: "1000000000", duration_seconds: 86400 } },
}],
requiredApprovals: 2,
executorAddress: "msg1...multisig",
timeLockSeconds: 0,
};
export const FREEZE_ASSET_PROPOSAL: EmergencyProposal = {
id: "EMERGENCY-FREEZE-001",
title: "Emergency Freeze Asset Transfers",
description: "冻结特定代币或合约的转账功能。",
actions: [{ contract: "msg1...token", message: { freeze_transfers: { until: 0 } } }],
requiredApprovals: 3,
executorAddress: "msg1...multisig",
timeLockSeconds: 0,
};
export const REPLACE_ORACLE_PROPOSAL: EmergencyProposal = {
id: "EMERGENCY-ORACLE-001",
title: "Replace Compromised Oracle",
description: "切换到备用预言机。",
actions: [{ contract: "msg1...lending", message: { set_oracle: { oracle_address: "msg1...backup-oracle", reason: "Primary oracle compromised" } } }],
requiredApprovals: 2,
executorAddress: "msg1...multisig",
timeLockSeconds: 0,
};
export const EMERGENCY_UPGRADE_PROPOSAL: EmergencyProposal = {
id: "EMERGENCY-UPGRADE-001",
title: "Emergency Contract Upgrade (Security Patch)",
description: "部署紧急安全补丁。",
actions: [{ contract: "msg1...vulnerable-contract", message: { migrate: { code_id: 42, msg: {} } } }],
requiredApprovals: 4,
executorAddress: "msg1...multisig",
timeLockSeconds: 86400,
};
export async function executeEmergencyProposal(client: SigningCosmWasmClient, proposer: string, proposal: EmergencyProposal): Promise<string> {
const msgs = proposal.actions.map(action => ({
typeUrl: "/cosmwasm.wasm.v1.MsgExecuteContract",
value: MsgExecuteContract.fromPartial({ sender: proposal.executorAddress, contract: action.contract, msg: Buffer.from(JSON.stringify(action.message)), funds: action.funds || [] }),
}));
const result = await client.signAndBroadcast(proposer, msgs, "auto", `Emergency: ${proposal.title}`);
return result.transactionHash;
}
2.6 漏洞赏金计划
# MSG Chain 漏洞赏金计划
## 计划概述
MSG Chain 运营持续漏洞赏金计划，鼓励安全研究人员发现并报告合约漏洞。
## 奖励范围
| 漏洞类型 | 严重等级 | 奖励范围 (USD) |
|----------|----------|----------------|
| 直接影响用户资金安全的漏洞 | Critical | $50,000 - $500,000 |
| 可导致合约状态异常的漏洞 | High | $10,000 - $50,000 |
| 违反最佳实践的安全缺陷 | Medium | $1,000 - $10,000 |
| 信息泄露或配置问题 | Low | $100 - $1,000 |
| 高质量审计报告或工具 | Info | 致谢 + $500 |
## 规则
1. **先报告后利用**: 任何在报告前利用漏洞的行为立即取消资格
2. **测试环境**: 优先在测试网 (testnet) 上验证
3. **最小影响**: 仅使用必要的最小测试资金
4. **保密期**: 漏洞公开前有 90 天保密期
5. **无竞争**: 同一漏洞按首次报告时间为准
## 报告渠道
- **主要**: security@msgchain.org (PGP 加密)
- **备用**: https://immunefi.com/project/msgchain
- **紧急**: Telegram @msgchain_security (仅 P0)
## 响应承诺
- **确认接收**: 24 小时内
- **漏洞分类**: 48 小时内
- **修复计划**: 5 个工作日内
- **赏金支付**: 修复确认后 14 天内
2.7 保险覆盖
// insurance-config.ts
export interface InsurancePolicy {
provider: string;
policyNumber: string;
coverageAmount: string;
premium: string;
deductible: string;
coverageType: string[];
exclusions: string[];
validFrom: string;
validTo: string;
claimsProcess: string;
}
export const RECOMMENDED_COVERAGE: InsurancePolicy[] = [
{
provider: "Nexus Mutual",
policyNumber: "NXM-MSG-001",
coverageAmount: "$5,000,000",
premium: "$125,000/年 (2.5%)",
deductible: "$50,000",
coverageType: ["smart_contract_failure", "price_oracle_failure", "economic_attack"],
exclusions: ["已知漏洞未修复", "治理攻击（若未启用多签）", "战争或政府行为"],
validFrom: "2026-01-01",
validTo: "2026-12-31",
claimsProcess: "1. 事件发生后 72 小时内提交初步报告\n2. 提供完整的事后分析报告\n3. 审计公司出具漏洞确认报告\n4. 评估赔付金额\n5. 14 个工作日内赔付",
},
{
provider: "Sherlock",
policyNumber: "SH-MSG-002",
coverageAmount: "$2,000,000",
premium: "$60,000/年 (3%)",
deductible: "$25,000",
coverageType: ["smart_contract_failure", "bridge_exploit"],
exclusions: ["外部合约依赖导致的损失", "前端安全漏洞"],
validFrom: "2026-03-01",
validTo: "2027-02-28",
claimsProcess: "1. 通过 Sherlock 平台提交理赔\n2. 上传安全事件报告和交易哈希\n3. Sherlock 安全团队审核\n4. 社区投票确认赔付",
},
];
2.8 安全演练计划
#!/usr/bin/env python3
"""security_drill.py — 安全事件模拟演练"""
import asyncio, json, logging, os, random, sys
from datetime import datetime
from typing import Any, Optional
logger = logging.getLogger("security_drill")
DRILL_SCENARIOS = {
"reentrancy_attack": {
"name": "重入攻击模拟",
"description": "模拟攻击者利用 reply handler 进行重入攻击",
"steps": ["部署包含重入漏洞的测试合约", "攻击者合约发起 withdraw 调用", "在 reply 中递归调用 withdraw", "检测监控系统是否告警"],
"expected_alerts": ["重入攻击模式", "大额转账"],
"expected_response_time": 300,
},
"price_oracle_manipulation": {
"name": "预言机价格操纵模拟",
"description": "模拟攻击者操纵预言机价格进行套利",
"steps": ["部署测试合约和模拟预言机", "发送大额 swap 交易操纵价格", "攻击者在价格偏离后套利", "检测价格偏差告警"],
"expected_alerts": ["价格异常偏离", "闪电贷攻击模式"],
"expected_response_time": 300,
},
"mass_withdrawal": {
"name": "集中提现模拟",
"description": "模拟多个账户同时提现",
"steps": ["创建 20 个测试账户并充值", "同时在 10 秒内发起全部提现", "检测集中提现告警"],
"expected_alerts": ["集中提现"],
"expected_response_time": 120,
},
"unauthorized_mint": {
"name": "非授权铸造模拟",
"description": "模拟攻击者利用权限漏洞铸造代币",
"steps": ["部署测试合约（模拟权限错误）", "非授权账户调用 mint 函数", "铸造大额代币", "检测非预期铸造告警"],
"expected_alerts": ["非预期代币铸造"],
"expected_response_time": 120,
},
"admin_role_takeover": {
"name": "管理员权限接管模拟",
"description": "模拟攻击者获取合约管理员权限",
"steps": ["模拟合约所有权转移操作", "新管理员执行敏感操作", "检测管理员权限变更告警"],
"expected_alerts": ["管理员权限变更"],
"expected_response_time": 300,
},
}
class SecurityDrillRunner:
def __init__(self, scenario_name: str):
self.scenario = DRILL_SCENARIOS.get(scenario_name)
if not self.scenario:
raise ValueError(f"Unknown scenario: {scenario_name}")
self.start_time: Optional[datetime] = None
self.results: dict = {}
async def run(self):
self.start_time = datetime.utcnow()
logger.info(f"=== 开始演练: {self.scenario['name']} ===")
print(f"安全演练: {self.scenario['name']}")
print(f"描述: {self.scenario['description']}")
print(f"预期告警: {', '.join(self.scenario['expected_alerts'])}")
print(f"预期响应: {self.scenario['expected_response_time']} 秒以内")
for i, step in enumerate(self.scenario["steps"], 1):
print(f"[{i}/{len(self.scenario['steps'])}] {step}")
self.results = {"scenario": self.scenario["name"], "started_at": self.start_time.isoformat(), "status": "in_progress"}
async def complete(self, success: bool):
elapsed = (datetime.utcnow() - self.start_time).total_seconds()
self.results["completed_at"] = datetime.utcnow().isoformat()
self.results["elapsed_seconds"] = elapsed
self.results["status"] = "success" if success else "failure"
print(f"演练完成: {'成功' if success else '失败'}")
print(f"耗时: {elapsed:.0f} 秒")
report_path = f"/var/log/drill_{self.scenario['name']}_{self.start_time.strftime('%Y%m%d_%H%M%S')}.json"
with open(report_path, "w") as f:
json.dump(self.results, f, indent=2)
print(f"报告已保存: {report_path}")
return self.results
async def main():
if len(sys.argv) < 2:
print(f"用法: python security_drill.py <scenario_name>")
print(f"可用场景: {', '.join(DRILL_SCENARIOS.keys())}")
sys.exit(1)
runner = SecurityDrillRunner(sys.argv[1])
await runner.run()
input("\n演练步骤已完成，按 Enter 继续...")
await runner.complete(success=True)
if __name__ == "__main__":
asyncio.run(main())
2.9 事前准备检查清单
## 事前准备检查清单
### □ 监控系统
- [ ] 链上事件监控已部署并运行 (> 7 天)
- [ ] 所有被监控合约的初始余额已记录
- [ ] 告警阈值已根据 TVL 和交易量调整
- [ ] 所有告警通道已测试 (Telegram/PagerDuty/Slack)
- [ ] WebSocket 订阅已建立并验证
- [ ] 监控系统自身有健康检查和告警
- [ ] 日志持久化和备份已配置
- [ ] 历史数据可查询 (≥ 90 天)
### □ 响应团队
- [ ] 所有角色已分配且成员确认
- [ ] 24/7 值班表已制定并公告
- [ ] 备份联系人已确认
- [ ] 加密通信渠道已建立
- [ ] 每季度演练已完成
- [ ] 角色权限矩阵已更新
### □ 合约能力
- [ ] 紧急暂停功能已实现并测试
- [ ] 速率限制功能已实现
- [ ] 多签治理已配置
- [ ] 合约升级路径已定义
- [ ] 资金快照功能已实现
- [ ] 所有管理员函数有多签保护
- [ ] 紧急提现功能已测试
### □ 外部联系
- [ ] 交易所联系人已确认
- [ ] 审计公司 retainer 已签署
- [ ] 保险公司保单已生效
- [ ] 法律顾问已就位
- [ ] 执法部门联系方式已备案
- [ ] 漏洞赏金平台已激活
### □ 文档
- [ ] 本应急响应指南已分发给所有成员
- [ ] 技术架构图已更新
- [ ] 私钥管理系统已审计
- [ ] 部署流程文档化
- [ ] 通信模板已准备
继续阅读第二部分...
3. 攻击检测
3.1 检测层次
检测层次
├── 第 1 层: 链上事件监控 (实时)
│ ├── wasm 事件订阅 (WebSocket)
│ ├── 交易内容分析
│ └── 状态差异检测
├── 第 2 层: 异常行为检测 (准实时)
│ ├── 交易模式分析
│ ├── 关联地址分析
│ └── 时序分析
├── 第 3 层: 经济安全监控 (分钟级)
│ ├── 价格偏差检测
│ ├── 流动性异常
│ └── 套利模式识别
├── 第 4 层: 链下数据交叉验证 (分钟-小时级)
│ ├── CEX 价格对比
│ ├── 社交媒体监控
│ └── 安全社区情报
└── 第 5 层: 安全研究 (被动)
├── Mempool 分析
├── 合约字节码扫描
└── 已知漏洞模式匹配
3.2 TypeScript 攻击检测引擎
// attack-detection-engine.ts
import { CosmWasmClient, IndexedTx } from "@cosmjs/cosmwasm-stargate";
import { Event } from "@cosmjs/stargate";
export interface SecurityEvent {
id: string;
type: SecurityEventType;
severity: "critical" | "high" | "medium" | "low";
title: string;
description: string;
txHash?: string;
contractAddress?: string;
blockHeight?: number;
timestamp: number;
affectedUsers?: string[];
estimatedLoss?: string;
rawData?: Record<string, unknown>;
}
export enum SecurityEventType {
LARGE_TRANSFER = "large_transfer",
UNEXPECTED_MINT = "unexpected_mint",
PRICE_DEVIATION = "price_deviation",
MASS_WITHDRAWAL = "mass_withdrawal",
REENTRANCY = "reentrancy",
FLASH_LOAN_ATTACK = "flash_loan_attack",
ADMIN_ROLE_CHANGE = "admin_role_change",
CONTRACT_MIGRATION = "contract_migration",
SUSPICIOUS_CODE_DEPLOY = "suspicious_code_deploy",
RAPID_CALLS = "rapid_calls",
PAUSE_BYPASS = "pause_bypass",
ORACLE_PRICE_ZERO = "oracle_price_zero",
UNUSUAL_GAS = "unusual_gas",
}
export interface DetectionRule {
id: string;
type: SecurityEventType;
enabled: boolean;
threshold: number;
windowMs: number;
cooldownMs: number;
severity: SecurityEvent["severity"];
}
export class AttackDetectionEngine {
private client: CosmWasmClient;
private rules: Map<string, DetectionRule>;
private state: Map<string, unknown>;
private eventCache: SecurityEvent[] = [];
private cooldowns: Map<string, number> = new Map();
constructor(rpcEndpoint: string, rules: DetectionRule[] = defaultRules) {
this.client = new CosmWasmClient(rpcEndpoint);
this.rules = new Map(rules.map(r => [r.id, r]));
this.state = new Map();
}
async connect(): Promise<void> {
await this.client.connect();
console.log(`Connected to MSG Chain at ${this.client.url}`);
}
private isInCooldown(ruleId: string): boolean {
const cooldownUntil = this.cooldowns.get(ruleId);
if (!cooldownUntil) return false;
if (Date.now() < cooldownUntil) return true;
this.cooldowns.delete(ruleId);
return false;
}
private setCooldown(ruleId: string, ms: number): void {
this.cooldowns.set(ruleId, Date.now() + ms);
}
private eventAttributes(event: Event): Record<string, string> {
const attrs: Record<string, string> = {};
for (const attr of event.attributes) {
attrs[attr.key] = attr.value;
}
return attrs;
}
async analyzeTransaction(tx: IndexedTx): Promise<SecurityEvent[]> {
const events: SecurityEvent[] = [];
const txHash = tx.hash;
const height = tx.height;
const wasmEvents = tx.events.filter(e => e.type === "wasm");
const transferEvents = tx.events.filter(e => e.type === "transfer");
const detections = await Promise.allSettled([
this.detectLargeTransfer(transferEvents, txHash, height),
this.detectUnexpectedMint(wasmEvents, txHash, height),
this.detectReentrancyPattern(wasmEvents, txHash, height),
this.detectFlashLoanPattern(tx, wasmEvents, txHash, height),
this.detectAdminChange(wasmEvents, txHash, height),
this.detectOracleAnomaly(wasmEvents, txHash, height),
this.detectSuspiciousDeploy(tx, txHash, height),
this.detectUnusualGas(tx, txHash, height),
]);
for (const result of detections) {
if (result.status === "fulfilled" && result.value) {
events.push(result.value);
}
}
return events;
}
private async detectLargeTransfer(transferEvents: Event[], txHash: string, height: number): Promise<SecurityEvent | null> {
const rule = this.rules.get("large_transfer");
if (!rule?.enabled || this.isInCooldown("large_transfer")) return null;
for (const event of transferEvents) {
const attrs = this.eventAttributes(event);
const amountStr = attrs.amount || "0";
const match = amountStr.match(/^(\d+)(\w+)$/);
if (!match) continue;
const amount = BigInt(match[1]);
if (amount > BigInt(rule.threshold)) {
this.setCooldown("large_transfer", rule.cooldownMs);
return {
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.LARGE_TRANSFER, severity: rule.severity,
title: "大额转账检测",
description: `从 ${attrs.from} 向 ${attrs.to} 转账 ${amountStr}`,
txHash, blockHeight: height, timestamp: Date.now(),
estimatedLoss: amountStr, rawData: attrs,
};
}
}
return null;
}
private async detectUnexpectedMint(wasmEvents: Event[], txHash: string, height: number): Promise<SecurityEvent | null> {
const rule = this.rules.get("unexpected_mint");
if (!rule?.enabled) return null;
for (const event of wasmEvents) {
const attrs = this.eventAttributes(event);
if (attrs.action === "mint" || attrs.action === "mint_tokens") {
const amount = BigInt(attrs.amount || "0");
if (amount > BigInt(rule.threshold)) {
return {
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.UNEXPECTED_MINT, severity: rule.severity,
title: "非预期代币铸造",
description: `检测到非授权铸造: ${amount} 由 ${attrs.minter || attrs.sender} 铸造`,
txHash, contractAddress: attrs._contract_address, blockHeight: height, timestamp: Date.now(),
estimatedLoss: amount.toString(), rawData: attrs,
};
}
}
}
return null;
}
private async detectReentrancyPattern(wasmEvents: Event[], txHash: string, height: number): Promise<SecurityEvent | null> {
const rule = this.rules.get("reentrancy");
if (!rule?.enabled) return null;
const contractCalls = new Map<string, string[]>();
for (const event of wasmEvents) {
const attrs = this.eventAttributes(event);
const contract = attrs._contract_address;
const action = attrs.action;
if (contract && action) {
if (!contractCalls.has(contract)) contractCalls.set(contract, []);
contractCalls.get(contract)!.push(action);
}
}
for (const [contract, actions] of contractCalls.entries()) {
if (actions.length > 3 && new Set(actions).size <= 2) {
return {
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.REENTRANCY, severity: rule.severity,
title: "潜在重入攻击",
description: `合约 ${contract} 在同一交易中被调用 ${actions.length} 次`,
txHash, contractAddress: contract, blockHeight: height, timestamp: Date.now(),
rawData: { callCount: actions.length, actions: [...new Set(actions)] },
};
}
}
return null;
}
private async detectFlashLoanPattern(tx: IndexedTx, wasmEvents: Event[], txHash: string, height: number): Promise<SecurityEvent | null> {
const rule = this.rules.get("flash_loan");
if (!rule?.enabled) return null;
const actions: string[] = [];
for (const event of wasmEvents) {
const attrs = this.eventAttributes(event);
if (attrs.action) actions.push(attrs.action);
}
if (actions.includes("borrow") && actions.some(a => a.includes("swap")) && actions.includes("repay")) {
return {
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.FLASH_LOAN_ATTACK, severity: rule.severity,
title: "闪电贷攻击模式",
description: "检测到 borrow → swap → repay 模式",
txHash, blockHeight: height, timestamp: Date.now(), rawData: { actions },
};
}
return null;
}
private async detectAdminChange(wasmEvents: Event[], txHash: string, height: number): Promise<SecurityEvent | null> {
const rule = this.rules.get("admin_change");
if (!rule?.enabled) return null;
const adminActions = ["update_owner", "transfer_ownership", "add_admin", "remove_admin", "set_guardian", "update_config"];
for (const event of wasmEvents) {
const attrs = this.eventAttributes(event);
if (adminActions.includes(attrs.action)) {
return {
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.ADMIN_ROLE_CHANGE, severity: rule.severity,
title: "管理员权限变更",
description: `操作 ${attrs.action} 由 ${attrs.sender || "unknown"} 执行`,
txHash, contractAddress: attrs._contract_address, blockHeight: height, timestamp: Date.now(), rawData: attrs,
};
}
}
return null;
}
private async detectOracleAnomaly(wasmEvents: Event[], txHash: string, height: number): Promise<SecurityEvent | null> {
const rule = this.rules.get("oracle_anomaly");
if (!rule?.enabled) return null;
for (const event of wasmEvents) {
const attrs = this.eventAttributes(event);
if (attrs.action === "oracle_update" && (attrs.price === "0" || attrs.price === "0.0")) {
return {
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.ORACLE_PRICE_ZERO, severity: rule.severity,
title: "预言机价格归零",
description: `资产 ${attrs.asset} 的预言机价格被设置为 0`,
txHash, contractAddress: attrs._contract_address, blockHeight: height, timestamp: Date.now(), rawData: attrs,
};
}
}
return null;
}
private async detectSuspiciousDeploy(tx: IndexedTx, txHash: string, height: number): Promise<SecurityEvent | null> {
const rule = this.rules.get("suspicious_deploy");
if (!rule?.enabled) return null;
for (const event of tx.events) {
if (event.type === "instantiate" || event.type === "store_code") {
const attrs = this.eventAttributes(event);
if (KNOWN_MALICIOUS_CODE_IDS.includes(attrs.code_id)) {
return {
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.SUSPICIOUS_CODE_DEPLOY, severity: "critical",
title: "已知恶意合约部署",
description: `已知恶意 code_id ${attrs.code_id} 被部署`,
txHash, blockHeight: height, timestamp: Date.now(), rawData: attrs,
};
}
}
}
return null;
}
private async detectUnusualGas(tx: IndexedTx, txHash: string, height: number): Promise<SecurityEvent | null> {
const rule = this.rules.get("unusual_gas");
if (!rule?.enabled) return null;
if (tx.gasUsed > rule.threshold) {
return {
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.UNUSUAL_GAS, severity: rule.severity,
title: "异常 Gas 消耗",
description: `交易消耗 ${tx.gasUsed} gas, 超过阈值 ${rule.threshold}`,
txHash, blockHeight: height, timestamp: Date.now(),
rawData: { gasUsed: tx.gasUsed, gasWanted: tx.gasWanted },
};
}
return null;
}
async analyzeBlock(blockEvents: Event[], height: number): Promise<SecurityEvent[]> {
const events: SecurityEvent[] = [];
const rule = this.rules.get("mass_withdrawal");
if (!rule?.enabled) return events;
const withdrawals: { user: string; amount: string }[] = [];
for (const event of blockEvents) {
if (event.type === "wasm") {
const attrs = this.eventAttributes(event);
if (attrs.action === "withdraw" || attrs.action === "withdraw_funds") {
withdrawals.push({ user: attrs.sender || attrs.from || "unknown", amount: attrs.amount || "0" });
}
}
}
const uniqueUsers = new Set(withdrawals.map(w => w.user));
if (uniqueUsers.size >= rule.threshold) {
events.push({
id: `alert-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`,
type: SecurityEventType.MASS_WITHDRAWAL, severity: rule.severity,
title: "集中提现检测",
description: `${uniqueUsers.size} 个用户集中提现`,
blockHeight: height, timestamp: Date.now(),
affectedUsers: [...uniqueUsers],
});
}
return events;
}
async startPolling(intervalMs = 6_000): Promise<void> {
console.log("Starting MSG Chain attack detection engine...");
let lastHeight = 0;
while (true) {
try {
const block = await this.client.getBlock();
const currentHeight = block.header.height;
if (currentHeight > lastHeight) {
for (let h = lastHeight + 1; h <= currentHeight; h++) {
const txs = await this.client.searchTx(`tx.height=${h}`);
await Promise.allSettled(txs.map(tx => this.analyzeTransaction(tx)));
const blockEvents = txs.flatMap(tx => tx.events);
await this.analyzeBlock(blockEvents, h);
}
lastHeight = currentHeight;
}
} catch (error) {
console.error("Polling error:", error);
await this.reconnect();
}
await new Promise(resolve => setTimeout(resolve, intervalMs));
}
}
private async reconnect(): Promise<void> {
for (let i = 0; i < 5; i++) {
try {
await this.client.disconnect();
await this.client.connect();
return;
} catch {
await new Promise(resolve => setTimeout(resolve, 5000));
}
}
throw new Error("Failed to reconnect");
}
}
const defaultRules: DetectionRule[] = [
{ id: "large_transfer", type: SecurityEventType.LARGE_TRANSFER, enabled: true, threshold: 100_000_000_000, windowMs: 60_000, cooldownMs: 300_000, severity: "high" },
{ id: "unexpected_mint", type: SecurityEventType.UNEXPECTED_MINT, enabled: true, threshold: 10_000_000_000, windowMs: 0, cooldownMs: 0, severity: "critical" },
{ id: "reentrancy", type: SecurityEventType.REENTRANCY, enabled: true, threshold: 3, windowMs: 0, cooldownMs: 60_000, severity: "critical" },
{ id: "flash_loan", type: SecurityEventType.FLASH_LOAN_ATTACK, enabled: true, threshold: 0, windowMs: 0, cooldownMs: 300_000, severity: "high" },
{ id: "mass_withdrawal", type: SecurityEventType.MASS_WITHDRAWAL, enabled: true, threshold: 10, windowMs: 6_000, cooldownMs: 60_000, severity: "critical" },
{ id: "admin_change", type: SecurityEventType.ADMIN_ROLE_CHANGE, enabled: true, threshold: 0, windowMs: 0, cooldownMs: 60_000, severity: "high" },
{ id: "oracle_anomaly", type: SecurityEventType.ORACLE_PRICE_ZERO, enabled: true, threshold: 0, windowMs: 0, cooldownMs: 60_000, severity: "high" },
{ id: "suspicious_deploy", type: SecurityEventType.SUSPICIOUS_CODE_DEPLOY, enabled: true, threshold: 0, windowMs: 0, cooldownMs: 0, severity: "critical" },
{ id: "unusual_gas", type: SecurityEventType.UNUSUAL_GAS, enabled: true, threshold: 2_000_000, windowMs: 0, cooldownMs: 60_000, severity: "medium" },
];
const KNOWN_MALICIOUS_CODE_IDS: string[] = [];
3.3 Webhook 告警服务
// webhook-alert-service.ts
import express from "express";
import axios from "axios";
interface WebhookPayload {
eventType: string;
contractAddress?: string;
txHash?: string;
blockHeight: number;
attributes: Record<string, string>;
timestamp: number;
}
interface NotificationChannel {
name: string;
type: "telegram" | "slack" | "discord" | "email" | "pagerduty";
config: Record<string, string>;
severityFilter: string[];
}
const CHANNELS: NotificationChannel[] = [
{ name: "Telegram Emergency", type: "telegram", config: { botToken: process.env.TELEGRAM_BOT_TOKEN || "", chatId: process.env.TELEGRAM_CHAT_ID || "" }, severityFilter: ["critical", "high"] },
{ name: "PagerDuty", type: "pagerduty", config: { routingKey: process.env.PAGERDUTY_KEY || "" }, severityFilter: ["critical", "high"] },
{ name: "Slack All Alerts", type: "slack", config: { webhookUrl: process.env.SLACK_WEBHOOK_URL || "" }, severityFilter: ["critical", "high", "medium", "low"] },
{ name: "Discord Community", type: "discord", config: { webhookUrl: process.env.DISCORD_WEBHOOK_URL || "" }, severityFilter: ["critical"] },
];
class WebhookAlertService {
private channels: NotificationChannel[];
private alertHistory: Map<string, number> = new Map();
constructor(channels: NotificationChannel[]) { this.channels = channels; }
async processAlert(payload: WebhookPayload): Promise<void> {
const dedupKey = `${payload.txHash}:${payload.eventType}`;
const now = Date.now();
const lastSent = this.alertHistory.get(dedupKey);
if (lastSent && now - lastSent < 60_000) return;
this.alertHistory.set(dedupKey, now);
const severity = this.inferSeverity(payload);
const title = this.formatTitle(payload, severity);
const desc = this.formatDescription(payload);
await Promise.allSettled(
this.channels.filter(c => c.severityFilter.includes(severity)).map(c => this.sendToChannel(c, severity, title, desc))
);
}
private inferSeverity(p: WebhookPayload): string {
return { unexpected_mint: "critical", pause_bypass: "critical", unauthorized_migration: "critical",
large_transfer: "high", price_deviation: "high", admin_role_change: "high", mass_withdrawal: "high",
rapid_calls: "medium", oracle_price_zero: "medium", unusual_gas: "medium" }[p.eventType] || "low";
}
private formatTitle(p: WebhookPayload, s: string): string {
const prefix = { critical: "🚨 [CRITICAL]", high: "🔴 [HIGH]", medium: "🟡 [MEDIUM]", low: "ℹ️ [LOW]" }[s] || "ℹ️";
const labels: Record<string, string> = { large_transfer: "大额转账", unexpected_mint: "非预期铸造", price_deviation: "价格异常", mass_withdrawal: "集中提现", reentrancy: "重入攻击", flash_loan_attack: "闪电贷", admin_role_change: "权限变更", contract_migration: "合约迁移", suspicious_code_deploy: "可疑合约", rapid_calls: "高频调用", pause_bypass: "暂停绕过", oracle_price_zero: "预言机归零", unusual_gas: "异常 Gas" };
return `${prefix} ${labels[p.eventType] || p.eventType}`;
}
private formatDescription(p: WebhookPayload): string {
const lines: string[] = [];
for (const [k, v] of Object.entries(p.attributes)) if (k !== "_contract_address") lines.push(`• ${k}: ${v}`);
if (p.txHash) lines.push(`• 交易: ${p.txHash}`);
if (p.contractAddress) lines.push(`• 合约: ${p.contractAddress}`);
if (p.blockHeight) lines.push(`• 区块: ${p.blockHeight}`);
return lines.join("\n");
}
private async sendToChannel(ch: NotificationChannel, severity: string, title: string, desc: string): Promise<void> {
try {
if (ch.type === "telegram") await this.sendTelegram(ch.config, title, desc);
else if (ch.type === "slack") await this.sendSlack(ch.config, severity, title, desc);
else if (ch.type === "discord") await this.sendDiscord(ch.config, title, desc);
else if (ch.type === "pagerduty") await this.sendPagerDuty(ch.config, severity, title, desc);
} catch (e) { console.error(`Failed to send to ${ch.name}:`, e); }
}
private async sendTelegram(config: Record<string, string>, title: string, desc: string): Promise<void> {
const { botToken, chatId } = config;
if (!botToken || !chatId) return;
await axios.post(`https://api.telegram.org/bot${botToken}/sendMessage`, { chat_id: chatId, text: `*${title}*\n\n${desc}`, parse_mode: "Markdown" });
}
private async sendSlack(config: Record<string, string>, severity: string, title: string, desc: string): Promise<void> {
if (!config.webhookUrl) return;
await axios.post(config.webhookUrl, { attachments: [{ color: { critical: "#FF0000", high: "#FF6600", medium: "#FFD700", low: "#36C5F0" }[severity] || "#36C5F0", title, text: desc }] });
}
private async sendDiscord(config: Record<string, string>, title: string, desc: string): Promise<void> {
if (!config.webhookUrl) return;
await axios.post(config.webhookUrl, { embeds: [{ title, description: desc, color: 0xFF0000, timestamp: new Date().toISOString() }] });
}
private async sendPagerDuty(config: Record<string, string>, severity: string, title: string, desc: string): Promise<void> {
if (!config.routingKey) return;
await axios.post("https://events.pagerduty.com/v2/enqueue", { routing_key: config.routingKey, event_action: "trigger", payload: { summary: title, severity: { critical: "critical", high: "error", medium: "warning", low: "info" }[severity] || "info", source: "msg-chain-webhook", custom_details: { description: desc } } });
}
}
const app = express();
app.use(express.json());
const alertService = new WebhookAlertService(CHANNELS);
app.post("/webhook/alert", async (req, res) => {
try { await alertService.processAlert(req.body); res.status(200).json({ status: "ok" }); }
catch (e) { res.status(500).json({ status: "error", message: String(e) }); }
});
app.get("/health", (_req, res) => res.json({ status: "healthy", timestamp: new Date().toISOString() }));
app.listen(process.env.PORT || 8080, () => console.log(`Alert service running on port ${process.env.PORT || 8080}`));
3.4 Mempool 监控
// mempool-monitor.ts
import { WebsocketClient } from "@cosmjs/tendermint-rpc";
interface MempoolTransaction {
hash: string;
sender: string;
contract?: string;
msgs: any[];
gasPrice: string;
nonce: number;
timestamp: number;
}
class MempoolMonitor {
private wsClient: WebsocketClient;
private pendingTxs: Map<string, MempoolTransaction> = new Map();
constructor(rpcEndpoint: string) {
this.wsClient = new WebsocketClient(rpcEndpoint);
}
async start(): Promise<void> {
console.log("Starting mempool monitor...");
const stream = this.wsClient.subscribe("tm.event='Tx'");
for await (const event of stream) {
try {
const tx = event.events[0];
if (tx) await this.analyzeMempoolTx(tx);
} catch (error) { console.error("Mempool error:", error); }
}
}
private async analyzeMempoolTx(tx: MempoolTransaction): Promise<void> {
if (this.isSandwichAttack(tx)) console.warn("Potential sandwich attack:", tx.hash);
if (this.isFrontRunning(tx)) console.warn("Potential front-running:", tx.hash);
if (this.isAbnormalGasPrice(tx)) console.warn("Abnormal gas price:", tx.hash);
this.pendingTxs.set(tx.hash, tx);
setTimeout(() => this.pendingTxs.delete(tx.hash), 30_000);
}
private isSandwichAttack(tx: MempoolTransaction): boolean {
if (!tx.contract) return false;
const related = Array.from(this.pendingTxs.values()).filter(t => t.contract === tx.contract && t.hash !== tx.hash);
return related.length >= 2;
}
private isFrontRunning(tx: MempoolTransaction): boolean {
return [/mev/i, /sandwich/i, /bot/i].some(p => p.test(tx.sender));
}
private isAbnormalGasPrice(tx: MempoolTransaction): boolean {
return BigInt(tx.gasPrice) > BigInt(1_000_000);
}
}
3.5 链下监控
// off-chain-monitor.ts
import axios from "axios";
class OffChainMonitor {
private priceFeeds: Map<string, { source: string; price: number; timestamp: number }[]> = new Map();
async start(): Promise<void> {
await Promise.all([this.monitorCEXPrices(), this.monitorSocialMedia(), this.monitorSecurityFeeds()]);
}
private async monitorCEXPrices(): Promise<void> {
const assets = ["MSG", "USDC", "USDT"];
while (true) {
for (const asset of assets) {
for (const exchange of ["binance", "coinbase"]) {
try {
const price = await this.fetchCEXPrice(exchange, asset);
const feeds = this.priceFeeds.get(asset) || [];
feeds.push({ source: exchange, price, timestamp: Date.now() });
if (feeds.length > 100) feeds.shift();
this.priceFeeds.set(asset, feeds);
} catch {}
}
}
await new Promise(resolve => setTimeout(resolve, 30_000));
}
}
private async fetchCEXPrice(exchange: string, asset: string): Promise<number> {
const urls: Record<string, string> = {
binance: `https://api.binance.com/api/v3/ticker/price?symbol=${asset}USDT`,
coinbase: `https://api.coinbase.com/v2/prices/${asset}-USD/spot`,
};
const resp = await axios.get(urls[exchange]);
if (exchange === "binance") return parseFloat(resp.data.price);
if (exchange === "coinbase") return parseFloat(resp.data.data.amount);
return 0;
}
private async monitorSocialMedia(): Promise<void> {
const keywords = ["MSG Chain hack", "MSG Chain exploit", "msgchain rug"];
while (true) {
for (const kw of keywords) {
console.log(`Monitoring social: ${kw}`);
}
await new Promise(resolve => setTimeout(resolve, 60_000));
}
}
private async monitorSecurityFeeds(): Promise<void> {
const feeds = ["https://rekt.news/feed.json"];
while (true) {
for (const feed of feeds) {
try {
const resp = await axios.get(feed, { timeout: 10_000 });
const content = JSON.stringify(resp.data).toLowerCase();
if (content.includes("cosmwasm") || content.includes("msg chain")) {
console.warn(`Security feed alert from ${feed}`);
}
} catch {}
}
await new Promise(resolve => setTimeout(resolve, 300_000));
}
}
}
3.6 攻击模式签名参考
## 常见攻击模式与链上签名
| 攻击类型 | 描述 | 链上签名 | 检测指标 |
|----------|------|----------|----------|
| 重入攻击 | 利用 reply handler 递归调用 | 同一合约在同一交易中被多次调用 (> 3 次) | wasm._contract_address 重复 |
| 闪电贷攻击 | 同一交易内借贷 + 价格操纵 | wasm.action 含 borrow/swap/repay 序列 | 大额借贷 + 价格偏差 |
| 三明治攻击 | 用户交易前后插入交易 | 连续交易中发送者地址交替 | gas 价格模式异常 |
| 预言机操纵 | 操纵链上价格源 | wasm.action=oracle_update 价格剧烈变化 | 价格偏离 TWAP > 15% |
| 非授权铸造 | 利用权限漏洞铸造 | wasm.action=mint 且发送者非授权 | wasm.amount 超过阈值 |
| 权限提升 | 利用存储/迁移漏洞 | wasm.action 涉及 update_owner 等 | 调用者非预期 |
| Gas Griefing | 消耗大量 gas 使执行失败 | gas_used > 2M 但状态变更极小 | gas_used/gas_wanted 异常 |
| IBC 超时攻击 | 利用 IBC 超时盗取资金 | ibc_packet_timeout 后非预期退款 | 退款地址非原发送者 |
| 悬崖抛售 | 大量抛售导致价格崩溃 | 短时间内大量卖出 | 价格 1 小时内跌 > 50% |
| 治理攻击 | 利用治理漏洞通过恶意提案 | 新创建地址集中投票 | 投票通过率突增 |
| Agent 宪法违规 | 绕过宪法限制执行操作 | CONSTITUTION_VIOLATION 后成功执行 | Agent 操作未通过验证 |
4. 应急响应流程
4.1 流程总览
事故报告/告警
│
▼
┌──────────────────┐
│ 第 1 阶段: Triage │ ≤ 5 分钟
│ ──────────────── │
│ □ 确认事件 │
│ □ 评估严重等级 │
│ □ 组建响应团队 │
│ □ 创建事件日志 │
└────────┬─────────┘
│
P0/P1 │ P2/P3
│
┌────────▼─────────┐
│ 第 2 阶段: 遏制 │ ≤ 15 分钟
│ ────────────── │
│ □ 暂停合约 │
│ □ 修改断路器 │
│ □ 速率限制 │
│ □ 联系验证者 │
│ □ 资金快照 │
└────────┬──────────┘
│
┌────────▼─────────┐
│ 第 3 阶段: 调查 │ ≤ 4 小时
│ ────────────── │
│ □ 重放攻击交易 │
│ □ 分析攻击向量 │
│ □ 追踪资金流向 │
│ □ 量化影响 │
└────────┬──────────┘
│
┌────────▼─────────┐
│ 第 4 阶段: 恢复 │ 24h - 2 周
│ ────────────── │
│ □ 开发修复补丁 │
│ □ 审计补丁 │
│ □ 部署修复 │
│ □ 追回资金 │
└────────┬──────────┘
│
┌────────▼─────────┐
│ 第 5 阶段: 复盘 │ 1 - 2 周
│ ────────────── │
│ □ 编写事故报告 │
│ □ 复盘会议 │
│ □ 更新安全流程 │
└────────────────────┘
4.2 第 1 阶段: Triage (≤ 5 分钟)
#!/usr/bin/env python3
"""triage.py — 事件分类分级工具"""
import json, sys
from datetime import datetime
from enum import Enum
from typing import Optional
class IncidentSeverity(Enum):
P0_CRITICAL = ("P0", "Critical", 0, "立即响应 (≤5分钟)")
P1_HIGH = ("P1", "High", 1, "紧急 (≤30分钟)")
P2_MEDIUM = ("P2", "Medium", 2, "常规 (≤4小时)")
P3_LOW = ("P3", "Low", 3, "排期 (≤1周)")
def __init__(self, code, name, priority, response_time):
self.code, self.severity_name, self.priority, self.response_time = code, name, priority, response_time
P0_CONDITIONS = {
"ongoing_fund_drain": "资金正在被提取",
"unauthorized_mint": "非授权代币铸造",
"contract_fully_controlled": "合约被完全控制",
"price_oracle_compromised": "预言机被操纵",
"mass_user_fund_loss": "大量用户同时报告资产丢失",
}
P1_CONDITIONS = {
"vulnerability_disclosed_unexploited": "漏洞被发现但未利用",
"funds_at_risk": "资金可被提取但尚未发生",
"admin_key_compromised": "管理员密钥泄露",
"governance_attack_in_progress": "治理攻击进行中",
"bridge_anomaly": "跨链桥异常",
}
class IncidentTriage:
def __init__(self):
self.incident_id: Optional[str] = None
self.severity: Optional[IncidentSeverity] = None
self.title: str = ""
self.description: str = ""
self.reported_by: str = ""
self.reported_at: str = ""
self.affected_contracts: list[str] = []
self.initial_evidence: dict = {}
self.team_members: list[str] = []
def create_incident(self, title: str, description: str, reported_by: str, evidence: Optional[dict] = None) -> str:
self.incident_id = f"INC-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}"
self.title, self.description, self.reported_by, self.reported_at = title, description, reported_by, datetime.utcnow().isoformat()
self.initial_evidence = evidence or {}
print(f"新安全事件: {self.incident_id}\n标题: {title}\n报告人: {reported_by}\n时间: {self.reported_at}")
return self.incident_id
def assess_severity(self) -> IncidentSeverity:
for condition, desc in P0_CONDITIONS.items():
if condition in str(self.initial_evidence).lower() or condition in self.description.lower():
self.severity = IncidentSeverity.P0_CRITICAL
print(f"P0 判定: {desc}\n响应要求: {self.severity.response_time}")
return self.severity
for condition, desc in P1_CONDITIONS.items():
if condition in str(self.initial_evidence).lower() or condition in self.description.lower():
self.severity = IncidentSeverity.P1_HIGH
print(f"P1 判定: {desc}\n响应要求: {self.severity.response_time}")
return self.severity
self.severity = IncidentSeverity.P2_MEDIUM
print(f"P2 判定: 非关键安全事件\n响应要求: {self.severity.response_time}")
return self.severity
def assemble_team(self) -> list[str]:
team_map = {IncidentSeverity.P0_CRITICAL: ["事件指挥官", "安全负责人", "通讯负责人", "法务负责人", "合约专家 (×2)", "基础设施工程师", "追回专家"],
IncidentSeverity.P1_HIGH: ["事件指挥官", "安全负责人", "通讯负责人", "合约专家"],
IncidentSeverity.P2_MEDIUM: ["安全负责人", "合约专家"],
IncidentSeverity.P3_LOW: ["安全负责人"]}
self.team_members = team_map.get(self.severity, ["安全负责人"])
print(f"\n响应团队 ({len(self.team_members)} 人):")
for m in self.team_members: print(f" • {m}")
return self.team_members
def run(self) -> dict:
if not self.incident_id: raise ValueError("必须先调用 create_incident()")
self.assess_severity()
self.assemble_team()
entry = {"incident_id": self.incident_id, "severity": self.severity.code, "title": self.title,
"description": self.description, "reported_by": self.reported_by, "reported_at": self.reported_at,
"team_members": self.team_members, "evidence": self.initial_evidence, "status": "triage_completed"}
log_path = f"/var/log/incidents/{self.incident_id}.json"
with open(log_path, "w") as f: json.dump(entry, f, indent=2, default=str)
print(f"事件日志已创建: {log_path}")
if self.severity in (IncidentSeverity.P0_CRITICAL, IncidentSeverity.P1_HIGH):
print("需要立即进入 Containment 阶段!")
return entry
def main():
if len(sys.argv) < 3:
print("用法: python triage.py <title> <description> [--evidence key=val ...]")
sys.exit(1)
triage = IncidentTriage()
triage.create_incident(sys.argv[1], sys.argv[2], "system")
triage.run()
if __name__ == "__main__":
main()
4.3 第 2 阶段: Containment
// containment.ts
import { SigningCosmWasmClient } from "@cosmjs/cosmwasm-stargate";
import { DirectSecp256k1HdWallet } from "@cosmjs/proto-signing";
import { calculateFee, GasPrice } from "@cosmjs/stargate";
interface ContainmentActions {
pauseContracts: string[];
rateLimit?: { maxOpsPerMin: number; durationSec: number };
freezeAssets?: string[];
notifyValidators: boolean;
snapshotFunds: boolean;
}
class IncidentContainment {
private client!: SigningCosmWasmClient;
private multisigAddress: string;
constructor(mnemonic: string, multisigAddress: string) {
this.multisigAddress = multisigAddress;
this.initialize(mnemonic);
}
private async initialize(mnemonic: string): Promise<void> {
const wallet = await DirectSecp256k1HdWallet.fromMnemonic(mnemonic, { prefix: "msg" });
this.client = await SigningCosmWasmClient.connectWithSigner("https://rpc.msgchain.org", wallet);
}
async executeContainment(actions: ContainmentActions): Promise<{ pauseResults: string[] }> {
const results: { pauseResults: string[] } = { pauseResults: [] };
const fee = calculateFee(500_000, GasPrice.fromString("1000000000umsg"));
if (actions.pauseContracts.length > 0) {
console.log(`Pausing ${actions.pauseContracts.length} contracts...`);
results.pauseResults = await Promise.all(
actions.pauseContracts.map(async (contract) => {
try {
const result = await this.client.execute(this.multisigAddress, contract,
{ emergency_pause: { reason: "Security incident" } }, fee, "Emergency pause");
console.log(` Paused: ${contract} (tx: ${result.transactionHash})`);
return result.transactionHash;
} catch (error) {
console.error(`Failed to pause ${contract}:`, error);
return `FAILED: ${error}`;
}
})
);
}
if (actions.rateLimit) {
try {
const result = await this.client.execute(this.multisigAddress, actions.pauseContracts[0],
{ set_rate_limit: { max_operations_per_minute: actions.rateLimit.maxOpsPerMin, duration_seconds: actions.rateLimit.durationSec } },
fee, "Emergency rate limit");
console.log(`Rate limit set: tx ${result.transactionHash}`);
} catch (error) {
console.error("Failed to set rate limit:", error);
}
}
if (actions.freezeAssets && actions.freezeAssets.length > 0) {
for (const asset of actions.freezeAssets) {
try {
const result = await this.client.execute(this.multisigAddress, asset,
{ freeze_transfers: { until: 0 } }, fee, "Emergency asset freeze");
console.log(`Frozen: ${asset} (tx: ${result.transactionHash})`);
} catch (error) {
console.error(`Failed to freeze ${asset}:`, error);
}
}
}
if (actions.snapshotFunds) {
console.log("Funds snapshot triggered");
}
return results;
}
}
4.4 第 3 阶段: Investigation
#!/usr/bin/env python3
"""investigation.py — 攻击分析工具"""
import json, os, sys
from datetime import datetime
from typing import Optional
import httpx
REST_ENDPOINT = os.getenv("MSG_REST", "https://rest.msgchain.org")
class IncidentInvestigation:
def __init__(self, incident_id: str):
self.incident_id = incident_id
self.attack_tx_hashes: list[str] = []
self.attacker_addresses: list[str] = []
self.affected_contracts: list[str] = []
self.root_cause: Optional[str] = None
self.attack_vector: Optional[str] = None
self.stolen_funds: dict[str, str] = {}
self.fund_flow: list[dict] = []
async def add_attack_tx(self, tx_hash: str):
self.attack_tx_hashes.append(tx_hash)
async with httpx.AsyncClient() as client:
resp = await client.get(f"{REST_ENDPOINT}/cosmos/tx/v1beta1/txs/{tx_hash}")
data = resp.json()
tx = data.get("tx_response", {})
events = tx.get("events", [])
for event in events:
if event.get("type") == "wasm":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
print(f" Event: {attrs.get('action', 'unknown')} - {attrs}")
if "sender" in attrs and attrs["sender"] not in self.attacker_addresses:
self.attacker_addresses.append(attrs["sender"])
if "_contract_address" in attrs and attrs["_contract_address"] not in self.affected_contracts:
self.affected_contracts.append(attrs["_contract_address"])
async def trace_funds(self, from_address: str, depth: int = 3):
"""追踪资金流向"""
print(f"Tracing funds from {from_address} (depth={depth})...")
if depth <= 0:
return
async with httpx.AsyncClient() as client:
resp = await client.get(f"{REST_ENDPOINT}/cosmos/tx/v1beta1/txs?events=transfer.sender={from_address}&pagination.limit=50")
data = resp.json()
txs = data.get("tx_responses", [])
for tx in txs:
for event in tx.get("events", []):
if event.get("type") == "transfer":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
self.fund_flow.append({
"from": attrs.get("sender", from_address),
"to": attrs.get("recipient", ""),
"amount": attrs.get("amount", ""),
"tx_hash": tx.get("txhash", ""),
"height": tx.get("height", ""),
"timestamp": datetime.utcnow().isoformat(),
})
# 递归追踪
if attrs.get("recipient"):
await self.trace_funds(attrs["recipient"], depth - 1)
def analyze_attack_vector(self) -> str:
"""分析攻击向量"""
# 重入检测
if any("reply" in str(tx).lower() for tx in self.attack_tx_hashes):
self.attack_vector = "reentrancy"
self.root_cause = "合约未遵循 Checks-Effects-Interactions 模式"
# 闪电贷检测
elif any("borrow" in str(self.fund_flow).lower()):
self.attack_vector = "flash_loan"
self.root_cause = "预言机价格未使用 TWAP 或缺乏滑点保护"
# 权限攻击
elif any("unauthorized" in str(self.fund_flow).lower()):
self.attack_vector = "unauthorized_access"
self.root_cause = "管理员函数缺少权限检查"
else:
self.attack_vector = "unknown"
self.root_cause = "待进一步分析"
print(f"攻击向量: {self.attack_vector}")
print(f"根因: {self.root_cause}")
return self.attack_vector
def quantify_impact(self) -> dict:
"""量化影响"""
total_stolen = sum(
int(v.replace("umsg", "").replace("uconst", ""))
for v in self.stolen_funds.values()
if v.replace("umsg", "").replace("uconst", "").isdigit()
)
return {
"incident_id": self.incident_id,
"total_stolen_umsg": total_stolen,
"attacker_addresses": self.attacker_addresses,
"affected_contracts": self.affected_contracts,
"transaction_count": len(self.attack_tx_hashes),
"attack_vector": self.attack_vector,
"root_cause": self.root_cause,
}
async def generate_report(self) -> dict:
report = {
"incident_id": self.incident_id,
"investigation_completed_at": datetime.utcnow().isoformat(),
"attack_tx_hashes": self.attack_tx_hashes,
"attacker_addresses": self.attacker_addresses,
"affected_contracts": self.affected_contracts,
"fund_flow": self.fund_flow,
"attack_vector": self.attack_vector,
"root_cause": self.root_cause,
"stolen_funds": self.stolen_funds,
}
report_path = f"/var/log/incidents/{self.incident_id}_investigation.json"
with open(report_path, "w") as f:
json.dump(report, f, indent=2, default=str)
print(f"调查报告已保存: {report_path}")
return report
4.5 Containment 检查清单
# P0 事件 Containment 检查清单 (15 分钟内完成)
## 0-2 分钟: 确认和通知
- [ ] 确认事件真实存在 (非误报)
- [ ] 通知事件指挥官 (IC)
- [ ] 在 Signal/Telegram 群组发第一条消息
- [ ] 确定攻击是否仍在进行
## 2-5 分钟: 快速遏制
- [ ] 暂停所有非关键合约 (通过多签)
- [ ] 调用 emergency_pause()
- [ ] 设置全局速率限制 (max 5 ops/min)
- [ ] 记录当前区块高度的合约余额快照
## 5-10 分钟: 扩大遏制
- [ ] 通知验证者做好链暂停准备
- [ ] 联系交易所冻结可疑地址
- [ ] 启动资金追踪
- [ ] 记录攻击者地址和交易哈希
## 10-15 分钟: 评估和计划
- [ ] 确认遏制措施有效
- [ ] 更新社区 (说明已暂停)
- [ ] 分配后续任务
- [ ] 开始详细调查
5. 合约级应急措施
5.1 紧急暂停实现
// emergency_pause.rs
// MSG Chain 合约紧急暂停机制
use cosmwasm_std::{
entry_point, to_binary, Binary, Deps, DepsMut, Env, MessageInfo,
Response, StdError, StdResult, Storage,
};
use cw_storage_plus::Item;
use serde::{Deserialize, Serialize};
// ─── 数据结构 ─────────────────────────────────────────────────
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct EmergencyConfig {
pub guardian: String, // 监护人地址 (多签)
pub backup_guardian: String, // 备用监护人
pub governance: Option<String>, // 治理合约地址 (可选)
pub timelock_seconds: u64, // 时间锁 (紧急情况下可设为 0)
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct PauseInfo {
pub paused: bool,
pub paused_by: Option<String>,
pub paused_at: Option<u64>,
pub pause_reason: Option<String>,
pub operations_paused: Vec<String>, // 被暂停的具体操作
}
// ─── 存储 ──────────────────────────────────────────────────────
const EMERGENCY_CONFIG: Item<EmergencyConfig> = Item::new("emergency_config_v1");
const PAUSE_INFO: Item<PauseInfo> = Item::new("pause_info_v1");
// ─── 初始化 ─────────────────────────────────────────────────────
pub fn initialize_emergency(
store: &mut dyn Storage,
guardian: String,
backup: String,
governance: Option<String>,
) -> StdResult<()> {
let config = EmergencyConfig {
guardian,
backup_guardian: backup,
governance,
timelock_seconds: 86400, // 默认 24 小时时间锁
};
EMERGENCY_CONFIG.save(store, &config)?;
let pause = PauseInfo {
paused: false,
paused_by: None,
paused_at: None,
pause_reason: None,
operations_paused: vec![],
};
PAUSE_INFO.save(store, &pause)?;
Ok(())
}
// ─── 权限检查 ──────────────────────────────────────────────────
pub fn assert_guardian(deps: &Deps, sender: &str) -> StdResult<()> {
let config = EMERGENCY_CONFIG.load(deps.storage)?;
if sender != config.guardian && sender != config.backup_guardian {
return Err(StdError::generic_err("Unauthorized: not a guardian"));
}
Ok(())
}
pub fn assert_not_paused(deps: &Deps) -> StdResult<()> {
let pause = PAUSE_INFO.load(deps.storage)?;
if pause.paused {
return Err(StdError::generic_err("Contract is paused"));
}
Ok(())
}
// ─── Execute ───────────────────────────────────────────────────
#[entry_point]
pub fn execute(deps: DepsMut, env: Env, info: MessageInfo, msg: ExecuteMsg) -> Result<Response, ContractError> {
match msg {
ExecuteMsg::EmergencyPause { reason, operations } => {
execute_pause(deps, env, info, reason, operations)
}
ExecuteMsg::EmergencyUnpause => execute_unpause(deps, env, info),
ExecuteMsg::SetGuardian { new_guardian } => set_guardian(deps, env, info, new_guardian),
ExecuteMsg::EmergencyWithdraw { recipient, denom, amount } => {
emergency_withdraw(deps, env, info, recipient, denom, amount)
}
ExecuteMsg::UpdateOperationsPaused { operations } => {
update_paused_ops(deps, env, info, operations)
}
_ => execute_normal(deps, env, info, msg),
}
}
// ─── 暂停 ──────────────────────────────────────────────────────
pub fn execute_pause(
deps: DepsMut,
env: Env,
info: MessageInfo,
reason: Option<String>,
operations: Option<Vec<String>>,
) -> Result<Response, ContractError> {
// 仅 guardian 或 governance 可暂停
let config = EMERGENCY_CONFIG.load(deps.storage)?;
let is_guardian = info.sender.as_str() == config.guardian
|| info.sender.as_str() == config.backup_guardian;
let is_governance = config.governance
.as_ref()
.map(|g| info.sender.as_str() == g.as_str())
.unwrap_or(false);
if !is_guardian && !is_governance {
return Err(ContractError::Unauthorized("Only guardian or governance can pause".into()));
}
let mut pause = PAUSE_INFO.load(deps.storage)?;
if pause.paused {
return Err(ContractError::AlreadyPaused {});
}
pause.paused = true;
pause.paused_by = Some(info.sender.to_string());
pause.paused_at = Some(env.block.time.seconds());
pause.pause_reason = reason;
pause.operations_paused = operations.unwrap_or_default();
PAUSE_INFO.save(deps.storage, &pause)?;
Ok(Response::new()
.add_attribute("action", "emergency_pause")
.add_attribute("paused_by", info.sender)
.add_attribute("paused_at", env.block.time.seconds().to_string()))
}
// ─── 解暂停 ────────────────────────────────────────────────────
pub fn execute_unpause(
deps: DepsMut,
env: Env,
info: MessageInfo,
) -> Result<Response, ContractError> {
let config = EMERGENCY_CONFIG.load(deps.storage)?;
if info.sender.as_str() != config.guardian
&& info.sender.as_str() != config.backup_guardian
{
return Err(ContractError::Unauthorized("Only guardian can unpause".into()));
}
let mut pause = PAUSE_INFO.load(deps.storage)?;
if !pause.paused {
return Err(ContractError::NotPaused {});
}
pause.paused = false;
pause.paused_by = None;
pause.paused_at = None;
pause.pause_reason = None;
pause.operations_paused = vec![];
PAUSE_INFO.save(deps.storage, &pause)?;
Ok(Response::new()
.add_attribute("action", "emergency_unpause")
.add_attribute("unpaused_by", info.sender))
}
// ─── 紧急提现 ───────────────────────────────────────────────────
pub fn emergency_withdraw(
deps: DepsMut,
env: Env,
info: MessageInfo,
recipient: String,
denom: String,
amount: Uint128,
) -> Result<Response, ContractError> {
let config = EMERGENCY_CONFIG.load(deps.storage)?;
if info.sender.as_str() != config.guardian {
return Err(ContractError::Unauthorized("Only guardian can withdraw".into()));
}
let recipient_addr = deps.api.addr_validate(&recipient)?;
let bank_msg = BankMsg::Send {
to_address: recipient_addr.to_string(),
amount: vec![Coin::new(amount.u128(), &denom)],
};
Ok(Response::new()
.add_message(bank_msg)
.add_attribute("action", "emergency_withdraw")
.add_attribute("recipient", recipient)
.add_attribute("amount", amount.to_string())
.add_attribute("denom", denom))
}
// ─── 更新监护人 ────────────────────────────────────────────────
pub fn set_guardian(
deps: DepsMut,
_env: Env,
info: MessageInfo,
new_guardian: String,
) -> Result<Response, ContractError> {
let config = EMERGENCY_CONFIG.load(deps.storage)?;
if info.sender.as_str() != config.guardian {
return Err(ContractError::Unauthorized("Only current guardian can transfer".into()));
}
deps.api.addr_validate(&new_guardian)?;
EMERGENCY_CONFIG.update(deps.storage, |mut c| -> StdResult<_> {
c.guardian = new_guardian.clone();
Ok(c)
})?;
Ok(Response::new()
.add_attribute("action", "set_guardian")
.add_attribute("new_guardian", new_guardian))
}
// ─── 操作级暂停 ────────────────────────────────────────────────
pub fn update_paused_ops(
deps: DepsMut,
_env: Env,
info: MessageInfo,
operations: Vec<String>,
) -> Result<Response, ContractError> {
assert_guardian(&deps.as_ref(), info.sender.as_str())?;
PAUSE_INFO.update(deps.storage, |mut p| -> StdResult<_> {
p.operations_paused = operations.clone();
Ok(p)
})?;
Ok(Response::new()
.add_attribute("action", "update_paused_operations")
.add_attribute("count", operations.len().to_string()))
}
// ─── 通用操作检查 ──────────────────────────────────────────────
pub fn check_operation_allowed(deps: &Deps, operation: &str) -> StdResult<()> {
let pause = PAUSE_INFO.load(deps.storage)?;
if pause.paused {
// 如果暂停列表为空，暂停所有操作
if pause.operations_paused.is_empty() {
return Err(StdError::generic_err("Contract is paused"));
}
// 如果操作在暂停列表中
if pause.operations_paused.contains(&operation.to_string()) {
return Err(StdError::generic_err(format!("Operation '{}' is paused", operation)));
}
}
Ok(())
}
5.2 速率限制
// rate_limiter.rs
use cosmwasm_std::{DepsMut, Env, MessageInfo, Response, StdError, StdResult, Storage};
use cw_storage_plus::{Item, Map};
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct RateLimitConfig {
pub enabled: bool,
pub max_operations_per_minute: u32,
pub max_transfer_amount: Uint128,
pub cooldown_seconds: u64,
pub whitelist: Vec<String>, // 白名单地址
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct UserRateLimit {
pub operation_count: u32,
pub window_start: u64,
pub total_transferred: Uint128,
}
const RATE_LIMIT_CONFIG: Item<RateLimitConfig> = Item::new("ratelimit_config_v1");
const USER_RATE_LIMITS: Map<&Addr, UserRateLimit> = Map::new("user_ratelimit_v1");
pub fn check_rate_limit(deps: &DepsMut, env: &Env, sender: &Addr, amount: Uint128) -> StdResult<()> {
let config = RATE_LIMIT_CONFIG.load(deps.storage)?;
if !config.enabled {
return Ok(());
}
// 白名单跳过
if config.whitelist.contains(&sender.to_string()) {
return Ok(());
}
let mut user_limit = USER_RATE_LIMITS
.may_load(deps.storage, sender)?
.unwrap_or(UserRateLimit {
operation_count: 0,
window_start: env.block.time.seconds(),
total_transferred: Uint128::zero(),
});
let now = env.block.time.seconds();
let window_elapsed = now - user_limit.window_start;
// 如果窗口过期，重置
if window_elapsed >= 60 {
user_limit.operation_count = 0;
user_limit.total_transferred = Uint128::zero();
user_limit.window_start = now;
}
// 检查操作频率
if user_limit.operation_count >= config.max_operations_per_minute {
return Err(StdError::generic_err("RATE_LIMIT_EXCEEDED: Too many operations"));
}
// 检查转账总额
let new_total = user_limit.total_transferred.checked_add(amount)
.map_err(|_| StdError::generic_err("Transfer amount overflow"))?;
if new_total > config.max_transfer_amount {
return Err(StdError::generic_err("RATE_LIMIT_EXCEEDED: Transfer limit exceeded"));
}
// 更新状态
user_limit.operation_count += 1;
user_limit.total_transferred = new_total;
USER_RATE_LIMITS.save(deps.storage, sender, &user_limit)?;
Ok(())
}
5.3 断路器 (Circuit Breaker)
// circuit_breaker.rs
use cosmwasm_std::{DepsMut, Env, MessageInfo, Response, StdError, StdResult, Uint128};
use cw_storage_plus::Item;
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct CircuitBreakerState {
pub tripped: bool,
pub tripped_at: Option<u64>,
pub tripped_by: Option<String>,
pub reason: Option<String>,
pub threshold_balance_change_pct: u64, // 余额变化触发百分比
pub threshold_call_frequency: u32, // 调用频率阈值
pub last_balance_check: Uint128,
}
const CIRCUIT_BREAKER: Item<CircuitBreakerState> = Item::new("circuit_breaker_v1");
pub fn check_circuit_breaker(
deps: &DepsMut,
env: &Env,
current_balance: Uint128,
call_count: u32,
) -> StdResult<()> {
let state = CIRCUIT_BREAKER.load(deps.storage)?;
if state.tripped {
return Err(StdError::generic_err("Circuit breaker is tripped"));
}
// 检查余额变化
if !state.last_balance_check.is_zero() {
let diff = if current_balance > state.last_balance_check {
current_balance - state.last_balance_check
} else {
state.last_balance_check - current_balance
};
let change_pct = diff * Uint128::new(100) / state.last_balance_check;
if change_pct.u128() > state.threshold_balance_change_pct as u128 {
trip_circuit_breaker(deps.storage, env, "Balance change exceeded threshold")?;
return Err(StdError::generic_err("Circuit breaker tripped: balance anomaly"));
}
}
// 检查调用频率
if call_count > state.threshold_call_frequency {
trip_circuit_breaker(deps.storage, env, "Call frequency exceeded threshold")?;
return Err(StdError::generic_err("Circuit breaker tripped: high call frequency"));
}
Ok(())
}
fn trip_circuit_breaker(
store: &mut dyn Storage,
env: &Env,
reason: &str,
) -> StdResult<()> {
CIRCUIT_BREAKER.update(store, |mut s| -> StdResult<_> {
s.tripped = true;
s.tripped_at = Some(env.block.time.seconds());
s.reason = Some(reason.to_string());
Ok(s)
})?;
Ok(())
}
pub fn reset_circuit_breaker(
deps: &DepsMut,
env: &Env,
caller: &str,
guardian: &str,
) -> StdResult<Response> {
if caller != guardian {
return Err(StdError::generic_err("Unauthorized"));
}
CIRCUIT_BREAKER.update(deps.storage, |mut s| -> StdResult<_> {
s.tripped = false;
s.tripped_at = None;
s.tripped_by = Some(caller.to_string());
s.reason = None;
Ok(s)
})?;
Ok(Response::new()
.add_attribute("action", "circuit_breaker_reset")
.add_attribute("reset_by", caller))
}
5.4 资金快照
// fund_snapshot.rs
use cosmwasm_std::{DepsMut, Env, MessageInfo, Response, StdResult, Uint128, Storage};
use cw_storage_plus::Map;
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct BalanceSnapshot {
pub block_height: u64,
pub timestamp: u64,
pub balances: Vec<BalanceEntry>,
pub total_value_locked: Uint128,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct BalanceEntry {
pub address: String,
pub denom: String,
pub amount: Uint128,
}
const SNAPSHOTS: Map<u64, BalanceSnapshot> = Map::new("balance_snapshots_v1");
pub fn take_snapshot(
deps: DepsMut,
env: Env,
info: MessageInfo,
addresses: Vec<String>,
) -> StdResult<Response> {
// 仅 guardian 或授权地址可触发
let mut balances = vec![];
for addr in addresses {
let addr_validated = deps.api.addr_validate(&addr)?;
let balance = deps.querier.query_all_balances(&addr_validated)?;
for coin in balance {
balances.push(BalanceEntry {
address: addr.clone(),
denom: coin.denom,
amount: coin.amount,
});
}
}
let total_value_locked = balances.iter()
.fold(Uint128::zero(), |acc, b| acc + b.amount);
let snapshot = BalanceSnapshot {
block_height: env.block.height,
timestamp: env.block.time.seconds(),
balances,
total_value_locked,
};
SNAPSHOTS.save(deps.storage, &env.block.height, &snapshot)?;
Ok(Response::new()
.add_attribute("action", "take_snapshot")
.add_attribute("height", env.block.height.to_string())
.add_attribute("tvl", total_value_locked.to_string()))
}
pub fn get_snapshot(
store: &dyn Storage,
height: u64,
) -> StdResult<BalanceSnapshot> {
SNAPSHOTS.load(store, &height)
}
5.5 事件日志
// security_events.rs
use cosmwasm_std::{Storage, Uint128};
use cw_storage_plus::Map;
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct SecurityEventLog {
pub id: u64,
pub event_type: String, // pause, unpause, withdraw, rate_limit, circuit_breaker
pub severity: String, // info, warning, high, critical
pub actor: String,
pub target: Option<String>,
pub amount: Option<Uint128>,
pub reason: Option<String>,
pub block_height: u64,
pub timestamp: u64,
pub tx_hash: Option<String>,
}
const EVENT_LOGS: Map<u64, SecurityEventLog> = Map::new("security_events_v1");
const EVENT_COUNTER: Item<u64> = Item::new("event_counter_v1");
pub fn log_security_event(
store: &mut dyn Storage,
event_type: &str,
severity: &str,
actor: &str,
target: Option<&str>,
amount: Option<Uint128>,
reason: Option<&str>,
block_height: u64,
timestamp: u64,
tx_hash: Option<&str>,
) -> StdResult<u64> {
let mut counter = EVENT_COUNTER.may_load(store)?.unwrap_or(0);
counter += 1;
let event = SecurityEventLog {
id: counter,
event_type: event_type.to_string(),
severity: severity.to_string(),
actor: actor.to_string(),
target: target.map(|s| s.to_string()),
amount,
reason: reason.map(|s| s.to_string()),
block_height,
timestamp,
tx_hash: tx_hash.map(|s| s.to_string()),
};
EVENT_LOGS.save(store, &counter, &event)?;
EVENT_COUNTER.save(store, &counter)?;
Ok(counter)
}
5.6 合约级应急措施检查清单
# 合约级应急措施实现检查清单
### □ 紧急暂停
- [ ] emergency_pause() 仅 guardian 可调用
- [ ] emergency_unpause() 仅 guardian 可调用
- [ ] 暂停后所有非关键操作返回错误
- [ ] 暂停状态可查询
- [ ] 暂停原因记录在事件中
- [ ] 支持操作级暂停 (选择性暂停)
### □ 速率限制
- [ ] max_operations_per_minute 可配置
- [ ] max_transfer_amount 可配置
- [ ] 白名单地址可跳过限制
- [ ] 窗口自动重置 (每分钟)
- [ ] 达到限制后返回 RATE_LIMIT_EXCEEDED
### □ 断路器
- [ ] 余额突变阈值可配置
- [ ] 调用频率阈值可配置
- [ ] 触发后所有操作被阻止
- [ ] 仅 guardian 可重置断路器
- [ ] 断路器状态可查询
### □ 资金快照
- [ ] 可手动触发快照
- [ ] 快照包含所有用户余额
- [ ] 快照按区块高度索引
- [ ] 快照可查询和导出
### □ 安全事件日志
- [ ] 所有安全操作被记录
- [ ] 日志不可篡改 (仅 append)
- [ ] 日志可查询 (分页)
- [ ] 日志包含完整上下文
6. 资金追回
6.1 资金追踪流程
资金追踪流程
├── 第 1 步: 识别初始攻击交易
│ ├── 从告警中获取攻击者地址
│ ├── 从事件日志中找到首笔异常交易
│ └── 确认攻击合约地址
├── 第 2 步: 链上追踪
│ ├── 追踪攻击者地址的所有转账
│ ├── 识别 layer-2 桥/混币器
│ ├── 追踪到 CEX 充值地址
│ └── 绘制资金流向图
├── 第 3 步: 交易所联络
│ ├── 识别资金进入的交易所
│ ├── 发送正式冻结请求
│ ├── 提供交易哈希和地址证明
│ └── 确认冻结执行
├── 第 4 步: 执法协作
│ ├── 整理完整的证据链
│ ├── 联系当地执法部门
│ ├── 通过 Chainalysis/慢雾 等机构
│ └── 获取法律冻结令
└── 第 5 步: 资金回收
├── 白帽回收操作
├── 治理投票回收
├── 链升级/回滚
└── 保险理赔
6.2 资金追踪工具
#!/usr/bin/env python3
"""fund_tracker.py — 资金追踪工具"""
import asyncio
import json
import os
from datetime import datetime
from typing import Optional
import httpx
REST_ENDPOINT = os.getenv("MSG_REST", "https://rest.msgchain.org")
EXPLORER_API = os.getenv("EXPLORER_API", "https://explorer.msgchain.org/api")
# 已知交易所充值地址前缀（示例）
EXCHANGE_ADDRESSES = {
"binance": ["msg1binance..."],
"coinbase": ["msg1coinbase..."],
"okx": ["msg1okx..."],
}
class FundTracker:
def __init__(self):
self.flow_graph: dict[str, list[dict]] = {}
self.exchange_detected: list[dict] = []
self.total_stolen: int = 0
async def trace_address(
self,
address: str,
depth: int = 5,
max_txs: int = 100,
) -> list[dict]:
"""追踪地址资金流向"""
print(f"Tracing: {address} (depth={depth})")
if depth <= 0:
return []
async with httpx.AsyncClient() as client:
# 查询地址的所有转账
url = f"{REST_ENDPOINT}/cosmos/tx/v1beta1/txs"
params = {
"events": f"transfer.sender={address}",
"pagination.limit": min(max_txs, 100),
"order_by": "ORDER_BY_DESC",
}
try:
resp = await client.get(url, params=params)
data = resp.json()
except Exception as e:
print(f" Error querying {address}: {e}")
return []
txs = data.get("tx_responses", [])
results = []
for tx in txs:
tx_hash = tx.get("txhash", "")
height = tx.get("height", "")
for event in tx.get("events", []):
if event.get("type") == "transfer":
attrs = {a["key"]: a["value"] for a in event.get("attributes", [])}
from_addr = attrs.get("sender", "")
to_addr = attrs.get("recipient", "")
amount = attrs.get("amount", "")
if from_addr.lower() == address.lower():
entry = {
"from": from_addr,
"to": to_addr,
"amount": amount,
"tx_hash": tx_hash,
"height": height,
"timestamp": datetime.utcnow().isoformat(),
}
results.append(entry)
if address not in self.flow_graph:
self.flow_graph[address] = []
self.flow_graph[address].append(entry)
# 检测是否进入交易所
for exchange, addrs in EXCHANGE_ADDRESSES.items():
if to_addr.lower() in [a.lower() for a in addrs]:
self.exchange_detected.append({
"exchange": exchange,
"address": to_addr,
"amount": amount,
"tx_hash": tx_hash,
})
print(f" Funds reached exchange: {exchange} ({amount})")
# 递归追踪下一跳
nested = await self.trace_address(to_addr, depth - 1, max_txs)
results.extend(nested)
return results
def calculate_total_stolen(self) -> int:
"""计算总被盗金额"""
total = 0
for addr, txs in self.flow_graph.items():
for tx in txs:
amount_str = tx.get("amount", "0")
if "umsg" in amount_str:
try:
total += int(amount_str.replace("umsg", ""))
except ValueError:
pass
self.total_stolen = total
return total
def generate_flow_report(self) -> dict:
"""生成资金流向报告"""
return {
"total_stolen_umsg": self.total_stolen,
"unique_addresses_involved": len(self.flow_graph),
"flow_graph": self.flow_graph,
"exchange_detected": self.exchange_detected,
"generated_at": datetime.utcnow().isoformat(),
}
def export_for_exchange(self, exchange: str) -> dict:
"""生成发送给交易所的证据包"""
relevant = [e for e in self.exchange_detected if e["exchange"] == exchange]
return {
"incident_report": {
"chain": "MSG Chain",
"chain_id": "msgchain-1",
"bech32_prefix": "msg",
},
"suspicious_addresses": list(set(
e["address"] for e in relevant
)),
"transactions": [
{
"tx_hash": e["tx_hash"],
"amount": e["amount"],
"timestamp": datetime.utcnow().isoformat(),
}
for e in relevant
],
"requested_action": "freeze",
"legal_basis": "Unauthorized access to smart contract funds",
"contact": {
"name": "[团队名称]",
"email": "security@example.com",
"phone": "[联系电话]",
},
}
6.3 交易所联络模板
#!/usr/bin/env python3
"""exchange_contact.py — 交易所资金冻结请求生成器"""
from datetime import datetime
class ExchangeContactGenerator:
@staticmethod
def generate_freeze_request(
exchange_name: str,
exchange_contact_email: str,
incident_id: str,
suspicious_addresses: list[str],
transaction_hashes: list[str],
estimated_loss_usd: str,
) -> str:
return f"""
Subject: URGENT - Asset Freeze Request - {incident_id} - MSG Chain Security Incident
To: {exchange_contact_email}
From: security@example.com
Date: {datetime.utcnow().isoformat()}
Priority: HIGH
Dear {exchange_name} Security Team,
We are writing to request immediate freezing of funds associated with the following addresses on MSG Chain. This request relates to an active security incident.
Incident Details
----------------
Incident ID: {incident_id}
Chain: MSG Chain (msgchain-1)
Date/Time: {datetime.utcnow().isoformat()}
Estimated Loss: ${estimated_loss_usd} USD
Suspicious Addresses
-------------------
{chr(10).join(f'- {addr}' for addr in suspicious_addresses)}
Related Transactions
------------------
{chr(10).join(f'- {tx}' for tx in transaction_hashes)}
Requested Action
---------------
Please freeze all assets associated with the above addresses on your platform, including any deposits, withdrawals, and trading activity.
Legal Basis
----------
This request is made pursuant to unauthorized access to smart contract funds constituting theft of digital assets. We are working with law enforcement and will provide any additional documentation required.
Contact Information
------------------
Security Lead: [Name]
Email: security@example.com
Phone: [Phone Number]
24/7 Emergency: [Emergency Number]
We appreciate your prompt assistance in this matter.
Best regards,
[Team Name] Security Team
Attachments:
- Incident investigation report
- Transaction evidence (JSON)
- Legal documentation
"""
@staticmethod
def generate_update_template(
exchange_name: str,
incident_id: str,
status: str,
additional_addresses: list[str] = None,
) -> str:
return f"""
Subject: UPDATE - {incident_id} - {exchange_name} Freeze Request Update
To: {exchange_name} Security Team
From: security@example.com
Incident ID: {incident_id}
Status: {status}
Additional Addresses: {', '.join(additional_addresses) if additional_addresses else 'None'}
Please find the latest update regarding the ongoing investigation. We request continued freezing of all identified addresses.
Best regards,
[Team Name] Security Team
"""
6.4 执法协作
#!/usr/bin/env python3
"""law_enforcement.py — 执法机构协作工具"""
from datetime import datetime
class LawEnforcementKit:
@staticmethod
def generate_case_package(
incident_id: str,
attack_tx_hashes: list[str],
attacker_addresses: list[str],
stolen_amount_usd: str,
affected_users_count: int,
vulnerability_type: str,
) -> dict:
return {
"case_reference": incident_id,
"submission_date": datetime.utcnow().isoformat(),
"reporting_entity": "[团队名称/公司名]",
"contact_person": "[联系人姓名]",
"contact_email": "security@example.com",
"incident_summary": {
"chain": "MSG Chain",
"chain_id": "msgchain-1",
"incident_type": "Smart Contract Exploit",
"vulnerability_type": vulnerability_type,
"total_loss_usd": stolen_amount_usd,
"affected_users": affected_users_count,
"date_discovered": datetime.utcnow().isoformat(),
},
"technical_evidence": {
"attack_transactions": attack_tx_hashes,
"attacker_addresses": attacker_addresses,
"affected_contracts": [],
"timeline": [],
},
"legal_claims": [
"Unauthorized access to computer system",
"Theft of digital assets",
"Computer fraud",
],
"requested_assistance": [
"Freeze orders for identified addresses",
"Exchange account identification (KYC)",
"Asset seizure assistance",
],
"attachments": [
"technical_report.pdf",
"fund_flow_graph.json",
"transaction_evidence.zip",
],
}
@staticmethod
def generate_witness_statement(case_ref: str, witness_name: str, statement: str) -> str:
return f"""
WITNESS STATEMENT
Case Reference: {case_ref}
Witness Name: {witness_name}
Date: {datetime.utcnow().isoformat()}
Statement:
{statement}
I, {witness_name}, confirm that the above statement is true and accurate to the best of my knowledge.
Signed: ___________________
Date: {datetime.utcnow().isoformat()}
"""
6.5 白帽回收操作
// whitehat-recovery.ts
// 白帽回收 — 通过相同漏洞回收资金
interface WhitehatRecoveryPlan {
vulnerabilityType: string;
recoveryContract: string;
steps: RecoveryStep[];
riskAssessment: RiskAssessment;
}
interface RecoveryStep {
action: string;
description: string;
expectedOutcome: string;
fallbackPlan: string;
}
interface RiskAssessment {
couldLockFunds: boolean;
couldAlertAttacker: boolean;
legalRisk: string;
communitySupport: boolean;
}
class WhitehatRecovery {
private multisigAddress: string;
constructor(multisigAddress: string) {
this.multisigAddress = multisigAddress;
}
async planRecovery(vulnerability: string, attackerAddress: string): Promise<WhitehatRecoveryPlan> {
console.log(`Planning whitehat recovery for ${vulnerability}...`);
switch (vulnerability) {
case "reentrancy":
return {
vulnerabilityType: "reentrancy",
recoveryContract: "msg1...recovery-contract",
steps: [
{
action: "Deploy recovery contract",
description: "部署利用相同重入漏洞但优先执行的合约",
expectedOutcome: "在攻击者之前提取资金到安全地址",
fallbackPlan: "使用管理密钥直接调用 emergency_withdraw",
},
{
action: "Execute recovery transaction",
description: "发送包含 recovery 逻辑的交易",
expectedOutcome: "资金转移到多签安全地址",
fallbackPlan: "联系验证者进行链回滚",
},
],
riskAssessment: {
couldLockFunds: false,
couldAlertAttacker: true,
legalRisk: "中 — 需要社区共识确认",
communitySupport: true,
},
};
case "unauthorized_access":
return {
vulnerabilityType: "unauthorized_access",
recoveryContract: "msg1...admin-recovery",
steps: [
{
action: "Use admin backdoor",
description: "通过合约预留的管理后门回收资金",
expectedOutcome: "管理员可提取所有资金",
fallbackPlan: "提交治理提案进行资金回收",
},
],
riskAssessment: {
couldLockFunds: false,
couldAlertAttacker: false,
legalRisk: "低 — 管理员权限操作",
communitySupport: true,
},
};
default:
throw new Error(`Unknown vulnerability type: ${vulnerability}`);
}
}
async executeRecovery(plan: WhitehatRecoveryPlan): Promise<boolean> {
console.log("Executing whitehat recovery...");
// 实际实现中，这里会发送链上交易
for (const step of plan.steps) {
console.log(`Step: ${step.action} — ${step.description}`);
// 模拟执行
await new Promise(resolve => setTimeout(resolve, 1000));
}
return true;
}
async communityVote(recoveryPlan: WhitehatRecoveryPlan): Promise<boolean> {
console.log("Initiating community vote for recovery plan...");
// 模拟治理投票
console.log(`Vulnerability: ${recoveryPlan.vulnerabilityType}`);
console.log(`Steps: ${recoveryPlan.steps.length}`);
console.log("Risk: ", recoveryPlan.riskAssessment);
return true;
}
}
6.6 治理投票回收
// governance-recovery.ts
interface RecoveryProposal {
title: string;
description: string;
actions: RecoveryAction[];
votingPeriod: number; // 投票期（秒）
quorum: number; // 法定人数百分比
threshold: number; // 通过阈值百分比
}
interface RecoveryAction {
contract: string;
message: any;
justification: string;
}
class GovernanceRecovery {
async createRecoveryProposal(
stolenAmount: string,
affectedUsers: string[],
recoveryAddress: string,
): Promise<RecoveryProposal> {
return {
title: `Emergency Fund Recovery - ${stolenAmount} MSG`,
description: `Recovery proposal to return ${stolenAmount} MSG to affected users.\n` +
`Affected users: ${affectedUsers.length}\n` +
`Recovery address: ${recoveryAddress}`,
actions: [
{
contract: "msg1...token-contract",
message: {
mint_and_send: {
recipient: recoveryAddress,
amount: stolenAmount,
},
},
justification: "Reimbursing users affected by the security incident",
},
],
votingPeriod: 86400 * 3, // 3 天
quorum: 30, // 30%
threshold: 66, // 66%
};
}
async executeProposal(proposal: RecoveryProposal): Promise<string> {
console.log(`Executing recovery proposal: ${proposal.title}`);
// 模拟提案执行
return "proposal_tx_hash_here";
}
}
6.7 链升级/回滚 (最后手段)
// chain-recovery.ts
interface ChainRecoveryOption {
type: "fork" | "revert" | "halt" | "upgrade";
description: string;
impact: string;
requiredValidatorConsensus: number; // 需要验证者同意百分比
timeToExecute: string;
risks: string[];
}
class ChainLevelRecovery {
async assessForkOption(): Promise<ChainRecoveryOption> {
return {
type: "fork",
description: "创建链分叉，在攻击发生前的状态重建链",
impact: "所有攻击后的交易将被回滚，需要全节点升级",
requiredValidatorConsensus: 90,
timeToExecute: "72 小时",
risks: [
"社区分裂风险",
"需要所有验证者协调升级",
"跨链桥和交易所需要同步升级",
"长期声誉损害",
],
};
}
async assessRevertOption(attackHeight: number): Promise<ChainRecoveryOption> {
return {
type: "revert",
description: `回滚到区块 ${attackHeight} 之前的状态`,
impact: "攻击后的区块将被丢弃，其他正常交易也会丢失",
requiredValidatorConsensus: 80,
timeToExecute: "48 小时",
risks: [
"正常交易丢失",
"需要链协调暂停",
"CEX/DEX 可能出现状态不一致",
],
};
}
async assessHaltOption(): Promise<ChainRecoveryOption> {
return {
type: "halt",
description: "暂停链运行，防止进一步损失",
impact: "所有链上活动停止",
requiredValidatorConsensus: 95,
timeToExecute: "立即 (需验证者配合)",
risks: [
"紧急暂停可能引起恐慌",
"恢复时间不确定",
"验证者需手动操作",
],
};
}
async coordinateWithValidators(option: ChainRecoveryOption): Promise<boolean> {
console.log(`Coordinating with validators for: ${option.type}`);
console.log(`Required consensus: ${option.requiredValidatorConsensus}%`);
console.log(`Risks: ${option.risks.join(", ")}`);
// 模拟验证者协调
const validatorCount = 50;
const agreed = Math.floor(validatorCount * option.requiredValidatorConsensus / 100);
console.log(`Need ${agreed}/${validatorCount} validators to agree`);
return true;
}
}
6.8 资金追回总结
# 资金追回策略对比
| 策略 | 成功率 | 时间 | 风险 | 适用场景 |
|------|--------|------|------|----------|
| 链上追踪 + 交易所冻结 | 中 (40-60%) | 1-24 小时 | 低 | 资金进入 CEX |
| 白帽回收 | 高 (70-90%) | 1-4 小时 | 中 | 漏洞仍可利用 |
| 治理投票回收 | 中 (50-70%) | 3-7 天 | 低 | 社区支持充足 |
| 执法协作 | 低 (20-40%) | 月-年 | 低 | 涉及 KYC/AML |
| 链回滚 | 高 (80-95%) | 48-72 小时 | 高 | P0 级别, 损失巨大 |
| 链分叉 | 中 (50-70%) | 72+ 小时 | 极高 | 最后手段 |
| 保险理赔 | 高 (80-90%) | 14-30 天 | 低 | 有保险覆盖 |
## 优先级建议
1. **立即 (0-1 小时)**: 交易所冻结 + 白帽回收
2. **短期 (1-24 小时)**: 链上追踪 + 执法联络
3. **中期 (24h - 1 周)**: 治理投票 + 保险理赔
4. **长期 (1 周+)**: 链升级/分叉 (仅当其他方案失败)
7. 事后复盘
7.1 事故报告模板
# Security Incident Report
**Incident ID**: INC-2026MMDD-HHMMSS
**Severity**: P0 (Critical)
**Status**: Resolved
**Date**: 2026-07-06
**Report Author**: [Name]
**Reviewers**: [IC, Security Lead, Comms Lead]
---
## Executive Summary
[1-2 段落描述事件概要：发生了什么、受影响范围、响应时间、修复状态]
## Timeline
| Time (UTC) | Event | Actor |
|------------|-------|-------|
| T+0:00 | Detection: Large transfer alert triggered | Monitor System |
| T+0:01 | Incident created (INC-2026MMDD-HHMMSS) | Security Lead |
| T+0:02 | Severity assessed: P0 Critical | IC |
| T+0:03 | Emergency pause initiated | Guardian Multisig |
| T+0:05 | Contracts paused | Guardian Multisig |
| T+0:10 | Validators notified for standby | IC |
| T+0:15 | Exchanges notified | Comms Lead |
| T+0:30 | Attack vector identified: Reentrancy | Security Lead |
| T+1:00 | Vulnerable code located (file:line) | Contract SME |
| T+2:00 | Fix developed and tested | Contract SME |
| T+3:00 | Fix audited by [Audit Firm] | External Auditor |
| T+4:00 | Governance vote for upgrade initiated | IC |
| T+12:00 | Upgrade approved and deployed | Validators |
| T+12:05 | Contracts unpaused | Guardian Multisig |
| T+24:00 | Funds recovery plan published | IC |
| T+72:00 | Post-mortem meeting held | Team |
## Root Cause Analysis
### Vulnerability Type
Reentrancy in `withdraw()` function
### Root Cause
The `withdraw` function sent funds to the caller (Interaction) before updating the internal balance (Effects), violating the Checks-Effects-Interactions pattern.
### Vulnerable Code
File: `src/contract.rs`, Lines 145-171
```rust
pub fn withdraw(deps: DepsMut, _env: Env, info: MessageInfo, amount: Uint128) -> StdResult<Response> {
let state = STATE.load(deps.storage)?;
// CHECK: 余额检查通过
// EFFECTS: 状态未更新 ← 漏洞！
// INTERACTION: 先发送资金
let bank_msg = BankMsg::Send { ... };
// 状态更新在发送资金之后 ← 错误顺序
STATE.save(deps.storage, &new_state)?;
}
Attack Flow
- Attacker deployed malicious contract with reentrant
replyhandler - Called
withdraw(500)on victim contract - Victim contract sent 500 MSG before updating balance
- Attacker's
replyhandler calledwithdraw(500)again - Since balance not yet updated, the second withdrawal succeeded
- Repeated until all funds drained
Impact Assessment
Financial Impact
- Total Loss: 1,250,000 MSG (~$2,500,000 USD)
- Recovered: 800,000 MSG (64%)
- Irrecoverable: 450,000 MSG ($900,000 USD)
- Insurance Claim: $500,000 (partial)
Affected Users
- Directly Impacted: 342 users
- Fully Reimbursed: 280 users (by governance vote)
- Partially Reimbursed: 62 users
Contracts Affected
msg1...lending-pool(primary)msg1...staking-rewards(secondary)
Reputational Impact
- TVL dropped 35% within 24 hours
- 15% recovery within 1 week
- Community trust maintained due to 5-minute pause response
Remediation
Immediate Fix
- Reordered state update before token transfer (Checks-Effects-Interactions)
- Added
ReplyOn::Neverto prevent reentrant reply handling - Added reentrancy guard (mutex flag)
Code Change
pub fn fixed_withdraw(deps: DepsMut, _env: Env, info: MessageInfo, amount: Uint128) -> StdResult<Response> {
let state = STATE.load(deps.storage)?;
// CHECKS
if amount > state.balances {
return Err(StdError::generic_err("Insufficient balance"));
}
// EFFECTS FIRST
let new_balance = state.balances.checked_sub(amount)?;
STATE.save(deps.storage, &State { balances: new_balance, locked: state.locked })?;
// INTERACTION SECOND
let bank_msg = BankMsg::Send { ... };
let sub_msg = SubMsg { reply_on: ReplyOn::Never, ... };
Ok(Response::new().add_submessage(sub_msg))
}
Systemic Fixes
- All
withdraw-like functions audited for similar patterns - Added reentrancy detector to CI pipeline
- Implemented emergency pause with < 1 minute response time
- Added automated transaction analysis for reentrancy patterns
Preventive Measures
| # | Measure | Owner | Deadline | Status |
|---|---|---|---|---|
| 1 | Add reentrancy guard to all state-changing functions | Contract SME | 2026-07-13 | ✅ Done |
| 2 | Integrate Slither/Clippy static analysis in CI | Infra | 2026-07-20 | ✅ Done |
| 3 | Implement Circuit Breaker with balance monitoring | Contract SME | 2026-08-01 | 🔄 In Progress |
| 4 | Monthly security drill for all response team | IC | 2026-08-01 | 📅 Planned |
| 5 | Third-party audit of all contract suite | Security Lead | 2026-09-01 | 📅 Planned |
Lessons Learned
What Went Well
- Detection within 1 minute (WebSocket event monitoring)
- Pause executed within 5 minutes (pre-approved DAO proposal)
- Communication was clear and timely
- Validator coordination was smooth
What Could Be Improved
- Reentrancy detector should have caught this in CI
- Rate limiting could have reduced loss even without pause
- Internal comms channel had 2-minute setup delay
Action Items
- Implement local reentrancy detector in CI pipeline
- Pre-heat all communication channels weekly
- Add rate limiting to all contracts by default
- Create automated incident log generator
Appendices
A. Transaction Hashes
- Attack Tx 1:
0xabc...def - Attack Tx 2:
0x123...456 - Pause Tx:
0x789...012 - Fix Deploy Tx:
0x345...678
B. Attacker Addresses
msg1attacker...address1msg1attacker...address2
C. Recovery Addresses
msg1recovery...multisig
D. Communication Log
- [Internal Comms Channel Export]
- [Public Announcement Thread]
- [Exchange Correspondence]
### 7.2 复盘会议议程
Post-Mortem Meeting Agenda
准备 (会议前)
- 收集所有时间线数据
- 准备技术分析幻灯片
- 汇总受影响用户数据
- 收集通信记录
会议议程 (60-90 分钟)
1. 概述 (5 分钟)
- IC 介绍事件概要
- 确认严重等级和时间线
2. 技术复盘 (20 分钟)
- 安全负责人: 攻击向量和根因
- 合约专家: 漏洞位置和修复
- 基础设施工程师: 监控和响应时间
3. 流程复盘 (15 分钟)
- IC: 响应时间线分析
- Comms: 通信效果评估
- Legal: 法规合规情况
4. 影响评估 (10 分钟)
- 财务影响
- 用户影响
- 声誉影响
5. 改进措施 (15 分钟)
- 技术改进清单
- 流程改进清单
- 人员/培训改进
6. 行动项分配 (5 分钟)
- 确认所有行动项
- 分配负责人和截止日期
- 确定跟踪机制
会议产出
- 事故事件报告 (终版)
- 行动项跟踪表
- 更新时间线 (最终版)
- 对外公告草稿 (如果需要)
### 7.3 行动项跟踪
```markdown
# Action Item Tracker
| ID | Action Item | Owner | Priority | Deadline | Status | Notes |
|----|-------------|-------|----------|----------|--------|-------|
| PM-001 | Add reentrancy guard to all contracts | @alice | P0 | 2026-07-13 | ✅ Done | |
| PM-002 | Integrate static analysis in CI | @bob | P0 | 2026-07-20 | ✅ Done | |
| PM-003 | Circuit breaker implementation | @charlie | P1 | 2026-08-01 | 🔄 WIP | Blocked by audit |
| PM-004 | Monthly security drill | @dave | P1 | 2026-08-01 | 📅 Planned | |
| PM-005 | Third-party audit | @alice | P1 | 2026-09-01 | 📅 Planned | Vendor selection |
| PM-006 | Update emergency contact list | @eve | P2 | 2026-07-15 | ✅ Done | |
| PM-007 | Create incident response playbook | @frank | P2 | 2026-08-15 | 🔄 WIP | |
| PM-008 | Rate limiting default implementation | @charlie | P2 | 2026-09-01 | 📅 Planned | |
7.4 根本原因分析 (RCA) 框架
## 根本原因分析框架
每起安全事件应从以下 5 个维度分析：
### 技术维度
- 漏洞类型与 CWE 分类
- 攻击向量和利用路径
- 代码提交历史 (何时引入漏洞)
- 测试覆盖缺失
### 流程维度
- 审计流程是否覆盖该漏洞
- 代码审查是否发现
- 测试是否充分
- 部署前检查是否执行
### 人员维度
- 安全培训是否覆盖
- 开发人员安全意识
- 团队沟通效果
- 决策速度和质量
### 工具维度
- 监控系统是否检测到
- 告警是否及时
- 自动化防护是否失效
- 日志是否充足
### 外部维度
- 是否已知的前例
- 生态中是否有类似漏洞
- 攻击者手法是否新颖
- 第三方依赖是否安全
8. 通讯策略
8.1 通讯原则
# 安全事件通讯原则
## 核心原则
1. **透明**: 不隐瞒、不拖延、不误导
2. **准确**: 只发布已确认事实，不猜测
3. **及时**: 在确认信息后尽快发布
4. **共情**: 理解用户情绪，表达关切
5. **统一**: 所有渠道信息一致，单一发声
## 信息分级
| 级别 | 定义 | 可公开内容 |
|------|------|-----------|
| 公开 | 可向所有人公开 | 事件状态、影响范围、修复进度 |
| 内部 | 仅限团队内部 | 漏洞细节、攻击者地址、恢复策略 |
| 保密 | 仅限核心团队 | 私钥、未部署的补丁、执法信息 |
## 对外发言规则
1. 仅 Comms Lead 或其指定人可对外发言
2. 所有对外信息需经 IC + Comms 双重确认
3. 不评论未确认的信息
4. 不讨论具体漏洞细节 (直到修复完成)
5. 不猜测损失金额 (用已确认数据)
6. 不使用 "rug" "scam" 等情绪化用语
8.2 内部通讯模板
# 内部通讯 — 事件响应启动
## 消息 1: 事件确认 (Triage)
**频道**: #incident-response (Signal/Telegram)
**发送者**: IC
**优先级**: 🚨 URGENT
🚨 [P0] 安全事件确认
事件ID: INC-通用维护记录-143000
严重等级: P0 (Critical)
标题: [攻击类型简述]
检测时间: 2026-07-06 14:30:00 UTC
当前状态:
- 攻击是否进行中: 是/否
- 受影响合约: [合约地址列表]
- 预估损失: [金额] (待确认)
已执行操作:
- [ ] 暂停合约
- [ ] 联系验证者
- [ ] 通知交易所
响应团队:
- IC: @name
- Security: @name
- Comms: @name
- Legal: @name
下一步:
- [5min] Containment 措施
- [15min] 向社区发布初步公告
- [30min] 详细调查开始
所有更新将在此频道发布。
## 消息 2: Containment 确认 (5-10 分钟)
**频道**: #incident-response
✅ Containment 措施已执行
时间: 2026-07-06 14:35:00 UTC
操作结果:
- 暂停合约: ✅ (tx: 0xabc...)
- 速率限制: ✅
- 快照完成: ✅
- 验证者通知: ✅
当前损失已停止: 是/否
下一步: 开始详细调查
## 消息 3: 状态更新 (每 30 分钟)
📊 事件更新 #2
时间: 2026-07-06 15:00:00 UTC
状态: 调查中
损失已遏制: ✅
调查进展:
- 漏洞类型: [重入/闪电贷/权限...]
- 攻击者地址: [已识别/待确认]
- 资金流向: [追踪中/已进入交易所]
沟通计划:
- 下次更新: 15:30 UTC
- 对外公告: [已发布/准备中]
- 社区问答: [安排中]
### 8.3 用户通知模板
```markdown
# 用户通知 — 安全事件公告
## 初步公告 (确认后 15 分钟内)
🚨 MSG Chain Security Update
We have detected suspicious activity on [Contract Name] and have paused all operations as a precaution.
What happened:
- Unusual transactions detected at [time] UTC
- Contracts paused immediately to protect user funds
- Investigation is ongoing
What we are doing:
- Security team is investigating the root cause
- Validators have been notified
- Exchanges have been contacted
User safety:
- All funds are SAFU
- Contracts are paused - no further activity possible
- No action required from users at this time
Next update: [time] UTC or sooner if significant developments
Stay tuned for updates.
## 调查公告 (1-4 小时)
🔍 MSG Chain Incident Update
We have identified the root cause of the incident.
Root Cause:
[简要技术描述，不包含可利用细节]
Impact:
- Total affected: [X] users
- Contract addresses: [list]
- Estimated loss: [amount] (confirmed)
Actions Taken:
- Contracts paused: ✅
- Root cause identified: ✅
- Fix developed: ✅/[in progress]
- Fix audited: ✅/[in progress]
- Governance vote: [submitted/scheduled]
- Funds recovery: [plan in place/in progress]
Next Steps:
- [Fix deployment timeline]
- [Funds recovery plan]
- [User reimbursement plan]
We will provide a detailed post-mortem after resolution.
Thank you for your patience and trust.
## 解决公告 (修复完成后)
✅ MSG Chain Incident Resolved
The security incident has been fully resolved.
Timeline:
- Detection: [time]
- Pause: [time]
- Fix verified: [time]
- Contracts unpaused: [time]
Resolution:
- Vulnerability fixed and audited
- Contracts restored to normal operation
- [Recovery/reimbursement plan]
Post-Mortem:
Full incident report will be published at [link] within [timeframe]
We thank our community for your understanding and support.
### 8.4 社交媒体声明模板
```typescript
// social-media-templates.ts
interface SocialMediaPost {
platform: "twitter" | "discord" | "telegram";
content: string;
scheduledTime?: string;
mediaAttachments?: string[];
}
class SocialMediaManager {
generateThread(posts: SocialMediaPost[]): SocialMediaPost[] {
return posts;
}
// 初步声明
initialStatement(incidentId: string, detectedAt: string): SocialMediaPost[] {
return [
{
platform: "twitter",
content: `🚨 [${incidentId}] We've detected unusual activity on MSG Chain. Contracts have been paused. Investigation ongoing. Funds are SAFU. Next update in 30 min.`,
},
{
platform: "discord",
content: `**🚨 Security Alert**\n\nIncident: ${incidentId}\nTime: ${detectedAt}\n\nWe have paused all contracts as a precaution. Our team is investigating. We will keep you updated here.\n\n**Do NOT send funds to any contracts at this time.**`,
},
{
platform: "telegram",
content: `🚨 Security Alert\n\nIncident: ${incidentId}\n\nContracts paused. Investigation in progress. Funds are SAFU.\n\nNext update: 30 minutes.`,
},
];
}
// 状态更新
statusUpdate(phase: string, details: string): SocialMediaPost[] {
return [
{
platform: "twitter",
content: `📊 Update: ${phase}\n\n${details}\n\nRead full update: [link]`,
},
{
platform: "discord",
content: `**📊 Update: ${phase}**\n\n${details}\n\nFull details: [link]`,
},
];
}
// 解决声明
resolutionStatement(timeToResolve: string): SocialMediaPost[] {
return [
{
platform: "twitter",
content: `✅ Resolved\n\nIncident fully resolved in ${timeToResolve}.\n\nContracts restored. Post-mortem coming soon.\n\nThank you for your patience.`,
},
{
platform: "discord",
content: `**✅ Incident Resolved**\n\nAll contracts have been restored to normal operation. The vulnerability has been fixed and audited.\n\nA full post-mortem will be published within [timeframe].\n\nThank you for your support.`,
},
];
}
// FAQ 回复
faqResponse(question: string, answer: string): SocialMediaPost {
return {
platform: "twitter",
content: `FAQ: ${question}\n\n${answer}\n\nMore FAQs: [link]`,
};
}
}
8.5 监管通知
#!/usr/bin/env python3
"""regulatory_notification.py — 监管通知模板"""
from datetime import datetime
from typing import Optional
class RegulatoryNotification:
def __init__(self, jurisdiction: str, regulator_name: str, contact_email: str):
self.jurisdiction = jurisdiction
self.regulator_name = regulator_name
self.contact_email = contact_email
def generate_initial_notification(
self,
incident_id: str,
incident_type: str,
estimated_loss_usd: str,
affected_users: int,
date_occurred: str,
) -> str:
return f"""
{self.regulator_name}
Initial Security Incident Notification
Date: {datetime.utcnow().isoformat()}
From: [Organization Name]
Subject: Initial Notification of Security Incident - {incident_id}
Regulator: {self.regulator_name}
Jurisdiction: {self.jurisdiction}
We are writing to notify you of a security incident affecting our smart contract platform on MSG Chain.
Incident Summary
----------------
- Incident ID: {incident_id}
- Type: {incident_type}
- Date Occurred: {date_occurred}
- Estimated Loss: ${estimated_loss_usd}
- Affected Users: {affected_users}
- Status: Contained (contracts paused)
Actions Taken
-------------
1. Immediate contract pause
2. Root cause investigation
3. Law enforcement notification
4. User communication initiated
Next Steps
----------
1. Complete investigation and root cause analysis
2. Implement fix and resume operations
3. Submit detailed report within [timeframe]
We will provide a detailed incident report within [timeframe].
Contact
-------
Security Lead: [Name]
Email: security@example.com
Phone: [Phone]
"""
def generate_detailed_report(
self,
incident_id: str,
root_cause: str,
remediation: str,
preventive_measures: list[str],
user_reimbursement_plan: str,
) -> str:
return f"""
Detailed Security Incident Report
Incident ID: {incident_id}
Date: {datetime.utcnow().isoformat()}
1. Executive Summary
{root_cause[:200]}
2. Technical Details
Root Cause: {root_cause}
3. Remediation
{remediation}
4. Preventive Measures
{chr(10).join(f' {i+1}. {m}' for i, m in enumerate(preventive_measures))}
5. User Impact and Reimbursement
{user_reimbursement_plan}
6. Regulatory Compliance
- Data breach notification: [Yes/No/N/A]
- User data affected: [Yes/No/N/A]
- Financial loss reported: Yes
We confirm no personal user data was compromised in this incident.
Contact: security@example.com
"""
8.6 通讯总结
# 安全事件通讯时间线
| 时间 | 动作 | 渠道 | 内容 | 审批人 |
|------|------|------|------|--------|
| T+0 | 内部通知 | Signal/Telegram | 事件确认, 团队召集 | IC |
| T+5min | 内部确认 | Signal/Telegram | Containment 执行 | Sec Lead |
| T+15min | 初步公告 | Twitter/Discord/Telegram | 事件已知, 合约暂停 | IC + Comms |
| T+30min | 内部更新 | Signal/Telegram | 调查进展 | Sec Lead |
| T+1h | 调查公告 | Twitter/Discord/Blog | 根因, 影响范围 | IC + Comms |
| T+2h | FAQ 发布 | Discord/Telegram | 用户常见问题 | Comms |
| T+4h | 修复公告 | Twitter/Discord/Blog | 修复进展, 恢复计划 | IC + Comms |
| T+resolved | 解决公告 | 全渠道 | 事件解决, 感谢用户 | IC + Comms |
| T+1 week | 事后分析 | Blog/GitHub | 完整事故报告 | IC + Sec Lead |
## 公告审批矩阵
| 内容类型 | 起草 | 技术审核 | 法务审核 | 最终审批 |
|----------|------|----------|----------|----------|
| 初步公告 | Comms | Security | - | IC |
| 调查公告 | Comms | Security | Legal | IC |
| 修复公告 | Comms | Security | - | IC |
| 解决公告 | Comms | Security | Legal | IC + CEO |
| 事后分析 | IC + Sec | Security | Legal | CEO |
9. 附录
附录 A: P0 事件响应快速检查清单
# P0/Critical 事件响应快速检查清单
## 0-2 分钟: 确认和召集
- [ ] 确认事件 (非误报)
- [ ] Signal 群组发第一条消息: @all P0 事件
- [ ] 评估是否仍在攻击中
- [ ] 分配 IC 角色
## 2-5 分钟: 快速遏制
- [ ] 调用 emergency_pause() (多签)
- [ ] 设置速率限制 (5 ops/min)
- [ ] 记录攻击者地址
- [ ] 记录攻击交易哈希
- [ ] 执行资金快照
## 5-15 分钟: 扩大遏制
- [ ] 通知验证者 (链暂停备选)
- [ ] 通知交易所 (冻结地址)
- [ ] 发布初步用户公告
- [ ] 启动详细调查
- [ ] 联系备用审计公司
## 15-60 分钟: 调查
- [ ] 重放攻击交易
- [ ] 分析漏洞类型
- [ ] 追踪资金流向
- [ ] 量化损失
- [ ] 识别所有受影响合约
## 1-4 小时: 修复
- [ ] 开发修复补丁
- [ ] 内部审查补丁
- [ ] 外部审计补丁
- [ ] 提交治理升级提案
- [ ] 发布调查公告
## 4-12 小时: 恢复
- [ ] 验证者批准升级
- [ ] 部署修复合约
- [ ] 测试修复后合约
- [ ] 取消合约暂停
- [ ] 恢复正常操作
- [ ] 发布解决公告
## 24 小时+
- [ ] 启动资金追回
- [ ] 完成事后分析报告
- [ ] 召开复盘会议
- [ ] 更新安全流程
- [ ] 更新监控指标
附录 B: 联系人模板
# 安全事件响应联系人清单
## [项目名称] — 紧急响应联系人
### 一级联系人 (P0/P1)
| 角色 | 姓名 | 电话 | 备用渠道 | 时区 | 备注 |
|------|------|------|----------|------|------|
| IC | ________ | ________ | @_______ | UTC__ | |
| IC Backup | ________ | ________ | @_______ | UTC__ | |
| Security Lead | ________ | ________ | @_______ | UTC__ | |
| Security Backup | ________ | ________ | @_______ | UTC__ | |
| Comms Lead | ________ | ________ | @_______ | UTC__ | |
| Comms Backup | ________ | ________ | @_______ | UTC__ | |
### 二级联系人 (P2)
| 角色 | 姓名 | 联系方式 | 说明 |
|------|------|----------|------|
| Contract SME | ________ | @_______ | |
| Contract SME | ________ | @_______ | |
| Infra Engineer | ________ | @_______ | |
| Legal Lead | ________ | ________ | |
### 外部联系人
| 机构 | 联系方式 | 用途 | 备注 |
|------|----------|------|------|
| MSG Chain Validators | @_______ | 链暂停/升级 | |
| Exchange A | ________ | 资金冻结 | |
| Exchange B | ________ | 资金冻结 | |
| Auditor Firm | ________ | 紧急审计 | |
| Insurance | ________ | 理赔 | |
| Law Enforcement | ________ | 报案 | |
### 通信渠道
| 渠道 | 链接/ID | 用途 | 管理员 |
|------|---------|------|--------|
| Signal | ________ | 核心团队 | |
| Telegram | ________ | 内部协调 | |
| Discord | ________ | 社区沟通 | |
| PagerDuty | ________ | 告警推送 | |
| Email | ________ | 正式通信 | |
附录 C: 攻击模式参考
# 常见攻击模式速查
## CosmWasm 专用攻击模式
### 1. 重入攻击 (Reentrancy)
- 利用 CosmWasm 的 SubMsg + Reply 机制
- 攻击者在 reply handler 中重新进入合约
- 修复: Checks-Effects-Interactions 模式, ReplyOn::Never
- 参考: 审计清单 VULN-01
### 2. 未授权访问 (Unauthorized Access)
- 管理员/敏感函数缺少权限检查
- 任意调用者可执行铸币、提现等操作
- 修复: 每处 execute 入口验证调用者权限
- 参考: 审计清单 VULN-02
### 3. 整数溢出/下溢 (Integer Overflow)
- Rust release 模式下整数回绕
- checked_add/checked_sub 未使用
- 修复: 使用 cosmwasm_std::Uint128 + checked 操作
- 参考: 审计清单 VULN-03
### 4. 存储碰撞 (Storage Collision)
- 合约迁移时新旧存储键冲突
- 数据损坏或权限提升
- 修复: 版本化存储键名 (v1_ / v2_ 前缀)
- 参考: 审计清单 VULN-04
### 5. Reply 处理漏洞 (Reply-Based Attacks)
- 未检查子消息执行结果
- 子消息失败仍更新状态
- 修复: 检查 SubMsgResult::Ok/Error
- 参考: 审计清单 VULN-05
### 6. IBC 超时漏洞 (IBC Timeout)
- IBC 包超时处理不完整
- 未验证通道和序列号
- 修复: 验证端口、通道、序列号、超时
- 参考: 审计清单 VULN-06
### 7. 价格预言机操纵 (Price Oracle Manipulation)
- 使用单一来源即时价格
- 攻击者通过大额交易操纵价格
- 修复: TWAP + 多重价格源 + 偏差检查
- 参考: 审计清单 VULN-07
### 8. 闪电贷攻击 (Flash Loan)
- CosmWasm 通过复合消息实现闪电贷模式
- 同一交易内借贷+操纵+还款
- 修复: TWAP + 滑点保护 + 交易哈希检查
- 参考: 审计清单 VULN-08
### 9. 签名重放 (Signature Replay)
- 自定义签名验证缺少 nonce/ChainID 检查
- 相同签名可在不同上下文再次使用
- 修复: ChainID + nonce + deadline + 有效期
- 参考: 审计清单 VULN-09
### 10. 抢先交易 / MEV
- Mempool 交易可被搜索者/验证者读取
- 三明治攻击: 在用户交易前后插入交易
- 修复: 批量拍卖、提交-揭示模式、FCFS
- 参考: 审计清单 VULN-11
### 11. Gas Griefing
- 攻击者触发高 gas 操作使交易失败
- 用户损失 gas 费
- 修复: 限制迭代 + 分页 + gas 限制
- 参考: 审计清单 VULN-12
### 12. Agent API 滥用
- AI Agent 绕过宪法限制执行操作
- 未验证 Agent 身份和权限
- 修复: 完整的 Agent 验证流程
- 参考: 审计清单 VULN-20
## 检测查询速查
```sql
-- MSG Chain 事件查询 (使用 REST API)
-- 1. 查询所有 wasm 事件 (最近 100 笔交易)
GET /cosmos/tx/v1beta1/txs?events=wasm&pagination.limit=100
-- 2. 查询特定合约的事件
GET /cosmos/tx/v1beta1/txs?events=wasm._contract_address='msg1...contract'
-- 3. 查询 transfer 事件
GET /cosmos/tx/v1beta1/txs?events=transfer.recipient='msg1...address'
-- 4. 查询特定高度的交易
GET /cosmos/tx/v1beta1/txs?events=tx.height=123456
-- 5. 查询特定操作
GET /cosmos/tx/v1beta1/txs?events=wasm.action='mint'
-- 6. 查询合约余额
GET /cosmos/bank/v1beta1/balances/msg1...contract
-- 7. 查询合约信息
GET /cosmwasm/wasm/v1/contract/msg1...contract
-- 8. 查询合约状态
GET /cosmwasm/wasm/v1/contract/msg1...contract/smart/{"status":{}}
-- 9. 查询最新区块
GET /cosmos/base/tendermint/v1beta1/blocks/latest
-- 10. 查询交易详情
GET /cosmos/tx/v1beta1/txs/0x...txhash
附录 D: 工具安装和配置
#!/bin/bash
# setup-response-tools.sh — 应急响应工具安装脚本
set -euo pipefail
echo "=== MSG Chain 应急响应工具安装 ===\n"
# 1. 安装 Python 依赖
echo "1. 安装 Python 依赖..."
pip3 install aiohttp httpx websockets prometheus-client pagerduty
# 2. 安装 Node.js 依赖
echo "2. 安装 Node.js 依赖..."
npm install -g @cosmjs/cosmwasm-stargate @cosmjs/proto-signing
npm install -g typescript ts-node
# 3. 安装 Rust 工具链 (如需编译合约)
echo "3. 安装 Rust 工具链..."
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
rustup target add wasm32-unknown-unknown
cargo install cosmwasm-check cargo-audit
# 4. 创建日志目录
echo "4. 创建日志目录..."
mkdir -p /var/log/incidents
mkdir -p /var/log/monitor
# 5. 配置监控脚本
echo "5. 配置监控脚本..."
mkdir -p /opt/msg-chain-monitor
# 复制监控脚本到该目录
# 6. 创建 cron 任务 (健康检查)
echo "6. 配置健康检查..."
cat > /etc/cron.d/msg-chain-health << 'EOF'
*/5 * * * * root /usr/bin/python3 /opt/msg-chain-monitor/msg_health_check.py >> /var/log/monitor/health.log 2>&1
EOF
# 7. 创建 systemd 服务 (监控守护进程)
echo "7. 创建监控服务..."
cat > /etc/systemd/system/msg-monitor.service << 'EOF'
[Unit]
Description=MSG Chain Security Monitor
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt/msg-chain-monitor
ExecStart=/usr/bin/python3 /opt/msg-chain-monitor/msg_chain_monitor.py
Restart=always
RestartSec=10
Environment=MSG_RPC=https://rpc.msgchain.org
Environment=MSG_REST=https://rest.msgchain.org
Environment=TELEGRAM_BOT_TOKEN=
Environment=TELEGRAM_CHAT_ID=
Environment=PAGERDUTY_KEY=
Environment=SLACK_WEBHOOK_URL=
[Install]
WantedBy=multi-user.target
EOF
systemctl daemon-reload
systemctl enable msg-monitor.service
echo "=== 安装完成 ==="
echo "启动监控: systemctl start msg-monitor"
echo "查看日志: journalctl -u msg-monitor -f"
附录 E: 关键术语表
# 术语表
| 中文 | English | 定义 |
|------|---------|------|
| 事故响应 | Incident Response | 针对安全事件的系统化响应流程 |
| 事件指挥官 | Incident Commander (IC) | 负责协调事件响应的总负责人 |
| 遏制 | Containment | 阻止损失扩大的措施 |
| 根因分析 | Root Cause Analysis (RCA) | 确定漏洞根本原因的过程 |
| 事后复盘 | Post-Mortem | 事件结束后的问题回顾会议 |
| 断路器 | Circuit Breaker | 检测到异常时自动停止操作的保护机制 |
| 多签 | Multisig | 需要多个私钥签名的地址 |
| 治理 | Governance | 通过投票机制管理协议的过程 |
| 白帽 | Whitehat | 利用漏洞保护资金的安全研究员 |
| 时间锁 | Timelock | 延迟交易执行的保护机制 |
| TWAP | Time-Weighted Average Price | 时间加权平均价格 |
| MEV | Miner/Maximal Extractable Value | 验证者可从交易中提取的价值 |
| SubMsg | SubMessage | CosmWasm 中的子消息 (可包含 Reply) |
| Reply | Reply Handler | CosmWasm 中处理子消息结果的函数 |
| Checks-Effects-Interactions | CEI | 先检查-再更新状态-最后交互的安全模式 |
| Dilithium-5 | Dilithium-5 | NIST 标准化的后量子签名算法 |
| DAR | Dynamic Authority Rotation | MSG Chain 的动态验证者轮换共识 |
| BadgerDB | BadgerDB | MSG Chain 使用的 LSM-Tree 键值存储 |
文档版本: v1.0.0
维护者: MSG Chain Security Team
配套文档: 安全审计清单与常见漏洞.md
贡献指南: 请通过 GitHub Issues 提交改进建议