Bridge Security Watchdog

Bridge Security Watchdog

Overview

The Bridge Security Watchdog is a critical safety tool that monitors Lock-and-Mint bridge protocols (Stargate, Wormhole/Portal, Across, Hop) for large or suspicious withdrawals before users interact with them. By combining TVL monitoring, withdrawal pattern detection, and historical stability analysis, it provides actionable safety scores (0-100) to help users avoid compromised bridges.

What Makes This Unique

Unlike simple TVL trackers, this skill actively watches for exploit indicators:

• Real-time TVL drain detection: Spots sudden liquidity drops that may indicate exploits
• Large withdrawal monitoring: Tracks unusual token movements from bridge contracts
• Pattern recognition: Identifies rapid withdrawal sequences characteristic of exploits
• Multi-dimensional scoring: Combines TVL, withdrawals, history, and volume for comprehensive safety assessment

Problem Solved

Bridge exploits have resulted in billions in losses (Wormhole: $326M, Nomad: $190M, Ronin: $625M). Users need pre-transaction safety checks before committing funds to bridges. This skill provides:

1. Pre-Bridge Safety Check: "Is this bridge safe to use RIGHT NOW?"
2. Exploit Detection: Identifies bridges actively being drained
3. Comparative Analysis: "Which bridge is safest for my transfer?"
4. Risk Quantification: Clear 0-100 safety scores with actionable recommendations

Core Features

1. TVL Monitoring (bridge_tvl_monitor.py)

Monitors Total Value Locked changes across bridge protocols using DefiLlama API.

Capabilities:

• Real-time TVL tracking across all major bridges
• Chain-by-chain TVL breakdown
• Historical comparison (24h default, configurable)
• Alert generation for TVL drops (>2%, >5%, >10%, >20%)
• Severity classification (SAFE, LOW, MEDIUM, HIGH, CRITICAL)

Detection Patterns:

TVL Drop Thresholds:
- <2% drop: SAFE (normal fluctuation)
- 2-5% drop: LOW RISK 
- 5-10% drop: MEDIUM RISK
- 10-20% drop: HIGH RISK
- >20% drop: CRITICAL (potential exploit)

Usage:

from scripts.bridge_tvl_monitor import BridgeTVLMonitor

monitor = BridgeTVLMonitor()

# Monitor specific bridge
result = monitor.monitor_bridge("stargate", time_window_hours=24)
print(f"TVL Change: {result['tvl_analysis']['change_percent']}%")
print(f"Risk Level: {result['risk_level']}")

# Monitor all bridges
all_bridges = monitor.monitor_all_bridges()
print(f"Critical Alerts: {all_bridges['monitoring_summary']['critical_alerts']}")

2. Withdrawal Detection (withdrawal_detector.py)

Monitors on-chain Transfer events from bridge contracts to detect suspicious withdrawals.

Capabilities:

• Scans recent blocks (default: 1000 blocks ~3-4 hours)
• Tracks large withdrawals ($1M+, $5M+, $10M+)
• Identifies withdrawals to non-CEX addresses (higher risk)
• Detects rapid withdrawal patterns
• Distinguishes between normal activity and potential exploits

Detection Patterns:

Withdrawal Thresholds:
- $1M+ to non-CEX: MEDIUM ALERT
- $5M+ anywhere: HIGH ALERT
- $10M+ anywhere: CRITICAL ALERT
- 3+ large transfers in short time: RAPID WITHDRAWAL ALERT

Known Addresses:

• Tracks 14+ major CEX addresses (Binance, Coinbase, Kraken)
• Withdrawals to CEXs treated as lower risk
• Withdrawals to unknown addresses flagged for review

Usage:

from scripts.withdrawal_detector import WithdrawalDetector

detector = WithdrawalDetector()

# Monitor Stargate on Ethereum
result = detector.monitor_bridge(
    bridge_id="stargate",
    chain="ethereum",
    blocks_to_scan=1000
)

print(f"Alerts: {result['monitoring_summary']['alerts_triggered']}")
print(f"Total Volume: ${result['monitoring_summary']['total_volume_usd']:,.0f}")

