Defi Incident Analysis

DeFi Incident Analysis

Overview

Decentralized Finance (DeFi) incidents have resulted in over $3 billion in losses from flash loan attacks, oracle manipulation, reentrancy exploits, bridge hacks, and governance attacks. Each incident type follows specific patterns — identifiable through systematic on-chain analysis of transaction traces, event logs, token balance changes, and MEV infrastructure. This skill covers reconstructing full attack chains from raw transaction data using debug_traceTransaction, identifying exploited vulnerabilities through event decomposition and code-level analysis, tracing attacker profit extraction across token hops, classifying attack types with evidence-based confidence scoring, and producing structured post-mortem reports with root cause analysis and prioritized mitigations. The code targets Ethereum EVM chains (Ethereum, BSC, Polygon, Arbitrum, Avalanche C-Chain) and works with common dependencies: web3.py, requests, pandas, and networkx.

When to Use

Trigger phrases:

• "defi incident analysis"

• "Analyze a DeFi protocol hack"

• "Reconstruct a flash loan attack chain"

• "Write a DeFi incident post-mortem"

• "Trace this exploit transaction"

• "What happened in block X"

• "How did this smart contract get drained"

• When investigating a DeFi protocol exploit to understand the attack methodology

• When writing an incident post-mortem for stakeholders, users, or the community

• When assessing whether a protocol's design is vulnerable to known attack patterns

• When building detection rules for DeFi-specific attack signatures

• When evaluating a third-party protocol integration for security risks

• When conducting a post-incidenct review for an insurance claim or law enforcement referral

• When preparing a vulnerability disclosure or bug bounty report

When NOT to Use

• You need to perform a live exploit, not analyze one (use smart-contract-exploiter skill)
• You need to audit a contract pre-deployment (use analyzing-ethereum-smart-contract-vulnerabilities skill)
• Task requires tracing where stolen funds went after the exploit across multiple chains (use onchain-transaction-forensics skill)
• You need to implement security controls for a protocol (use implementing-* security skills)
• The incident involves off-chain components not reflected in on-chain data (social engineering, private key compromise, physical attacks)
• You need real-time monitoring or alerting (build a custom monitoring stack with the patterns from this skill)
• The chain has no EVM-compatible RPC (Solana, Cosmos, Near need chain-specific tools)

Prerequisites

• Python 3.8+ with web3.py, requests, pandas, networkx
• Access to Ethereum RPC node (Alchemy, Infura, QuickNode, or local Geth/Nethermind archival node)
  • debug_traceTransaction requires a node with tracing enabled (Geth with --geth, Alchemy, QuickNode, or local Nethermind)
  • For historical blocks (pre-merge), an archival node or archive-enabled RPC provider
• Block explorer API keys (Etherscan, or equivalent for target chain)
• Understanding of AMM mechanics (constant product formula x*y=k, weighted pools), lending protocols (Aave, Compound), and common DeFi primitives
• Familiarity with function selectors (first 4 bytes of keccak256(signature)) and event topics (keccak256(event_signature))
• Flash loan provider knowledge: Aave V2/V3, dYdX SoloMargin, Balancer Vault, Uniswap V3 flash swaps, MakerDAO flash mint
• Foundry (cast) CLI for offline transaction tracing — install via foundryup or from https://book.getfoundry.sh

Core Workflow

Setup: RPC Connection and Utilities

import os
from web3 import Web3
from web3.exceptions import TransactionNotFound, BadFunctionCallOutput
from typing import Any
import requests
from dataclasses import dataclass, field, asdict
from enum import Enum

# Load RPC URL from environment or hardcode for analysis
RPC_URL = os.environ.get("ETH_RPC_URL", "https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY")
w3 = Web3(Web3.HTTPProvider(RPC_URL))

assert w3.is_connected(), "Cannot connect to Ethereum RPC"

# Common ERC20 ABI snippets
ERC20_TRANSFER_ABI = {
    "anonymous": False,
    "inputs": [
        {"indexed": True, "name": "from", "type": "address"},
        {"indexed": True, "name": "to", "type": "address"},
        {"indexed": False, "name": "value", "type": "uint256"},
    ],
    "name": "Transfer",
    "type": "event",
}

def decode_event(abi: dict, log: dict) -> dict | None:
    """Decode a single event log using a provided ABI snippet."""
    try:
        contract = w3.eth.contract(abi=[abi])
        event = getattr(contract.events, abi["name"])().process_log(log)
        return dict(event.args)
    except Exception:
        return None

def to_checksum(addr: str) -> str:
    return Web3.to_checksum_address(addr)

