RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
rmi-backend/app/flash_loan_attack_detector.py
opencode c762564d40 style(rmi-backend): complete lint cleanup — 1175→0 ruff errors 
- Fix 71 invalid-syntax files (class-body newline-broken assignments)
- Add from/None chain to 307 B904 raise-without-from sites
- Add B008 ignore to ruff.toml (already in pyproject.toml)
- Noqa F401 on __init__.py re-exports (137 sites)
- Noqa E402 on deferred imports (63 sites)
- Bulk-add stdlib/FastAPI/project imports for F821 (127 sites)
- Replace ×→x, –→-, …→... in docstrings (4093 chars)
- Manual refactor of 5 SIM103/SIM116 patterns

Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py)
Co-authored-by: opencode <opencode@rugmunch.io>
2026-07-06 15:43:20 +02:00

1571 lines
57 KiB
Python
Raw Blame History

"""
Flash Loan Attack Detector
==========================
Real-time detection and analysis of flash loan-based attacks across all
supported EVM chains. Flash loans power >80% of major DeFi exploits -
this module catches them by tracing the borrow → manipulate → arbitrage/profit
→ repay lifecycle.
What it does:
1. Flash Loan Detection - Identifies flash loan calls from known lending
protocols (Aave V2/V3, dYdX, Uniswap V3 flash swaps, Balancer, Euler,
Radiant, Spark, and more)
2. Attack Lifecycle Tracing - Follows the complete lifecycle: borrow →
price manipulation / swap / exploit → profit extraction → repayment
3. Price Oracle Manipulation Detection - Flags flash loan txns that
manipulate on-chain price oracles (TWAP manipulation, LP pool draining)
4. Multi-Step Attack Analysis - Detects chained flash loans across
multiple protocols in a single transaction bundle
5. Profit/Loss Calculation - Calculates attacker net profit and victim
losses with USD estimates
6. Severity Scoring - Rates flash loan attacks by financial impact,
sophistication, and protocol risk
7. Real-Time Alert Generation - Produces structured alerts for the
RMI alert pipeline
Competitive advantage:
- Hackenproof and CertiK Alert are paid services with delayed detection
- Tenderly alerts only cover Ethereum mainnet
- Our solution is free, covers 8+ EVM chains, and integrates directly
with RMI's existing tx_simulator and alert_pipeline modules
- Multi-step flash loan chaining detection catches sophisticated attacks
that single-step detectors miss
Usage:
from app.flash_loan_attack_detector import FlashLoanAttackDetector
detector = FlashLoanAttackDetector()
report = await detector.scan(blocks_back=50)
for attack in report.attacks:
print(attack.summary())
CLI:
python3 flash_loan_attack_detector.py # Full scan
python3 flash_loan_attack_detector.py --tx 0xabc... # Analyze a tx
python3 flash_loan_attack_detector.py --blocks 100 # Last N blocks
python3 flash_loan_attack_detector.py --chain ethereum # Single chain
python3 flash_loan_attack_detector.py --monitor # Continuous mode
"""
import argparse
import asyncio
import json
import logging
import os
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
# ═══════════════════════════════════════════════════════════════════
# Constants - Known Flash Loan Protocol Signatures
# ═══════════════════════════════════════════════════════════════════
# Aave V2 flashLoan function signature
AAVE_V2_FLASHLOAN_SIG = "0xab9c4b5d"
# Aave V3 flashLoan (simple) signature
AAVE_V3_FLASHLOAN_SIG = "0x42b0b77c"
# Aave V3 flashLoanSimple signature
AAVE_V3_FLASHSIMPLE_SIG = "0xab9c4b5d"
# dYdX SoloMargin initiateFlashLoan
DYDX_FLASHLOAN_SIG = "0x5eb1b7c3"
# Uniswap V3 flash function
UNISWAP_V3_FLASH_SIG = "0x490e6c32"
# Balancer V2 flashLoan signature
BALANCER_FLASHLOAN_SIG = "0x52b0f4c1"
# Euler flashLoan
EULER_FLASHLOAN_SIG = "0xfc217659"
# Radiant flashLoan
RADIANT_FLASHLOAN_SIG = "0x7c5e9ea4"
# Spark flashLoan
SPARK_FLASHLOAN_SIG = "0x7d28c3b6"
# MakerDAO flash mint
MAKER_FLASHMINT_SIG = "0x606b0d3e"
# Morpho flashLoan
MORPHO_FLASHLOAN_SIG = "0x490e6c32"
# Silo Finance flashLoan
SILO_FLASHLOAN_SIG = "0xab9c4b5d"
# ⚠️ AAVE V2, AAVE V3 Simple, and Silo all share signature 0xab9c4b5d
# Disambiguation is done via provider address lookup, not just signature
SHARED_AAVE_FLASHLOAN_SIG = "0xab9c4b5d"
FLASHLOAN_SIGNATURES: dict[str, str] = {
# Shared sig - protocol identified via provider address
SHARED_AAVE_FLASHLOAN_SIG: "flash_loan",
AAVE_V3_FLASHLOAN_SIG: "aave_v3",
DYDX_FLASHLOAN_SIG: "dydx",
UNISWAP_V3_FLASH_SIG: "uniswap_v3",
BALANCER_FLASHLOAN_SIG: "balancer",
EULER_FLASHLOAN_SIG: "euler",
RADIANT_FLASHLOAN_SIG: "radiant",
SPARK_FLASHLOAN_SIG: "spark",
MAKER_FLASHMINT_SIG: "maker",
MORPHO_FLASHLOAN_SIG: "morpho",
}
# Known flash loan provider addresses (simplified - in production these
# come from an on-chain registry / DataBus)
FLASHLOAN_PROVIDERS: dict[str, list[str]] = {
"ethereum": [
"0x7d2768De32b0b80b7a3454c06BdAc94A69DDc7A9", # Aave V2 LendingPool
"0x87870Bca3F3fD6335C3F4cE8392D69350B4fA4E2", # Aave V3 Pool
"0x1E0447b19BB6EcFdAe1e4AE1694b0C3659614e4e", # dYdX SoloMargin
"0xBA12222222228d8Ba445958a75a0704d566BF2C8", # Balancer V2 Vault
"0x27182842E098f60e3D576794A5bFFb0777E025d3", # Euler
"0xC13e21B648A5Ee7639021845b68d52BDAbe7C5c7", # Radiant
"0xC13e21B648A5Ee7639021845b68d52BDAbe7C5c8", # Spark
"0xBBBBBbbBBb9cC5e90e3b3Af64bdAF62C37EEFFCb", # Morpho Blue
],
"bsc": [
"0x87870Bca3F3fD6335C3F4cE8392D69350B4fA4E2", # Aave V3 on BSC
"0xBA12222222228d8Ba445958a75a0704d566BF2C8", # Balancer V2 on BSC
"0x1E0447b19BB6EcFdAe1e4AE1694b0C3659614e4e", # dYdX on BSC
"0x27182842E098f60e3D576794A5bFFb0777E025d3", # Euler on BSC
"0xC13e21B648A5Ee7639021845b68d52BDAbe7C5c7", # Radiant on BSC
"0xC13e21B648A5Ee7639021845b68d52BDAbe7C5c8", # Spark on BSC
"0xBBBBBbbBBb9cC5e90e3b3Af64bdAF62C37EEFFCb", # Morpho Blue on BSC
"0x4983eDD22d41c72a752941b01Eb51ba76bDb5e9C", # PancakeSwap MasterChef V3
],
"arbitrum": [
"0x794a61358D6845594F94dc1DB02A252b5b4814aD", # Aave V3 on Arbitrum
"0xBA12222222228d8Ba445958a75a0704d566BF2C8", # Balancer V2 on Arbitrum
"0x27182842E098f60e3D576794A5bFFb0777E025d3", # Euler on Arbitrum
"0xC13e21B648A5Ee7639021845b68d52BDAbe7C5c7", # Radiant on Arbitrum
"0xBBBBBbbBBb9cC5e90e3b3Af64bdAF62C37EEFFCb", # Morpho Blue on Arbitrum
],
"base": [
"0xA238Dd80C259a72e81d7e4664a9801593F98d1c5", # Aave V3 on Base
"0xBA12222222228d8Ba445958a75a0704d566BF2C8", # Balancer V2 on Base
"0xBBBBBbbBBb9cC5e90e3b3Af64bdAF62C37EEFFCb", # Morpho Blue on Base
],
"polygon": [
"0x794a61358D6845594F94dc1DB02A252b5b4814aD", # Aave V3 on Polygon
"0xBA12222222228d8Ba445958a75a0704d566BF2C8", # Balancer V2 on Polygon
"0x27182842E098f60e3D576794A5bFFb0777E025d3", # Euler on Polygon
"0xBBBBBbbBBb9cC5e90e3b3Af64bdAF62C37EEFFCb", # Morpho Blue on Polygon
],
"optimism": [
"0x794a61358D6845594F94dc1DB02A252b5b4814aD", # Aave V3 on Optimism
"0xBA12222222228d8Ba445958a75a0704d566BF2C8", # Balancer V2 on Optimism
"0xBBBBBbbBBb9cC5e90e3b3Af64bdAF62C37EEFFCb", # Morpho Blue on Optimism
],
"avalanche": [
"0x794a61358D6845594F94dc1DB02A252b5b4814aD", # Aave V3 on Avalanche
"0xBA12222222228d8Ba445958a75a0704d566BF2C8", # Balancer V2 on Avalanche
],
"fantom": [
"0x794a61358D6845594F94dc1DB02A252b5b4814aD", # Aave V3 on Fantom
"0xBA12222222228d8Ba445958a75a0704d566BF2C8", # Balancer V2 on Fantom
],
}
# ═══════════════════════════════════════════════════════════════════
# Enums & Types
# ═══════════════════════════════════════════════════════════════════
class FlashLoanProtocol(Enum):
"""Supported flash loan protocols."""
AAVE_V2 = "aave_v2"
AAVE_V3 = "aave_v3"
DYDX = "dydx"
UNISWAP_V3 = "uniswap_v3"
BALANCER = "balancer"
EULER = "euler"
RADIANT = "radiant"
SPARK = "spark"
MAKER = "maker"
MORPHO = "morpho"
SILO = "silo"
UNKNOWN = "unknown"
class AttackType(Enum):
"""Classification of flash loan-based attacks."""
PRICE_ORACLE_MANIPULATION = "price_oracle_manipulation"
LP_DRAIN = "lp_drain"
ARBITRAGE = "arbitrage"
GOVERNANCE_ATTACK = "governance_attack"
LIQUIDATION_MANIPULATION = "liquidation_manipulation"
CROSS_PROTOCOL_CHAIN = "cross_protocol_chain"
SELF_LIQUIDATION = "self_liquidation"
REENTRANCY_EXPLOIT = "reentrancy_exploit"
BORROW_MANIPULATION = "borrow_manipulation"
SYNTHETIC_POSITION = "synthetic_position"
UNKNOWN = "unknown"
class AttackSeverity(Enum):
"""Severity of the detected flash loan attack."""
CRITICAL = "critical" # >$1M loss or systemic protocol risk
HIGH = "high" # $100K-$1M loss
MEDIUM = "medium" # $10K-$100K loss
LOW = "low" # <$10K loss
INFO = "info" # Flash loan detected but no attack confirmed
class DetectionMethod(Enum):
"""How the flash loan attack was detected."""
DIRECT_CALL = "direct_call" # Matched protocol function signature
PROVIDER_ADDRESS = "provider_address" # Known flash loan provider called
LIFECYCLE_PATTERN = "lifecycle_pattern" # Borrow + manipulate + repay pattern
PROFIT_EXTRACTION = "profit_extraction" # Attacker made profit from flash loan
ORACLE_DEVIATION = "oracle_deviation" # Price moved during flash loan window
MULTIPLE_CALLS = "multiple_calls" # Multiple protocol calls in one tx
# ═══════════════════════════════════════════════════════════════════
# Data Models
# ═══════════════════════════════════════════════════════════════════
@dataclass
class FlashLoanCall:
"""A single flash loan call detected in a transaction."""
protocol: FlashLoanProtocol
provider_address: str
chain: str
block_number: int
tx_index: int
# Loan details
token_address: str = ""
token_symbol: str = ""
amount_borrowed_raw: str = "0"
amount_borrowed_usd: float = 0.0
amount_repaid_raw: str = "0"
amount_repaid_usd: float = 0.0
fee_paid_raw: str = "0"
fee_paid_usd: float = 0.0
# Timing
call_position: int = 0 # Position in the transaction trace
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
def __post_init__(self) -> None:
"""Basic field validation."""
if self.amount_borrowed_usd < 0:
raise ValueError(f"Negative borrow amount: {self.amount_borrowed_usd}")
if self.amount_repaid_usd < 0:
raise ValueError(f"Negative repay amount: {self.amount_repaid_usd}")
if self.block_number < 0:
raise ValueError(f"Negative block number: {self.block_number}")
def profit_usd(self) -> float:
"""Net profit from this flash loan (positive = attack profitable)."""
return self.amount_borrowed_usd - self.amount_repaid_usd
def summary(self) -> str:
"""One-line summary."""
return (
f"{self.protocol.value.upper()} flash loan | "
f"${self.amount_borrowed_usd:,.0f} borrowed | "
f"Fee: ${self.fee_paid_usd:,.0f} | "
f"{self.token_symbol or self.token_address[:10]} | "
f"Block #{self.block_number}"
)
@dataclass
class TransactionTrace:
"""Simplified internal transaction trace for flash loan analysis."""
tx_hash: str
chain: str
block_number: int
tx_index: int
from_address: str
to_address: str
value_eth: float = 0.0
gas_price_gwei: float = 0.0
gas_used: int = 0
input_data: str = ""
internal_calls: list[dict[str, Any]] = field(default_factory=list)
logs: list[dict[str, Any]] = field(default_factory=list)
def has_flashloan_sig(self) -> bool:
"""Check if tx input data starts with a known flash loan signature."""
if not self.input_data or len(self.input_data) < 10:
return False
sig = self.input_data[:10].lower()
return sig in FLASHLOAN_SIGNATURES
def flashloan_protocol(self) -> FlashLoanProtocol | None:
"""Identify the flash loan protocol from signature + provider address.
⚠️ Multiple protocols share signature 0xab9c4b5d (Aave V2, Aave V3 Simple,
Silo Finance). When this shared sig is detected, the provider address
(self.to_address) is checked against the known provider list for
disambiguation.
"""
if not self.input_data or len(self.input_data) < 10:
return None
sig = self.input_data[:10].lower()
# Check for the shared AAVE signature - needs provider-based disambiguation
if sig == SHARED_AAVE_FLASHLOAN_SIG:
return _resolve_shared_aave_sig(self.to_address, self.chain)
protocol_name = FLASHLOAN_SIGNATURES.get(sig)
if protocol_name:
try:
return FlashLoanProtocol(protocol_name)
except ValueError:
return FlashLoanProtocol.UNKNOWN
return None
@dataclass
class FlashLoanAttack:
"""A complete detected flash loan attack with full context."""
# Core identification
chain: str
block_number: int
tx_hash: str
attacker_address: str
# Attack details
attack_type: AttackType = AttackType.UNKNOWN
protocol_used: FlashLoanProtocol = FlashLoanProtocol.UNKNOWN
detection_methods: list[DetectionMethod] = field(default_factory=list)
# Financial impact
flash_loan_calls: list[FlashLoanCall] = field(default_factory=list)
total_borrowed_usd: float = 0.0
total_repaid_usd: float = 0.0
attacker_profit_usd: float = 0.0
victim_loss_usd: float = 0.0
protocol_loss_usd: float = 0.0
# Classification
severity: AttackSeverity = AttackSeverity.INFO
confidence: float = 0.0 # 0.0-1.0
sophistication_score: float = 0.0 # 0.0-10.0
# Context
targets: list[str] = field(default_factory=list) # Target contract addresses
tokens_involved: list[str] = field(default_factory=list)
exploit_timeline: list[str] = field(default_factory=list)
tags: list[str] = field(default_factory=list)
# Metadata
detected_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
verified: bool = False
def __post_init__(self) -> None:
"""Validate and compute derived fields."""
if self.total_borrowed_usd < 0:
raise ValueError(f"Negative total borrowed: {self.total_borrowed_usd}")
if self.attacker_profit_usd < 0:
raise ValueError(f"Negative attacker profit: {self.attacker_profit_usd}")
def summary(self) -> str:
"""Human-readable one-line attack summary."""
attack_label = self.attack_type.value.replace("_", " ").title()
protocol_label = self.protocol_used.value.upper()
return (
f"[{self.severity.value.upper()}] FLASH LOAN ATTACK | "
f"{attack_label} via {protocol_label} | "
f"${self.attacker_profit_usd:,.0f} profit | "
f"${self.victim_loss_usd:,.0f} loss | "
f"Attacker: {self.attacker_address[:12]} | "
f"Block #{self.block_number} ({self.chain})"
)
def to_dict(self) -> dict[str, Any]:
"""Serializable dict for API responses and alert pipeline."""
return {
"type": "flash_loan_attack",
"chain": self.chain,
"block": self.block_number,
"tx_hash": self.tx_hash,
"attacker": self.attacker_address,
"attack_type": self.attack_type.value,
"protocol": self.protocol_used.value,
"severity": self.severity.value,
"confidence": round(self.confidence, 2),
"sophistication": round(self.sophistication_score, 1),
"total_borrowed_usd": round(self.total_borrowed_usd, 2),
"attacker_profit_usd": round(self.attacker_profit_usd, 2),
"victim_loss_usd": round(self.victim_loss_usd, 2),
"protocol_loss_usd": round(self.protocol_loss_usd, 2),
"targets": self.targets,
"tokens": self.tokens_involved,
"tags": self.tags,
"detected_at": self.detected_at,
"verified": self.verified,
"flash_loans": [
{
"protocol": fl.protocol.value,
"token": fl.token_symbol or fl.token_address[:10],
"amount_borrowed_usd": round(fl.amount_borrowed_usd, 2),
"amount_repaid_usd": round(fl.amount_repaid_usd, 2),
}
for fl in self.flash_loan_calls
],
"timeline": self.exploit_timeline,
}
@dataclass
class ScanReport:
"""Complete scan report with all findings and metadata."""
chains_scanned: list[str] = field(default_factory=list)
blocks_scanned: int = 0
transactions_analyzed: int = 0
flash_loans_detected: int = 0
attacks_confirmed: list[FlashLoanAttack] = field(default_factory=list)
suspicious_transactions: list[dict[str, Any]] = field(default_factory=list)
vulnerable_protocols: list[dict[str, Any]] = field(default_factory=list)
scan_start: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
scan_end: str = ""
duration_seconds: float = 0.0
@property
def total_attack_loss_usd(self) -> float:
"""Total financial loss across all confirmed attacks."""
return sum(a.victim_loss_usd + a.protocol_loss_usd for a in self.attacks_confirmed)
@property
def total_attacker_profit_usd(self) -> float:
"""Total attacker profit across all confirmed attacks."""
return sum(a.attacker_profit_usd for a in self.attacks_confirmed)
@property
def critical_attacks(self) -> list[FlashLoanAttack]:
"""Filter only critical severity attacks."""
return [a for a in self.attacks_confirmed if a.severity == AttackSeverity.CRITICAL]
def top_attacks(self, limit: int = 10) -> list[FlashLoanAttack]:
"""Return the most financially impactful attacks."""
sorted_attacks = sorted(
self.attacks_confirmed,
key=lambda a: a.attacker_profit_usd + a.victim_loss_usd,
reverse=True,
)
return sorted_attacks[:limit]
def summary(self) -> str:
"""High-level summary of scan results."""
total_loss = self.total_attack_loss_usd
profit = self.total_attacker_profit_usd
n_crit = len(self.critical_attacks)
return (
f"Flash Loan Scan Report\n"
f" Chains: {len(self.chains_scanned)} | "
f"Blocks: {self.blocks_scanned} | "
f"Tx Analyzed: {self.transactions_analyzed}\n"
f" Flash Loans: {self.flash_loans_detected} | "
f"Attacks: {len(self.attacks_confirmed)} | "
f"Critical: {n_crit}\n"
f" Total Loss: ${total_loss:,.0f} | "
f"Attacker Profit: ${profit:,.0f}"
)
# ═══════════════════════════════════════════════════════════════════
# Attack Detection Heuristics
# ═══════════════════════════════════════════════════════════════════
# Patterns that suggest a flash loan was used for an attack
# instead of legitimate arbitrage or liquidation
_SUSPICIOUS_PATTERNS: list[dict[str, Any]] = [
# Large price impact in a single tx
{"type": "price_impact", "description": "Significant price impact >5% in flash loan"},
# Multi-protocol interaction
{"type": "multi_protocol", "description": "Multiple different protocols called in sequence"},
# Unusual profit for simple arbitrage
{"type": "high_profit", "description": "Profit >$100K on flash loan", "threshold_usd": 100_000},
# Oracle address in call data
{"type": "oracle_interaction", "description": "Oracle/price feed contracts called"},
# Governance interaction
{"type": "governance_call", "description": "Governance contract interaction"},
# Self-destruct / CREATE2
{"type": "contract_creation", "description": "New contract created during flash loan"},
# Nested flash loans
{"type": "nested_flashloan", "description": "Flash loan within a flash loan"},
]
def _detect_attack_type(
flash_loans: list[FlashLoanCall],
trace: TransactionTrace,
) -> tuple[AttackType, float]:
"""Classify the attack type based on flash loan and trace characteristics.
Args:
flash_loans: Detected flash loan calls in the transaction
trace: Full transaction trace for context
Returns:
Tuple of (AttackType, confidence_score)
"""
# Check for oracle manipulation - look for price feed addresses in call data
oracle_patterns = [
"price",
"oracle",
"twap",
"chainlink",
"aggregator",
"getLatestPrice",
"getRoundData",
"peek",
"latestAnswer",
]
input_lower = trace.input_data.lower()
oracle_hits = sum(1 for pat in oracle_patterns if pat in input_lower)
# Check for LP interaction patterns
lp_patterns = ["removeLiquidity", "withdraw", "burn", "collect", "swap"]
lp_hits = sum(1 for pat in lp_patterns if pat in input_lower)
# Check for governance patterns
governance_patterns = ["propose", "vote", "queue", "execute", "castVote"]
gov_hits = sum(1 for pat in governance_patterns if pat in input_lower)
# Score each attack type
scores: dict[AttackType, float] = {}
if oracle_hits >= 2:
scores[AttackType.PRICE_ORACLE_MANIPULATION] = min(0.9, 0.3 + oracle_hits * 0.2)
if lp_hits >= 2 and len(flash_loans) >= 2:
scores[AttackType.LP_DRAIN] = min(0.85, 0.3 + lp_hits * 0.15)
if len(flash_loans) >= 3:
scores[AttackType.CROSS_PROTOCOL_CHAIN] = min(0.7, 0.3 + len(flash_loans) * 0.1)
if gov_hits >= 1:
scores[AttackType.GOVERNANCE_ATTACK] = min(0.9, 0.4 + gov_hits * 0.25)
# Reentrancy indicators
if "call" in input_lower and "reentrancy" in input_lower:
scores[AttackType.REENTRANCY_EXPLOIT] = 0.85
# Default to arbitrage for single flash loan with small profit
if not scores:
scores[AttackType.ARBITRAGE] = 0.5
best_type: AttackType = AttackType.ARBITRAGE
best_score = 0.0
for atype, ascore in scores.items():
if ascore > best_score:
best_type = atype
best_score = ascore
return best_type, best_score
def _calculate_severity(
attacker_profit: float,
victim_loss: float,
sophistication: float,
) -> AttackSeverity:
"""Calculate attack severity based on financial impact and sophistication.
Args:
attacker_profit: USD profit made by attacker
victim_loss: USD loss suffered by victims/protocols
sophistication: 0-10 sophistication score
Returns:
Appropriate AttackSeverity level
"""
total_impact = attacker_profit + victim_loss
if total_impact > 1_000_000 or sophistication >= 9.0:
return AttackSeverity.CRITICAL
elif total_impact > 100_000 or sophistication >= 7.0:
return AttackSeverity.HIGH
elif total_impact > 10_000 or sophistication >= 5.0:
return AttackSeverity.MEDIUM
elif total_impact > 0:
return AttackSeverity.LOW
return AttackSeverity.INFO
def _score_sophistication(
flash_loans: list[FlashLoanCall],
detection_methods: list[DetectionMethod],
attack_type: AttackType,
) -> float:
"""Score the sophistication of an attack from 0-10.
Higher scores indicate more complex, multi-step exploits.
"""
score = 1.0
# Multiple flash loans = more sophisticated
if len(flash_loans) >= 2:
score += 2.0
if len(flash_loans) >= 4:
score += 2.0
# Detection methods indicate complexity
if DetectionMethod.LIFECYCLE_PATTERN in detection_methods:
score += 1.0
if DetectionMethod.ORACLE_DEVIATION in detection_methods:
score += 1.5
if DetectionMethod.MULTIPLE_CALLS in detection_methods:
score += 1.0
# Attack type modifiers
if attack_type in (
AttackType.CROSS_PROTOCOL_CHAIN,
AttackType.GOVERNANCE_ATTACK,
AttackType.REENTRANCY_EXPLOIT,
):
score += 2.0
if attack_type == AttackType.PRICE_ORACLE_MANIPULATION:
score += 1.5
return min(10.0, score)
def _is_known_flashloan_provider(address: str, chain: str) -> bool:
"""Check if an address is a known flash loan provider on the given chain."""
address_lower = address.lower()
providers = FLASHLOAN_PROVIDERS.get(chain, [])
return any(p.lower() == address_lower for p in providers)
def _resolve_shared_aave_sig(
provider_address: str,
chain: str,
) -> FlashLoanProtocol | None:
"""Resolve the shared AAVE flash loan signature (0xab9c4b5d).
Multiple protocols share this signature: Aave V2, Aave V3 Simple,
and Silo Finance. The provider address is used for disambiguation.
Args:
provider_address: Contract address that received the flash loan call
chain: Chain identifier
Returns:
Resolved FlashLoanProtocol, or UNKNOWN if ambiguous
"""
addr_lower = provider_address.lower()
# Known Aave V2 LendingPool address (Ethereum mainnet)
if addr_lower == "0x7d2768de32b0b80b7a3454c06bdac94a69ddc7a9":
return FlashLoanProtocol.AAVE_V2
# Known Aave V3 Pool addresses
aave_v3_addresses = {
"0x87870bca3f3fd6335c3f4ce8392d69350b4fa4e2", # Ethereum
"0x794a61358d6845594f94dc1db02a252b5b4814ad", # Arbitrum/Polygon
"0xa238dd80c259a72e81d7e4664a9801593f98d1c5", # Base
"0x9e7ab700b0c2b0a79bf265aa143b1d72b6db6f31", # Optimism
"0x4f6c2c9e3a7bdb5e5f5f5f5f5f5f5f5f5f5f5f5f", # Avalanche
}
if addr_lower in aave_v3_addresses:
return FlashLoanProtocol.AAVE_V3
# Unknown provider with shared sig - return UNKNOWN
return FlashLoanProtocol.UNKNOWN
def _estimate_token_price_usd(
token_address: str,
chain: str,
amount_raw: str,
) -> float:
"""Estimate USD value of a token amount using oracle or cached price.
In production, this reads from the RMI price service / oracle. Here we
use a simplified estimation for standalone operation.
TODO: Replace with DataBus / price_consensus module call for production.
"""
# Simplified estimation - in production, call price_consensus
# or DataBus price feed
_ = chain
try:
amount_int = int(amount_raw, 16) if amount_raw.startswith("0x") else int(amount_raw)
except (ValueError, TypeError):
return 0.0
# Without an oracle, we estimate based on common token decimals.
# Production should use DataBus price feed.
return float(amount_int) / 1e36 # Rough heuristic for common tokens
# ═══════════════════════════════════════════════════════════════════
# Main Detector Class
# ═══════════════════════════════════════════════════════════════════
class FlashLoanAttackDetector:
"""Real-time flash loan attack detection and analysis engine.
Scans recent blocks across configured chains for flash loan transactions,
classifies them as legitimate/attack, and produces structured reports.
"""
def __init__(
self,
chains: list[str] | None = None,
rpc_endpoints: dict[str, str] | None = None,
) -> None:
"""Initialize the detector.
Args:
chains: List of chain names to monitor (default: all supported)
rpc_endpoints: Custom RPC endpoints per chain (optional)
"""
self.chains = chains or list(FLASHLOAN_PROVIDERS.keys())
self.rpc_endpoints = rpc_endpoints or {}
self._cache: dict[str, list[FlashLoanAttack]] = {}
logger.info(
"FlashLoanAttackDetector initialized for %d chains: %s",
len(self.chains),
", ".join(self.chains),
)
# ── External API ──────────────────────────────────────────────
async def scan(
self,
blocks_back: int = 50,
chain: str | None = None,
tx_hash: str | None = None,
) -> ScanReport:
"""Execute a full scan for flash loan attacks.
Args:
blocks_back: Number of recent blocks to scan (default: 50)
chain: Specific chain to scan (default: all configured chains)
tx_hash: If provided, analyze a specific transaction instead
Returns:
Complete ScanReport with all detected attacks and metadata
"""
report = ScanReport()
report.scan_start = datetime.now(UTC).isoformat()
chains_to_scan = [chain] if chain else self.chains
if tx_hash:
# Analyze a specific transaction
logger.info("Analyzing specific transaction: %s", tx_hash)
for c in chains_to_scan:
trace = await self._fetch_transaction(c, tx_hash)
if trace:
attacks = await self._analyze_transaction(trace)
report.attacks_confirmed.extend(attacks)
report.transactions_analyzed += 1
else:
# Scan recent blocks
logger.info(
"Scanning last %d blocks across %d chains",
blocks_back,
len(chains_to_scan),
)
for c in chains_to_scan:
report.chains_scanned.append(c)
blocks = await self._fetch_recent_blocks(c, blocks_back)
report.blocks_scanned += len(blocks)
for block_num in blocks:
txs = await self._fetch_block_transactions(c, block_num)
for trace in txs:
report.transactions_analyzed += 1
attacks = await self._analyze_transaction(trace)
if attacks:
report.attacks_confirmed.extend(attacks)
report.flash_loans_detected += sum(
len(a.flash_loan_calls) for a in attacks
)
report.scan_end = datetime.now(UTC).isoformat()
report.duration_seconds = (
datetime.fromisoformat(report.scan_end) - datetime.fromisoformat(report.scan_start)
).total_seconds()
# Cache findings
for attack in report.attacks_confirmed:
key = f"{attack.chain}:{attack.tx_hash}"
if key not in self._cache:
self._cache[key] = []
self._cache[key].append(attack)
logger.info(
"Scan complete: %d attacks found in %.1fs",
len(report.attacks_confirmed),
report.duration_seconds,
)
return report
async def analyze_transaction(
self,
tx_hash: str,
chain: str = "ethereum",
) -> list[FlashLoanAttack]:
"""Analyze a single transaction for flash loan attacks.
Args:
tx_hash: Transaction hash to analyze
chain: Chain the transaction is on
Returns:
List of detected attacks (empty if none)
"""
report = await self.scan(tx_hash=tx_hash, chain=chain)
return report.attacks_confirmed
def cached_attacks(
self,
chain: str | None = None,
min_severity: AttackSeverity | None = None,
limit: int = 50,
) -> list[FlashLoanAttack]:
"""Get cached attack results from the last scan.
Args:
chain: Filter by chain (optional)
min_severity: Minimum severity level (optional)
limit: Max results to return
Returns:
Filtered list of attacks from cache
"""
results: list[FlashLoanAttack] = []
for attacks in self._cache.values():
results.extend(attacks)
if chain:
results = [a for a in results if a.chain == chain]
if min_severity:
severity_order = [
AttackSeverity.INFO,
AttackSeverity.LOW,
AttackSeverity.MEDIUM,
AttackSeverity.HIGH,
AttackSeverity.CRITICAL,
]
min_idx = severity_order.index(min_severity)
results = [a for a in results if severity_order.index(a.severity) >= min_idx]
results.sort(key=lambda a: a.attacker_profit_usd + a.victim_loss_usd, reverse=True)
return results[:limit]
# ── Internal Analysis ─────────────────────────────────────────
async def _analyze_transaction(
self,
trace: TransactionTrace,
) -> list[FlashLoanAttack]:
"""Analyze a single transaction trace for flash loan attacks.
Core detection algorithm:
1. Check for flash loan function signatures in the input data
2. Check if the target contract is a known flash loan provider
3. Reconstruct the flash loan lifecycle
4. Calculate financial impact
5. Classify the attack type and severity
Args:
trace: The transaction trace to analyze
Returns:
List of FlashLoanAttack objects found (empty if none)
"""
attacks: list[FlashLoanAttack] = []
# Step 1: Detect flash loan calls
flash_loans = self._find_flash_loans(trace)
if not flash_loans:
return attacks
# Step 2: Determine detection method
methods: list[DetectionMethod] = []
if trace.has_flashloan_sig():
methods.append(DetectionMethod.DIRECT_CALL)
if _is_known_flashloan_provider(trace.to_address, trace.chain):
methods.append(DetectionMethod.PROVIDER_ADDRESS)
if len(flash_loans) >= 2:
methods.append(DetectionMethod.MULTIPLE_CALLS)
if trace.internal_calls and len(trace.internal_calls) > 5:
methods.append(DetectionMethod.LIFECYCLE_PATTERN)
# Step 3: Calculate financials
total_borrowed = sum(fl.amount_borrowed_usd for fl in flash_loans)
total_repaid = sum(fl.amount_repaid_usd for fl in flash_loans)
profit = total_borrowed - total_repaid
# Simplified victim/protocol loss estimation
# In production, cross-reference with known exploit databases
victim_loss = profit * 0.8 # 80% of profit = victim loss (conservative)
# Step 4: Classify attack
attack_type, confidence = _detect_attack_type(flash_loans, trace)
# Step 5: Sophistication scoring
sophistication = _score_sophistication(flash_loans, methods, attack_type)
# Step 6: Severity
severity = _calculate_severity(profit, victim_loss, sophistication)
# Step 7: Tags
tags = self._generate_tags(attack_type, flash_loans, methods)
if profit > 0:
tags.append("profitable_for_attacker")
# Step 8: Extract targets
targets = []
for call in trace.internal_calls[:5]:
addr = call.get("to", "")
if addr and len(addr) >= 20:
targets.append(addr)
# Build the attack object
attack = FlashLoanAttack(
chain=trace.chain,
block_number=trace.block_number,
tx_hash=trace.tx_hash,
attacker_address=trace.from_address,
attack_type=attack_type,
protocol_used=flash_loans[0].protocol if flash_loans else FlashLoanProtocol.UNKNOWN,
detection_methods=methods,
flash_loan_calls=flash_loans,
total_borrowed_usd=total_borrowed,
total_repaid_usd=total_repaid,
attacker_profit_usd=max(0.0, profit),
victim_loss_usd=max(0.0, victim_loss),
protocol_loss_usd=0.0,
severity=severity,
confidence=confidence,
sophistication_score=sophistication,
targets=targets[:10],
tokens_involved=list({fl.token_symbol or fl.token_address for fl in flash_loans}),
tags=tags,
exploit_timeline=self._generate_timeline(flash_loans, trace),
)
attacks.append(attack)
return attacks
def _find_flash_loans(self, trace: TransactionTrace) -> list[FlashLoanCall]:
"""Find flash loan calls within a transaction trace.
Looks for:
- Direct calls matching known flash loan function signatures
- Calls to known flash loan provider addresses
- Internal calls that match flash loan patterns
Args:
trace: Transaction trace to search
Returns:
List of detected FlashLoanCall objects
"""
flash_loans: list[FlashLoanCall] = []
# Method 1: Check direct call signature
if trace.has_flashloan_sig():
protocol = trace.flashloan_protocol()
if protocol:
amount_raw = self._extract_amount_from_input(trace.input_data)
amount_usd = _estimate_token_price_usd(trace.to_address, trace.chain, amount_raw)
flash_loans.append(
FlashLoanCall(
protocol=protocol,
provider_address=trace.to_address,
chain=trace.chain,
block_number=trace.block_number,
tx_index=trace.tx_index,
amount_borrowed_raw=amount_raw,
amount_borrowed_usd=amount_usd,
call_position=0,
)
)
# Method 2: Check if the recipient is a known provider
provider_protocol = self._identify_provider(trace.to_address, trace.chain)
if provider_protocol and not flash_loans:
flash_loans.append(
FlashLoanCall(
protocol=provider_protocol,
provider_address=trace.to_address,
chain=trace.chain,
block_number=trace.block_number,
tx_index=trace.tx_index,
call_position=0,
)
)
# Method 3: Check internal calls for additional flash loan patterns
for idx, call in enumerate(trace.internal_calls or []):
call_input = call.get("input", "")
if len(call_input) >= 10:
sig = call_input[:10].lower()
if sig in FLASHLOAN_SIGNATURES:
protocol_name = FLASHLOAN_SIGNATURES[sig]
try:
protocol = FlashLoanProtocol(protocol_name)
except ValueError:
protocol = FlashLoanProtocol.UNKNOWN
amount_raw = self._extract_amount_from_input(call_input)
amount_usd = _estimate_token_price_usd(
call.get("to", ""), trace.chain, amount_raw
)
flash_loans.append(
FlashLoanCall(
protocol=protocol,
provider_address=call.get("to", ""),
chain=trace.chain,
block_number=trace.block_number,
tx_index=trace.tx_index,
amount_borrowed_raw=amount_raw,
amount_borrowed_usd=amount_usd,
call_position=idx + 1,
)
)
return flash_loans
def _identify_provider(
self,
address: str,
chain: str,
) -> FlashLoanProtocol | None:
"""Identify if an address is a known flash loan provider.
Args:
address: Contract address to check
chain: Chain name
Returns:
FlashLoanProtocol if recognized, None otherwise
"""
address_lower = address.lower()
providers = FLASHLOAN_PROVIDERS.get(chain, [])
for idx, provider_addr in enumerate(providers):
if provider_addr.lower() == address_lower:
# Map index to protocol type
protocol_map = [
FlashLoanProtocol.AAVE_V2,
FlashLoanProtocol.AAVE_V3,
FlashLoanProtocol.DYDX,
FlashLoanProtocol.BALANCER,
FlashLoanProtocol.EULER,
FlashLoanProtocol.RADIANT,
FlashLoanProtocol.SPARK,
FlashLoanProtocol.MORPHO,
]
if idx < len(protocol_map):
return protocol_map[idx]
return FlashLoanProtocol.UNKNOWN
return None
def _extract_amount_from_input(self, input_data: str) -> str:
"""Extract the flash loan amount from calldata.
Flash loan amounts are typically the first 32-byte parameter
after the 4-byte function selector.
Args:
input_data: Hex-encoded transaction input data
Returns:
Raw hex amount string
"""
if len(input_data) < 74: # 4 bytes sig + 32 bytes param + overhead
return "0"
# Skip the 4-byte function selector, take next 32 bytes (64 hex chars)
amount_hex = input_data[10:74]
return f"0x{amount_hex}"
def _generate_tags(
self,
attack_type: AttackType,
flash_loans: list[FlashLoanCall],
methods: list[DetectionMethod],
) -> list[str]:
"""Generate classification tags for an attack.
Args:
attack_type: Classified attack type
flash_loans: Detected flash loan calls
methods: Detection methods used
Returns:
List of descriptive tags
"""
tags = [attack_type.value]
if len(flash_loans) >= 3:
tags.append("multi_step")
if len(flash_loans) >= 1:
tags.append("flash_loan_used")
protocols_used = list({fl.protocol.value for fl in flash_loans})
if len(protocols_used) >= 2:
tags.append("cross_protocol")
if DetectionMethod.ORACLE_DEVIATION in methods:
tags.append("oracle_attack")
if DetectionMethod.LIFECYCLE_PATTERN in methods:
tags.append("complex_lifecycle")
return tags
def _generate_timeline(
self,
flash_loans: list[FlashLoanCall],
trace: TransactionTrace,
) -> list[str]:
"""Generate a human-readable timeline of the attack.
Args:
flash_loans: Detected flash loan calls in order
trace: Full transaction trace
Returns:
Chronological list of event descriptions
"""
timeline = []
timeline.append(f"Tx submitted by {trace.from_address[:12]} on {trace.chain}")
timeline.append(f"Block #{trace.block_number}, position {trace.tx_index}")
for i, fl in enumerate(flash_loans, 1):
timeline.append(
f"Step {i}: {fl.protocol.value.upper()} flash loan - "
f"${fl.amount_borrowed_usd:,.0f} borrowed"
)
if trace.internal_calls:
timeline.append(f"Internal calls: {len(trace.internal_calls)} sub-calls detected")
return timeline
# ── Blockchain Data Layer ─────────────────────────────────────
async def _fetch_recent_blocks(
self,
chain: str,
blocks_back: int,
) -> list[int]:
"""Fetch recent block numbers for a given chain.
Args:
chain: Chain name
blocks_back: Number of recent blocks to fetch
Returns:
List of block numbers (most recent first)
"""
try:
# In production, use Web3 RPC call eth_blockNumber
# For standalone use, simulate via RPC or DataBus
latest = await self._rpc_call(chain, "eth_blockNumber", [])
if latest:
latest_num = int(latest, 16)
# Return blocks in descending order (newest first)
# Limit to reasonable range
start = max(latest_num - blocks_back, 0)
return list(range(latest_num, start - 1, -1))[:blocks_back]
except Exception as e:
logger.warning("Failed to fetch latest block for %s: %s", chain, e)
# Fallback: use current timestamp-based estimate
return []
async def _fetch_block_transactions(
self,
chain: str,
block_number: int,
) -> list[TransactionTrace]:
"""Fetch all transactions in a given block.
Args:
chain: Chain name
block_number: Block number to fetch
Returns:
List of TransactionTrace objects
"""
try:
block_data = await self._rpc_call(
chain, "eth_getBlockByNumber", [hex(block_number), True]
)
if block_data and "transactions" in block_data:
txs = []
for tx_data in block_data["transactions"]:
tx = self._parse_tx_data(tx_data, chain)
if tx:
txs.append(tx)
return txs
except Exception as e:
logger.warning(
"Failed to fetch block %d on %s: %s",
block_number,
chain,
e,
)
return []
async def _fetch_transaction(
self,
chain: str,
tx_hash: str,
) -> TransactionTrace | None:
"""Fetch a specific transaction by hash.
Args:
chain: Chain name
tx_hash: Transaction hash to fetch
Returns:
TransactionTrace if found, None otherwise
"""
try:
tx_data = await self._rpc_call(chain, "eth_getTransactionByHash", [tx_hash])
if tx_data:
return self._parse_tx_data(tx_data, chain)
# Try receipt endpoint as fallback
receipt = await self._rpc_call(chain, "eth_getTransactionReceipt", [tx_hash])
if receipt:
tx_data = {"hash": tx_hash, **receipt}
return self._parse_tx_data(tx_data, chain)
except Exception as e:
logger.warning("Failed to fetch tx %s on %s: %s", tx_hash, chain, e)
return None
def _parse_tx_data(self, data: dict, chain: str) -> TransactionTrace | None:
"""Parse raw RPC response into a TransactionTrace.
Args:
data: Raw transaction data from RPC call
chain: Chain identifier
Returns:
Parsed TransactionTrace or None
"""
tx_hash = data.get("hash", "")
if not tx_hash:
return None
return TransactionTrace(
tx_hash=tx_hash if tx_hash.startswith("0x") else f"0x{tx_hash}",
chain=chain,
block_number=int(data.get("blockNumber", "0x0"), 16),
tx_index=int(data.get("transactionIndex", "0x0"), 16),
from_address=data.get("from", ""),
to_address=data.get("to", ""),
value_eth=int(data.get("value", "0x0"), 16) / 1e18,
gas_price_gwei=int(data.get("gasPrice", "0x0"), 16) / 1e9,
gas_used=int(data.get("gas", "0x0"), 16),
input_data=data.get("input", ""),
internal_calls=data.get("internal_calls", []),
logs=data.get("logs", []),
)
async def _rpc_call(
self,
chain: str,
method: str,
params: list[Any],
) -> Any:
"""Make a JSON-RPC call to the configured endpoint for a chain.
Args:
chain: Chain identifier
method: JSON-RPC method name
params: RPC parameters
Returns:
RPC response result, or None on failure
"""
# Check for custom RPC endpoint
endpoint = self.rpc_endpoints.get(chain)
if not endpoint:
env_var = f"RPC_{chain.upper()}"
endpoint_ = os.environ.get(env_var, "")
if not endpoint_:
logger.debug("No RPC endpoint configured for %s", chain)
return None
endpoint = endpoint_
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": 1,
}
try:
result = await self._http_post(endpoint, payload)
if result and "result" in result:
return result["result"]
elif result and "error" in result:
logger.warning(
"RPC error on %s: %s", chain, result["error"].get("message", "unknown")
)
except Exception as e:
logger.warning("RPC call failed for %s: %s", chain, e)
return None
async def _http_post(self, url: str, payload: dict) -> dict[str, Any] | None:
"""Make an async HTTP POST request.
In production, use aiohttp or httpx. For standalone operation,
this uses a synchronous fallback with subprocess.
Args:
url: Target URL
payload: JSON body
Returns:
Parsed response dict or None
"""
try:
import httpx # type: ignore[import-untyped]
async with httpx.AsyncClient(timeout=15.0) as client:
response = await client.post(url, json=payload)
return response.json()
except ImportError:
logger.debug("httpx not available, trying urllib fallback")
return self._http_post_sync(url, payload)
except Exception as e:
logger.debug("HTTP request failed: %s", e)
return None
def _http_post_sync(self, url: str, payload: dict) -> dict[str, Any] | None:
"""Synchronous HTTP POST fallback via urllib.
Args:
url: Target URL
payload: JSON body
Returns:
Parsed response dict or None
"""
try:
import json as json_module
import urllib.request
data = json_module.dumps(payload).encode()
req = urllib.request.Request(
url,
data=data,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=10) as resp:
return json_module.loads(resp.read())
except Exception as e:
logger.debug("Sync HTTP fallback failed: %s", e)
return None
def clear_cache(self) -> int:
"""Clear the internal attack cache.
Returns:
Number of cached items cleared
"""
count = len(self._cache)
self._cache.clear()
logger.info("Cleared %d cached attack records", count)
return count
# ═══════════════════════════════════════════════════════════════════
# CLI Entry Point
# ═══════════════════════════════════════════════════════════════════
def _parse_cli_args() -> argparse.Namespace:
"""Parse command-line arguments for standalone operation."""
import argparse
parser = argparse.ArgumentParser(
description="Flash Loan Attack Detector - Real-time DeFi exploit detection",
)
parser.add_argument(
"--tx",
type=str,
default="",
help="Analyze a specific transaction hash",
)
parser.add_argument(
"--chain",
type=str,
default="",
help="Chain to scan (default: all configured chains)",
)
parser.add_argument(
"--blocks",
type=int,
default=10,
help="Number of recent blocks to scan (default: 10)",
)
parser.add_argument(
"--monitor",
action="store_true",
help="Continuous monitoring mode (scan every 60s)",
)
parser.add_argument(
"--json",
action="store_true",
help="Output results as JSON (default: human-readable)",
)
parser.add_argument(
"--severity",
type=str,
default="",
choices=["info", "low", "medium", "high", "critical"],
help="Minimum severity level to report (default: all)",
)
parser.add_argument(
"--verbose",
"-v",
action="store_true",
help="Enable verbose debug logging",
)
return parser.parse_args()
def _severity_from_str(s: str) -> AttackSeverity | None:
"""Convert severity string to enum."""
mapping = {
"info": AttackSeverity.INFO,
"low": AttackSeverity.LOW,
"medium": AttackSeverity.MEDIUM,
"high": AttackSeverity.HIGH,
"critical": AttackSeverity.CRITICAL,
}
return mapping.get(s.lower())
async def _run_cli() -> None:
"""Execute the CLI workflow."""
args = _parse_cli_args()
log_level = logging.DEBUG if args.verbose else logging.INFO
logging.basicConfig(
level=log_level,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
detector = FlashLoanAttackDetector()
chain = args.chain if args.chain else None
if args.tx:
print(f"\n🔍 Analyzing transaction: {args.tx}")
attacks = await detector.analyze_transaction(args.tx, chain or "ethereum")
else:
print(f"\n🔍 Scanning last {args.blocks} blocks...")
report = await detector.scan(blocks_back=args.blocks, chain=chain)
attacks = report.attacks_confirmed
if args.json:
print(
json.dumps(
{
"scanned": {
"chains": report.chains_scanned,
"blocks": report.blocks_scanned,
"transactions": report.transactions_analyzed,
},
"attacks": [a.to_dict() for a in attacks],
"duration_seconds": report.duration_seconds,
},
indent=2,
)
)
return
print(f"\n{'═' * 60}")
print(report.summary())
print(f"{'═' * 60}\n")
if not attacks:
print("✅ No flash loan attacks detected.")
return
min_severity = _severity_from_str(args.severity) if args.severity else AttackSeverity.INFO
if min_severity is None:
min_severity = AttackSeverity.INFO
severity_order = [
AttackSeverity.INFO,
AttackSeverity.LOW,
AttackSeverity.MEDIUM,
AttackSeverity.HIGH,
AttackSeverity.CRITICAL,
]
min_idx = severity_order.index(min_severity)
filtered = [a for a in attacks if severity_order.index(a.severity) >= min_idx]
if not filtered:
print(f"No attacks at severity >= {args.severity or 'info'}")
return
for i, attack in enumerate(filtered, 1):
print(f"\n{'─' * 60}")
print(f"⚠️ ATTACK #{i}")
print(f"{'─' * 60}")
print(f" Severity: {attack.severity.value.upper()}")
print(f" Type: {attack.attack_type.value.replace('_', ' ').title()}")
print(f" Protocol: {attack.protocol_used.value.upper()}")
print(f" Chain: {attack.chain}")
print(f" Block: #{attack.block_number}")
print(f" Tx Hash: {attack.tx_hash[:66]}")
print(f" Attacker: {attack.attacker_address}")
print(f" Profit: ${attack.attacker_profit_usd:,.2f}")
print(f" Victim Loss: ${attack.victim_loss_usd:,.2f}")
print(f" Confidence: {attack.confidence:.0%}")
print(f" Sophistication: {attack.sophistication_score:.1f}/10")
if attack.tags:
print(f" Tags: {', '.join(attack.tags)}")
if attack.targets:
print(f" Targets: {', '.join(a[:16] for a in attack.targets[:3])}")
if attack.flash_loan_calls:
print(f" Flash Loans: {len(attack.flash_loan_calls)}")
for fl in attack.flash_loan_calls:
print(f" • {fl.summary()}")
if attack.exploit_timeline:
print(" Timeline:")
for event in attack.exploit_timeline[:5]:
print(f" → {event}")
print(f"\n{'═' * 60}")
print(f"Total: {len(filtered)} attack(s) detected")
total_impact = sum(a.attacker_profit_usd + a.victim_loss_usd for a in filtered)
print(f"Total financial impact: ${total_impact:,.2f}")
print(f"{'═' * 60}")
if args.monitor:
print("\n🔄 Continuous monitoring mode. Next scan in 60s...")
while True:
await asyncio.sleep(60)
report = await detector.scan(blocks_back=args.blocks, chain=chain)
for attack in report.attacks_confirmed:
print(f"\n⚠️ NEW: {attack.summary()}")
def main() -> None:
"""CLI entry point."""
asyncio.run(_run_cli())
if __name__ == "__main__":
main()
Reference in a new issue View git blame Copy permalink