# Check specific alerts
for alert in result['alerts']:
    print(f"{alert['severity']}: ${alert['amount_usd']:,.0f} {alert['token']}")

3. Safety Scorer (bridge_safety_scorer.py)

Main orchestrator that combines all analyses to generate comprehensive safety scores.

Scoring System (0-100 points):

Component Weights:
├─ TVL Stability: 40 points
│  ├─ TVL increasing: 40 pts
│  ├─ <2% drop: 36 pts
│  ├─ 2-5% drop: 28 pts
│  ├─ 5-10% drop: 16 pts
│  ├─ 10-20% drop: 8 pts
│  └─ >20% drop: 0 pts
│
├─ Withdrawal Safety: 35 points
│  ├─ No alerts: 35 pts
│  ├─ Minor alerts: 30 pts
│  ├─ Medium alerts: 25 pts
│  ├─ 1 high alert: 18 pts
│  ├─ 2+ high alerts: 7 pts
│  └─ Critical alerts: 0 pts
│
├─ Historical Stability: 15 points
│  ├─ Stargate: 15 pts (best track record)
│  ├─ Across: 14 pts
│  ├─ Hop: 13 pts
│  ├─ Synapse: 12 pts
│  ├─ Portal: 11 pts
│  ├─ Wormhole: 10 pts
│  └─ Unknown: 9 pts
│
└─ Volume Health: 10 points
   ├─ 20+ withdrawals + $1M+ volume: 10 pts
   ├─ 10+ withdrawals + $500k+ volume: 8 pts
   ├─ 5+ withdrawals: 6 pts
   └─ <5 withdrawals: 4 pts

Risk Levels:

90-100: SAFE ✅
  → "Bridge is safe to use"
  
70-89: LOW RISK ✅
  → "Safe with normal precautions"
  
50-69: MEDIUM RISK ⚡
  → "Use with caution, reduce amounts"
  
30-49: HIGH RISK ⚠️
  → "Avoid if possible"
  
0-29: CRITICAL 🚨
  → "DO NOT USE"

Usage:

from scripts.bridge_safety_scorer import BridgeSafetyScorer

scorer = BridgeSafetyScorer()

# Analyze single bridge
result = scorer.calculate_safety_score(
    bridge_id="stargate",
    chain="ethereum",
    detailed=True
)

print(f"Safety Score: {result['safety_score']}/100")
print(f"Risk Level: {result['risk_level']}")
print(f"Recommendation: {result['recommendation']}")

# Compare multiple bridges
comparison = scorer.compare_bridges(
    ["stargate", "across", "hop"],
    chain="ethereum"
)

for rank in comparison['rankings']:
    print(f"{rank['rank']}. {rank['bridge']}: {rank['safety_score']}/100")

Supported Bridges & Chains

Monitored Bridge Protocols

Chain Support

API Dependencies

DefiLlama API (Free, No Key Required)

Endpoints Used:

GET https://bridges.llama.fi/bridges
  → List all bridge protocols (alternative endpoint)

GET https://api.llama.fi/protocol/{bridge_id}
  → Get bridge TVL history and chain breakdown
  → Primary endpoint used in production

GET https://api.llama.fi/protocols
  → Get protocol list data

Rate Limits:

• Free tier: 300 requests/5min (not strictly enforced)
• No authentication required
• No fallback data - fails gracefully with error messages

Blockchain RPC (Free Public Endpoints)

Used For:

• On-chain Transfer event scanning
• Token balance queries
• Transaction data retrieval

Error Handling:

• Returns empty results when RPC fails
• Graceful failure with clear error messages
• No dummy/fallback data - production-grade error handling

Configuration

Environment Variables (Optional)

# Optional: Paid RPC for higher rate limits
export ALCHEMY_API_KEY="your_key"
export INFURA_API_KEY="your_key"
export QUICKNODE_ENDPOINT="your_endpoint"

# Optional: Explorer API keys for enhanced data
export ETHERSCAN_API_KEY="your_key"
export ARBISCAN_API_KEY="your_key"

Customizable Parameters

# TVL Monitor Configuration
time_window_hours = 24  # How far back to compare TVL
alert_thresholds = {
    "CRITICAL": 20.0,   # % TVL drop
    "HIGH": 10.0,
    "MEDIUM": 5.0,
    "LOW": 2.0
}

