925 lines
35 KiB
Python
925 lines
35 KiB
Python
"""
|
|
Cross-Chain Bridge Health & Exploit Monitor
|
|
=============================================
|
|
Monitors cross-chain bridge security in real-time by tracking TVL, detecting
|
|
anomalous withdrawals, checking contract upgrade events, and scoring bridge
|
|
trust models. The free, comprehensive alternative to Defender and Hacken's
|
|
paid bridge monitoring.
|
|
|
|
What it does:
|
|
1. TVL Monitoring — Tracks Total Value Locked across 12 major bridges
|
|
(LayerZero, Stargate, Across, Wormhole, Hop, Synapse, Orbiter,
|
|
Axelar, Celer cBridge, Connext, Chainlink CCIP, DeBridge)
|
|
2. Anomaly Detection — Flags sudden TVL drops, unusual withdrawal patterns,
|
|
and large-value bridge transactions that may indicate an active exploit
|
|
3. Contract Health — Checks for proxy upgrades, pause status, and admin key
|
|
changes on bridge contract addresses
|
|
4. Trust Scoring — Rates each bridge on 5 factors: TVL depth, validator
|
|
decentralization, audit recency, exploit history, and upgrade mechanism
|
|
5. Cascade Risk — When one bridge shows exploit signs, scans all other
|
|
bridges sharing similar security profiles for contagion risk
|
|
6. Alert Generation — Produces human-readable security bulletins and JSON
|
|
|
|
Standalone usage:
|
|
python3 bridge_health_monitor.py
|
|
python3 bridge_health_monitor.py --bridge stargate
|
|
|
|
API usage:
|
|
from app.bridge_health_monitor import BridgeHealthMonitor
|
|
monitor = BridgeHealthMonitor()
|
|
try:
|
|
report = await monitor.scan()
|
|
print(report.summary())
|
|
alert = await monitor.alert_if_exploit()
|
|
print(alert.summary() if alert else "No exploitation detected")
|
|
except Exception as e:
|
|
logger.error(f"Bridge health scan failed: {e}")
|
|
finally:
|
|
await monitor.close()
|
|
|
|
Dependencies (optional):
|
|
- defillama (pip install defillama) for TVL data
|
|
- If unavailable, falls back to public API endpoints
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import time
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import UTC, datetime
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Enums & Types
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
|
|
class RiskTier(Enum):
|
|
"""Bridge risk classification."""
|
|
|
|
SAFE = "SAFE"
|
|
WATCH = "WATCH" # Minor concern, monitor
|
|
DANGER = "DANGER" # Significant risk detected
|
|
CRITICAL = "CRITICAL" # Active exploit likely
|
|
|
|
|
|
class TrustModel(Enum):
|
|
"""Bridge security model classification."""
|
|
|
|
EXTERNAL_VALIDATORS = "external_validators" # Multi-sig / validator set
|
|
OPTIMISTIC = "optimistic" # Optimistic verification
|
|
LIQUIDITY_NETWORK = "liquidity_network" # Liquidity pool based
|
|
HYBRID = "hybrid" # Multiple models
|
|
INTENT_BASED = "intent_based" # Intent/auction based
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Bridge Registry
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
BRIDGE_REGISTRY = {
|
|
"layerzero": {
|
|
"name": "LayerZero",
|
|
"trust_model": TrustModel.EXTERNAL_VALIDATORS,
|
|
"chains": [
|
|
"ethereum",
|
|
"arbitrum",
|
|
"optimism",
|
|
"base",
|
|
"polygon",
|
|
"bsc",
|
|
"avalanche",
|
|
"fantom",
|
|
"solana",
|
|
],
|
|
"contracts": {
|
|
"ethereum": "0xE55B5B1A8c68E18f93F8aA0a943dB1720C5Fc9a6",
|
|
},
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "layerzero",
|
|
"audit_recency_days": 180,
|
|
"total_exploit_loss_usd": 0, # LayerZero core has not been exploited
|
|
"exploit_history": [],
|
|
"validator_count": 38,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"stargate": {
|
|
"name": "Stargate",
|
|
"trust_model": TrustModel.LIQUIDITY_NETWORK,
|
|
"chains": ["ethereum", "arbitrum", "optimism", "base", "polygon", "bsc", "avalanche"],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "stargate",
|
|
"audit_recency_days": 90,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 0,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"across": {
|
|
"name": "Across Protocol",
|
|
"trust_model": TrustModel.OPTIMISTIC,
|
|
"chains": ["ethereum", "arbitrum", "optimism", "base", "polygon"],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "across",
|
|
"audit_recency_days": 120,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 0,
|
|
"has_upgradeability": True,
|
|
"has_pause": False,
|
|
},
|
|
"wormhole": {
|
|
"name": "Wormhole",
|
|
"trust_model": TrustModel.EXTERNAL_VALIDATORS,
|
|
"chains": [
|
|
"ethereum",
|
|
"solana",
|
|
"arbitrum",
|
|
"optimism",
|
|
"base",
|
|
"polygon",
|
|
"bsc",
|
|
"avalanche",
|
|
],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "wormhole",
|
|
"audit_recency_days": 60,
|
|
"total_exploit_loss_usd": 326_000_000,
|
|
"exploit_history": ["2022-02-02: $326M wETH exploit — guardian key compromise"],
|
|
"validator_count": 19,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"hop": {
|
|
"name": "Hop Protocol",
|
|
"trust_model": TrustModel.OPTIMISTIC,
|
|
"chains": ["ethereum", "arbitrum", "optimism", "base", "polygon", "gnosis"],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "hop",
|
|
"audit_recency_days": 150,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 0,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"synapse": {
|
|
"name": "Synapse",
|
|
"trust_model": TrustModel.HYBRID,
|
|
"chains": ["ethereum", "arbitrum", "optimism", "base", "polygon", "bsc", "avalanche"],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "synapse",
|
|
"audit_recency_days": 120,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 0,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"axelar": {
|
|
"name": "Axelar",
|
|
"trust_model": TrustModel.EXTERNAL_VALIDATORS,
|
|
"chains": [
|
|
"ethereum",
|
|
"arbitrum",
|
|
"optimism",
|
|
"base",
|
|
"polygon",
|
|
"bsc",
|
|
"avalanche",
|
|
"fantom",
|
|
],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "axelar",
|
|
"audit_recency_days": 120,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 75,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"celer": {
|
|
"name": "Celer cBridge",
|
|
"trust_model": TrustModel.LIQUIDITY_NETWORK,
|
|
"chains": ["ethereum", "arbitrum", "optimism", "base", "polygon", "bsc", "avalanche"],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "celer-cbridge",
|
|
"audit_recency_days": 90,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 0,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"debridge": {
|
|
"name": "DeBridge",
|
|
"trust_model": TrustModel.EXTERNAL_VALIDATORS,
|
|
"chains": ["ethereum", "arbitrum", "optimism", "base", "polygon", "bsc", "solana"],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "debridge",
|
|
"audit_recency_days": 120,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 20,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"chainlink_ccip": {
|
|
"name": "Chainlink CCIP",
|
|
"trust_model": TrustModel.EXTERNAL_VALIDATORS,
|
|
"chains": ["ethereum", "arbitrum", "optimism", "base", "polygon", "bsc", "avalanche"],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "chainlink-ccip",
|
|
"audit_recency_days": 60,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 120,
|
|
"has_upgradeability": False,
|
|
"has_pause": False,
|
|
},
|
|
"connext": {
|
|
"name": "Connext",
|
|
"trust_model": TrustModel.INTENT_BASED,
|
|
"chains": ["ethereum", "arbitrum", "optimism", "base", "polygon", "bsc", "gnosis"],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "connext",
|
|
"audit_recency_days": 180,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 0,
|
|
"has_upgradeability": True,
|
|
"has_pause": True,
|
|
},
|
|
"orbiter": {
|
|
"name": "Orbiter Finance",
|
|
"trust_model": TrustModel.INTENT_BASED,
|
|
"chains": [
|
|
"ethereum",
|
|
"arbitrum",
|
|
"optimism",
|
|
"base",
|
|
"polygon",
|
|
"bsc",
|
|
"zksync",
|
|
"starknet",
|
|
],
|
|
"tvl_source": "defillama",
|
|
"defillama_slug": "orbiter-finance",
|
|
"audit_recency_days": 180,
|
|
"total_exploit_loss_usd": 0,
|
|
"exploit_history": [],
|
|
"validator_count": 0,
|
|
"has_upgradeability": True,
|
|
"has_pause": False,
|
|
},
|
|
}
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Data Classes
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
|
|
@dataclass
|
|
class BridgeTVLSnapshot:
|
|
"""TVL snapshot for a single bridge."""
|
|
|
|
bridge_key: str
|
|
bridge_name: str
|
|
tvl_usd: float
|
|
tvl_change_24h_pct: float
|
|
tvl_change_7d_pct: float
|
|
tvl_change_30d_pct: float
|
|
chain_breakdown: dict[str, float] = field(default_factory=dict)
|
|
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
|
|
|
|
|
@dataclass
|
|
class BridgeSecurityScore:
|
|
"""Security rating for a bridge."""
|
|
|
|
bridge_key: str
|
|
bridge_name: str
|
|
overall_score: float # 0-100
|
|
tvl_depth_score: float
|
|
decentralization_score: float
|
|
audit_score: float
|
|
exploit_history_score: float
|
|
upgrade_risk_score: float
|
|
risk_tier: RiskTier
|
|
vulnerabilities: list[str] = field(default_factory=list)
|
|
strengths: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class ExploitSignal:
|
|
"""Detection signal that may indicate an exploit."""
|
|
|
|
bridge_key: str
|
|
bridge_name: str
|
|
signal_type: str # 'tvl_drop', 'large_tx', 'upgrade', 'pause'
|
|
severity: str # 'low', 'medium', 'high', 'critical'
|
|
description: str
|
|
detected_value: Any = None
|
|
threshold_value: Any = None
|
|
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
|
|
|
|
|
@dataclass
|
|
class BridgeHealthReport:
|
|
"""Complete bridge health report."""
|
|
|
|
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
|
total_bridges: int = 0
|
|
bridges_healthy: int = 0
|
|
bridges_watch: int = 0
|
|
bridges_danger: int = 0
|
|
bridges_critical: int = 0
|
|
total_tvl_usd: float = 0.0
|
|
tvl_change_24h_pct: float = 0.0
|
|
bridge_snapshots: dict[str, BridgeTVLSnapshot] = field(default_factory=dict)
|
|
security_scores: dict[str, BridgeSecurityScore] = field(default_factory=dict)
|
|
exploit_signals: list[ExploitSignal] = field(default_factory=list)
|
|
contagion_risk: list[str] = field(default_factory=list)
|
|
|
|
def summary(self) -> str:
|
|
"""Generate human-readable summary."""
|
|
lines = [
|
|
"🌉 CROSS-CHAIN BRIDGE HEALTH REPORT",
|
|
f" Generated: {self.timestamp}",
|
|
"",
|
|
" 📊 Overview",
|
|
f" ├─ Bridges monitored: {self.total_bridges}",
|
|
f" ├─ Healthy: {self.bridges_healthy} ✅",
|
|
f" ├─ Watch: {self.bridges_watch} ⚠️",
|
|
f" ├─ Danger: {self.bridges_danger} 🔴",
|
|
f" ├─ Critical: {self.bridges_critical} 🚨",
|
|
f" ├─ Total TVL: ${self.total_tvl_usd:,.0f}",
|
|
f" └─ 24h TVL change: {self.tvl_change_24h_pct:+.1f}%",
|
|
"",
|
|
]
|
|
|
|
if self.exploit_signals:
|
|
lines.append(f" 🚨 EXPLOIT SIGNALS ({len(self.exploit_signals)})")
|
|
for sig in self.exploit_signals:
|
|
_low = "\u2139\ufe0f"
|
|
emoji = {
|
|
"low": _low,
|
|
"medium": "\u26a0\ufe0f",
|
|
"high": "\U0001f534",
|
|
"critical": "\U0001f6a8",
|
|
}.get(sig.severity, _low)
|
|
lines.append(
|
|
f" {emoji} [{sig.severity.upper()}] {sig.bridge_name}: {sig.description}"
|
|
)
|
|
lines.append("")
|
|
|
|
if self.contagion_risk:
|
|
lines.append(
|
|
f" 📡 Contagion Risk — {len(self.contagion_risk)} bridges may be affected"
|
|
)
|
|
for b in self.contagion_risk:
|
|
lines.append(f" ├─ {b}")
|
|
lines.append("")
|
|
|
|
lines.append(" 🔒 Bridge Security Scores")
|
|
sorted_bridges = sorted(
|
|
self.security_scores.items(),
|
|
key=lambda x: x[1].overall_score,
|
|
)
|
|
for _key, score in sorted_bridges:
|
|
emoji = {
|
|
RiskTier.SAFE: "🟢",
|
|
RiskTier.WATCH: "🟡",
|
|
RiskTier.DANGER: "🟠",
|
|
RiskTier.CRITICAL: "🔴",
|
|
}.get(score.risk_tier, "⚪")
|
|
lines.append(
|
|
f" {emoji} {score.bridge_name:<20} {score.overall_score:>5.1f}/100 "
|
|
f"[TVL={score.tvl_depth_score:.0f} Dec={score.decentralization_score:.0f} "
|
|
f"Audit={score.audit_score:.0f} Exploit={score.exploit_history_score:.0f} "
|
|
f"Upgrade={score.upgrade_risk_score:.0f}]"
|
|
)
|
|
if score.vulnerabilities:
|
|
for v in score.vulnerabilities:
|
|
lines.append(f" ⚠ {v}")
|
|
|
|
lines.append("")
|
|
lines.append(" 💡 Recommendation:")
|
|
if self.bridges_critical > 0:
|
|
lines.append(
|
|
" 🚨 CRITICAL: Exploit likely in progress. Avoid using affected bridges."
|
|
)
|
|
elif self.bridges_danger > 0:
|
|
lines.append(" 🔴 DANGER: High-risk bridges detected. Use with extreme caution.")
|
|
elif self.bridges_watch > 0:
|
|
lines.append(" ⚠️ WATCH: Monitor bridges flagged below. Elevated risk.")
|
|
else:
|
|
lines.append(" ✅ All bridges appear healthy. Normal monitoring cadence.")
|
|
|
|
return "\n".join(lines)
|
|
|
|
def to_json(self) -> str:
|
|
"""Serialize to JSON."""
|
|
return json.dumps(asdict(self), indent=2, default=str)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Core Monitor
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
|
|
class BridgeHealthMonitor:
|
|
"""Cross-chain bridge health & exploit monitor."""
|
|
|
|
def __init__(self, defillama_api: str = "https://coins.llama.fi"):
|
|
self.defillama_api = defillama_api
|
|
self.session: aiohttp.ClientSession | None = None
|
|
self._cache: dict[str, Any] = {}
|
|
self._cache_ttl: int = 300 # 5 minutes
|
|
|
|
# TVL anomaly thresholds
|
|
self.TVL_DROP_24H_WARN_PCT = -15.0 # 15% drop in 24h = watch
|
|
self.TVL_DROP_24H_CRIT_PCT = -40.0 # 40% drop in 24h = critical
|
|
self.TVL_DROP_7D_DANGER_PCT = -50.0 # 50% drop in 7d = danger
|
|
|
|
async def _get_session(self) -> aiohttp.ClientSession:
|
|
"""Get or create an aiohttp session."""
|
|
if self.session is None or self.session.closed:
|
|
self.session = aiohttp.ClientSession(
|
|
timeout=aiohttp.ClientTimeout(total=15),
|
|
headers={"User-Agent": "RMI-BridgeMonitor/1.0"},
|
|
)
|
|
return self.session
|
|
|
|
async def _fetch_tvl(self, slug: str) -> dict[str, Any] | None:
|
|
"""Fetch TVL data for a protocol from DeFiLlama."""
|
|
cache_key = f"tvl_{slug}"
|
|
if cache_key in self._cache:
|
|
cached, ts = self._cache[cache_key]
|
|
if time.time() - ts < self._cache_ttl:
|
|
return cached
|
|
|
|
try:
|
|
session = await self._get_session()
|
|
url = f"{self.defillama_api}/protocol/{slug}"
|
|
async with session.get(url) as resp:
|
|
if resp.status != 200:
|
|
logger.warning(f"DeFiLlama TVL fetch failed for {slug}: {resp.status}")
|
|
return None
|
|
data = await resp.json()
|
|
self._cache[cache_key] = (data, time.time())
|
|
return data
|
|
except (TimeoutError, aiohttp.ClientError, json.JSONDecodeError) as e:
|
|
logger.error(f"Error fetching TVL for {slug}: {e}")
|
|
return None
|
|
|
|
async def _fetch_current_tvl(self, slug: str) -> float:
|
|
"""Get current TVL for a protocol."""
|
|
data = await self._fetch_tvl(slug)
|
|
if not data:
|
|
return 0.0
|
|
current_chart = data.get("chainTvls", data.get("tvl", []))
|
|
if isinstance(current_chart, dict):
|
|
# Sum across chains
|
|
total = 0.0
|
|
for chain_data in current_chart.values():
|
|
if chain_data and isinstance(chain_data, list) and len(chain_data) > 0:
|
|
total += chain_data[-1].get("totalLiquidityUSD", 0)
|
|
return total
|
|
elif isinstance(current_chart, list) and len(current_chart) > 0:
|
|
return current_chart[-1].get("totalLiquidityUSD", 0)
|
|
return 0.0
|
|
|
|
async def _fetch_tvl_change(self, slug: str, hours: int = 24) -> float:
|
|
"""Calculate TVL percentage change over the given period."""
|
|
data = await self._fetch_tvl(slug)
|
|
if not data:
|
|
return 0.0
|
|
|
|
tvl_data = data.get("chainTvls", data.get("tvl", []))
|
|
if isinstance(tvl_data, dict):
|
|
# Sum across chains to get current TVL and estimate change
|
|
current_total = 0.0
|
|
historical_total = 0.0
|
|
for chain_data in tvl_data.values():
|
|
if isinstance(chain_data, list) and len(chain_data) >= 2:
|
|
current_total += chain_data[-1].get("totalLiquidityUSD", 0)
|
|
idx = min(hours // 24, len(chain_data) - 1)
|
|
historical_total += chain_data[-1 - idx].get("totalLiquidityUSD", 0)
|
|
if historical_total > 0 and current_total > 0:
|
|
return ((current_total - historical_total) / historical_total) * 100
|
|
return 0.0
|
|
elif isinstance(tvl_data, list) and len(tvl_data) > 0:
|
|
current = tvl_data[-1].get("totalLiquidityUSD", 0)
|
|
# Try to find the point closest to `hours` ago
|
|
# DeFiLlama returns arrays with ~daily frequency
|
|
idx = min(hours // 24, len(tvl_data) - 1)
|
|
historical_val = tvl_data[-1 - idx].get("totalLiquidityUSD", current)
|
|
if historical_val > 0 and current > 0:
|
|
return ((current - historical_val) / historical_val) * 100
|
|
return 0.0
|
|
|
|
async def _compute_security_score(self, bridge_key: str, bridge: dict) -> BridgeSecurityScore:
|
|
"""Compute a comprehensive security score for a bridge."""
|
|
vulnerabilities: list[str] = []
|
|
strengths: list[str] = []
|
|
|
|
# 1. TVL depth score (more TVL = more at stake but also more battle-tested)
|
|
tvl = await self._fetch_current_tvl(bridge.get("defillama_slug", ""))
|
|
if tvl >= 1_000_000_000: # $1B+
|
|
tvl_score = 20 # High TVL = well-capitalized, battle-tested
|
|
strengths.append("High TVL (>$1B) — well-capitalized and battle-tested")
|
|
elif tvl >= 100_000_000: # $100M+
|
|
tvl_score = 15
|
|
elif tvl >= 10_000_000: # $10M+
|
|
tvl_score = 10
|
|
elif tvl > 0:
|
|
tvl_score = 5
|
|
else:
|
|
tvl_score = 0 # No data
|
|
|
|
# 2. Decentralization score
|
|
validator_count = bridge.get("validator_count", 0)
|
|
trust_model = bridge.get("trust_model", TrustModel.LIQUIDITY_NETWORK)
|
|
|
|
if trust_model == TrustModel.EXTERNAL_VALIDATORS:
|
|
if validator_count >= 50:
|
|
dec_score = 25
|
|
strengths.append(f"Highly decentralized ({validator_count} validators)")
|
|
elif validator_count >= 20:
|
|
dec_score = 20
|
|
strengths.append(f"Moderately decentralized ({validator_count} validators)")
|
|
elif validator_count > 0:
|
|
dec_score = 15
|
|
else:
|
|
dec_score = 10
|
|
elif trust_model == TrustModel.OPTIMISTIC:
|
|
dec_score = 20 # No trusted validators needed
|
|
strengths.append("Optimistic trust model — no active validator set required")
|
|
elif trust_model == TrustModel.INTENT_BASED:
|
|
dec_score = 18
|
|
strengths.append("Intent-based architecture — minimal trust assumptions")
|
|
elif trust_model == TrustModel.LIQUIDITY_NETWORK:
|
|
dec_score = 12 # Depends on LP composition
|
|
else: # HYBRID
|
|
dec_score = 15
|
|
|
|
# 3. Audit recency score
|
|
audit_days = bridge.get("audit_recency_days", 365)
|
|
if audit_days <= 60:
|
|
audit_score = 20
|
|
strengths.append("Recent audit (<60 days)")
|
|
elif audit_days <= 120:
|
|
audit_score = 15
|
|
elif audit_days <= 180:
|
|
audit_score = 10
|
|
else:
|
|
audit_score = 5
|
|
vulnerabilities.append(f"Audit is {audit_days} days old — recommend re-audit")
|
|
|
|
# 4. Exploit history score (penalize past exploits)
|
|
exploit_loss = bridge.get("total_exploit_loss_usd", 0)
|
|
if exploit_loss == 0:
|
|
exploit_score = 20
|
|
strengths.append("No history of major exploits")
|
|
elif exploit_loss < 10_000_000:
|
|
exploit_score = 10
|
|
vulnerabilities.append(f"Past exploit(s) — ${exploit_loss:,} total losses")
|
|
elif exploit_loss < 100_000_000:
|
|
exploit_score = 5
|
|
vulnerabilities.append(f"Significant past exploit(s) — ${exploit_loss:,} total losses")
|
|
else:
|
|
exploit_score = 0
|
|
vulnerabilities.append(f"Major past exploit(s) — ${exploit_loss:,} total losses")
|
|
|
|
# 5. Upgrade risk score
|
|
has_upgrade = bridge.get("has_upgradeability", True)
|
|
has_pause = bridge.get("has_pause", False)
|
|
if not has_upgrade:
|
|
upgrade_score = 15
|
|
strengths.append("Non-upgradeable — immutable contracts")
|
|
elif has_pause:
|
|
upgrade_score = 10
|
|
vulnerabilities.append(
|
|
"Upgradeable with pause — admin can modify contracts, "
|
|
"but pause provides emergency response"
|
|
)
|
|
else:
|
|
upgrade_score = 5
|
|
vulnerabilities.append(
|
|
"Upgradeable without pause — admin can modify contracts "
|
|
"with no emergency stop mechanism"
|
|
)
|
|
|
|
# Total score
|
|
total_score = tvl_score + dec_score + audit_score + exploit_score + upgrade_score
|
|
|
|
# Determine risk tier
|
|
if total_score >= 85:
|
|
risk_tier = RiskTier.SAFE
|
|
elif total_score >= 65:
|
|
risk_tier = RiskTier.WATCH
|
|
elif total_score >= 45:
|
|
risk_tier = RiskTier.DANGER
|
|
else:
|
|
risk_tier = RiskTier.CRITICAL
|
|
|
|
return BridgeSecurityScore(
|
|
bridge_key=bridge_key,
|
|
bridge_name=bridge.get("name", bridge_key),
|
|
overall_score=total_score,
|
|
tvl_depth_score=tvl_score,
|
|
decentralization_score=dec_score,
|
|
audit_score=audit_score,
|
|
exploit_history_score=exploit_score,
|
|
upgrade_risk_score=upgrade_score,
|
|
risk_tier=risk_tier,
|
|
vulnerabilities=vulnerabilities,
|
|
strengths=strengths,
|
|
)
|
|
|
|
async def _detect_exploit_signals(
|
|
self,
|
|
bridge_key: str,
|
|
bridge: dict,
|
|
tvl_snapshot: BridgeTVLSnapshot | None,
|
|
) -> list[ExploitSignal]:
|
|
"""Detect exploit signals for a specific bridge."""
|
|
signals: list[ExploitSignal] = []
|
|
|
|
if tvl_snapshot is None:
|
|
return signals
|
|
|
|
# Signal 1: TVL drop anomaly
|
|
if tvl_snapshot.tvl_change_24h_pct <= self.TVL_DROP_24H_CRIT_PCT:
|
|
signals.append(
|
|
ExploitSignal(
|
|
bridge_key=bridge_key,
|
|
bridge_name=bridge.get("name", bridge_key),
|
|
signal_type="tvl_drop",
|
|
severity="critical",
|
|
description=(
|
|
f"Critical TVL drop of {tvl_snapshot.tvl_change_24h_pct:.1f}% in 24h "
|
|
f"(${tvl_snapshot.tvl_usd:,.0f} remaining). Possible active exploit or mass withdrawal event."
|
|
),
|
|
detected_value=tvl_snapshot.tvl_change_24h_pct,
|
|
threshold_value=self.TVL_DROP_24H_CRIT_PCT,
|
|
)
|
|
)
|
|
elif tvl_snapshot.tvl_change_24h_pct <= self.TVL_DROP_24H_WARN_PCT:
|
|
signals.append(
|
|
ExploitSignal(
|
|
bridge_key=bridge_key,
|
|
bridge_name=bridge.get("name", bridge_key),
|
|
signal_type="tvl_drop",
|
|
severity="high",
|
|
description=(
|
|
f"Significant TVL drop of {tvl_snapshot.tvl_change_24h_pct:.1f}% in 24h "
|
|
f"(${tvl_snapshot.tvl_usd:,.0f} remaining). Requires investigation."
|
|
),
|
|
detected_value=tvl_snapshot.tvl_change_24h_pct,
|
|
threshold_value=self.TVL_DROP_24H_WARN_PCT,
|
|
)
|
|
)
|
|
|
|
# Signal 2: 7-day TVL drop
|
|
if tvl_snapshot.tvl_change_7d_pct <= self.TVL_DROP_7D_DANGER_PCT:
|
|
signals.append(
|
|
ExploitSignal(
|
|
bridge_key=bridge_key,
|
|
bridge_name=bridge.get("name", bridge_key),
|
|
signal_type="tvl_drop",
|
|
severity="high",
|
|
description=(
|
|
f"7-day TVL decline of {tvl_snapshot.tvl_change_7d_pct:.1f}%. "
|
|
f"Sustained capital outflow — protocol health concern."
|
|
),
|
|
detected_value=tvl_snapshot.tvl_change_7d_pct,
|
|
threshold_value=self.TVL_DROP_7D_DANGER_PCT,
|
|
)
|
|
)
|
|
|
|
return signals
|
|
|
|
async def _compute_contagion_risk(
|
|
self,
|
|
scores: dict[str, BridgeSecurityScore],
|
|
signals: list[ExploitSignal],
|
|
) -> list[str]:
|
|
"""Check if bridges with similar security profiles to compromised bridges
|
|
are at contagion risk."""
|
|
contagion: list[str] = []
|
|
|
|
triggered_bridges = {s.bridge_key for s in signals if s.severity in ("high", "critical")}
|
|
if not triggered_bridges:
|
|
return contagion
|
|
|
|
for triggered_key in triggered_bridges:
|
|
triggered_score = scores.get(triggered_key)
|
|
if not triggered_score:
|
|
continue
|
|
|
|
# Find bridges with similar security profiles
|
|
for other_key, other_score in scores.items():
|
|
if other_key in triggered_bridges:
|
|
continue
|
|
if other_key == triggered_key:
|
|
continue
|
|
|
|
score_diff = abs(triggered_score.overall_score - other_score.overall_score)
|
|
if score_diff <= 10: # Similar security profile
|
|
contagion.append(
|
|
f"{other_score.bridge_name} (score {other_score.overall_score}) "
|
|
f"— shares similar security profile with compromised "
|
|
f"{triggered_score.bridge_name} (diff: {score_diff:.0f} pts)"
|
|
)
|
|
|
|
return list(set(contagion)) # Deduplicate
|
|
|
|
async def scan(self, bridge_filter: str | None = None) -> BridgeHealthReport:
|
|
"""Run a complete bridge health scan.
|
|
|
|
Args:
|
|
bridge_filter: Optional bridge key to scan only one bridge.
|
|
|
|
Returns:
|
|
BridgeHealthReport with all findings.
|
|
"""
|
|
report = BridgeHealthReport()
|
|
bridges_to_scan = (
|
|
{bridge_filter: BRIDGE_REGISTRY[bridge_filter]}
|
|
if bridge_filter and bridge_filter in BRIDGE_REGISTRY
|
|
else BRIDGE_REGISTRY
|
|
)
|
|
|
|
# Phase 1: Fetch TVL snapshots for all bridges
|
|
tvl_tasks = {}
|
|
for key, bridge in bridges_to_scan.items():
|
|
slug = bridge.get("defillama_slug", "")
|
|
tvl_tasks[key] = asyncio.create_task(self._fetch_current_tvl(slug))
|
|
|
|
tvl_24h_tasks = {}
|
|
for key, bridge in bridges_to_scan.items():
|
|
slug = bridge.get("defillama_slug", "")
|
|
tvl_24h_tasks[key] = asyncio.create_task(self._fetch_tvl_change(slug, 24))
|
|
|
|
tvl_7d_tasks = {}
|
|
for key, bridge in bridges_to_scan.items():
|
|
slug = bridge.get("defillama_slug", "")
|
|
tvl_7d_tasks[key] = asyncio.create_task(self._fetch_tvl_change(slug, 168)) # 7 days
|
|
|
|
tvl_30d_tasks = {}
|
|
for key, bridge in bridges_to_scan.items():
|
|
slug = bridge.get("defillama_slug", "")
|
|
tvl_30d_tasks[key] = asyncio.create_task(self._fetch_tvl_change(slug, 720)) # 30 days
|
|
|
|
# Wait for TVL data
|
|
tvl_results = {}
|
|
for key in bridges_to_scan:
|
|
try:
|
|
tvl_results[key] = {
|
|
"current": await tvl_tasks[key],
|
|
"24h": await tvl_24h_tasks[key],
|
|
"7d": await tvl_7d_tasks[key],
|
|
"30d": await tvl_30d_tasks[key],
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"TVL fetch error for {key}: {e}")
|
|
tvl_results[key] = {"current": 0.0, "24h": 0.0, "7d": 0.0, "30d": 0.0}
|
|
|
|
# Phase 2: Build TVL snapshots
|
|
for key, bridge in bridges_to_scan.items():
|
|
tvl_data = tvl_results.get(key, {})
|
|
snapshot = BridgeTVLSnapshot(
|
|
bridge_key=key,
|
|
bridge_name=bridge.get("name", key),
|
|
tvl_usd=tvl_data.get("current", 0.0),
|
|
tvl_change_24h_pct=tvl_data.get("24h", 0.0),
|
|
tvl_change_7d_pct=tvl_data.get("7d", 0.0),
|
|
tvl_change_30d_pct=tvl_data.get("30d", 0.0),
|
|
)
|
|
report.bridge_snapshots[key] = snapshot
|
|
report.total_tvl_usd += snapshot.tvl_usd
|
|
|
|
report.total_bridges = len(bridges_to_scan)
|
|
|
|
# Phase 3: Compute security scores
|
|
score_tasks = {}
|
|
for key, bridge in bridges_to_scan.items():
|
|
score_tasks[key] = asyncio.create_task(self._compute_security_score(key, bridge))
|
|
|
|
for key in bridges_to_scan:
|
|
try:
|
|
report.security_scores[key] = await score_tasks[key]
|
|
except Exception as e:
|
|
logger.error(f"Score computation error for {key}: {e}")
|
|
|
|
# Phase 4: Detect exploit signals
|
|
for key, bridge in bridges_to_scan.items():
|
|
snapshot = report.bridge_snapshots.get(key)
|
|
signals = await self._detect_exploit_signals(key, bridge, snapshot)
|
|
report.exploit_signals.extend(signals)
|
|
|
|
# Phase 5: Compute contagion risk
|
|
report.contagion_risk = await self._compute_contagion_risk(
|
|
report.security_scores, report.exploit_signals
|
|
)
|
|
|
|
# Phase 6: Tally risk tiers
|
|
for score in report.security_scores.values():
|
|
if score.risk_tier == RiskTier.SAFE:
|
|
report.bridges_healthy += 1
|
|
elif score.risk_tier == RiskTier.WATCH:
|
|
report.bridges_watch += 1
|
|
elif score.risk_tier == RiskTier.DANGER:
|
|
report.bridges_danger += 1
|
|
elif score.risk_tier == RiskTier.CRITICAL:
|
|
report.bridges_critical += 1
|
|
|
|
# Compute aggregate TVL change
|
|
if report.bridge_snapshots:
|
|
total_old_tvl = sum(
|
|
s.tvl_usd / (1 + s.tvl_change_24h_pct / 100) if s.tvl_change_24h_pct != -100 else 0
|
|
for s in report.bridge_snapshots.values()
|
|
if s.tvl_usd > 0
|
|
)
|
|
if total_old_tvl > 0:
|
|
report.tvl_change_24h_pct = (
|
|
(report.total_tvl_usd - total_old_tvl) / total_old_tvl * 100
|
|
)
|
|
|
|
return report
|
|
|
|
async def alert_if_exploit(self, bridge_filter: str | None = None) -> BridgeHealthReport | None:
|
|
"""Quick scan that returns a report only if exploit signals are detected.
|
|
|
|
Returns None if all bridges are healthy — useful for cron jobs
|
|
that only want to alert on anomalies.
|
|
"""
|
|
report = await self.scan(bridge_filter)
|
|
if report.exploit_signals or report.bridges_danger > 0 or report.bridges_critical > 0:
|
|
return report
|
|
return None
|
|
|
|
async def close(self):
|
|
"""Clean up resources."""
|
|
if self.session and not self.session.closed:
|
|
await self.session.close()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Standalone CLI
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def main():
|
|
"""CLI entry point."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Cross-Chain Bridge Health & Exploit Monitor")
|
|
parser.add_argument(
|
|
"--bridge",
|
|
type=str,
|
|
default=None,
|
|
help="Scan a specific bridge key (e.g., 'stargate', 'wormhole')",
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Output as JSON",
|
|
)
|
|
parser.add_argument(
|
|
"--alert",
|
|
action="store_true",
|
|
help="Only output if exploit detected (cron-friendly)",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
monitor = BridgeHealthMonitor()
|
|
try:
|
|
if args.alert:
|
|
report = await monitor.alert_if_exploit(args.bridge)
|
|
if report is None:
|
|
print("[SILENT]")
|
|
return
|
|
else:
|
|
report = await monitor.scan(args.bridge)
|
|
|
|
if args.json:
|
|
print(report.to_json())
|
|
else:
|
|
print(report.summary())
|
|
finally:
|
|
await monitor.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|