Web3智能合约与DeFi安全取证深度分析

import requests
from datetime import datetime

ETHERSCAN_API_KEY = "YOUR_API_KEY"
CONTRACT_ADDRESS = "0x_target_contract_address"

def find_anomalous_outflows(contract_address, api_key, min_value_eth=100):
    url = f"https://api.etherscan.io/api"
    params = {
        "module": "account",
        "action": "txlist",
        "address": contract_address,
        "startblock": 0,
        "endblock": 99999999,
        "page": 1,
        "offset": 100,
        "sort": "desc",
        "apikey": api_key
    }
    response = requests.get(url, params=params)
    txs = response.json().get("result", [])
    anomalous = []
    for tx in txs:
        value_eth = int(tx["value"]) / 1e18
        if value_eth >= min_value_eth and tx["to"] != contract_address:
            anomalous.append({
                "hash": tx["hash"],
                "from": tx["from"],
                "to": tx["to"],
                "value": value_eth,
                "block": int(tx["blockNumber"]),
                "timestamp": datetime.fromtimestamp(int(tx["timeStamp"])).isoformat(),
                "gasUsed": int(tx["gasUsed"])
            })
    return anomalous

results = find_anomalous_outflows(CONTRACT_ADDRESS, ETHERSCAN_API_KEY)
for r in results:
    print(f"[ALERT] TX: {r['hash'][:16]}... | Value: {r['value']:.2f} ETH | Block: {r['block']}")

from web3 import Web3

w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"))

def analyze_call_trace(tx_hash):
    tx = w3.eth.get_transaction(tx_hash)
    receipt = w3.eth.get_transaction_receipt(tx_hash)
    
    trace_url = f"https://api.etherscan.io/api?module=proxy&action=debug_traceTransaction&txhash={tx_hash}&apikey={ETHERSCAN_API_KEY}"
    trace_resp = requests.get(trace_url).json()
    
    call_count = {}
    for log_entry in receipt.get("logs", []):
        topic = log_entry["topics"][0].hex() if log_entry["topics"] else ""
        if topic not in call_count:
            call_count[topic] = 0
        call_count[topic] += 1
    
    repeated_topics = {k: v for k, v in call_count.items() if v > 2}
    if repeated_topics:
        print(f"[REENTRANCY INDICATOR] Repeated event topics: {repeated_topics}")
    
    gas_used = receipt["gasUsed"]
    if gas_used > 5000000:
        print(f"[HIGH GAS] TX used {gas_used} gas units, possible complex attack")
    
    return {"topics": call_count, "gasUsed": gas_used}

def trace_fund_flow(start_address, depth=5):
    visited = set()
    flow = []
    
    def _trace(addr, current_depth):
        if current_depth > depth or addr in visited:
            return
        visited.add(addr)
        url = f"https://api.etherscan.io/api?module=account&action=txlist&address={addr}&sort=desc&offset=20&apikey={ETHERSCAN_API_KEY}"
        resp = requests.get(url).json()
        for tx in resp.get("result", []):
            value = int(tx["value"]) / 1e18
            if value > 10 and tx["from"].lower() == addr.lower():
                flow.append({"from": addr, "to": tx["to"], "value": value, "hash": tx["hash"]})
                _trace(tx["to"], current_depth + 1)
    
    _trace(start_address, 0)
    return flow

from web3 import Web3
import json

w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"))

CHAINLINK_ETH_USD = "0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419"

def detect_price_anomaly(aggregator_address, threshold_pct=5.0):
    abi = json.loads('[{"anonymous":false,"inputs":[{"indexed":false,"name":"current","type":"int256"},{"indexed":false,"name":"roundId","type":"uint80"}],"name":"AnswerUpdated","type":"event"}]')
    
    contract = w3.eth.contract(address=Web3.to_checksum_address(aggregator_address), abi=abi)
    latest_round = contract.functions.latestRound().call()
    latest_price = contract.functions.getRoundData(latest_round).call()[1]
    
    previous_round = latest_round - 10 if latest_round > 10 else 0
    previous_price = contract.functions.getRoundData(previous_round).call()[1]
    
    if previous_price > 0:
        change_pct = abs(latest_price - previous_price) / previous_price * 100
        if change_pct > threshold_pct:
            print(f"[ANOMALY] Price deviation: {change_pct:.2f}% | Current: {latest_price/1e8:.2f} | Previous: {previous_price/1e8:.2f}")
            return True
    return False