# Withdrawal Detector Configuration
blocks_to_scan = 1000  # Recent blocks to analyze
thresholds = {
    "large_withdrawal_usd": 1000000,
    "very_large_withdrawal_usd": 5000000,
    "critical_withdrawal_usd": 10000000,
    "rapid_withdrawal_count": 5
}

# Safety Scorer Weights
weights = {
    "tvl_stability": 40,
    "withdrawal_safety": 35,
    "historical_stability": 15,
    "volume_health": 10
}

Integration Examples

1. Pre-Bridge Safety Check

from scripts.bridge_safety_scorer import BridgeSafetyScorer

def check_bridge_before_use(bridge_name, chain, amount_usd):
    """Check if bridge is safe before committing funds"""
    scorer = BridgeSafetyScorer()
    
    result = scorer.calculate_safety_score(
        bridge_id=bridge_name.lower(),
        chain=chain.lower()
    )
    
    safety_score = result['safety_score']
    risk_level = result['risk_level']
    
    # Apply transfer amount-based logic
    if risk_level == "CRITICAL":
        return False, "DO NOT USE - Bridge compromised"
    
    if risk_level == "HIGH RISK":
        return False, "Bridge unstable - Use alternative"
    
    if risk_level == "MEDIUM RISK" and amount_usd > 10000:
        return False, "Reduce amount or use safer bridge"
    
    if safety_score >= 70:
        return True, f"Safe to bridge (Score: {safety_score}/100)"
    
    return False, "Safety check failed - Monitor situation"

# Usage
safe, message = check_bridge_before_use("Stargate", "Ethereum", 50000)
if safe:
    print(f"✅ {message}")
    # Proceed with bridge transaction
else:
    print(f"🚨 {message}")
    # Block transaction or suggest alternative

2. Bridge Selection Helper

from scripts.bridge_safety_scorer import BridgeSafetyScorer

def recommend_safest_bridge(source_chain, dest_chain):
    """Recommend the safest bridge for a route"""
    scorer = BridgeSafetyScorer()
    
    # Bridges supporting the route
    available_bridges = ["stargate", "across", "hop", "synapse"]
    
    comparison = scorer.compare_bridges(
        available_bridges,
        chain=source_chain
    )
    
    # Get top 3 safest bridges
    rankings = comparison['rankings'][:3]
    
    print(f"Recommended Bridges for {source_chain} → {dest_chain}:\n")
    for rank in rankings:
        print(f"{rank['rank']}. {rank['bridge']}")
        print(f"   Safety: {rank['safety_score']}/100 ({rank['risk_level']})")
        print(f"   {rank['recommendation']}\n")
    
    return rankings[0]['bridge']  # Return safest

# Usage
best_bridge = recommend_safest_bridge("ethereum", "arbitrum")
print(f"Use: {best_bridge}")

3. Continuous Monitoring Dashboard

from scripts.bridge_tvl_monitor import BridgeTVLMonitor
import time

def monitor_bridges_continuously(interval_seconds=300):
    """Monitor all bridges and alert on issues"""
    monitor = BridgeTVLMonitor()
    
    while True:
        print(f"\n[{datetime.now()}] Scanning bridges...")
        
        results = monitor.monitor_all_bridges()
        summary = results['monitoring_summary']
        
        if summary['critical_alerts'] > 0:
            print(f"🚨 CRITICAL: {summary['critical_alerts']} bridge(s) at risk!")
            # Send alert notification
            
        if summary['high_risk_alerts'] > 0:
            print(f"⚠️ WARNING: {summary['high_risk_alerts']} bridge(s) unstable")
        
        # Wait before next scan
        time.sleep(interval_seconds)

# Run continuous monitoring
monitor_bridges_continuously(interval_seconds=300)  # Every 5 minutes

Security Considerations

Data Sources

1. DefiLlama (TVL Data):

   • Aggregates from official bridge APIs
   • Historical data validated against multiple sources
   • Updates every 5-15 minutes

2. On-Chain Data (Withdrawals):

   • Direct from blockchain RPC
   • Transfer events are immutable
   • Real-time or near-real-time (<1 min delay)