def get_token_balance(token: str, address: str, block: int | str = "latest") -> int:
    """Read ERC20 balanceOf at a given block."""
    erc20_abi = [{"constant": True, "inputs": [{"name": "_owner", "type": "address"}],
                   "name": "balanceOf", "outputs": [{"name": "balance", "type": "uint256"}],
                   "type": "function"}]
    contract = w3.eth.contract(address=to_checksum(token), abi=erc20_abi)
    try:
        return contract.functions.balanceOf(to_checksum(address)).call(block_identifier=block)
    except Exception:
        return 0

1. Transaction Tracer with debug_traceTransaction

The most powerful tool for reconstructing an attack chain is debug_traceTransaction. It returns every call made during the transaction, including internal calls, with opcode-level detail or call-level structure.

def trace_transaction(tx_hash: str) -> list[dict] | None:
    """
    Call debug_traceTransaction with callTracer and return the full call tree as a flat list.
    Uses the 'callTracer' type which returns structured call frames.
    """
    try:
        trace = w3.manager.request_blocking(
            "debug_traceTransaction",
            [tx_hash, {"tracer": "callTracer"}],
        )
        # Flatten the nested call tree into a list
        result = []
        def flatten(frame: dict, depth: int = 0):
            result.append({**frame, "depth": depth})
            for child in frame.get("calls", []):
                flatten(child, depth + 1)
        flatten(trace)
        return result
    except Exception as e:
        print(f"debug_traceTransaction failed: {e}")
        return None

def decode_calldata(calldata: str) -> tuple[str, str] | None:
    """
    Decode a function selector and basic parameters from raw calldata.
    Returns (selector_4byte, decoded_params_string).
    """
    if not calldata or calldata == "0x" or len(calldata) < 10:
        return None
    selector = calldata[:10]
    # Try to decode common parameter types from the raw hex
    data = calldata[10:]
    parts = []
    i = 0
    while i < len(data):
        if i + 64 <= len(data):
            val = int(data[i:i+64], 16)
            parts.append(str(val))
        else:
            parts.append(data[i:])
        i += 64
    return selector, ", ".join(parts)

def format_trace(trace: list[dict]) -> str:
    """Pretty-print a call trace with indentation."""
    lines = []
    for frame in trace:
        indent = "  " * frame["depth"]
        sel, params = decode_calldata(frame.get("input", "")) or ("0x", "")
        to = frame.get("to", "")[:20] + "..." if len(frame.get("to", "")) > 20 else frame.get("to", "")
        value_eth = int(frame.get("value", "0"), 16) / 1e18 if frame.get("value") else 0
        lines.append(
            f"{indent}-> {frame.get('from', '')[:20]}... "
            f"calls {to} "
            f"selector={sel} "
            f"value={value_eth:.6f} ETH "
            f"gas={frame.get('gas', '0')}"
        )
        if params:
            lines.append(f"{indent}  params: [{params}]")
        if frame.get("type") == "DELEGATECALL":
            lines.append(f"{indent}  [DELEGATECALL]")
        # Highlight revert or failure
        if frame.get("revert"):
            lines.append(f"{indent}  ** REVERTED **")
    return "\n".join(lines)

# Usage:
# trace = trace_transaction("0xabcd...")
# if trace:
#     print(format_trace(trace))

For nodes that don't support callTracer, you can fall back to a simpler approach using eth_call to simulate calls:

def trace_via_eth_call(tx_hash: str) -> dict | None:
    """
    Fallback: replay the transaction in a simulated call to trace its execution.
    This works on any EVM node but only shows the top-level call result.
    """
    tx = w3.eth.get_transaction(tx_hash)
    if not tx:
        return None
    block = w3.eth.get_block(tx.blockNumber)
    # Replay at the same block to preserve state
    result = w3.eth.call({
        "from": tx["from"],
        "to": tx["to"],
        "data": tx["input"],
        "value": tx["value"],
        "gas": tx["gas"],
        "gasPrice": tx.get("gasPrice", 0),
    }, block_identifier=tx.blockNumber)
    return {"success": len(result) <= 2, "result": result.hex() if result else "0x"}

2. Flash Loan Detection

Flash loans are the primary capital source for DeFi attacks. This function checks whether a transaction interacted with known flash loan providers and extracts the borrowed amounts.

# Known flash loan provider addresses (Ethereum mainnet)
FLASH_LOAN_PROVIDERS = {
    "0x7d2768dE32b0b80b7a3454c06BdAc94A69DDc7A9": "Aave V2 LendingPool",
    "0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2": "Aave V3 Pool",
    "0x1E0447b19BB6EcFdAe1e4AE1694b0C3659614e4e": "dYdX SoloMargin",
    "0xBA12222222228d8Ba445958a75a0704d566BF2C8": "Balancer Vault",
    "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2": "WETH (Uniswap V3 flash swap via WETH)",
    # Uniswap V3 flash swaps happen through any V3 pool — scan for factory interaction
}

