如何使用 Web3.py 审核以太坊钱包活动

概述

审计公司经常联系我们的企业与专业服务团队，寻求帮助以检索与客户的钱包地址相关的区块链活动。这种用例通常是在报税季节进行常规报税或当客户收到监管机构的行动通知或罚款时所需的。访问这些区块链活动传统上是一项挑战性任务，但我们在这里通过这份逐步指南来简化它。我们将指导你如何使用 Python 对 Ethereum 上的任何钱包地址进行彻底审计。

如果你对本指南中的步骤有任何疑问，或是作为审计公司希望获得帮助以检索你客户的区块链活动，请快速联系（QuickNode），我们非常乐意提供帮助！请联系 victor@quicknode.com，我们期待与你交谈。

现在，让我们开始进行审计吧！

你将会做什么

在本指南中，你将获取与钱包相关的所有交易活动，包括：

• 交易历史
• 代币转移历史 (ERC20)
• 内部交易历史

你将需要的东西

在你开始之前，请确保你具备以下条件：

• 对 Ethereum 交易的基本了解
• Python 3+
• Web3.py (v5)
• QuickNode 上的 Ethereum 端点

配置你的脚本

在编写将为我们执行审计的函数之前，我们需要确保将所需的 Python 包导入到我们的环境中。创建一个 Python 文件，并在文件顶部包含以下代码：

from web3 import Web3 # 我们将使用 Web3.py 库进行本指南
import json # 我们需要这个来解析区块链节点的响应
from tqdm import tqdm # 这个库帮助我们跟踪脚本的进度

为了从区块链查询数据，你需要 API 端点与 Ethereum 网络进行通信。为此，确保你在 这里 创建一个免费的 QuickNode 账户，登录后，点击 创建一个端点 按钮，然后选择 Ethereum 链和 Mainnet 网络。

创建端点后，复制 HTTP Provider 链接并将其添加到你的脚本中：

## 配置 Ethereum 端点
w3 = Web3(Web3.HTTPProvider("https://{your-endpoint-name}.quiknode.pro/{your-token}/"))

获取钱包交易

为了识别钱包交易活动，我们将从解析网络上的每个区块开始，然后解析每个交易，检查关注的地址是否在交易的 from 或 to 字段中。以下主函数将完成这一操作：

## 主函数以提取一系列区块的钱包活动
def get_transactions_for_addresses(addresses, from_block, to_block):
    transactions = []

    # 计算要处理的区块总数
    total_blocks = to_block - from_block + 1

    with tqdm(total=total_blocks, desc="处理区块") as pbar:
        for block_num in range(from_block, to_block + 1):
            # 请求区块数据
            block = w3.eth.getBlock(block_num, full_transactions=True)

            # 确认区块交易中是否找到关注地址
            for tx in block.transactions:
                if tx["from"] in addresses or tx["to"] in addresses:
                    tx_details = {
                        "block": block_num,
                        "hash": tx.hash.hex(),
                        "from": tx["from"],
                        "to": tx["to"],
                        "value": tx["value"],
                        "gas": tx["gas"],
                        "gasPrice": tx["gasPrice"],
                        "input": tx["input"],
                    }

                    transactions.append(tx_details)

            # 更新进度条
            pbar.update(1)

    return transactions

获取代币转移

请注意，主函数未能完成两件事：捕获代币转移并识别内部交易。现在我们将创建一个 get_token_transfers 函数，以捕获 ERC20 代币转移。你会注意到，为了实现这一点，我们需要解析特定区块的日志并调查符合条件的 topics。更具体地说，我们将查找 ERC20 转移事件签名，其中我们的钱包地址是转移的发送方或接收方。