detect_price_anomaly(CHAINLINK_ETH_USD, threshold_pct=5.0)

def trace_oracle_manipulation(pool_address, block_start, block_end):
    url = f"https://api.etherscan.io/api?module=account&action=tokentx&address={pool_address}&startblock={block_start}&endblock={block_end}&sort=asc&apikey={ETHERSCAN_API_KEY}"
    resp = requests.get(url).json()
    
    swap_events = []
    for tx in resp.get("result", []):
        value = int(tx["value"]) / (10 ** int(tx["tokenDecimal"]))
        swap_events.append({
            "hash": tx["hash"],
            "token": tx["tokenName"],
            "value": value,
            "from": tx["from"],
            "to": tx["to"],
            "block": int(tx["blockNumber"])
        })
    
    large_swaps = [e for e in swap_events if e["value"] > 1000000]
    if large_swaps:
        print(f"[MANIPULATION] Found {len(large_swaps)} large swaps potentially manipulating price")
        for s in large_swaps:
            print(f"  Block {s['block']}: {s['value']:.2f} {s['token']} via TX {s['hash'][:16]}...")
    
    return large_swaps

def detect_sandwich_attacks(block_number):
    url = f"https://api.etherscan.io/api?module=proxy&action=eth_getBlockByNumber&tag={hex(block_number)}&boolean=true&apikey={ETHERSCAN_API_KEY}"
    resp = requests.get(url).json()
    block = resp["result"]
    
    txs = block["transactions"]
    tx_hashes = [tx["hash"] for tx in txs]
    
    suspected = []
    for i in range(len(txs) - 2):
        tx_from_0 = txs[i]["from"].lower()
        tx_from_1 = txs[i+1]["from"].lower()
        tx_from_2 = txs[i+2]["from"].lower()
        
        if tx_from_0 == tx_from_2 and tx_from_0 != tx_from_1:
            if txs[i].get("to") and txs[i+2].get("to"):
                suspected.append({
                    "front_runner": tx_from_0,
                    "victim_tx": txs[i+1]["hash"],
                    "back_runner": tx_from_2,
                    "block": block_number,
                    "front_tx": txs[i]["hash"],
                    "back_tx": txs[i+2]["hash"]
                })
    
    return suspected

suspected = detect_sandwich_attacks(19000000)
for s in suspected:
    print(f"[SANDWICH] Front: {s['front_tx'][:16]}... | Victim: {s['victim_tx'][:16]}... | Back: {s['back_tx'][:16]}...")

TIMELOCK_ABI = json.loads('[{"anonymous":false,"inputs":[{"indexed":true,"name":"target","type":"address"},{"indexed":false,"name":"value","type":"uint256"},{"indexed":true,"name":"signature","type":"bytes32"},{"indexed":true,"name":"dataHash","type":"bytes32"}],"name":"ProposalScheduled","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"proposalId","type":"bytes32"}],"name":"ProposalExecuted","type": "event"}]')

def analyze_timelock_operations(timelock_address, start_block, end_block):
    contract = w3.eth.contract(address=Web3.to_checksum_address(timelock_address), abi=TIMELOCK_ABI)
    
    scheduled = contract.events.ProposalScheduled.create_filter(
        fromBlock=start_block, toBlock=end_block
    ).get_all_entries()
    
    executed = contract.events.ProposalExecuted.create_filter(
        fromBlock=start_block, toBlock=end_block
    ).get_all_entries()
    
    schedule_blocks = {e["blockNumber"]: e for e in scheduled}
    execute_blocks = {e["blockNumber"]: e for e in executed}
    
    rapid_executions = []
    for s_block, s_event in schedule_blocks.items():
        for e_block, e_event in execute_blocks.items():
            delay = e_block - s_block
            if delay < 100 and s_event["args"]["proposalId"] == e_event["args"]["proposalId"]:
                rapid_executions.append({
                    "proposalId": s_event["args"]["proposalId"].hex(),
                    "scheduled_block": s_block,
                    "executed_block": e_block,
                    "delay_blocks": delay,
                    "target": s_event["args"]["target"]
                })
    
    return rapid_executions