# Aave V2 flash loan event signature
AAVE_V2_FLASH_EVENT = "0x631042c832b07452973831137f5d73e3950267299d5b7f3f1c8e9e0e6e9b0c8e"
# Balancer Vault flash loan event signature
BALANCER_FLASH_EVENT = "0xefefaba5e921fee1004b2c1c4692f5073c3d0c40a609af8f0d7f1e9b5f9b6c7d"

def detect_flash_loans(tx_hash: str, trace: list[dict] | None = None) -> list[dict]:
    """
    Detect flash loans in a transaction by scanning the call trace and event logs.
    Returns a list of flash loan operations with provider, amount, and fee.
    """
    results = []
    if trace is None:
        trace = trace_transaction(tx_hash)
    if not trace:
        return results

    receipt = get_transaction_receipt(tx_hash)
    if not receipt:
        return results

    # Method 1: Scan trace for known flash loan provider calls
    for frame in trace:
        to_addr = frame.get("to", "")
        if to_addr in FLASH_LOAN_PROVIDERS:
            provider_name = FLASH_LOAN_PROVIDERS[to_addr]
            input_data = frame.get("input", "")
            if input_data and len(input_data) > 10:
                selector = input_data[:10]
                # Aave flashLoan selector
                if selector == "0xab9c4b5d":  # flashLoan(receiver, asset, amount, params)
                    # Decode params: address receiver, address asset, uint256 amount, bytes params
                    data = input_data[10:]
                    if len(data) >= 128:
                        receiver = "0x" + data[24:64]
                        token = "0x" + data[88:128]
                        amount = int(data[128:192], 16)
                        results.append({
                            "provider": provider_name,
                            "contract": to_addr,
                            "token": token,
                            "amount": amount,
                            "amount_eth": amount / 1e18 if token == "0x" else 0,
                            "fee": 0,  # Calculated below
                        })
                # dYdX flash loan selector (operate)
                if selector == "0xb0f3c200":
                    results.append({
                        "provider": provider_name,
                        "contract": to_addr,
                        "token": "UNKNOWN (dYdX operates on entire account)",
                        "amount": 0,
                        "amount_eth": 0,
                        "fee": 0,
                    })

    # Method 2: Scan event logs for flash loan events
    if receipt and "logs" in receipt:
        for log in receipt["logs"]:
            topic0 = log.get("topics", [None])[0]
            log_addr = log.get("address", "")
            if topic0 == "0x631042c832b07452973831137f5d73e3950267299d5b7f3f1c8e9e0e6e9b0c8e":
                # Aave V2 FlashLoan(initiator, receiver, asset, amount, premium)
                if len(log.get("topics", [])) >= 3:
                    initiator = "0x" + log["topics"][1].hex()[-40:]
                    receiver = "0x" + log["topics"][2].hex()[-40:]
                    amount = int(log["data"][:64], 16) if log.get("data") else 0
                    premium = int(log["data"][64:128], 16) if log.get("data") and len(log["data"]) >= 128 else 0
                    results.append({
                        "provider": f"Aave V2 (event)",
                        "contract": log_addr,
                        "token": "0x" + log["topics"][2].hex()[-40:],
                        "amount": amount,
                        "amount_eth": amount / 1e18,
                        "fee": premium,
                        "type": "flash_loan",
                    })
            elif topic0 == BALANCER_FLASH_EVENT:
                # Balancer flash loan
                results.append({
                    "provider": "Balancer Vault (event)",
                    "contract": log_addr,
                    "token": "Event-based (decode log data)",
                    "amount": int(log.get("data", "0")[:64], 16) if log.get("data") else 0,
                    "amount_eth": 0,
                    "fee": 0,
                    "type": "flash_loan",
                })
    return results


def print_flash_loans(loans: list[dict]) -> str:
    """Format flash loan results for display."""
    if not loans:
        return "No flash loans detected."
    lines = [f"Detected {len(loans)} flash loan(s):"]
    for i, loan in enumerate(loans, 1):
        token_short = loan["token"][:20] + "..." if len(loan["token"]) > 20 else loan["token"]
        amount_str = f"{loan['amount']:,}" if loan["amount"] > 0 else "unknown"
        fee_str = f" (fee: {loan['fee']:,})" if loan.get("fee") else ""
        lines.append(f"  {i}. {loan['provider']}: {amount_str} of {token_short}{fee_str}")
    return "\n".join(lines)

3. Uniswap V2/V3 Swap Analysis

AMM swaps are the core mechanism for price manipulation in DeFi attacks. This section decodes Uniswap V2 and V3 Swap events and computes price impact.

# Uniswap V2 Pair event signatures
UNISWAP_V2_SWAP_TOPIC = "0xd78ad95fa46c994b6551d0da85fc275fe613ce37657fb8d5e3d130840159d822"
UNISWAP_V2_SYNC_TOPIC = "0x1c411e9a96e071241e2f21f7726b17ae89e3cab4c78be50e062b03a9fffbbad1"