## 函数以获取钱包代币转移
def get_token_transfers(address, block_num):
    # 将地址转换为其 32 字节表示
    padded_address = address.lower().replace("0x", "0x" + "0" * 24)

    # 将 block_num 转换为十六进制字符串
    block_hex = hex(block_num)

    filter_params = {
        "fromBlock": block_hex,
        "toBlock": block_hex,
        "topics": [\
            "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",  # ERC20 转移事件签名\
            None,\
            None\
        ]
    }

    # 查找地址作为发送方
    filter_params["topics"][1] = padded_address
    sent_transfers = w3.eth.get_logs(filter_params)

    # 解析API响应并记录转移详情
    sent_transfers_list = []
    for entry in sent_transfers:
        modified_entry = {
            "contractAddress": entry["address"],
            "from": w3.toChecksumAddress(entry["topics"][1].hex().lstrip("0x")),
            "to": w3.toChecksumAddress(entry["topics"][2].hex().lstrip("0x")),
            "value": int(entry["data"], 16),
            "topics": [topic.hex() if isinstance(topic, bytes) else topic for topic in entry["topics"]],
            "data": entry["data"],
            "blockNumber": entry["blockNumber"],
            "logIndex": entry["logIndex"],
            "transactionIndex": entry["transactionIndex"],
            "transactionHash": entry["transactionHash"].hex(),
            "blockHash": entry["blockHash"].hex(),
            "removed": entry["removed"]
        }
        sent_transfers_list.append(modified_entry)

    # 查找地址作为接收方
    filter_params["topics"][1] = None
    filter_params["topics"][2] = padded_address
    received_transfers = w3.eth.getLogs(filter_params)

    # 解析API响应并记录转移详情
    received_transfers_list = []
    for entry in received_transfers:
        modified_entry = {
            "contractAddress": entry["address"],
            "from": w3.toChecksumAddress(entry["topics"][1].hex().lstrip("0x")),
            "to": w3.toChecksumAddress(entry["topics"][2].hex().lstrip("0x")),
            "value": int(entry["data"], 16),
            "topics": [topic.hex() if isinstance(topic, bytes) else topic for topic in entry["topics"]],
            "data": entry["data"],
            "blockNumber": entry["blockNumber"],
            "logIndex": entry["logIndex"],
            "transactionIndex": entry["transactionIndex"],
            "transactionHash": entry["transactionHash"].hex(),
            "blockHash": entry["blockHash"].hex(),
            "removed": entry["removed"]
        }
        received_transfers_list.append(modified_entry)

    return sent_transfers_list + received_transfers_list

现在，我们只需更新我们的 get_transactions_for_addresses 函数，以调用 get_token_transfers 函数：

## 主函数以提取一系列区块的钱包活动
def get_transactions_for_addresses(addresses, from_block, to_block):
    transactions = []

    # 计算要处理的区块总数
    total_blocks = to_block - from_block + 1

    with tqdm(total=total_blocks, desc="处理区块") as pbar:
        for block_num in range(from_block, to_block + 1):
            # 请求区块数据
            block = w3.eth.getBlock(block_num, full_transactions=True)

            # 确认区块交易中是否找到关注地址
            for tx in block.transactions:
                if tx["from"] in addresses or tx["to"] in addresses:
                    tx_details = {
                        "block": block_num,
                        "hash": tx.hash.hex(),
                        "from": tx["from"],
                        "to": tx["to"],
                        "value": tx["value"],
                        "gas": tx["gas"],
                        "gasPrice": tx["gasPrice"],
                        "input": tx["input"],
                        "token_transfers": [],
                    }

                    # 获取代币转移
                    tx_details["token_transfers"].extend(get_token_transfers(tx["from"], block_num))
                    tx_details["token_transfers"].extend(get_token_transfers(tx["to"], block_num))

                    transactions.append(tx_details)

            # 更新进度条
            pbar.update(1)

    return transactions

获取内部交易

正如我们在前面的部分提到的，主函数也错过了捕获内部交易。内部交易发生在 EOA（钱包）与智能合约交互时，而这种交互的元数据详情仅通过调查交易的痕迹来获取。因此，我们将创建一个 get_internal_transactions 函数，该函数将运行 debug_traceTransaction 并在适用时获取那些详情。Trace/Debug API 专属于 Build 计划及以上。有关计划及其功能的更多信息，请访问 QuickNode 定价页面。