UPGRADEABLE_ABI = json.loads('[{"anonymous":false,"inputs":[{"indexed":false,"name":"implementation","type":"address"}],"name":"Upgraded","type":"event"}]')

def detect_suspicious_upgrades(proxy_address, window_blocks=1000):
    contract = w3.eth.contract(address=Web3.to_checksum_address(proxy_address), abi=UPGRADEABLE_ABI)
    
    current_block = w3.eth.block_number
    events = contract.events.Upgraded.create_filter(
        fromBlock=current_block - window_blocks, toBlock=current_block
    ).get_all_entries()
    
    upgrades = []
    for event in events:
        block = event["blockNumber"]
        impl = event["args"]["implementation"]
        
        impl_code = w3.eth.get_code(Web3.to_checksum_address(impl))
        has_selfdestruct = b"\xff" in impl_code[-3:]
        
        tx = w3.eth.get_transaction(event["transactionHash"])
        deployer = tx["from"]
        
        upgrades.append({
            "block": block,
            "implementation": impl,
            "deployer": deployer,
            "has_selfdestruct": has_selfdestruct,
            "code_size": len(impl_code),
            "tx_hash": event["transactionHash"].hex()
        })
        
        if has_selfdestruct:
            print(f"[CRITICAL] Upgrade at block {block} to implementation with SELFDESTRUCT: {impl}")
    
    return upgrades

WORMHOLE_CORE = "0x98f3097C3664d46bb06f14B16933c2Eb21b9c089"

def verify_crosschain_message(tx_hash):
    receipt = w3.eth.get_transaction_receipt(tx_hash)
    
    wormhole_events = []
    for log in receipt["logs"]:
        if log["address"].lower() == WORMHOLE_CORE.lower():
            topic = log["topics"][0].hex()
            wormhole_events.append({
                "topic": topic,
                "data": log["data"].hex(),
                "block": receipt["blockNumber"]
            })
    
    if wormhole_events:
        print(f"[CROSSCHAIN] Found {len(wormhole_events)} Wormhole events in TX {tx_hash[:16]}...")
        for e in wormhole_events:
            print(f"  Topic: {e['topic'][:16]}... | Data length: {len(e['data'])//2} bytes")
    
    return wormhole_events

def analyze_multisig_signatures(tx_hash):
    tx = w3.eth.get_transaction(tx_hash)
    receipt = w3.eth.get_transaction_receipt(tx_hash)
    
    nonce = tx["nonce"]
    signer_count = 0
    unique_signers = set()
    
    for log in receipt["logs"]:
        if "0x" in log["data"]:
            signer = "0x" + log["data"][26:66]
            unique_signers.add(signer.lower())
            signer_count += 1
    
    return {
        "tx_hash": tx_hash,
        "unique_signers": list(unique_signers),
        "signer_count": signer_count,
        "from_address": tx["from"].lower()
    }

title: DeFi Flash Loan Large Value Borrow Detection
id: 9a7e3f21-4b8c-4d12-a5e6-7f8b9c0d1e2f
status: experimental
description: Detects flash loan borrow operations exceeding threshold values in a single transaction
references:
  - https://docs.aave.com/developers/v/2.0/guides/flash-loans
  - https://rekt.news/
author: Security Forensics Team
date: 2026/07/17
tags:
  - attack.defi
  - attack.flash_loan
  - tactic.impact
  - technique.t1485
logsource:
  product: blockchain
  service: ethereum
detection:
  selection_event:
    event_name: "FlashBorrow"
    event_source|endswith:
      - "0x7d2768de32b0b80b7a3454c06bdac94a69ddc7a9"
      - "0x398ec734b42e7c2d2d67f4928aa28f5c1c15295"
  selection_value:
    value_eth|gte: 1000000
  condition: selection_event and selection_value
level: high
falsepositives:
  - Legitimate protocol testing
  - White-hat security researchers

title: MEV Sandwich Attack Pattern Detection
id: 2c4e6f8a-1b3d-5e7f-9a0b-c2d4e6f8a1b3
status: experimental
description: Detects sandwich attack patterns where same address executes front and back run transactions
references:
  - https://wikipedia.org/wiki/Maximal_extractable_value
  - https://libmev.org/