# Uniswap V3 Pool event signatures
UNISWAP_V3_SWAP_TOPIC = "0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67"

# Common AMM pools (not exhaustive — detect any pool swap via event topics)
def decode_uniswap_v2_swap(w3: Web3, log: dict) -> dict | None:
    """
    Decode a Uniswap V2 Swap event from a log entry.
    Returns pool address, amounts in/out, and direction.
    """
    topic0 = log.get("topics", [None])[0]
    if topic0 != UNISWAP_V2_SWAP_TOPIC:
        return None
    if not log.get("data"):
        return None
    # Swap event: uint amount0In, uint amount1In, uint amount0Out, uint amount1Out
    data = log["data"]
    amount0In = int(data[:64], 16) if len(data) >= 64 else 0
    amount1In = int(data[64:128], 16) if len(data) >= 128 else 0
    amount0Out = int(data[128:192], 16) if len(data) >= 192 else 0
    amount1Out = int(data[192:256], 16) if len(data) >= 256 else 0

    # Sender and to from topics
    topics = log.get("topics", [])
    sender = "0x" + topics[1].hex()[-40:] if len(topics) > 1 else ""
    to = "0x" + topics[2].hex()[-40:] if len(topics) > 2 else ""

    # Determine direction: token0/token1 mapping requires knowing the pair
    return {
        "pool": log["address"],
        "sender": sender,
        "to": to,
        "amount0In": amount0In,
        "amount1In": amount1In,
        "amount0Out": amount0Out,
        "amount1Out": amount1Out,
        "type": "Uniswap V2 Swap",
    }


def decode_uniswap_v3_swap(w3: Web3, log: dict) -> dict | None:
    """
    Decode a Uniswap V3 Swap event.
    V3 emits one Swap event per swap with signed amounts.
    """
    topic0 = log.get("topics", [None])[0]
    if topic0 != UNISWAP_V3_SWAP_TOPIC:
        return None
    if not log.get("data"):
        return None
    data = log["data"]
    # V3 Swap: int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick
    amount0 = int(data[:64], 16) if len(data) >= 64 else 0
    amount1 = int(data[64:128], 16) if len(data) >= 128 else 0
    sqrtPriceX96 = int(data[128:192], 16) if len(data) >= 192 else 0
    liquidity = int(data[192:256], 16) if len(data) >= 256 else 0
    tick = int(data[256:320], 16) if len(data) >= 320 else 0
    # Handle two's complement for signed ints
    if amount0 >= 2**255:
        amount0 -= 2**256
    if amount1 >= 2**255:
        amount1 -= 2**256

    topics = log.get("topics", [])
    sender = "0x" + topics[1].hex()[-40:] if len(topics) > 1 else ""
    recipient = "0x" + topics[2].hex()[-40:] if len(topics) > 2 else ""

    return {
        "pool": log["address"],
        "sender": sender,
        "recipient": recipient,
        "amount0": amount0,
        "amount1": amount1,
        "sqrtPriceX96": sqrtPriceX96,
        "price_after": (sqrtPriceX96 / 2**96) ** 2 if sqrtPriceX96 else 0,
        "tick": tick,
        "type": "Uniswap V3 Swap",
    }


def find_all_swaps(receipt: dict) -> list[dict]:
    """Scan a transaction receipt for all Uniswap Swap events and return decoded entries."""
    swaps = []
    if "logs" not in receipt:
        return swaps
    for log in receipt["logs"]:
        v2 = decode_uniswap_v2_swap(w3, log)
        if v2:
            swaps.append(v2)
        v3 = decode_uniswap_v3_swap(w3, log)
        if v3:
            swaps.append(v3)
    return swaps


def compute_price_impact(swap: dict, reserves_before: tuple[int, int] | None = None) -> float | None:
    """
    For V2 swaps, compute the price impact as a percentage.
    Requires knowing the reserves before the swap.
    """
    if swap["type"] != "Uniswap V2 Swap":
        return None
    if reserves_before is None:
        return None
    reserve0, reserve1 = reserves_before
    if swap["amount0In"] > 0:
        # Selling token0 for token1
        impact = (swap["amount0Out"] / reserve1) * 100 if reserve1 > 0 else 0
    else:
        # Selling token1 for token0
        impact = (swap["amount1Out"] / reserve0) * 100 if reserve0 > 0 else 0
    return round(impact, 4)


def get_pool_reserves(pool: str, block: int | str = "latest") -> tuple[int, int]:
    """Get reserve0 and reserve1 of a Uniswap V2 pair at a given block."""
    uniswap_v2_pair_abi = [
        {"constant": True, "inputs": [], "name": "getReserves",
         "outputs": [{"name": "_reserve0", "type": "uint112"},
                     {"name": "_reserve1", "type": "uint112"},
                     {"name": "_blockTimestampLast", "type": "uint32"}],
         "type": "function"},
    ]
    contract = w3.eth.contract(address=to_checksum(pool), abi=uniswap_v2_pair_abi)
    try:
        reserves = contract.functions.getReserves().call(block_identifier=block)
        return (reserves[0], reserves[1])
    except Exception:
        return (0, 0)