## 函数以获取钱包内部交易
def get_internal_transactions(tx_hash):
    try:
        # 发起请求
        trace = w3.provider.make_request("debug_traceTransaction", [tx_hash, {"tracer": "callTracer"}])

        internal_txs = []
        if "result" in trace:
            internal_txs.append(trace["result"]["calls"])
        return internal_txs

    except Exception as e:
        return str(e)

现在，我们只需更新我们的 get_transactions_for_addresses 函数，以调用 get_internal_transactions 函数：

## 主函数以提取一系列区块的钱包活动
def get_transactions_for_addresses(addresses, from_block, to_block):
    transactions = []

    # 计算要处理的区块总数
    total_blocks = to_block - from_block + 1

    with tqdm(total=total_blocks, desc="处理区块") as pbar:
        for block_num in range(from_block, to_block + 1):
            # 请求区块数据
            block = w3.eth.getBlock(block_num, full_transactions=True)

            # 确认区块交易中是否找到关注地址
            for tx in block.transactions:
                if tx["from"] in addresses or tx["to"] in addresses:
                    tx_details = {
                        "block": block_num,
                        "hash": tx.hash.hex(),
                        "from": tx["from"],
                        "to": tx["to"],
                        "value": tx["value"],
                        "gas": tx["gas"],
                        "gasPrice": tx["gasPrice"],
                        "input": tx["input"],
                        "token_transfers": [],
                        "internal_transactions": []
                    }

                    # 获取代币转移
                    tx_details["token_transfers"].extend(get_token_transfers(tx["from"], block_num))
                    tx_details["token_transfers"].extend(get_token_transfers(tx["to"], block_num))

                    # 检查与合约的交互并获取内部交易
                    if tx["to"] and w3.eth.getCode(tx["to"]).hex() != "0x":
                        tx_details["internal_transactions"].extend(get_internal_transactions(tx.hash.hex()))

                    transactions.append(tx_details)

            # 更新进度条
            pbar.update(1)

    return transactions

执行你的审计

现在我们已经构建了这三种函数，让我们在脚本中包括执行函数，该函数将接收运行所需的所有参数。

## 执行函数
def run(addresses, from_block, to_block):

    transactions = get_transactions_for_addresses(addresses, from_block, to_block)

    # 将输出写入 JSON 文件
    output_file_path = "wallet_audit_data.json"
    with open(output_file_path, "w") as json_file:
        json.dump(transactions, json_file, indent=4)

剩下的就是在脚本末尾添加使用示例，输入你想要审核的钱包地址和感兴趣的区块范围。使用你的命令行终端运行脚本（例如 Python3 wallet_auditor.py），或在 VS Code 中运行（例如单击编辑器右上角的播放按钮）。为了演示，我们将暴露我的朋友和同事 @bunsen：

## 使用示例:
## run(["{wallet_address}"], fromBlock, toBlock)
run(["0x91b51c173a4bDAa1A60e234fC3f705A16D228740"],17881437, 17881437)

执行后，你将在同一文件夹目录中找到一个 wallet_audit_data.json 文件，里面包含该钱包的整个活动轨迹，包括交易、代币转移和内部交易。做得好！

如果你想要全面参考我们的代码，请查看我们的 GitHub 页面 这里。

结论

恭喜你，现在能够对 Ethereum 上的任何钱包地址进行彻底审计！

这是正确进行区块链活动全审计的基础知识。以下是将其提升到更高水平的方法：

• 为一组钱包地址提取数据
• 通过利用多线程来减少预期的运行时间，从而并行运行你的脚本，处理区块范围的块
• 识别超出 ERC20 的代币转移，例如 ERC721 和 ERC1155

要了解更多关于 QuickNode 如何帮助审计公司从区块链中提取这种类型的数据，以确保数据的完整性和准确性，请随时通过 victor@quicknode.com 联系我，我们期待与你交谈！

我们 ❤️ 反馈！

让我们知道 如果你有任何反馈或对新主题的请求。我们期待你的消息。

• 原文链接： quicknode.com/guides/eth...
• 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～