author: Security Forensics Team
date: 2026/07/17
tags:
  - attack.mev
  - attack.sandwich
  - tactic.impact
  - technique.t1059
logsource:
  product: blockchain
  service: ethereum
detection:
  selection_block:
    event_name: "BlockTransactions"
  selection_pattern:
    tx_sequence:
      - position: "front"
        from_same_as: "back"
        function_selectors:
          - "0x38ed1739"
          - "0x8803dbee"
          - "0x7ff36ab5"
      - position: "target"
        from_different: true
      - position: "back"
        from_same_as: "front"
  condition: selection_block and selection_pattern
level: critical
falsepositives:
  - Legitimate arbitrage bots
  - Market maker rebalancing

#!/bin/bash

ETHERSCAN_API_KEY="${ETHERSCAN_API_KEY:-YOUR_API_KEY}"
CONTRACT_ADDRESS="$1"
BLOCK_RANGE="${2:-10000}"
MIN_VALUE_ETH="${3:-100}"
OUTPUT_FILE="forensics_report_${CONTRACT_ADDRESS}.json"

if [ -z "$CONTRACT_ADDRESS" ]; then
    echo "Usage: $0 <contract_address> [block_range] [min_value_eth]"
    exit 1
fi

LATEST_BLOCK=$(curl -s "https://api.etherscan.io/api?module=proxy&action=eth_blockNumber&apikey=${ETHERSCAN_API_KEY}" | jq -r '.result' | printf '%d\n' "0x$(cat /dev/stdin)")
START_BLOCK=$((LATEST_BLOCK - BLOCK_RANGE))

echo "[*] Querying transactions for ${CONTRACT_ADDRESS} from block ${START_BLOCK} to ${LATEST_BLOCK}"

curl -s "https://api.etherscan.io/api?module=account&action=txlist&address=${CONTRACT_ADDRESS}&startblock=${START_BLOCK}&endblock=${LATEST_BLOCK}&sort=desc&apikey=${ETHERSCAN_API_KEY}" | \
jq -r '.result[] | select(.value | tonumber > '"$(echo "$MIN_VALUE_ETH * 10^18" | bc)"')' > /tmp/large_txs.json

LARGE_TX_COUNT=$(jq 'length' /tmp/large_txs.json)
echo "[*] Found ${LARGE_TX_COUNT} transactions exceeding ${MIN_VALUE_ETH} ETH"

echo "[*] Analyzing transaction patterns..."
cat /tmp/large_txs.json | jq -r '.[] | "\(.hash)\t\(.from)\t\(.to)\t\(.value | tonumber / 10^18 | tostring) ETH\t\(.blockNumber)\t\(.gasUsed)"' > /tmp/tx_analysis.txt

echo "[*] Checking for repeated caller patterns..."
awk -F'\t' '{print $2}' /tmp/tx_analysis.txt | sort | uniq -c | sort -rn | head -20

echo "[*] Checking for self-destruct patterns..."
SELFDESTRUCT_SELECTOR="0xff"
cat /tmp/large_txs.json | jq -r '.[] | select(.input | startswith("0x3d346885")) | "\(.hash) - Potential SELFDESTRUCT"' 2>/dev/null

echo "[*] Generating report..."
echo "{" > "${OUTPUT_FILE}"
echo "  \"contract\": \"${CONTRACT_ADDRESS}\"," >> "${OUTPUT_FILE}"
echo "  \"block_range\": \"${START_BLOCK}-${LATEST_BLOCK}\"," >> "${OUTPUT_FILE}"
echo "  \"large_tx_count\": ${LARGE_TX_COUNT}," >> "${OUTPUT_FILE}"
echo "  \"generated_at\": \"$(date -u +%Y-%m-%dT%H:%M:%SZ)\"" >> "${OUTPUT_FILE}"
echo "}" >> "${OUTPUT_FILE}"

echo "[+] Forensics report generated: ${OUTPUT_FILE}"
echo "[+] Analysis complete. Review /tmp/tx_analysis.txt for details."

import requests
import json
from collections import defaultdict
from datetime import datetime

ETHERSCAN_API_KEY = "YOUR_API_KEY"