3. Known Addresses (CEX Detection):

   • Manually curated list of major CEX wallets
   • Regularly updated from Etherscan labels
   • Reduces false positives for large transfers

Limitations

1. False Positives:

   • Large legitimate withdrawals may trigger alerts
   • CEX rebalancing can look like exploits
   • Mitigation: CEX address whitelist, severity levels

2. Detection Delay:

   • TVL data updates every 5-15 minutes
   • On-chain scanning limited by RPC rate limits
   • Mitigation: Use paid RPC for production monitoring

3. Chain Coverage:

   • Withdrawal monitoring requires RPC access
   • Free RPCs have rate limits (may return 0 results)
   • Mitigation: Use Alchemy/Infura/QuickNode for production

Best Practices

• Never rely solely on automated scoring
• Cross-reference with official bridge announcements
• Monitor multiple indicators before large transfers
• Use paid RPCs for production/high-volume monitoring
• Whitelist trusted addresses to reduce false positives

Testing & Validation

Manual Testing

# Test TVL monitor with real data
from scripts.bridge_tvl_monitor import BridgeTVLMonitor

monitor = BridgeTVLMonitor()
result = monitor.monitor_bridge('stargate', time_window_hours=24)
print(f"TVL: ${result['tvl_analysis']['current_tvl']:,.2f}")
print(f"Change: {result['tvl_analysis']['change_percent']:.2f}%")

# Test withdrawal detector
from scripts.withdrawal_detector import WithdrawalDetector

detector = WithdrawalDetector()
result = detector.monitor_bridge('stargate', 'ethereum', blocks_to_scan=1000)
print(f"Withdrawals: {result['monitoring_summary']['total_withdrawals']}")

# Test complete safety analysis
from scripts.bridge_safety_scorer import BridgeSafetyScorer

scorer = BridgeSafetyScorer()
result = scorer.calculate_safety_score('stargate', 'ethereum')
print(f"Safety Score: {result['safety_score']}/100")

Expected Outputs

Healthy Bridge (Real Example - Stargate):

Safety Score: 82/100
Risk Level: LOW RISK
Current TVL: $9,753,697
24h Change: -3.42% (normal fluctuation)
Withdrawals: 0 detected in last 1000 blocks
Alerts: 0
Recommendation: ✅ Bridge appears safe with minor concerns

High-TVL Bridge (Real Example - Portal/Wormhole):

Bridge: Portal
Current TVL: $1,279,654,159
24h Change: -2.72% (normal fluctuation)
Risk Level: LOW
Recommendation: ✅ Safe for large transfers

Note: When RPC fails or no activity detected:

• Withdrawal count: 0
• Alerts: 0
• Status: Returns empty results (not fake data)
• Safety score still calculated using TVL + historical data

Performance Metrics

Real Test Results:

• TVL Analysis: 2-3 seconds per bridge
• Withdrawal Detection: 12-14 seconds (1000 blocks on Ethereum)
• Complete Safety Analysis: 14-16 seconds (TVL + Withdrawals + Scoring)
• Multi-Bridge Comparison: 30-40 seconds (3 bridges)

Data Accuracy:

• TVL Data: Direct from DefiLlama Protocol API (real-time)
• Withdrawal Detection: Direct from blockchain via RPC (on-chain verified)
• Memory Usage: <50MB per analysis
• Network Bandwidth: ~500KB per comprehensive analysis

API Performance:

• DefiLlama API: 1-3 seconds latency, 99%+ uptime
• Ethereum RPC (0xrpc.io): 10-15 seconds for 1000 block scan
• No Rate Limits: Encountered in testing (free tier sufficient for development)

Future Enhancements

• Machine learning for exploit pattern recognition
• Integration with more bridge protocols
• Historical exploit database for pattern matching
• WebSocket real-time monitoring
• Discord/Telegram alert bot
• Multi-chain transfer route optimization
• Bridge insurance integration
• Automated transaction blocking

License & Attribution

Created for the SpoonOS Skills ecosystem demonstrating production-ready Web3 data intelligence and on-chain analysis.

Data Sources:

• DefiLlama (https://defillama.com)
• Public blockchain RPC endpoints
• Etherscan address labels