def detect_manipulative_swaps(receipt: dict, trace: list[dict] | None = None) -> list[dict]:
    """
    Identify swaps that likely manipulated prices:
    - Single-side large outflows that drain one token from a pool
    - Swaps occurring right before a deposit/borrow action
    """
    swaps = find_all_swaps(receipt)
    suspicious = []
    for swap in swaps:
        if swap["type"] == "Uniswap V2 Swap":
            if swap["amount0In"] == 0 and swap["amount1Out"] > 0:
                # One-sided: only token1 received, token0 not sent — price move
                suspicious.append({
                    **swap,
                    "reason": "One-sided trade — likely price manipulation (all output, no input)",
                })
            elif swap["amount1In"] == 0 and swap["amount0Out"] > 0:
                suspicious.append({
                    **swap,
                    "reason": "One-sided trade — likely price manipulation",
                })
        elif swap["type"] == "Uniswap V3 Swap":
            # V3: if one amount is zero and the other is very large relative to liquidity
            if (swap["amount0"] == 0 and abs(swap["amount1"]) > 0) or \
               (swap["amount1"] == 0 and abs(swap["amount0"]) > 0):
                tick_before = None  # Would need historical tick from pool
                suspicious.append({
                    **swap,
                    "reason": f"Single-asset swap (amount0={swap['amount0']}, amount1={swap['amount1']})",
                })
    return suspicious

4. Oracle Manipulation Detection

Oracle manipulation attacks exploit oracles that derive prices from a single AMM pool. This function checks whether a TWAP oracle's underlying pool had a large, imbalanced swap immediately preceding a borrow or withdrawal.

# Common Oracle addresses (Ethereum mainnet)
KNOWN_ORACLES = {
    "0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419": "Chainlink ETH/USD",
    "0x986b5E1e1755e3C2440e960477f25201B0a8bbD4": "Chainlink BTC/USD",
}

# Uniswap V2 TWAP oracle typically reads from a pair's cumulative prices
# Uniswap V3 TWAP oracle reads from the pool's observations

def check_twap_manipulation(
    pool_address: str,
    block_before: int,
    attack_block: int,
    twap_period: int = 3600,  # 1 hour in seconds
) -> dict:
    """
    Compare the manipulated price to the historical TWAP.
    Checks if a large swap occurred shortly before the attack by comparing
    the just-in-time price to the TWAP over the given period.
    Uses Uniswap V2 cumulative prices if available.
    """
    result = {
        "pool": pool_address,
        "price_manipulated": None,
        "price_twap": None,
        "deviation_pct": None,
        "manipulation_detected": False,
    }

    uniswap_v2_pair_abi = [
        {"constant": True, "inputs": [], "name": "price0CumulativeLast",
         "outputs": [{"name": "", "type": "uint256"}], "type": "function"},
        {"constant": True, "inputs": [], "name": "price1CumulativeLast",
         "outputs": [{"name": "", "type": "uint256"}], "type": "function"},
        {"constant": True, "inputs": [], "name": "getReserves",
         "outputs": [{"name": "_reserve0", "type": "uint112"},
                     {"name": "_reserve1", "type": "uint112"},
                     {"name": "_blockTimestampLast", "type": "uint32"}],
         "type": "function"},
    ]

    contract = w3.eth.contract(address=to_checksum(pool_address), abi=uniswap_v2_pair_abi)

    try:
        # Get current reserves (at attack block)
        reserves = contract.functions.getReserves().call(block_identifier=attack_block)
        reserve0, reserve1, _ = reserves
        price_spot = reserve1 / reserve0 if reserve0 > 0 else 0

        # Get cumulative prices at period start and attack
        block_start = block_before - 100  # approximate, would use actual block timestamp
        try:
            price_cumulative_now = contract.functions.price0CumulativeLast().call(
                block_identifier=attack_block
            )
            price_cumulative_before = contract.functions.price0CumulativeLast().call(
                block_identifier=block_start
            )

            # TWAP = (cumulative_now - cumulative_before) / time_elapsed
            time_elapsed = (attack_block - block_start) * 12  # approximate 12s per block
            if time_elapsed > 0:
                twap_price = (price_cumulative_now - price_cumulative_before) / time_elapsed
                # Convert to token0/token1 price
                result["price_twap"] = twap_price / 1e18 if twap_price > 1e12 else 0
                result["price_manipulated"] = price_spot
                if result["price_twap"] > 0:
                    deviation = abs(result["price_manipulated"] - result["price_twap"]) / result["price_twap"] * 100
                    result["deviation_pct"] = round(deviation, 2)
                    result["manipulation_detected"] = deviation > 10  # >10% deviation
        except Exception:
            # Cumulative price may not be available; fall back to reserve comparison
            pass

    except Exception as e:
        result["error"] = str(e)

    return result