class DeFiForensicsAnalyzer:
    def __init__(self, api_key):
        self.api_key = api_key
        self.graph = defaultdict(list)
        self.address_labels = {}
    
    def fetch_transactions(self, address, block_start=0, block_end=99999999):
        url = "https://api.etherscan.io/api"
        params = {
            "module": "account",
            "action": "txlist",
            "address": address,
            "startblock": block_start,
            "endblock": block_end,
            "page": 1,
            "offset": 500,
            "sort": "desc",
            "apikey": self.api_key
        }
        resp = requests.get(url, params=params)
        return resp.json().get("result", [])
    
    def build_graph(self, start_address, max_depth=4):
        queue = [(start_address, 0)]
        visited = set()
        
        while queue:
            addr, depth = queue.pop(0)
            if depth > max_depth or addr in visited:
                continue
            visited.add(addr)
            
            txs = self.fetch_transactions(addr)
            for tx in txs:
                value_eth = int(tx["value"]) / 1e18
                if tx["from"].lower() == addr.lower() and value_eth > 1:
                    self.graph[addr].append({
                        "to": tx["to"],
                        "value": value_eth,
                        "hash": tx["hash"],
                        "block": int(tx["blockNumber"]),
                        "timestamp": datetime.fromtimestamp(int(tx["timeStamp"])).isoformat()
                    })
                    if tx["to"] and tx["to"].lower() not in visited:
                        queue.append((tx["to"], depth + 1))
        
        return self.graph
    
    def identify_hub_addresses(self):
        in_degree = defaultdict(int)
        out_degree = defaultdict(int)
        total_flow = defaultdict(float)
        
        for addr, edges in self.graph.items():
            out_degree[addr] += len(edges)
            for edge in edges:
                in_degree[edge["to"]] += 1
                total_flow[addr] += edge["value"]
        
        hubs = []
        for addr in set(list(in_degree.keys()) + list(out_degree.keys())):
            score = in_degree.get(addr, 0) + out_degree.get(addr, 0)
            if score >= 3:
                hubs.append({
                    "address": addr,
                    "in_degree": in_degree.get(addr, 0),
                    "out_degree": out_degree.get(addr, 0),
                    "total_flow_eth": total_flow.get(addr, 0),
                    "hub_score": score
                })
        
        hubs.sort(key=lambda x: x["hub_score"], reverse=True)
        return hubs
    
    def generate_report(self, start_address, output_path="forensics_graph_report.json"):
        self.build_graph(start_address)
        hubs = self.identify_hub_addresses()
        
        report = {
            "start_address": start_address,
            "generated_at": datetime.utcnow().isoformat(),
            "total_nodes": len(self.graph),
            "hub_addresses": hubs,
            "graph_edges": sum(len(edges) for edges in self.graph.values()),
            "graph": {addr: [{"to": e["to"], "value": e["value"], "hash": e["hash"]} for e in edges] 
                      for addr, edges in self.graph.items()}
        }
        
        with open(output_path, "w") as f:
            json.dump(report, f, indent=2)
        
        print(f"[+] Report generated: {output_path}")
        print(f"[+] Total nodes: {len(self.graph)}")
        print(f"[+] Hub addresses: {len(hubs)}")
        for h in hubs[:5]:
            print(f"    {h['address'][:16]}... | Score: {h['hub_score']} | Flow: {h['total_flow_eth']:.2f} ETH")
        
        return report

if __name__ == "__main__":
    analyzer = DeFiForensicsAnalyzer(ETHERSCAN_API_KEY)
    report = analyzer.generate_report("0xATTACKER_ADDRESS")

攻击者合约(0x304a8507...) → 子DAO(0x4aed5f14...) → 白帽回收合约 → 新以太坊链(ETH)
                                                                 └→ 原始链(ETC)

Aave FlashLoan(3000万DAI)
  → Euler合约(存款/捐赠/清算)
    → 攻击者合约(0xc8a2c00...)
      → Tornado Cash(~5000 ETH)
      → BNB Chain(跨链桥)
      → 部分资金归还Euler
      → CEX(币安等充值地址)

Solana链: 攻击合约(伪造签名铸造)
  → wETH Mint Event(Ethereum)
    → 攻击者Ethereum地址(0x629e7...)
      → 多笔分拆转账
        → 中间地址(0x3bfc...)
        → 中间地址(0x5789...)
        → DeFi协议交互(部分)