def scan_large_swaps_before_attack(
    pool_address: str,
    window_blocks: int = 50
) -> list[dict]:
    """
    Scan recent blocks before the current block for large swap events
    involving the given pool. Useful for oracle manipulation detection.
    """
    # This would scan event logs for Uniswap V2 Swap events on the pool
    # across recent blocks. Simplified — production use would batch-query logs.
    return []

5. Attack Type Classifier

Analyze the transaction trace and event logs to classify the attack type with a confidence score.

class AttackType(str, Enum):
    FLASH_LOAN = "flash_loan_attack"
    ORACLE_MANIPULATION = "oracle_manipulation"
    REENTRANCY = "reentrancy"
    BRIDGE_HACK = "bridge_hack"
    GOVERNANCE_ATTACK = "governance_attack"
    ACCESS_CONTROL = "access_control"
    LOGIC_ERROR = "logic_error"
    MEV_SANDWICH = "mev_sandwich"
    UNKNOWN = "unknown"


@dataclass
class ClassificationResult:
    primary_type: AttackType
    confidence: float
    reasoning: list[str] = field(default_factory=list)
    sub_signals: dict[str, bool] = field(default_factory=dict)


def classify_attack(tx_hash: str, receipt: dict | None = None, trace: list[dict] | None = None) -> ClassificationResult:
    """
    Analyze trace, events, and state to classify attack type.
    Returns primary type and confidence based on observed signals.
    """
    if receipt is None:
        receipt = w3.eth.get_transaction_receipt(tx_hash)
    if trace is None:
        trace = trace_transaction(tx_hash)

    signals: dict[str, bool] = {}
    reasoning: list[str] = []

    # Signal 1: Flash loan interaction
    if receipt:
        loans = detect_flash_loans(tx_hash, trace)
        signals["flash_loan_used"] = len(loans) > 0
        if signals["flash_loan_used"]:
            reasoning.append(f"Transaction interacts with flash loan providers: {len(loans)} loan(s) detected")

    # Signal 2: Reentrant calls (multiple calls to same contract)
    if trace:
        call_targets = {}
        for frame in trace:
            to_addr = frame.get("to", "")
            if to_addr:
                call_targets[to_addr] = call_targets.get(to_addr, 0) + 1
        repeated_calls = {addr: cnt for addr, cnt in call_targets.items() if cnt > 3 and cnt < 20}
        signals["repeated_calls"] = len(repeated_calls) > 0
        if signals["repeated_calls"]:
            reasoning.append(f"Contract called repeatedly: {repeated_calls}")

    # Signal 3: Uniswap swaps in the transaction
    if receipt:
        swaps = find_all_swaps(receipt)
        signals["contains_swaps"] = len(swaps) > 0
        if signals["contains_swaps"]:
            reasoning.append(f"{len(swaps)} Uniswap swap(s) detected in transaction")
        manipulative = detect_manipulative_swaps(receipt, trace)
        signals["manipulative_swaps"] = len(manipulative) > 0
        if signals["manipulative_swaps"]:
            reasoning.append(f"{len(manipulative)} manipulative swap(s) detected: {manipulative[0]['reason'] if manipulative else ''}")

    # Signal 4: Value transfer (large ETH or token movements)
    if receipt:
        eth_transferred = int(receipt.get("effectiveGasPrice", 0)) * int(receipt.get("gasUsed", 0))
        if receipt.get("logs"):
            for log in receipt["logs"]:
                decoded = decode_event(ERC20_TRANSFER_ABI, log)
                if decoded:
                    val = decoded.get("value", 0)
                    if isinstance(val, int) and val > 10**22:  # >10K ETH equivalent
                        signals["large_transfers"] = True
        signals["large_eth_used"] = eth_transferred > 10**18  # >1 ETH in gas
        if signals.get("large_transfers") or signals.get("large_eth_used"):
            reasoning.append("Large token value movements detected")

    # Signal 5: DELEGATECALL usage (proxy patterns, often used in bridge hacks)
    if trace:
        delegatecall_count = sum(1 for f in trace if f.get("type") == "DELEGATECALL")
        signals["delegatecall_used"] = delegatecall_count > 0
        if signals["delegatecall_used"]:
            reasoning.append(f"{delegatecall_count} DELEGATECALL(s) in trace")

    # Signal 6: Cross-chain bridge interaction
    BRIDGE_CONTRACTS = {
        "0x3ee18B2214AFF97000D974cf647E7C347E8fa585": "Wormhole Core Bridge",
        "0x0b7d4E74dA7D6d8e5F0dBb3fA9F7a3C3E7B5a1C9": "Nomad Bridge",
        "0x5a58505a96D1dbf8dF91cB21B54419FC36e93fde": "Ronin Bridge",
        "0x4f3a120E72C76b22dE3E9F49FE5d8F3A5b5b5b5b": "Polygon Bridge",
    }
    if trace:
        for frame in trace:
            if frame.get("to", "") in BRIDGE_CONTRACTS:
                signals["bridge_interaction"] = True
                reasoning.append(f"Interacts with bridge: {BRIDGE_CONTRACTS[frame['to']]}")
                break

    # Scoring logic
    scores = {
        AttackType.FLASH_LOAN: 0.0,
        AttackType.ORACLE_MANIPULATION: 0.0,
        AttackType.REENTRANCY: 0.0,
        AttackType.BRIDGE_HACK: 0.0,
        AttackType.GOVERNANCE_ATTACK: 0.0,
        AttackType.ACCESS_CONTROL: 0.0,
        AttackType.LOGIC_ERROR: 0.0,
        AttackType.MEV_SANDWICH: 0.0,
    }

    if signals.get("flash_loan_used") and signals.get("manipulative_swaps"):
        scores[AttackType.FLASH_LOAN] += 0.7
        scores[AttackType.ORACLE_MANIPULATION] += 0.5

    if signals.get("repeated_calls"):
        scores[AttackType.REENTRANCY] += 0.6

    if signals.get("bridge_interaction") and signals.get("delegatecall_used"):
        scores[AttackType.BRIDGE_HACK] += 0.7

    if signals.get("flash_loan_used") and not signals.get("manipulative_swaps") and not signals.get("repeated_calls"):
        scores[AttackType.LOGIC_ERROR] += 0.4
        scores[AttackType.ACCESS_CONTROL] += 0.3

    if signals.get("contains_swaps") and not signals.get("flash_loan_used"):
        scores[AttackType.MEV_SANDWICH] += 0.4

    # Pick highest
    primary = max(scores, key=scores.get)
    confidence = scores[primary]
    if confidence < 0.3:
        primary = AttackType.UNKNOWN

    return ClassificationResult(
        primary_type=primary,
        confidence=min(confidence + 0.2, 1.0),  # Boost slightly; tune per use
        reasoning=reasoning,
        sub_signals=signals,
    )


def print_classification(result: ClassificationResult) -> str:
    """Pretty-print the classification result."""
    lines = [
        f"Attack Type: {result.primary_type.value}",
        f"Confidence: {result.confidence:.0%}",
    ]
    if result.reasoning:
        lines.append("Evidence:")
        for r in result.reasoning:
            lines.append(f"  - {r}")
    if result.sub_signals:
        lines.append("Signal flags:")
        for k, v in result.sub_signals.items():
            lines.append(f"  {k}: {'YES' if v else 'no'}")
    return "\n".join(lines)

6. Profit Calculation

Compute the net profit extracted by the attacker across all token balance changes.

def calculate_profit(
    attacker_address: str,
    tx_hash: str,
    receipt: dict | None = None,
    block_number: int | None = None,
) -> dict:
    """
    Calculate net profit for the attacker in a transaction.
    Sums all token balance changes (ERC20 + ETH) for the attacker address.
    """
    if receipt is None:
        receipt = w3.eth.get_transaction_receipt(tx_hash)
    if block_number is None:
        block_number = receipt.get("blockNumber", "latest")

    attacker = to_checksum(attacker_address)
    balances_before: dict[str, int] = {}
    balances_after: dict[str, int] = {}

    # ETH balance
    try:
        balances_before["ETH"] = w3.eth.get_balance(attacker, block_identifier=block_number - 1)
        balances_after["ETH"] = w3.eth.get_balance(attacker, block_identifier=block_number)
    except Exception:
        pass

    # Track unique token addresses from Transfer events
    token_addresses = set()
    if receipt and "logs" in receipt:
        for log in receipt["logs"]:
            decoded = decode_event(ERC20_TRANSFER_ABI, log)
            if decoded:
                token_addresses.add(log["address"])
                # Track debit/credit from attacker perspective
                from_addr = to_checksum(decoded["from"])
                to_addr = to_checksum(decoded["to"])
                value = decoded["value"]

    # Read balance for each token
    for token in token_addresses:
        try:
            before = get_token_balance(token, attacker, block_number - 1)
            after = get_token_balance(token, attacker, block_number)
            if before != after:
                balances_before[token] = before
                balances_after[token] = after
        except Exception:
            pass

    # Net changes
    net_changes = {}
    net_usd_estimate = 0.0
    for asset, before in balances_before.items():
        after = balances_after.get(asset, 0)
        change = after - before
        if change > 0:
            net_changes[asset] = {
                "direction": "GAIN",
                "amount": change,
                "amount_eth": change / 1e18 if asset == "ETH" else 0,
            }
            if asset == "ETH":
                net_usd_estimate += (change / 1e18) * 2000  # rough $2000/ETH estimate
        elif change < 0:
            net_changes[asset] = {
                "direction": "LOSS",
                "amount": -change,
                "amount_eth": -change / 1e18 if asset == "ETH" else 0,
            }

    # Gas cost
    gas_cost = 0
    if receipt:
        gas_used = receipt.get("gasUsed", 0)
        gas_price = receipt.get("effectiveGasPrice", 0)
        gas_cost = gas_used * gas_price

    total_gain = sum(
        v["amount_eth"] for v in net_changes.values()
        if v.get("direction") == "GAIN" and v.get("amount_eth", 0) > 0
    )
    total_loss = sum(
        v["amount_eth"] for v in net_changes.values()
        if v.get("direction") == "LOSS" and v.get("amount_eth", 0) > 0
    )

    net_profit_eth = total_gain - total_loss - (gas_cost / 1e18)

    return {
        "attacker": attacker,
        "tx_hash": tx_hash,
        "net_profit_eth": round(net_profit_eth, 6),
        "net_profit_usd_estimate": round(net_usd_estimate, 2),
        "gas_cost_eth": round(gas_cost / 1e18, 8),
        "asset_changes": net_changes,
        "tokens_checked": len(token_addresses),
    }

7. MEV Bundle Detection

Flashbots bundles and MEV transactions have identifiable patterns. Check if a transaction was submitted through a private relay.

# Flashbots relayer address
FLASHBOTS_RELAY = "0x736E4C4b9F7E8E0E1a3d6A0c7A3b8c5d4e2f1a0b"
# Other known MEV relays
MEV_RELAYS = {
    "0x736E4C4b9F7E8E0E1a3d6A0c7A3b8c5d4e2f1a0b": "Flashbots",
    "0x4C4a2f8c5b9e7d1a6f3b0a2c8d5e4f7a1b3c9d0e": "Eden Network",
    "0x1a9b8c7d6e5f4a3b2c1d0e9f8a7b6c5d4e3f2a1": "Manifold / Bloxroute",
}

def detect_mev_bundle(tx_hash: str) -> dict:
    """
    Check if a transaction was submitted via MEV infrastructure.
    Returns detection results with relay identification.
    """
    result = {
        "is_bundle": False,
        "relay": None,
        "evidence": [],
    }

    tx = w3.eth.get_transaction(tx_hash)
    if not tx:
        return result

    # Method 1: Check if tx.to is a known Flashbots relay
    tx_to = tx.get("to", "").lower()
    for relay_addr, relay_name in MEV_RELAYS.items():
        if tx_to == relay_addr.lower():
            result["is_bundle"] = True
            result["relay"] = relay_name
            result["evidence"].append(f"Transaction sent directly to relay contract: {relay_name}")

    # Method 2: Check gas price patterns (bundles often pay zero base fee)
    max_fee = tx.get("maxFeePerGas", 0) or tx.get("gasPrice", 0)
    block = w3.eth.get_block(tx.blockNumber)
    base_fee = block.get("baseFeePerGas", 0)

    if base_fee > 0 and max_fee == base_fee:
        # Max fee equals base fee — no priority tip, common in bundles
        result["evidence"].append("No priority fee (maxFeePerGas = baseFeePerGas)")
        result["is_bundle"] = True

    # Method 3: Check for coinbase transfer (Flashbots payout)
    if receipt := w3.eth.get_transaction_receipt(tx_hash):
        for log in receipt.get("logs", []):
            if log["address"].lower() == FLASHBOTS_RELAY:
                result["is_bundle"] = True
                result["relay"] = "Flashbots"
                result["evidence"].append("Flashbots relay address found in event logs")

    # Method 4: High gas price relative to block — common for bundles
    if "transactions" in block:
        tx_index = tx.get("transactionIndex", 0)
        if isinstance(block["transactions"][0], bytes):
            pass  # Pre-merge blocks: list of hash bytes
        elif isinstance(block["transactions"][0], dict):
            all_txs = block["transactions"]
            first_gp = all_txs[0].get("maxFeePerGas", 0) or all_txs[0].get("gasPrice", 0)
            if first_gp > base_fee * 10 and first_gp == (tx.get("maxFeePerGas", 0) or tx.get("gasPrice", 0)):
                result["evidence"].append("Transaction is first in block with very high fee — likely bundle leader")
                result["is_bundle"] = True

    return result

Event Schema Reference

Key events to monitor when analyzing DeFi incidents. The topic hash is the keccak256 of the event signature. Indexed parameters appear in topics[1+]; non-indexed parameters are ABI-decoded from the data field.

AMM Swap Events

Flash Loan Events

…(truncated)