diff --git a/.gitignore b/.gitignore index 45c0524..4fa103c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ __pycache__/ +.vscode/ build/ dist/ *.egg-info/ diff --git a/Poseidon/Blockchain.py b/Poseidon/Blockchain.py index 0c775d8..c6808b4 100644 --- a/Poseidon/Blockchain.py +++ b/Poseidon/Blockchain.py @@ -1,46 +1,51 @@ """ -本模块可用于与任何以太坊同构链（即通常所说的 EVM 兼容链）进行交互，支持常用的链上交互操作。 +本模块可用于与任何以太坊同构链（即通常所说的 EVM 链）进行交互，支持常用的链上交互操作。 """ from web3 import Web3 from eth_account import Account as EthAccount from loguru import logger -from sys import exc_info -from json import dumps +from typing import Optional, Union, List, Any +from traceback import format_exc +from json import dump, dumps +import os -logger.add('logs\\Poseidon_{time}.log') +LogPath = os.path.join("logs", "Poseidon_{time}.log") +logger.add(LogPath) class Chain(): """ - Chain 是区块链实例，作为链上交互的基础。 + Chain 是区块链实例，后续的所有链上交互的操作都将经由该指定节点处理。 """ - def __init__(self, RPCUrl: str, RequestParams: dict = None): + def __init__(self, RPCUrl: str, RequestParams: Optional[dict] = None): """ - 初始化。当连接失败时会抛出异常。 + 初始化。根据给定的节点 RPC 地址进行连接，可通过代理访问。当连接节点失败时会抛出异常。 参数： - RPCUrl (str): 链 RPC 地址 - RequestParams (可选)(dict): 指定连接时使用的 request 参数，默认为 None。 - 例如当需要使用代理进行访问时，则传入 RequestParams={"proxies": {"http": "http://127.0.0.1:","https": "http://127.0.0.1:"}} + RPCUrl (str): 节点 RPC 地址 + RequestParams (Optional[dict]): 连接时使用的 request 参数，默认为 None。 + 例如当需要使用代理进行访问时，则传入 RequestParams={"proxies": {"http": "http://localhost:","https": "http://localhost:"}} 成员变量： - Net (Web3.HTTPProvider): web3.py 原生的链交互器对象 ChainId (int): 链 ID - ClientVersion (str): 链 RPC 的客户端软件版本号 + Node (Web3.HTTPProvider): web3.py 原生的 HTTP 交互器实例 + Eth (Web3.HTTPProvider.eth): HTTP 交互器实例中的 eth 模块 """ + + from time import time from web3 import HTTPProvider from web3.middleware import geth_poa_middleware - from time import time RequestParamsPrint = f"[RequestParams]{RequestParams}\n" if RequestParams else "" StartTime = time() - self.Net = Web3(HTTPProvider(RPCUrl, request_kwargs=RequestParams)) - if self.Net.isConnected(): + self.Node = Web3(HTTPProvider(RPCUrl, request_kwargs=RequestParams)) + if self.Node.isConnected(): FinishTime = time() Delay = round((FinishTime - StartTime) * 1000) logger.success(f"\n[Chain][Initialize]Connected to [{RPCUrl}] [{Delay} ms]\n{RequestParamsPrint}{'-'*80}") - self.Net.middleware_onion.inject(geth_poa_middleware, layer=0) + self.Node.middleware_onion.inject(geth_poa_middleware, layer=0) + self.Eth = self.Node.eth self.GetBasicInformation() else: logger.error(f"\n[Chain][Initialize]Failed to connect to [{RPCUrl}]\n{RequestParamsPrint}{'-'*80}") @@ -48,34 +53,37 @@ def __init__(self, RPCUrl: str, RequestParams: dict = None): def GetBasicInformation(self) -> dict: """ - 获取链的基本信息。包括链 ID 、区块高度、 GasPrice 、出块间隔、链 RPC 的客户端软件版本号。 + 获取区块链基本信息。包括链 ID 、区块高度、 GasPrice 、出块间隔、当前节点的客户端软件版本号。 返回值： - BasicInformation (dict): 链的基本信息构成的字典。 + BasicInformation (dict): 区块链基本信息构成的字典。 {"ChainId"|"BlockNumber"|"GasPrice"|"Timeslot"|"ClientVersion"} """ - self.ChainId = self.Net.eth.chain_id - self.ClientVersion = self.Net.clientVersion - BlockNumber = self.Net.eth.block_number - GasPrice = self.Net.eth.gas_price - Timeslot = self.Net.eth.get_block(BlockNumber).timestamp - self.Net.eth.get_block(BlockNumber - 1).timestamp + + self.ChainId = self.Eth.chain_id + BlockNumber = self.Eth.block_number + GasPrice = self.Eth.gas_price + Timeslot = self.Eth.get_block(BlockNumber).timestamp - self.Eth.get_block(BlockNumber - 1).timestamp + ClientVersion = self.Node.clientVersion logger.success( - f"\n[Chain][GetBasicInformation]\n[ChainId]{self.ChainId}\n[BlockNumber]{BlockNumber}\n[GasPrice]{Web3.fromWei(GasPrice, 'gwei')} Gwei\n[Timeslot]{Timeslot}s\n[ClientVersion]{self.ClientVersion}\n{'-'*80}") - return {"ChainId": self.ChainId, "BlockNumber": BlockNumber, "GasPrice": GasPrice, "Timeslot": Timeslot, "ClientVersion": self.ClientVersion} + f"\n[Chain][GetBasicInformation]\n[ChainId]{self.ChainId}\n[BlockNumber]{BlockNumber}\n[GasPrice]{Web3.fromWei(GasPrice, 'gwei')} Gwei\n[Timeslot]{Timeslot}s\n[ClientVersion]{ClientVersion}\n{'-'*80}" + ) + return {"ChainId": self.ChainId, "BlockNumber": BlockNumber, "GasPrice": GasPrice, "Timeslot": Timeslot, "ClientVersion": ClientVersion} def GetTransactionInformationByHash(self, TransactionHash: str) -> dict: """ - 根据交易哈希获取交易信息。包括交易哈希、所在区块号、交易索引号、交易状态、交易类型、交易行为、发送者、接收者、(部署的合约地址)、(GasPrice 或 (MaxFeePerGas 和 MaxPriorityFeePerGas))、GasLimit、GasUsed、Nonce、Value、Logs、InputData。 + 根据交易哈希查询该交易的详细回执信息。包括交易哈希、所在区块号、交易索引号、交易状态、交易类型、交易行为、发送者、接收者、(部署的合约地址)、(GasPrice 或 (MaxFeePerGas 和 MaxPriorityFeePerGas))、GasLimit、GasUsed、Nonce、Value、R、S、V、Logs、InputData。 参数： - TransactionHash (str): 要查询的交易哈希 + TransactionHash (str): 要查询的交易的哈希 返回值： TransactionInformation (dict): 交易信息构成的字典。当出现异常时返回 None 。 - {"TransactionHash"|"BlockNumber"|"TransactionIndex"|"Status"|"Type"|"Action"|"From"|"To"|("ContractAddress")|<"GasPrice"|("MaxFeePerGas"&"MaxPriorityFeePerGas")>|"GasLimit"|"GasUsed"|"Nonce"|"Value"|"Logs"|"InputData"} + {"TransactionHash"|"BlockNumber"|"TransactionIndex"|"Status"|"Type"|"Action"|"From"|"To"|("ContractAddress")|<"GasPrice"|("MaxFeePerGas"&"MaxPriorityFeePerGas")>|"GasLimit"|"GasUsed"|"Nonce"|"Value"|"R"|"S"|"V"|"Logs"|"InputData"} """ + try: - Info = self.Net.eth.wait_for_transaction_receipt(TransactionHash, timeout=90) + Info = self.Eth.wait_for_transaction_receipt(TransactionHash, timeout=90) BlockNumber = Info.blockNumber TransactionIndex = Info.transactionIndex Status = Info.status @@ -83,8 +91,8 @@ def GetTransactionInformationByHash(self, TransactionHash: str) -> dict: To = Info.to ContractAddress = Info.contractAddress GasUsed = Info.gasUsed - Logs = Info.logs - Info = self.Net.eth.get_transaction(TransactionHash) + Logs = Web3.toJSON(Info.logs) + Info = self.Eth.get_transaction(TransactionHash) TransactionHash = Info.hash.hex() GasPrice = Info.gasPrice MaxFeePerGas = Info.get("maxFeePerGas", None) @@ -92,38 +100,42 @@ def GetTransactionInformationByHash(self, TransactionHash: str) -> dict: GasLimit = Info.gas Nonce = Info.nonce Value = Info.value + R = Info.r.hex() + S = Info.s.hex() + V = Info.v InputData = Info.input - Type = "EIP-1559" if MaxFeePerGas else "Traditional" - Action = "Deploy Contract" if To == None else "Call Contract" if self.Net.eth.get_code(Web3.toChecksumAddress(To)).hex() != "0x" else "Normal Transfer" + Type = "EIP-1559" if MaxFeePerGas or MaxPriorityFeePerGas else "Traditional" + Action = "Deploy Contract" if To == None else "Call Contract" if self.Eth.get_code(Web3.toChecksumAddress(To)).hex() != "0x" else "Normal Transfer" ContractPrint = f"[ContractAddress]{ContractAddress}\n" if ContractAddress else "" GasPricePrint = f"[GasPrice]{Web3.fromWei(GasPrice, 'gwei')} Gwei" if Type == "Traditional" else f"[MaxFeePerGas]{Web3.fromWei(MaxFeePerGas, 'gwei')} Gwei\n[MaxPriorityFeePerGas]{Web3.fromWei(MaxPriorityFeePerGas, 'gwei')} Gwei" + GeneralPrint = f"\n[Chain][GetTransactionInformationByHash]\n[TransactionHash]{TransactionHash}\n[BlockNumber]{BlockNumber}\n[TransactionIndex]{TransactionIndex}\n[Status]{'Success' if Status else 'Fail'}\n[Type]{Type}\n[Action]{Action}\n[From]{From}\n[To]{To}\n{ContractPrint}{GasPricePrint}\n[GasLimit]{GasLimit} [GasUsed]{GasUsed}\n[Nonce]{Nonce} [Value]{Value}\n[R]{R}\n[S]{S}\n[V]{V}\n[Logs]{Logs}\n[InputData]{InputData}\n{'-'*80}" if Status: - logger.success( - f"\n[Chain][GetTransactionInformationByHash]\n[TransactionHash]{TransactionHash}\n[BlockNumber]{BlockNumber}\n[TransactionIndex]{TransactionIndex}\n[Status]Success\n[Type]{Type}\n[Action]{Action}\n[From]{From}\n[To]{To}\n{ContractPrint}{GasPricePrint}\n[GasLimit]{GasLimit} [GasUsed]{GasUsed}\n[Nonce]{Nonce} [Value]{Value}\n[Logs]{Logs}\n[InputData]{InputData}\n{'-'*80}") + logger.success(GeneralPrint) else: - logger.error( - f"\n[Chain][GetTransactionInformationByHash]\n[TransactionHash]{TransactionHash}\n[BlockNumber]{BlockNumber}\n[TransactionIndex]{TransactionIndex}\n[Status]Fail\n[Type]{Type}\n[Action]{Action}\n[From]{From}\n[To]{To}\n{ContractPrint}{GasPricePrint}\n[GasLimit]{GasLimit} [GasUsed]{GasUsed}\n[Nonce]{Nonce} [Value]{Value}\n[Logs]{Logs}\n[InputData]{InputData}\n{'-'*80}") - return {"TransactionHash": TransactionHash, "BlockNumber": BlockNumber, "TransactionIndex": TransactionIndex, "Status": Status, "Type": Type, "Action": Action, "From": From, "To": To, "ContractAddress": ContractAddress, "GasPrice": GasPrice, "GasLimit": GasLimit, "GasUsed": GasUsed, "Nonce": Nonce, "Value": Value, "Logs": Logs, "InputData": InputData} if Type == "Traditional" else {"TransactionHash": TransactionHash, "BlockNumber": BlockNumber, "TransactionIndex": TransactionIndex, "Status": Status, "Type": Type, "Action": Action, "From": From, "To": To, "ContractAddress": ContractAddress, "MaxFeePerGas": MaxFeePerGas, "MaxPriorityFeePerGas": MaxPriorityFeePerGas, "GasLimit": GasLimit, "GasUsed": GasUsed, "Nonce": Nonce, "Value": Value, "Logs": Logs, "InputData": InputData} + logger.error(GeneralPrint) + return {"TransactionHash": TransactionHash, "BlockNumber": BlockNumber, "TransactionIndex": TransactionIndex, "Status": Status, "Type": Type, "Action": Action, "From": From, "To": To, "ContractAddress": ContractAddress, "GasPrice": GasPrice, "MaxFeePerGas": MaxFeePerGas, "MaxPriorityFeePerGas": MaxPriorityFeePerGas, "GasLimit": GasLimit, "GasUsed": GasUsed, "Nonce": Nonce, "Value": Value, "R": R, "S": S, "V": V, "Logs": Logs, "InputData": InputData} except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Chain][GetTransactionInformationByHash]Failed to get transaction information\n[TransactionHash]{TransactionHash}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Chain][GetTransactionInformationByHash]Failed\n[TransactionHash]{TransactionHash}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None - def GetTransactionInformationByBlockIdAndIndex(self, BlockID, TransactionIndex: int) -> dict: + def GetTransactionInformationByBlockIdAndIndex(self, BlockID: Union[str, int], TransactionIndex: int) -> dict: """ - 根据区块 ID 和交易在块中的索引来获取交易信息。包括交易哈希、所在区块号、交易索引号、交易状态、交易类型、交易行为、发送者、接收者、(部署的合约地址)、(GasPrice 或 (MaxFeePerGas 和 MaxPriorityFeePerGas))、GasLimit、GasUsed、Nonce、Value、Logs、InputData。 + 根据区块 ID 和交易在块中的索引来查询该交易的详细回执信息。包括交易哈希、所在区块号、交易索引号、交易状态、交易类型、交易行为、发送者、接收者、(部署的合约地址)、(GasPrice 或 (MaxFeePerGas 和 MaxPriorityFeePerGas))、GasLimit、GasUsed、Nonce、Value、R、S、V、Logs、InputData。 参数： - BlockID (str|int): 区块 ID 。可为区块号或 'latest', 'earliest', 'pending' 。 + BlockID (Union[str,int]): 区块 ID 。可为区块号数值或 'latest', 'earliest', 'pending' 。 TransactionIndex (int): 交易在块中的索引 返回值： TransactionInformation (dict): 交易信息构成的字典。当出现异常时返回 None 。 - {"TransactionHash"|"BlockNumber"|"TransactionIndex"|"Status"|"Type"|"Action"|"From"|"To"|("ContractAddress")|<"GasPrice"|("MaxFeePerGas"&"MaxPriorityFeePerGas")>|"GasLimit"|"GasUsed"|"Nonce"|"Value"|"Logs"|"InputData"} + {"TransactionHash"|"BlockNumber"|"TransactionIndex"|"Status"|"Type"|"Action"|"From"|"To"|("ContractAddress")|<"GasPrice"|("MaxFeePerGas"&"MaxPriorityFeePerGas")>|"GasLimit"|"GasUsed"|"Nonce"|"Value"|"R"|"S"|"V"|"Logs"|"InputData"} """ + try: - Info = self.Net.eth.get_transaction_by_block(BlockID, TransactionIndex) + Info = self.Eth.get_transaction_by_block(BlockID, TransactionIndex) TransactionHash = Info.hash.hex() BlockNumber = Info.blockNumber TransactionIndex = Info.transactionIndex @@ -135,27 +147,62 @@ def GetTransactionInformationByBlockIdAndIndex(self, BlockID, TransactionIndex: GasLimit = Info.gas Nonce = Info.nonce Value = Info.value + R = Info.r.hex() + S = Info.s.hex() + V = Info.v InputData = Info.input - Info = self.Net.eth.wait_for_transaction_receipt(TransactionHash, timeout=90) + Info = self.Eth.wait_for_transaction_receipt(TransactionHash, timeout=90) Status = Info.status GasUsed = Info.gasUsed ContractAddress = Info.contractAddress Logs = Info.logs Type = "EIP-1559" if MaxFeePerGas else "Traditional" - Action = "Deploy Contract" if To == None else "Call Contract" if self.Net.eth.get_code(Web3.toChecksumAddress(To)).hex() != "0x" else "Normal Transfer" + Action = "Deploy Contract" if To == None else "Call Contract" if self.Eth.get_code(Web3.toChecksumAddress(To)).hex() != "0x" else "Normal Transfer" ContractPrint = f"[ContractAddress]{ContractAddress}\n" if ContractAddress else "" GasPricePrint = f"[GasPrice]{Web3.fromWei(GasPrice, 'gwei')} Gwei" if Type == "Traditional" else f"[MaxFeePerGas]{Web3.fromWei(MaxFeePerGas, 'gwei')} Gwei\n[MaxPriorityFeePerGas]{Web3.fromWei(MaxPriorityFeePerGas, 'gwei')} Gwei" + GeneralPrint = f"\n[Chain][GetTransactionInformationByBlockIdAndIndex]\n[TransactionHash]{TransactionHash}\n[BlockNumber]{BlockNumber}\n[TransactionIndex]{TransactionIndex}\n[Status]{'Success' if Status else 'Fail'}\n[Type]{Type}\n[Action]{Action}\n[From]{From}\n[To]{To}\n{ContractPrint}{GasPricePrint}\n[GasLimit]{GasLimit} [GasUsed]{GasUsed}\n[Nonce]{Nonce} [Value]{Value}\n[R]{R}\n[S]{S}\n[V]{V}\n[Logs]{Logs}\n[InputData]{InputData}\n{'-'*80}" if Status: - logger.success( - f"\n[Chain][GetTransactionInformationByBlockIdAndIndex]\n[TransactionHash]{TransactionHash}\n[BlockNumber]{BlockNumber}\n[TransactionIndex]{TransactionIndex}\n[Status]Success\n[Type]{Type}\n[Action]{Action}\n[From]{From}\n[To]{To}\n{ContractPrint}{GasPricePrint}\n[GasLimit]{GasLimit} [GasUsed]{GasUsed}\n[Nonce]{Nonce} [Value]{Value}\n[Logs]{Logs}\n[InputData]{InputData}\n{'-'*80}") + logger.success(GeneralPrint) else: - logger.error( - f"\n[Chain][GetTransactionInformationByBlockIdAndIndex]\n[TransactionHash]{TransactionHash}\n[BlockNumber]{BlockNumber}\n[TransactionIndex]{TransactionIndex}\n[Status]Fail\n[Type]{Type}\n[Action]{Action}\n[From]{From}\n[To]{To}\n{ContractPrint}{GasPricePrint}\n[GasLimit]{GasLimit} [GasUsed]{GasUsed}\n[Nonce]{Nonce} [Value]{Value}\n[Logs]{Logs}\n[InputData]{InputData}\n{'-'*80}") - return {"TransactionHash": TransactionHash, "BlockNumber": BlockNumber, "TransactionIndex": TransactionIndex, "Status": Status, "Type": Type, "Action": Action, "From": From, "To": To, "ContractAddress": ContractAddress, "GasPrice": GasPrice, "GasLimit": GasLimit, "GasUsed": GasUsed, "Nonce": Nonce, "Value": Value, "Logs": Logs, "InputData": InputData} if Type == "Traditional" else {"TransactionHash": TransactionHash, "BlockNumber": BlockNumber, "TransactionIndex": TransactionIndex, "Status": Status, "Type": Type, "Action": Action, "From": From, "To": To, "ContractAddress": ContractAddress, "MaxFeePerGas": MaxFeePerGas, "MaxPriorityFeePerGas": MaxPriorityFeePerGas, "GasLimit": GasLimit, "GasUsed": GasUsed, "Nonce": Nonce, "Value": Value, "Logs": Logs, "InputData": InputData} + logger.error(GeneralPrint) + return {"TransactionHash": TransactionHash, "BlockNumber": BlockNumber, "TransactionIndex": TransactionIndex, "Status": Status, "Type": Type, "Action": Action, "From": From, "To": To, "ContractAddress": ContractAddress, "GasPrice": GasPrice, "MaxFeePerGas": MaxFeePerGas, "MaxPriorityFeePerGas": MaxPriorityFeePerGas, "GasLimit": GasLimit, "GasUsed": GasUsed, "Nonce": Nonce, "Value": Value, "R": R, "S": S, "V": V, "Logs": Logs, "InputData": InputData} except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Chain][GetTransactionInformationByBlockIdAndIndex]Failed to get transaction information\n[BlockID]{BlockID}\n[TransactionIndex]{TransactionIndex}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Chain][GetTransactionInformationByBlockIdAndIndex]Failed\n[BlockID]{BlockID}\n[TransactionIndex]{TransactionIndex}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) + return None + + def GetBlockInformation(self, BlockID: Union[str, int]) -> dict: + """ + 根据区块 ID 获取该区块的详细信息。包括区块号、区块哈希、矿工、时间戳、GasLimit、GasUsed、块内交易的哈希集合。 + + 参数： + BlockID (Union[str,int]): 区块 ID 。可为区块号数值或 'latest', 'earliest', 'pending' 。 + + 返回值： + BlockInformation (dict): 区块信息构成的字典。当出现异常时返回 None 。 + {"BlockNumber"|"BlockHash"|"Miner"|"TimeStamp"|"GasLimit"|"GasUsed"|"Transactions"} + """ + + try: + Info = self.Eth.get_block(BlockID) + BlockNumber = Info.number + BlockHash = Info.hash.hex() + Miner = Info.miner + TimeStamp = Info.timestamp + GasLimit = Info.gasLimit + GasUsed = Info.gasUsed + Transactions = Web3.toJSON(Info.transactions) + logger.success( + f"\n[Chain][GetBlockInformation]\n[BlockNumber]{BlockNumber}\n[BlockHash]{BlockHash}\n[Miner]{Miner}\n[TimeStamp]{TimeStamp}\n[GasLimit]{GasLimit}\n[GasUsed]{GasUsed}\n[Transactions]{Transactions}" + ) + return {"BlockNumber": BlockNumber, "BlockHash": BlockHash, "Miner": Miner, "TimeStamp": TimeStamp, "GasLimit": GasLimit, "GasUsed": GasUsed, "Transactions": Transactions} + except Exception: + ExceptionInformation = format_exc() + logger.error( + f"\n[Chain][GetBlockInformation]Failed\n[BlockID]{BlockID}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None def GetBalance(self, Address: str) -> int: @@ -168,14 +215,17 @@ def GetBalance(self, Address: str) -> int: 返回值： Balance (int): 账户网络原生代币余额。单位为 wei ，当出现异常时返回 None 。 """ + try: Address = Web3.toChecksumAddress(Address) - Balance = self.Net.eth.get_balance(Address) - logger.success(f"\n[Chain][GetBalance]\n[Address]{Address}\n[Balance][{Balance} Wei]<=>[{Web3.fromWei(Balance,'ether')} Ether]\n{'-'*80}") + Balance = self.Eth.get_balance(Address) + logger.success( + f"\n[Chain][GetBalance]\n[Address]{Address}\n[Balance][{Balance} Wei]<=>[{Web3.fromWei(Balance,'ether')} Ether]\n{'-'*80}" + ) return Balance except Exception: - ExceptionInformation = exc_info() - logger.error(f"\n[Chain][GetBalance]Failed to get balance\n[Address]{Address}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + ExceptionInformation = format_exc() + logger.error(f"\n[Chain][GetBalance]Failed\n[Address]{Address}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}") return None def GetCode(self, Address: str) -> str: @@ -188,14 +238,15 @@ def GetCode(self, Address: str) -> str: 返回值： Code (str): 合约已部署字节码。含 0x 前缀的十六进制形式，当出现异常时返回 None 。 """ + try: Address = Web3.toChecksumAddress(Address) - Code = self.Net.eth.get_code(Address).hex() + Code = self.Eth.get_code(Address).hex() logger.success(f"\n[Chain][GetCode]\n[Address]{Address}\n[Code]{Code}\n{'-'*80}") return Code except Exception: - ExceptionInformation = exc_info() - logger.error(f"\n[Chain][GetCode]Failed to get code\n[Address]{Address}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + ExceptionInformation = format_exc() + logger.error(f"\n[Chain][GetCode]Failed\n[Address]{Address}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}") return None def GetStorage(self, Address: str, SlotIndex: int) -> str: @@ -209,14 +260,19 @@ def GetStorage(self, Address: str, SlotIndex: int) -> str: 返回值： Data (str): 存储值。含 0x 前缀的十六进制形式，当出现异常时返回 None 。 """ + try: Address = Web3.toChecksumAddress(Address) - Data = self.Net.eth.get_storage_at(Address, SlotIndex).hex() - logger.success(f"\n[Chain][GetStorage]\n[Address]{Address}\n[SlotIndex]{SlotIndex}\n[Value][Hex][{Data}]<=>[Dec][{int(Data,16)}]\n{'-'*80}") + Data = self.Eth.get_storage_at(Address, SlotIndex).hex() + logger.success( + f"\n[Chain][GetStorage]\n[Address]{Address}\n[SlotIndex]{SlotIndex}\n[Value][Hex][{Data}]<=>[Dec][{int(Data,16)}]\n{'-'*80}" + ) return Data except Exception: - ExceptionInformation = exc_info() - logger.error(f"\n[Chain][GetStorage]Failed to get storage\n[Address]{Address}\n[SlotIndex]{SlotIndex}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + ExceptionInformation = format_exc() + logger.error( + f"\n[Chain][GetStorage]Failed\n[Address]{Address}\n[SlotIndex]{SlotIndex}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None def DumpStorage(self, Address: str, Count: int) -> list: @@ -230,16 +286,18 @@ def DumpStorage(self, Address: str, Count: int) -> list: 返回值： Data (List[str]): 存储值列表。含 0x 前缀的十六进制形式，当出现异常时返回 None 。 """ + try: Address = Web3.toChecksumAddress(Address) - Data = [self.Net.eth.get_storage_at(Address, i).hex() for i in range(Count)] + Data = [self.Eth.get_storage_at(Address, i).hex() for i in range(Count)] Temp = '\n'.join([f"[Slot {i}]{Data[i]}" for i in range(len(Data))]) logger.success(f"\n[Chain][DumpStorage]\n[Address]{Address}\n{Temp}\n{'-'*80}") return Data except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Chain][DumpStorage]Failed to dump storage\n[Address]{Address}\n[slot 0 ... {Count-1}]\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Chain][DumpStorage]Failed\n[Address]{Address}\n[slot 0 ... {Count-1}]\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None def GetPublicKeyByTransactionHash(self, TransactionHash: str) -> tuple: @@ -252,10 +310,11 @@ def GetPublicKeyByTransactionHash(self, TransactionHash: str) -> tuple: 返回值： (Address, PublicKey) (tuple): 由账户地址和账户公钥组成的元组。当出现异常时返回 None 。 """ + try: from eth_account._utils.signing import to_standard_v, extract_chain_id, serializable_unsigned_transaction_from_dict - Transaction = self.Net.eth.get_transaction(TransactionHash) - Signature = self.Net.eth.account._keys.Signature(vrs=(to_standard_v(extract_chain_id(Transaction.v)[1]), Web3.toInt(Transaction.r), Web3.toInt(Transaction.s))) + Transaction = self.Eth.get_transaction(TransactionHash) + Signature = self.Eth.account._keys.Signature(vrs=(to_standard_v(extract_chain_id(Transaction.v)[1]), Web3.toInt(Transaction.r), Web3.toInt(Transaction.s))) UnsignedTransactionDict = {i: Transaction[i] for i in ['chainId', 'nonce', 'gasPrice' if int( Transaction.type, 0) != 2 else '', 'gas', 'to', 'value', 'accessList', 'maxFeePerGas', 'maxPriorityFeePerGas'] if i in Transaction} UnsignedTransactionDict['data'] = Transaction['input'] @@ -264,56 +323,55 @@ def GetPublicKeyByTransactionHash(self, TransactionHash: str) -> tuple: PublicKey = str(Temp).replace('0x', '0x04') # 比特币未压缩公钥格式 Address = Temp.to_checksum_address() logger.success( - f"\n[Chain][GetPublicKeyByTransactionHash]\n[TransactionHash]{TransactionHash}\n[Address]{Address}\n[PublicKey]{PublicKey}\n{'-'*80}") + f"\n[Chain][GetPublicKeyByTransactionHash]\n[TransactionHash]{TransactionHash}\n[Address]{Address}\n[PublicKey]{PublicKey}\n{'-'*80}" + ) return (Address, PublicKey) except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Chain][GetPublicKeyByTransactionHash]\nFailed to get public key by transaction hash\n[TransactionHash]{TransactionHash}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Chain][GetPublicKeyByTransactionHash]Failed\n[TransactionHash]{TransactionHash}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None class Account(): """ - Account 是账户实例，作为发起链上调用的基础。 + Account 是账户实例，后续的交易将经由该指定账户发送至链上。 """ def __init__(self, Chain: Chain, PrivateKey: str): """ - 初始化。通过私钥导入账户并与 Chain 实例绑定，后续的所有链上调用都会作用在 Chain 实例化表示的链上。当导入账户失败时将会抛出异常。 + 初始化。通过私钥导入账户并与 Chain 实例绑定，后续的交易将经由该指定账户发送至链上。当导入账户失败时将会抛出异常。 参数： Chain (Poseidon.Blockchain.Chain): 区块链实例 PrivateKey (str): 账户私钥。不含 0x 前缀的十六进制形式。 成员变量： - Chain (Poseidon.Blockchain.Chain): 区块链实例 - Address (str): 账户地址 - PrivateKey (str): 账户私钥 + EthAccount (eth_account.Account): eth_account 的原生 Account 对象实例 """ try: - self.Chain = Chain - self.Net = Chain.Net - Temp = EthAccount.from_key(PrivateKey) - self.Address = Web3.toChecksumAddress(Temp.address) - self.PrivateKey = Temp.privateKey - self.Net.eth.default_account = self.Address - logger.success(f"\n[Account][Initialize]Successfully import account [{self.Address}]\n{'-'*80}") + self.EthAccount = EthAccount.from_key(PrivateKey) + self._Chain = Chain + self._Eth = Chain.Eth + self._Eth.default_account = self.EthAccount.address + logger.success(f"\n[Account][Initialize]Successfully import account [{self.EthAccount.address}]\n{'-'*80}") self.GetSelfBalance() except: - ExceptionInformation = exc_info() - logger.error(f"\n[Account][Initialize]Failed to import account\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + ExceptionInformation = format_exc() + logger.error(f"\n[Account][Initialize]Failed to import account\n[ExceptionInformation]{ExceptionInformation}{'-'*80}") raise Exception("Failed to import account.") def GetSelfBalance(self) -> int: """ - 获取自身账户的网络原生代币余额。当余额为 0 时会触发无法发送交易的警告。 + 获取自身账户的网络原生代币余额。 返回值： Balance (int): 自身账户网络原生代币余额。单位为 wei ，当出现异常时返回 None 。 """ - Balance = self.Chain.GetBalance(self.Address) + + Balance = self._Chain.GetBalance(self.EthAccount.address) if Balance == 0: logger.warning(f"\n[Account][GetSelfBalance]\n[Warning]This account's balance is insufficient to send transactions\n{'-'*80}") return Balance @@ -331,29 +389,31 @@ def Transfer(self, To: str, Value: int, GasLimit: int = 100000, Data: str = "0x" 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 """ + try: - From = Web3.toChecksumAddress(self.Address) + From = self.EthAccount.address To = Web3.toChecksumAddress(To) Txn = { - "chainId": self.Chain.ChainId, + "chainId": self._Chain.ChainId, "from": From, - "to": To, + "to": Web3.toChecksumAddress(To), "value": Value, "gas": GasLimit, - "gasPrice": self.Net.eth.gas_price, - "nonce": self.Net.eth.get_transaction_count(self.Address), + "gasPrice": self._Eth.gas_price, + "nonce": self._Eth.get_transaction_count(From), "data": Data, } - SignedTxn = self.Net.eth.account.sign_transaction(Txn, self.PrivateKey) - TransactionHash = self.Net.eth.send_raw_transaction(SignedTxn.rawTransaction).hex() + SignedTxn = self.EthAccount.sign_transaction(Txn) + TransactionHash = self._Eth.send_raw_transaction(SignedTxn.rawTransaction).hex() Txn["gasPrice"] = f'{Web3.fromWei(Txn["gasPrice"],"gwei")} Gwei' logger.info(f"\n[Account][Transfer]\n[TransactionHash]{TransactionHash}\n[Txn]{dumps(Txn, indent=2)}\n{'-'*80}") - TransactionInformation = self.Chain.GetTransactionInformationByHash(TransactionHash) + TransactionInformation = self._Chain.GetTransactionInformationByHash(TransactionHash) return TransactionInformation except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Account][Transfer]Failed to transfer\n[From]{From}\n[To]{To}\n[Value]{Value}[GasLimit]{GasLimit}\n[Data]{Data}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Account][Transfer]Failed\n[From]{From}\n[To]{To}\n[Value]{Value}[GasLimit]{GasLimit}\n[Data]{Data}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None def SendTransaction(self, To: str, Data: str, Value: int = 0, GasLimit: int = 1000000) -> dict: @@ -369,29 +429,31 @@ def SendTransaction(self, To: str, Data: str, Value: int = 0, GasLimit: int = 10 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 """ + try: - From = Web3.toChecksumAddress(self.Address) + From = self.EthAccount.address To = Web3.toChecksumAddress(To) Txn = { - "chainId": self.Chain.ChainId, + "chainId": self._Chain.ChainId, "from": From, "to": To, "value": Value, "gas": GasLimit, - "gasPrice": self.Net.eth.gas_price, - "nonce": self.Net.eth.get_transaction_count(self.Address), + "gasPrice": self._Eth.gas_price, + "nonce": self._Eth.get_transaction_count(From), "data": Data, } - SignedTxn = self.Net.eth.account.sign_transaction(Txn, self.PrivateKey) - TransactionHash = self.Net.eth.send_raw_transaction(SignedTxn.rawTransaction).hex() + SignedTxn = self.EthAccount.sign_transaction(Txn) + TransactionHash = self._Eth.send_raw_transaction(SignedTxn.rawTransaction).hex() Txn["gasPrice"] = f'{Web3.fromWei(Txn["gasPrice"],"gwei")} Gwei' logger.info(f"\n[Account][SendTransaction][Traditional]\n[TransactionHash]{TransactionHash}\n[Txn]{dumps(Txn, indent=2)}\n{'-'*80}") - TransactionInformation = self.Chain.GetTransactionInformationByHash(TransactionHash) + TransactionInformation = self._Chain.GetTransactionInformationByHash(TransactionHash) return TransactionInformation except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Account][SendTransaction][Traditional]Failed to send transaction\n[From]{From}\n[To]{To}\n[Value]{Value}[GasLimit]{GasLimit}\n[Data]{Data}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Account][SendTransaction][Traditional]Failed\n[From]{From}\n[To]{To}\n[Value]{Value}[GasLimit]{GasLimit}\n[Data]{Data}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None def SendTransactionByEIP1559(self, To: str, Data: str, Value: int = 0, GasLimit: int = 1000000) -> dict: @@ -407,36 +469,38 @@ def SendTransactionByEIP1559(self, To: str, Data: str, Value: int = 0, GasLimit: 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 """ + try: - From = Web3.toChecksumAddress(self.Address) + From = self.EthAccount.address To = Web3.toChecksumAddress(To) - BaseFee = self.Net.eth.gas_price - MaxPriorityFee = self.Net.eth.max_priority_fee + Web3.toWei(1, "gwei") + BaseFee = self._Eth.gas_price + MaxPriorityFee = self._Eth.max_priority_fee + Web3.toWei(1, "gwei") Txn = { - "chainId": self.Chain.ChainId, + "chainId": self._Chain.ChainId, "from": From, "to": To, "value": Value, "gas": GasLimit, "maxFeePerGas": BaseFee + MaxPriorityFee, "maxPriorityFeePerGas": MaxPriorityFee, - "nonce": self.Net.eth.get_transaction_count(self.Address), + "nonce": self._Eth.get_transaction_count(From), "data": Data } - SignedTxn = self.Net.eth.account.sign_transaction(Txn, self.PrivateKey) - TransactionHash = self.Net.eth.send_raw_transaction(SignedTxn.rawTransaction).hex() + SignedTxn = self.EthAccount.sign_transaction(Txn) + TransactionHash = self._Eth.send_raw_transaction(SignedTxn.rawTransaction).hex() Txn["maxFeePerGas"] = f'{Web3.fromWei(Txn["maxFeePerGas"],"gwei")} Gwei' Txn["maxPriorityFeePerGas"] = f'{Web3.fromWei(Txn["maxPriorityFeePerGas"],"gwei")} Gwei' logger.info(f"\n[Account][SendTransaction][EIP-1559]\n[TransactionHash]{TransactionHash}\n[Txn]{dumps(Txn, indent=2)}\n{'-'*80}") - TransactionInformation = self.Chain.GetTransactionInformationByHash(TransactionHash) + TransactionInformation = self._Chain.GetTransactionInformationByHash(TransactionHash) return TransactionInformation except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Account][SendTransaction][EIP-1559]Failed to send transaction\n[From]{From}\n[To]{To}\n[Value]{Value}[GasLimit]{GasLimit}\n[Data]{Data}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Account][SendTransaction][EIP-1559]Failed\n[From]{From}\n[To]{To}\n[Value]{Value}[GasLimit]{GasLimit}\n[Data]{Data}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None - def DeployContract(self, ABI: dict, Bytecode: str, Value: int = 0, *Arguments) -> dict: + def DeployContract(self, ABI: dict, Bytecode: str, Value: int = 0, *Arguments: Optional[Any]) -> dict: """ 部署合约。若 90 秒内交易未确认则作超时处理。 @@ -444,29 +508,30 @@ def DeployContract(self, ABI: dict, Bytecode: str, Value: int = 0, *Arguments) - ABI (dict): 合约 ABI Bytecode (str): 合约部署字节码。含 0x 前缀的十六进制形式。 Value (可选)(int): 随交易发送给合约的网络原生代币数量。单位为 wei ，默认为 0 wei 。 - *Arguments (可选)(any): 传给合约构造函数的参数，默认为空。 + *Arguments (Optional[Any]): 传给合约构造函数的参数，默认为空。 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 当合约部署成功时，字典中会额外添加"Contract"字段，该变量是已实例化的 Contract 对象，失败时为 None。 """ + try: - DeployingContract = self.Net.eth.contract(abi=ABI, bytecode=Bytecode) - TransactionData = DeployingContract.constructor(*Arguments).buildTransaction({"gasPrice": self.Net.eth.gas_price, "value": Value}) + DeployingContract = self._Eth.contract(abi=ABI, bytecode=Bytecode) + TransactionData = DeployingContract.constructor(*Arguments).buildTransaction({"gasPrice": self._Eth.gas_price, "value": Value}) Txn = { - "chainId": self.Chain.ChainId, - "from": Web3.toChecksumAddress(self.Address), + "chainId": self._Chain.ChainId, + "from": self.EthAccount.address, "value": TransactionData["value"], "gas": TransactionData["gas"], "gasPrice": TransactionData["gasPrice"], - "nonce": self.Net.eth.get_transaction_count(self.Address), + "nonce": self._Eth.get_transaction_count(self.EthAccount.address), "data": TransactionData["data"] } - SignedTxn = self.Net.eth.account.sign_transaction(Txn, self.PrivateKey) - TransactionHash = self.Net.eth.send_raw_transaction(SignedTxn.rawTransaction).hex() + SignedTxn = self.EthAccount.sign_transaction(Txn) + TransactionHash = self._Eth.send_raw_transaction(SignedTxn.rawTransaction).hex() Txn["gasPrice"] = f'{Web3.fromWei(Txn["gasPrice"],"gwei")} Gwei' logger.info(f"\n[Account][DeployContract]\n[TransactionHash]{TransactionHash}\n[Txn]{dumps(Txn, indent=2)}\n{'-'*80}") - TransactionInformation = self.Chain.GetTransactionInformationByHash(TransactionHash) + TransactionInformation = self._Chain.GetTransactionInformationByHash(TransactionHash) if TransactionInformation["Status"]: DeployedContract = Contract(self, TransactionInformation["ContractAddress"], ABI) TransactionInformation["Contract"] = DeployedContract @@ -475,8 +540,10 @@ def DeployContract(self, ABI: dict, Bytecode: str, Value: int = 0, *Arguments) - TransactionInformation["Contract"] = None return TransactionInformation except Exception: - ExceptionInformation = exc_info() - logger.error(f"\n[Account][DeployContract]Failed to deploy contract\n[Value]{Value}\n[ABI]{ABI}\n[Bytecode]{Bytecode}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + ExceptionInformation = format_exc() + logger.error( + f"\n[Account][DeployContract]Failed\n[Value]{Value}\n[ABI]{ABI}\n[Bytecode]{Bytecode}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None def DeployContractWithoutABI(self, Bytecode: str, Value: int = 0, GasLimit: int = 10000000) -> dict: @@ -491,26 +558,28 @@ def DeployContractWithoutABI(self, Bytecode: str, Value: int = 0, GasLimit: int 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 """ + try: Txn = { - "chainId": self.Chain.ChainId, - "from": Web3.toChecksumAddress(self.Address), + "chainId": self._Chain.ChainId, + "from": self.EthAccount.address, "value": Value, "gas": GasLimit, - "gasPrice": self.Net.eth.gas_price, - "nonce": self.Net.eth.get_transaction_count(self.Address), + "gasPrice": self._Eth.gas_price, + "nonce": self._Eth.get_transaction_count(self.EthAccount.address), "data": Bytecode, } - SignedTxn = self.Net.eth.account.sign_transaction(Txn, self.PrivateKey) - TransactionHash = self.Net.eth.send_raw_transaction(SignedTxn.rawTransaction).hex() + SignedTxn = self.EthAccount.sign_transaction(Txn) + TransactionHash = self._Eth.send_raw_transaction(SignedTxn.rawTransaction).hex() Txn["gasPrice"] = f'{Web3.fromWei(Txn["gasPrice"],"gwei")} Gwei' logger.info(f"\n[Account][DeployContractWithoutABI]\n[TransactionHash]{TransactionHash}\n[Txn]{dumps(Txn, indent=2)}\n{'-'*80}") - TransactionInformation = self.Chain.GetTransactionInformationByHash(TransactionHash) + TransactionInformation = self._Chain.GetTransactionInformationByHash(TransactionHash) return TransactionInformation except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Account][DeployContractWithoutABI]Failed to deploy contract\n[Value]{Value}\n[GasLimit]{GasLimit}\n[Bytecode]{Bytecode}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Account][DeployContractWithoutABI]Failed\n[Value]{Value}\n[GasLimit]{GasLimit}\n[Bytecode]{Bytecode}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None def SignMessage(self, Message: str) -> dict: @@ -524,22 +593,24 @@ def SignMessage(self, Message: str) -> dict: SignatureData (str): 签名数据构成的字典。当出现异常时返回 None 。 {"Address"|"Message"|"MessageHash"|"Signature"|"R"|"S"|"V"} """ - from eth_account.messages import encode_defunct + try: - Temp = encode_defunct(text=Message) - SignedMessage = EthAccount.sign_message(Temp, private_key=self.PrivateKey) + from eth_account.messages import encode_defunct + SignedMessage = self.EthAccount.sign_message(encode_defunct(text=Message)) MessageHash = SignedMessage.messageHash.hex() Signature = SignedMessage.signature.hex() R = hex(SignedMessage.r) S = hex(SignedMessage.s) V = hex(SignedMessage.v) logger.success( - f"\n[Account][SignMessage]\n[Address]{self.Address}\n[Message]{Message}\n[MessageHash]{MessageHash}\n[Signature]{Signature}\n[R]{R}\n[S]{S}\n[V]{V}\n{'-'*80}") - return {"Address": self.Address, "Message": Message, "MessageHash": MessageHash, "Signature": Signature, "R": R, "S": S, "V": V} + f"\n[Account][SignMessage]\n[Address]{self.EthAccount.address}\n[Message]{Message}\n[MessageHash]{MessageHash}\n[Signature]{Signature}\n[R]{R}\n[S]{S}\n[V]{V}\n{'-'*80}" + ) + return {"Address": self.EthAccount.address, "Message": Message, "MessageHash": MessageHash, "Signature": Signature, "R": R, "S": S, "V": V} except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Account][SignMessage]Failed to sign message\n[Address]{self.Address}\n[Message]{Message}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Account][SignMessage]Failed to sign message\n[Address]{self.EthAccount.address}\n[Message]{Message}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None def SignMessageHash(self, MessageHash: str) -> dict: @@ -553,19 +624,22 @@ def SignMessageHash(self, MessageHash: str) -> dict: SignatureData (str): 签名数据构成的字典。当出现异常时返回 None 。 {"Address"|"MessageHash"|"Signature"|"R"|"S"|"V"} """ + try: - SignedMessage = EthAccount.signHash(MessageHash, self.PrivateKey) + SignedMessage = self.EthAccount.signHash(MessageHash) Signature = SignedMessage.signature.hex() R = hex(SignedMessage.r) S = hex(SignedMessage.s) V = hex(SignedMessage.v) logger.success( - f"\n[Account][SignMessageHash]\n[Address]{self.Address}\n[MessageHash]{MessageHash}\n[Signature]{Signature}\n[R]{R}\n[S]{S}\n[V]{V}\n{'-'*80}") - return {"Address": self.Address, "MessageHash": MessageHash, "Signature": Signature, "R": R, "S": S, "V": V} + f"\n[Account][SignMessageHash]\n[Address]{self.EthAccount.address}\n[MessageHash]{MessageHash}\n[Signature]{Signature}\n[R]{R}\n[S]{S}\n[V]{V}\n{'-'*80}" + ) + return {"Address": self.EthAccount.address, "MessageHash": MessageHash, "Signature": Signature, "R": R, "S": S, "V": V} except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Account][SignMessageHash]Failed to sign message hash\n[Address]{self.Address}\n[MessageHash]{MessageHash}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Account][SignMessageHash]Failed\n[Address]{self.EthAccount.address}\n[MessageHash]{MessageHash}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @@ -576,7 +650,7 @@ class Contract(): def __init__(self, Account: Account, Address: str, ABI: dict): """ - 初始化。通过合约地址与 ABI 来实例化合约，并与 Account 绑定，后续所有对该合约的调用都会由这一账户发起。当实例化失败时会抛出异常。 + 初始化。通过合约地址与 ABI 来实例化合约，并与 Account 绑定，后续所有对该合约的调用都会由这一账户发起。当合约实例化失败时会抛出异常。 参数： Account (Poseidon.Blockchain.Account): 账户实例 @@ -584,38 +658,41 @@ def __init__(self, Account: Account, Address: str, ABI: dict): ABI (str): 合约 ABI 成员变量： - Account (Poseidon.Blockchain.Account): 账户实例 - Address (str): 合约地址 Instance (Web3.eth.Contract): web3.py 原生 contract 对象实例 + Address (str): 合约地址 """ + try: - self.Account = Account - self.Net = Account.Net + self._Account = Account + self._Eth = Account._Eth self.Address = Web3.toChecksumAddress(Address) - self.Instance = self.Net.eth.contract(address=self.Address, abi=ABI) + self.Instance = self._Eth.contract(address=self.Address, abi=ABI) logger.success(f"\n[Contract][Initialize]Successfully instantiated contract [{self.Address}]\n{'-'*80}") except Exception: - ExceptionInformation = exc_info() - logger.error(f"\n[Contract][Initialize]Failed to instantiated contract [{self.Address}]\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + ExceptionInformation = format_exc() + logger.error( + f"\n[Contract][Initialize]Failed to instantiated contract [{self.Address}]\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) raise Exception("Failed to instantiate contract.") - def CallFunction(self, FunctionName: str, *FunctionArguments) -> dict: + def CallFunction(self, FunctionName: str, *FunctionArguments: Optional[Any]) -> dict: """ 通过传入函数名及参数来调用该合约内的函数。 参数： FunctionName (str): 函数名称 - *FunctionArguments (可选)(any): 函数参数，默认为空。 + *FunctionArguments (Optional[Any]): 函数参数，默认为空。 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 """ - TransactionData = self.Instance.functions[FunctionName](*FunctionArguments).buildTransaction({"gasPrice": self.Net.eth.gas_price}) + + TransactionData = self.Instance.functions[FunctionName](*FunctionArguments).buildTransaction({"gasPrice": self._Eth.gas_price}) logger.info(f"\n[Contract][CallFunction]\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n{'-'*80}") - TransactionInformation = self.Account.SendTransaction(self.Address, TransactionData["data"], TransactionData["value"], TransactionData["gas"]) + TransactionInformation = self._Account.SendTransaction(self.Address, TransactionData["data"], TransactionData["value"], TransactionData["gas"]) return TransactionInformation - def CallFunctionWithValueAndGasLimit(self, Value: int, GasLimit: int, FunctionName: str, *FunctionArguments) -> dict: + def CallFunctionWithValueAndGasLimit(self, Value: int, GasLimit: int, FunctionName: str, *FunctionArguments: Optional[Any]) -> dict: """ 通过传入函数名及参数来调用该合约内的函数。支持自定义 Value 和 GasLimit 。 @@ -623,114 +700,129 @@ def CallFunctionWithValueAndGasLimit(self, Value: int, GasLimit: int, FunctionNa Value (int): 随交易发送的网络原生代币数量。单位为 wei 。 GasLimit (int): Gas 最大使用量。单位为 wei 。 FunctionName (str): 函数名称 - *FunctionArguments (可选)(any): 函数参数，默认为空。 + *FunctionArguments (Optional[Any]): 函数参数，默认为空。 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 """ - TransactionData = self.Instance.functions[FunctionName](*FunctionArguments).buildTransaction({"gasPrice": self.Net.eth.gas_price, "gas": GasLimit, "value": Value}) + + TransactionData = self.Instance.functions[FunctionName](*FunctionArguments).buildTransaction({"gasPrice": self._Eth.gas_price, "gas": GasLimit, "value": Value}) logger.info( - f"\n[Contract][CallFunctionWithValueAndGasLimit]\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[Value]{TransactionData['value']} [GasLimit]{TransactionData['gas']}\n{'-'*80}") - TransactionInformation = self.Account.SendTransaction(self.Address, TransactionData["data"], TransactionData["value"], TransactionData["gas"]) + f"\n[Contract][CallFunctionWithValueAndGasLimit]\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[Value]{TransactionData['value']} [GasLimit]{TransactionData['gas']}\n{'-'*80}" + ) + TransactionInformation = self._Account.SendTransaction(self.Address, TransactionData["data"], TransactionData["value"], TransactionData["gas"]) return TransactionInformation - def ReadOnlyCallFunction(self, FunctionName: str, *FunctionArguments): + def ReadOnlyCallFunction(self, FunctionName: str, *FunctionArguments: Optional[Any]) -> Any: """ 通过传入函数名及参数来调用该合约内的只读函数。 参数： FunctionName (str): 函数名称 - *FunctionArguments (可选)(any): 函数参数，默认为空。 + *FunctionArguments (Optional[Any]): 函数参数，默认为空。 返回值： - Result (any): 调用函数后得到的返回值。当出现异常时返回 None 。 + Result (Any): 调用函数后得到的返回值。当出现异常时返回 None 。 """ + try: Result = self.Instance.functions[FunctionName](*FunctionArguments).call() - logger.success(f"\n[Contract][ReadOnlyCallFunction]\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[Result]{Result}\n{'-'*80}") + logger.success( + f"\n[Contract][ReadOnlyCallFunction]\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[Result]{Result}\n{'-'*80}" + ) return Result except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Contract][ReadOnlyCallFunction]Failed to call readonly function\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Contract][ReadOnlyCallFunction]Failed\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None - def EncodeABI(self, FunctionName: str, *FunctionArguments) -> str: + def EncodeABI(self, FunctionName: str, *FunctionArguments: Optional[Any]) -> str: """ 通过传入函数名及参数进行编码，相当于生成调用该函数的 CallData 。 参数： FunctionName (str): 函数名称 - *FunctionArguments (可选)(any): 函数参数，默认为空。 + *FunctionArguments (Optional[Any]): 函数参数，默认为空。 返回值： CallData (str): 调用数据编码。含 0x 前缀的十六进制形式。当出现异常时返回 None 。 """ + try: CallData = self.Instance.encodeABI(fn_name=FunctionName, args=FunctionArguments) - logger.success(f"\n[Contract][EncodeABI]\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[CallData]{CallData}\n{'-'*80}") + logger.success( + f"\n[Contract][EncodeABI]\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[CallData]{CallData}\n{'-'*80}" + ) return CallData except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[Contract][EncodeABI]Failed to encode abi\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[Contract][EncodeABI]Failed\n[ContractAddress]{self.Address}\n[Function]{FunctionName}{FunctionArguments}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None class BlockchainUtils(): """ - 通用工具集，整合了链下使用的常用功能。 + 通用工具集，整合了常用的链下操作。静态类，无需实例化。 """ @staticmethod def SwitchSolidityVersion(SolidityVersion: str): """ - 设置当前使用的 Solidity 版本，若该版本未安装则会自动安装。当设置版本失败时会抛出异常。 + 设置当前使用的 Solidity 版本，若该版本未安装则会自动安装。 参数： SolidityVersion (str): Solidity 版本号 """ - from solcx import install_solc, set_solc_version + try: + from solcx import install_solc, set_solc_version install_solc(SolidityVersion) set_solc_version(SolidityVersion) logger.success(f"\n[BlockchainUtils][SwitchSolidityVersion]Current Solidity Version [{SolidityVersion}]\n{'-'*80}") except Exception: - ExceptionInformation = exc_info() - logger.error(f"\n[BlockchainUtils][SwitchSolidityVersion]Failed to switch to version [{SolidityVersion}]\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") - raise Exception("Failed to switch solidity version.") + ExceptionInformation = format_exc() + logger.error( + f"\n[BlockchainUtils][SwitchSolidityVersion]Failed to switch to version [{SolidityVersion}]\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) @staticmethod - def Compile(FileCourse: str, ContractName: str, SolidityVersion: str = None, AllowPaths: str = None, Optimize: bool = False) -> tuple: + def Compile(FileCourse: str, ContractName: str, SolidityVersion: Optional[str] = None, AllowPaths: Optional[str] = None, Optimize: Optional[bool] = False) -> tuple: """ 根据给定的参数使用 py-solc-x 编译合约。当编译失败时会抛出异常。 参数： FileCourse (str): 合约文件完整路径。当合约文件与脚本文件在同一目录下时可直接使用文件名。 ContractName (str): 要编译的合约名称 - SolidityVersion (可选)(str): 指定使用的 Solidity 版本。若不指定则会使用当前已激活的 Solidity 版本进行编译。默认为 None 。 - AllowPaths (可选)(str): 指定许可路径。在编译时可能会出现 AllowPaths 相关错误可在这里解决。默认为 None 。 - Optimize (可选)(str): 是否开启优化器。默认为 False 。 + SolidityVersion (Optional[str]): 指定使用的 Solidity 版本。若不指定则会使用当前已激活的 Solidity 版本进行编译。默认为 None 。 + AllowPaths (Optional[str]): 指定许可路径。在编译时可能会出现 AllowPaths 相关错误可在这里解决。默认为 None 。 + Optimize (Optional[bool]): 是否开启优化器。默认为 False 。 返回值： (ABI, Bytecode) (tuple): 由 ABI 和 Bytecode 组成的元组 """ - from solcx import compile_source - from json import dump + try: + from solcx import compile_source with open(FileCourse, "r", encoding="utf-8") as sol: CompiledSol = compile_source(sol.read(), solc_version=SolidityVersion, allow_paths=AllowPaths, optimize=Optimize, output_values=['abi', 'bin']) ContractData = CompiledSol[f':{ContractName}'] ABI = ContractData['abi'] Bytecode = ContractData['bin'] - with open(f'{ContractName}_ABI.json', 'w') as f: + with open(f'{ContractName}_ABI.json', 'w', encoding="utf-8") as f: dump(ABI, f, indent=4) - logger.success(f"\n[BlockchainUtils][Compile]\n[FileCourse]{FileCourse}\n[ContractName]{ContractName}\n[ABI]{ABI}\n[Bytecode]{Bytecode}\n{'-'*80}") + logger.success( + f"\n[BlockchainUtils][Compile]\n[FileCourse]{FileCourse}\n[ContractName]{ContractName}\n[ABI]{ABI}\n[Bytecode]{Bytecode}\n{'-'*80}" + ) return (ABI, Bytecode) except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][Compile]Failed to compile the contract\n[FileCourse]{FileCourse}\n[ContractName]{ContractName}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][Compile]Failed\n[FileCourse]{FileCourse}\n[ContractName]{ContractName}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) raise Exception("Failed to compile the contract.") @staticmethod @@ -741,6 +833,7 @@ def CreateNewAccount() -> tuple: 返回值： (Address, PrivateKey) (tuple): 由账户地址和私钥组成的元组 """ + Temp = EthAccount.create() Address = Web3.toChecksumAddress(Temp.address) PrivateKey = Temp.privateKey.hex() @@ -758,17 +851,21 @@ def MnemonicToAddressAndPrivateKey(Mnemonic: str) -> tuple: 返回值： (Address, PrivateKey) (tuple): 由账户地址和私钥组成的元组。当出现异常时返回 None 。 """ + try: EthAccount.enable_unaudited_hdwallet_features() Temp = EthAccount.from_mnemonic(Mnemonic) Address = Web3.toChecksumAddress(Temp.address) PrivateKey = Temp.privateKey.hex() - logger.success(f"\n[BlockchainUtils][MnemonicToAddressAndPrivateKey]\n[Mnemonic]{Mnemonic}\n[Address]{Address}\n[PrivateKey]{PrivateKey}\n{'-'*80}") + logger.success( + f"\n[BlockchainUtils][MnemonicToAddressAndPrivateKey]\n[Mnemonic]{Mnemonic}\n[Address]{Address}\n[PrivateKey]{PrivateKey}\n{'-'*80}" + ) return (Address, PrivateKey) except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][MnemonicToAddressAndPrivateKey]Failed to convert mnemonic to address and private key\n[Mnemonic]{Mnemonic}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][MnemonicToAddressAndPrivateKey]Failed\n[Mnemonic]{Mnemonic}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @staticmethod @@ -783,16 +880,17 @@ def RecoverMessage(Message: str, Signature: str) -> str: 返回值： Signer (str): 签署者的账户地址。当出现异常时返回 None 。 """ - from eth_account.messages import encode_defunct + try: - Temp = encode_defunct(text=Message) - Signer = EthAccount.recover_message(Temp, signature=Signature) + from eth_account.messages import encode_defunct + Signer = EthAccount.recover_message(encode_defunct(text=Message), signature=Signature) logger.success(f"\n[BlockchainUtils][RecoverMessage]\n[Message]{Message}\n[Signature]{Signature}\n[Signer]{Signer}\n{'-'*80}") return Signer except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][RecoverMessage]Failed to recover message\n[Message]{Message}\n[Signature]{Signature}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][RecoverMessage]Failed\n[Message]{Message}\n[Signature]{Signature}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @staticmethod @@ -807,14 +905,18 @@ def RecoverMessageHash(MessageHash: str, Signature: str) -> str: 返回值： Signer (str): 签署者的账户地址。当出现异常时返回 None 。 """ + try: Signer = EthAccount.recoverHash(MessageHash, signature=Signature) - logger.success(f"\n[BlockchainUtils][RecoverMessageByHash]\n[MessageHash]{MessageHash}\n[Signature]{Signature}\n[Signer]{Signer}\n{'-'*80}") + logger.success( + f"\n[BlockchainUtils][RecoverMessageByHash]\n[MessageHash]{MessageHash}\n[Signature]{Signature}\n[Signer]{Signer}\n{'-'*80}" + ) return Signer except Exception: - ExceptionInformation = exc_info() + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][RecoverMessageByHash]Failed to recover message hash\n[MessageHash]{MessageHash}\n[Signature]{Signature}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][RecoverMessageByHash]Failed\n[MessageHash]{MessageHash}\n[Signature]{Signature}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @staticmethod @@ -828,18 +930,20 @@ def RecoverRawTransaction(RawTransactionData: str) -> str: 返回值： Address (str): 账户地址。当出现异常时返回 None 。 """ + try: Address = EthAccount.recover_transaction(RawTransactionData) logger.success(f"\n[BlockchainUtils][RecoverRawTransaction]\n[RawTransactionData]{RawTransactionData}\n[Address]{Address}\n{'-'*80}") return Address - except: - ExceptionInformation = exc_info() + except Exception: + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][RecoverRawTransaction]Failed to recover raw transaction\n[RawTransactionData]{RawTransactionData}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][RecoverRawTransaction]Failed\n[RawTransactionData]{RawTransactionData}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @staticmethod - def CrackSelector(SourceFunctionName: str, SourceFunctionParameters: list, ToGenerateFunctionParameters: list) -> str: + def CrackSelector(SourceFunctionName: str, SourceFunctionParameters: List[str], ToGenerateFunctionParameters: List[str]) -> str: """ 根据源函数名、参数与想要碰撞生成的函数的参数，碰撞生成出一个函数名，以使得这两个函数的选择器签名相等。 @@ -857,21 +961,26 @@ def CrackSelector(SourceFunctionName: str, SourceFunctionParameters: list, ToGen def Crack(SourceFunctionSelector: str, Temp: str) -> str: Charset = "0123456789abcdef" X = mbruteforce(lambda x: Web3.keccak(f"function_{x}({Temp})".encode())[:4].hex() == SourceFunctionSelector, Charset, 8, method='fixed') - return f"function_{x}({Temp})" + return f"function_{X}({Temp})" + try: SourceFunctionSelector = Web3.keccak(f"{SourceFunctionName}({','.join(SourceFunctionParameters)})".encode())[:4].hex() Temp = ','.join(ToGenerateFunctionParameters) logger.info( - f"\n[BlockchainUtils][CrackSelector]\n[SourceFunction]{SourceFunctionName}({','.join(SourceFunctionParameters)})\n[SourceFunctionSelector]{SourceFunctionSelector}\n[ToGenerateFunction]function_?({Temp})\nCrack start...") + f"\n[BlockchainUtils][CrackSelector]\n[SourceFunction]{SourceFunctionName}({','.join(SourceFunctionParameters)})\n[SourceFunctionSelector]{SourceFunctionSelector}\n[ToGenerateFunction]function_?({Temp})\nCrack start..." + ) ToGenerateFunction = Crack(SourceFunctionSelector, Temp) ToGenerateFunctionSelector = Web3.keccak(ToGenerateFunction.encode())[:4].hex() - logger.success(f"\n[BlockchainUtils][CrackSelector]\n[ToGenerateFunction]{ToGenerateFunction}\n[ToGenerateSelector]{ToGenerateSelector}\n{'-'*80}") + logger.success( + f"\n[BlockchainUtils][CrackSelector]\n[ToGenerateFunction]{ToGenerateFunction}\n[ToGenerateFunctionSelector]{ToGenerateFunctionSelector}\n{'-'*80}" + ) return ToGenerateFunction - except: - ExceptionInformation = exc_info() + except Exception: + ExceptionInformation = format_exc() Temp = ','.join(ToGenerateFunctionParameters) logger.error( - f"\n[BlockchainUtils][CrackSelector]Failed to crack selector\n[SourceFunction]{SourceFunctionName}({','.join(SourceFunctionParameters)})\n[ToGenerateFunction]{f'function_?({Temp})'}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][CrackSelector]Failed\n[SourceFunction]{SourceFunctionName}({','.join(SourceFunctionParameters)})\n[ToGenerateFunction]{f'function_?({Temp})'}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @staticmethod @@ -885,15 +994,17 @@ def AssemblyToBytecode(Assembly: str) -> str: 返回值： Bytecode (str): EVM Bytecode 。含 0x 前缀的六进制形式。当出现异常时返回 None 。 """ + try: from pyevmasm import assemble_hex Bytecode = assemble_hex(Assembly) logger.success(f"\n[BlockchainUtils][AssemblyToBytecode]\n[Bytecode]{Bytecode}\n{'-'*80}") return Bytecode - except: - ExceptionInformation = exc_info() + except Exception: + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][AssemblyToBytecod]Failed to transform assembly to bytecode\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][AssemblyToBytecod]Failed\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @staticmethod @@ -907,15 +1018,17 @@ def BytecodeToAssembly(Bytecode: str) -> str: 返回值： Assembly (str): EVM Assembly 。当出现异常时返回 None 。 """ + try: from pyevmasm import disassemble_hex Assembly = disassemble_hex(Bytecode) logger.success(f"\n[BlockchainUtils][AssemblyToBytecode]\n[Assembly]\n{Assembly}\n{'-'*80}") return Assembly - except: - ExceptionInformation = exc_info() + except Exception: + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][BytecodeToAssembly]Failed to transform bytecode to assembly\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][BytecodeToAssembly]Failed\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @staticmethod @@ -930,19 +1043,21 @@ def SignatureToRSV(Signature: str) -> dict: Result (dict): 解析结果。当出现异常时返回 None 。 {"Signature"|"R"|"S"|"V"} """ + try: Signature = hex(int(Signature, 16)) - assert (len(Signature) == 132) + assert (len(Signature) == 130 + 2) R = '0x' + Signature[2:66] S = '0x' + Signature[66:-2] V = '0x' + Signature[-2:] logger.success(f"\n[BlockchainUtils][SignatureToRSV]\n[Signature]{Signature}\n[R]{R}\n[S]{S}\n[V]{V}\n{'-'*80}") Result = {"Signature": Signature, "R": R, "S": S, "V": V} return Result - except: - ExceptionInformation = exc_info() + except Exception: + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][SignatureToRSV]Failed to transform signature to rsv\n[Signature]{Signature}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][SignatureToRSV]Failed\n[Signature]{Signature}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None @staticmethod @@ -959,6 +1074,7 @@ def RSVToSignature(R: str, S: str, V: str) -> dict: Result (dict): 合并结果。当出现异常时返回 None 。 {"R"|"S"|"V"|"Signature"} """ + try: R = hex(int(R, 16)) S = hex(int(S, 16)) @@ -968,8 +1084,9 @@ def RSVToSignature(R: str, S: str, V: str) -> dict: logger.success(f"\n[BlockchainUtils][RSVToSignature]\n[R]{R}\n[S]{S}\n[V]{V}\n[Signature]{Signature}\n{'-'*80}") Result = {"R": R, "S": S, "V": V, "Signature": Signature} return Result - except: - ExceptionInformation = exc_info() + except Exception: + ExceptionInformation = format_exc() logger.error( - f"\n[BlockchainUtils][RSVToSignature]Failed to transform rsv to signature\n[R]{R}\n[S]{S}\n[V]{V}\n[ExceptionInformation]{ExceptionInformation}\n{'-'*80}") + f"\n[BlockchainUtils][RSVToSignature]Failed\n[R]{R}\n[S]{S}\n[V]{V}\n[ExceptionInformation]{ExceptionInformation}{'-'*80}" + ) return None diff --git a/Poseidon/Cryptography.py b/Poseidon/Cryptography.py index 9f97783..b069b1b 100644 --- a/Poseidon/Cryptography.py +++ b/Poseidon/Cryptography.py @@ -6,11 +6,12 @@ import base64 from Crypto.Cipher import AES as aes from Crypto.Util.number import long_to_bytes, bytes_to_long +from typing import Optional, List class MiscUtils(): """ - 本模块用于处理进制转换和常用哈希。 + 本模块用于处理进制转换和常用哈希。静态类，无需实例化。 """ @staticmethod @@ -24,6 +25,7 @@ def Binary_String(Binary: str) -> str: 返回值： String (str): 转换得到的字符串 """ + if len(Binary) % 8 != 0: raise Exception("二进制字符串的长度应该为8的倍数") String = "".join([chr(int(Binary[i:i + 8], 2)) for i in range(0, len(Binary), 8)]) @@ -40,6 +42,7 @@ def Binary_Dec(Binary: str) -> int: 返回值： Dec (int): 转换得到的十进制整数 """ + Dec = int(Binary, 2) return Dec @@ -54,6 +57,7 @@ def Binary_Hex(Binary: str) -> str: 返回值： Hex (str): 转换得到的十六进制字符串 """ + Hex = hex(int(Binary, 2)) return Hex @@ -68,6 +72,7 @@ def Dec_String(Dec: int) -> str: 返回值： String (str): 转换得到的字符串 """ + from Crypto.Util.number import long_to_bytes String = long_to_bytes(Dec).decode() return String @@ -83,6 +88,7 @@ def Dec_Binary(Dec: int) -> str: 返回值： Binary (str): 转换得到的二进制字符串 """ + Binary = bin(Dec) return Binary @@ -97,6 +103,7 @@ def Dec_Hex(Dec: int) -> str: 返回值： Hex (str): 转换得到的十六进制字符串 """ + Hex = hex(Dec) return Hex @@ -111,6 +118,7 @@ def Hex_String(Hex: str) -> str: 返回值： String (str): 转换得到的字符串 """ + if len(Hex) % 2 != 0: raise Exception("十六进制字符串的长度应该为2的倍数") String = "".join([chr(int(Hex[i:i + 2], 16)) for i in range(0, len(Hex), 2)]) @@ -127,6 +135,7 @@ def Hex_Binary(Hex: str) -> str: 返回值： Binary (str): 转换得到的二进制字符串 """ + Binary = bin(int(Hex, 16)) return Binary @@ -141,6 +150,7 @@ def Hex_Dec(Hex: str) -> int: 返回值： Dec (int): 转换得到的十进制整数 """ + Dec = int(Hex, 16) return Dec @@ -155,6 +165,7 @@ def SHA1(Text: str) -> str: 返回值： Hash (str): 该字符串的 SHA1 哈希值（十六进制字符串，不含 0x 前缀） """ + from hashlib import sha1 Hash = sha1(Text.encode()).hexdigest() return Hash @@ -170,6 +181,7 @@ def SHA256(Text: str) -> str: 返回值： Hash (str): 该字符串的 SHA256 哈希值（十六进制字符串，不含 0x 前缀） """ + from hashlib import sha256 Hash = sha256(Text.encode()).hexdigest() return Hash @@ -185,6 +197,7 @@ def SHA512(Text: str) -> str: 返回值： Hash (str): 该字符串的 SHA512 哈希值（十六进制字符串，不含 0x 前缀） """ + from hashlib import sha512 Hash = sha512(Text.encode()).hexdigest() return Hash @@ -200,6 +213,7 @@ def MD5(Text: str) -> str: 返回值： Hash (str): 该字符串的 MD5 哈希值（十六进制字符串，不含 0x 前缀） """ + from hashlib import md5 Hash = md5(Text.encode()).hexdigest() return Hash @@ -221,6 +235,7 @@ def Base64_Encrypt(Text: str) -> str: 返回值： EncryptedText (str): Base64 编码后的字符串 """ + EncryptedText = base64.b64encode(Text.encode()).decode() return EncryptedText @@ -235,6 +250,7 @@ def Base64_Decrypt(Text: str) -> str: 返回值： DecryptedText (str): Base64 解码后的字符串 """ + DecryptedText = base64.b64decode(Text.encode()).decode() return DecryptedText @@ -249,6 +265,7 @@ def Base32_Encrypt(Text: str) -> str: 返回值： EncryptedText (str): Base32 编码后的字符串 """ + EncryptedText = base64.b32encode(Text.upper().encode()).decode() return EncryptedText @@ -263,6 +280,7 @@ def Base32_Decrypt(Text: str) -> str: 返回值： DecryptedText (str): Base32 解码后的字符串 """ + DecryptedText = base64.b32decode(Text.upper().encode()).decode() return DecryptedText @@ -277,6 +295,7 @@ def Base16_Encrypt(Text: str) -> str: 返回值： EncryptedText (str): Base16 编码后的字符串 """ + EncryptedText = base64.b16encode(Text.upper().encode()).decode() return EncryptedText @@ -291,6 +310,7 @@ def Base16_Decrypt(Text: str) -> str: 返回值： DecryptedText (str): Base16 解码后的字符串 """ + DecryptedText = base64.b16decode(Text.upper().encode()).decode() return DecryptedText @@ -306,6 +326,7 @@ def AES_Padding(Text: str, BlockSize: int = 16) -> bytes: 返回值： Fill (bytes): padding 后的字节数据 """ + Fill = Text.encode() while len(Fill) % BlockSize != 0: Fill += b'\x00' @@ -324,6 +345,7 @@ def AES_Encrypt(Text: str, Key: str, BlockSize: int = 16) -> str: 返回值： EncryptedText (str): AES 加密后的密文（Base64 编码形式） """ + AES = aes.new(ModernCryptoUtils.AES_Padding(Key, BlockSize), aes.MODE_ECB) EncryptedText = base64.b64encode(AES.encrypt(ModernCryptoUtils.AES_Padding(Text, BlockSize))).decode() return EncryptedText @@ -341,6 +363,7 @@ def AES_Decrypt(Text: str, Key: str, BlockSize: int = 16) -> str: 返回值： DecryptedText (str): AES 解密后得到的原文 """ + AES = aes.new(ModernCryptoUtils.AES_Padding(Key, BlockSize), aes.MODE_ECB) DecryptedText = AES.decrypt(base64.b64decode(ModernCryptoUtils.AES_Padding(Text, BlockSize))).decode() return DecryptedText @@ -359,6 +382,7 @@ def RSA_Encrypt(Text: str, p: int, q: int, e: int) -> str: 返回值： EncryptedText (str): RSA 加密后的密文（ Base64 编码形式） """ + m = bytes_to_long(Text.encode()) c = gmpy2.powmod(m, e, p * q) EncryptedText = base64.b64encode(long_to_bytes(c)).decode() @@ -378,6 +402,7 @@ def RSA_Base64_Decrypt(Base64Text: str, p: int, q: int, e: int) -> str: 返回值： DecryptedText (str): RSA 解密后得到的原文 """ + c = bytes_to_long(base64.b64decode(Base64Text.encode())) d = gmpy2.invert(e, (p - 1) * (q - 1)) m = gmpy2.powmod(c, d, p * q) @@ -398,6 +423,7 @@ def RSA_Long_Decrypt(Long: int, p: int, q: int, e: int) -> str: 返回值： DecryptedText (str): RSA 解密后得到的原文 """ + d = gmpy2.invert(e, (p - 1) * (q - 1)) m = gmpy2.powmod(Long, d, p * q) DecryptedText = long_to_bytes(m).decode() @@ -416,6 +442,7 @@ def RSA_Wiener_Attack(c: int, e: int, n: int) -> str: 返回值： m (str): RSA 维纳攻击后得到的原文 """ + def continuedFra(x, y): """ 计算连分数 @@ -427,6 +454,7 @@ def continuedFra(x, y): 返回值： cf: 连分数列表 """ + cf = [] while y: cf.append(x // y) @@ -443,6 +471,7 @@ def gradualFra(cf): 返回值： numerator, denominator: 该列表最后的渐近分数（这里的渐进分数分子分母要分开） """ + numerator = 0 denominator = 1 for x in cf[::-1]: @@ -461,6 +490,7 @@ def solve_pq(a, b, c): 返回值： p, q: p 和 q 的值 """ + par = gmpy2.isqrt(b * b - 4 * a * c) return (-b + par) // (2 * a), (-b - par) // (2 * a) @@ -474,6 +504,7 @@ def getGradualFra(cf): 返回值： gf: 该列表所有的渐近分数 """ + gf = [] for i in range(1, len(cf) + 1): gf.append(gradualFra(cf[:i])) @@ -493,7 +524,7 @@ def getGradualFra(cf): return m @staticmethod - def RSA_MultiPrime_Attack(c: int, e: int, n: int, primes: list, powers: list = None) -> str: + def RSA_MultiPrime_Attack(c: int, e: int, n: int, primes: List[int], powers: Optional[List[int]] = None) -> str: """ 用于对长整数形式的 RSA 密文进行多素数攻击并解出原文。 @@ -502,11 +533,12 @@ def RSA_MultiPrime_Attack(c: int, e: int, n: int, primes: list, powers: list = N e (int): e 值 n (int): n 值 primes (List[int]): 用于攻击的多素数列表 - powers (可选)(List[int]): 各素数对应的阶数，默认均为 1 次方 + powers (Optional[List[int]]): 各素数对应的阶数，默认均为 1 次方 返回值： m (str): RSA 多素数攻击后得到的原文 """ + from operator import mul from functools import reduce if powers == None: @@ -532,6 +564,7 @@ def RSA_LowEncryptionIndex_Attack(c: int, e: int, n: int) -> str: 返回值： m (str): RSA 低加密指数攻击后得到的原文 """ + k = 0 while k < 10**5: m = gmpy2.iroot(c + k * n, e)[0] @@ -558,12 +591,13 @@ def RSA_CommonMod_Attack(c1: int, c2: int, e1: int, e2: int, n: int) -> str: 返回值： m (str): RSA 共模攻击后得到的原文 """ + _, x, y = gmpy2.gcdext(e1, e2) m = long_to_bytes((gmpy2.powmod(c1, x, n) * gmpy2.powmod(c2, y, n)) % n).decode() return m @staticmethod - def RSA_Broadcast_Attack(cs: list, e: int, ns: list) -> str: + def RSA_Broadcast_Attack(cs: List[int], e: int, ns: List[int]) -> str: """ 用于对长整数形式的 RSA 密文列表进行广播攻击并解出原文。 @@ -575,6 +609,7 @@ def RSA_Broadcast_Attack(cs: list, e: int, ns: list) -> str: 返回值： m (str): RSA 广播攻击后得到的原文 """ + from functools import reduce if len(cs) != len(ns): raise Exception("密文列表长度与 n 值列表长度不一致") @@ -599,6 +634,7 @@ def RC4_Encrypt(Text: str, Key: str) -> str: 返回值： EncryptedText (str): RC4 加密后得到的密文（ Base64 编码形式） """ + from Crypto.Cipher import ARC4 RC4 = ARC4.new(Key.encode()) EncryptedText = base64.b64encode(RC4.encrypt(Text.encode())).decode() @@ -616,6 +652,7 @@ def RC4_Decrypt(Text: str, Key: str) -> str: 返回值： DecryptedText (str): RC4 解密后得到的原文 """ + from Crypto.Cipher import ARC4 RC4 = ARC4.new(Key.encode()) DecryptedText = RC4.decrypt(base64.b64decode(Text.encode())).decode() @@ -634,11 +671,12 @@ def Caesar_Encrypt(Text: str, Move: int = 3) -> str: 参数： Text (str): 待进行 Caesar 加密的字符串 - Move (int): 移位位数，默认为 3 + Move (可选)(int): 移位位数，默认为 3 返回值： EncryptedText (str): Caesar 加密后得到的密文 """ + EncryptedText = "" for i in Text: if i.isupper(): @@ -656,11 +694,12 @@ def Caesar_Decrypt(Text: str, Move: int = 3) -> str: 参数： Text (str): 待进行解密的 Caesar 密文 - Move (int): 移位位数，默认为 3 + Move (可选)(int): 移位位数，默认为 3 返回值： DecryptedText (str): Caesar 解密后得到的原文 """ + DecryptedText = "" for i in Text: if i.isupper(): @@ -672,7 +711,7 @@ def Caesar_Decrypt(Text: str, Move: int = 3) -> str: return DecryptedText @staticmethod - def Caesar_Attack(Text: str) -> list: + def Caesar_Attack(Text: str) -> List[str]: """ 用于对 Caesar 密文进行爆破攻击。 @@ -682,6 +721,7 @@ def Caesar_Attack(Text: str) -> list: 返回值： Result (List[str]): Caesar 爆破攻击后得到的字符串列表 """ + Result = [ClassicalCryptoUtils.Caesar_Decrypt(Text, i) for i in range(1, 27)] return Result @@ -696,6 +736,7 @@ def Morse_Encrypt(Text: str) -> str: 返回值： EncryptedText (str): Morse 加密后得到的密文（未找到映射关系的字符将保持不变） """ + MorseCode = { "A": ".-", "B": "-...", "C": "-.-.", "D": "-..", "E": ".", "F": "..-.", "G": "--.", "H": "....", "I": "..", "J": ".---", "K": "-.-", "L": ".-..", "M": "--", @@ -725,6 +766,7 @@ def Morse_Decrypt(Text: str) -> str: 返回值： DecryptedText (str): Morse 解密后得到的原文（未找到映射关系的字符将保持不变） """ + MorseCode = { ".-": "a", "-...": "b", "-.-.": "c", "-..": "d", ".": "e", "..-.": "f", "--.": "g", "....": "h", "..": "i", ".---": "j", "-.-": "k", ".-..": "l", @@ -750,6 +792,7 @@ def Bacon_Encrypt(Text: str) -> str: 返回值： EncryptedText (str): Bacon 加密后得到的密文（大写形式 未找到映射关系的字符将以[]包裹） """ + BaconCode = { "a": "aaaaa", "b": "aaaab", "c": "aaaba", "d": "aaabb", "e": "aabaa", "f": "aabab", "g": "aabba", "h": "aabbb", "i": "abaaa", "j": "abaab", "k": "ababa", "l": "ababb", "m": "abbaa", "n": "abbab", @@ -770,6 +813,7 @@ def Bacon_Decrypt(Text: str) -> str: 返回值： DecryptedText (str): Bacon 解密后得到的原文（大写形式 未找到映射关系的字符将以[]包裹） """ + if len(Text) % 5 != 0: raise Exception("Bacon 密文应该为 5 的倍数") BaconCode = { @@ -794,6 +838,7 @@ def Fence_Encrypt(Text: str, Fence: int) -> str: 返回值： EncryptedText (str): Fence 加密后得到的密文 """ + if len(Text) % Fence != 0: raise Exception("字符串长度应该为栏数的倍数") templist = [] @@ -818,6 +863,7 @@ def Fence_Decrypt(Text: str, Fence: int) -> str: 返回值： DecryptedText (str): Fence 解密后得到的原文 """ + if len(Text) % Fence != 0: raise Exception("字符串长度应该为栏数的倍数") templist = ["" for _ in range(Fence)] @@ -829,7 +875,7 @@ def Fence_Decrypt(Text: str, Fence: int) -> str: return DecryptedText @staticmethod - def Fence_Attack(Text: str) -> list: + def Fence_Attack(Text: str) -> List[tuple]: """ 用于对 Fence 密文进行爆破攻击。 @@ -839,12 +885,13 @@ def Fence_Attack(Text: str) -> list: 返回值： Result (List[tuple]): Fence 爆破攻击后得到的元组列表（字符串, 栏数） """ + Factors = [factor for factor in range(2, len(Text)) if len(Text) % factor == 0] Result = [(ClassicalCryptoUtils.Fence_Decrypt(Text, i), i) for i in Factors] return Result @staticmethod - def WFence_Generate(Text: str, Fence: int) -> list: + def WFence_Generate(Text: str, Fence: int) -> List[List[str]]: """ 用于生成 WFence 矩阵以便后续处理。 @@ -855,6 +902,7 @@ def WFence_Generate(Text: str, Fence: int) -> list: 返回值： Matrix (List[List[str]]): 生成的 WFence 矩阵 """ + Matrix = [['.'] * len(Text) for _ in range(Fence)] Row = 0 Up = False @@ -882,6 +930,7 @@ def WFence_Encrypt(Text: str, Fence: int) -> str: 返回值： EncryptedText (str): WFence 加密后得到的密文 """ + Matrix = ClassicalCryptoUtils.WFence_Generate(Text, Fence) EncryptedText = "" for Row in range(Fence): @@ -902,6 +951,7 @@ def WFence_Decrypt(Text: str, Fence: int) -> str: 返回值： DecryptedText (str): WFence 解密后得到的原文 """ + Matrix = ClassicalCryptoUtils.WFence_Generate(Text, Fence) index = 0 for Row in range(Fence): @@ -917,7 +967,7 @@ def WFence_Decrypt(Text: str, Fence: int) -> str: return DecryptedText @staticmethod - def WFence_Attack(Text: str) -> list: + def WFence_Attack(Text: str) -> List[tuple]: """ 用于对 WFence 密文进行爆破攻击。 @@ -927,5 +977,6 @@ def WFence_Attack(Text: str) -> list: 返回值： Result (List[tuple]): WFence 爆破攻击后得到的元组列表（字符串, 栏数） """ + Result = [(ClassicalCryptoUtils.WFence_Decrypt(Text, i), i) for i in range(2, len(Text))] return Result diff --git a/Poseidon/PoW.py b/Poseidon/PoW.py index 15b1d37..e4a1c2c 100644 --- a/Poseidon/PoW.py +++ b/Poseidon/PoW.py @@ -28,6 +28,7 @@ def ProofOfWork_SHA256_Full(Url: str, Port: int, HashBegin: str, HashEnd: str, T 返回值： Connection (pwn.remote): 与服务器建立的连接对象 """ + Connection = remote(Url, Port) Connection.recvuntil(HashBegin) Hash = Connection.recvuntil(HashEnd, drop=True).decode().strip() @@ -53,6 +54,7 @@ def ProofOfWork_SHA256_Prefix(Url: str, Port: int, PrefixBegin: str, PrefixEnd: 返回值： Connection (pwn.remote): 与服务器建立的连接对象 """ + Connection = remote(Url, Port) Connection.recvuntil(PrefixBegin) Prefix = Connection.recvuntil(PrefixEnd, drop=True).decode().strip() @@ -78,6 +80,7 @@ def ProofOfWork_SHA256_EndWithZero(Url: str, Port: int, KnownBegin: str, KnownEn 返回值： Connection (pwn.remote): 与服务器建立的连接对象 """ + Connection = remote(Url, Port) Connection.recvuntil(KnownBegin) Known = Connection.recvuntil(KnownEnd, drop=True).decode().strip() @@ -102,6 +105,7 @@ def ProofOfWork_MD5_Full(Url: str, Port: int, HashBegin: str, HashEnd: str, Text 返回值： Connection (pwn.remote): 与服务器建立的连接对象 """ + Connection = remote(Url, Port) Connection.recvuntil(HashBegin) Hash = Connection.recvuntil(HashEnd, drop=True).decode().strip() @@ -127,6 +131,7 @@ def ProofOfWork_MD5_Prefix(Url: str, Port: int, PrefixBegin: str, PrefixEnd: str 返回值： Connection (pwn.remote): 与服务器建立的连接对象 """ + Connection = remote(Url, Port) Connection.recvuntil(PrefixBegin) Prefix = Connection.recvuntil(PrefixEnd, drop=True).decode().strip() @@ -152,6 +157,7 @@ def ProofOfWork_MD5_EndWithZero(Url: str, Port: int, KnownBegin: str, KnownEnd: 返回值： Connection (pwn.remote): 与服务器建立的连接对象 """ + Connection = remote(Url, Port) Connection.recvuntil(KnownBegin) Known = Connection.recvuntil(KnownEnd, drop=True).decode().strip() diff --git a/Poseidon/__init__.py b/Poseidon/__init__.py index 2e2e794..c9b0481 100644 --- a/Poseidon/__init__.py +++ b/Poseidon/__init__.py @@ -1,4 +1,4 @@ __title__ = 'poseidon-python' __author__ = 'B1ue1nWh1te' -__version__ = '1.0.7' +__version__ = '1.1.0' __license__ = 'GPL-3.0' diff --git a/README.md b/README.md index e6ef044..ab59eba 100644 --- a/README.md +++ b/README.md @@ -12,20 +12,19 @@ [![Python Version](https://img.shields.io/badge/python-3.7+-blue)](https://www.python.org/) [![Release](https://img.shields.io/github/v/release/B1ue1nWh1te/Poseidon?include_prereleases)](https://github.com/B1ue1nWh1te/Poseidon/releases/) [![Visitors](https://visitor-badge.glitch.me/badge?page_id=B1ue1nWh1te-Poseidon&left_color=gray&right_color=orange)](https://github.com/B1ue1nWh1te/Poseidon) +![Downloads](https://img.shields.io/pypi/dm/poseidon-python) # 注意事项 -1. **本工具仅可用于 CTF 比赛解题，请勿在任何其他场景下使用。** +1. **本工具原则上仅可用于 CTF 比赛解题或测试链开发。但开源工具无法约束使用场景，若执意要在具有经济价值的公链中使用，所产生的影响将由你自行承担。** -2. 请尽可能使用最新版本解题。 +2. 在使用 `Blockchain` 模块时，你始终应该使用全新生成的账户，而不是导入常用的具有实际价值的账户，以确保你的账户安全。 -3. 在使用 `Blockchain` 模块时，你始终应该使用全新生成的账户，而不是导入常用的具有实际价值的账户，以确保你的账户安全。 +3. `Blockchain` 模块的所有功能在 `Goerli` 测试网络中均正常通过检验。 -4. `Blockchain` 模块的所有功能在`Goerli`测试网络中均正常通过检验。 - -5. 如果你在使用过程中遇到了其他问题，或者有任何好的想法和建议，欢迎提[issue](https://github.com/B1ue1nWh1te/Poseidon/issues)进行反馈。 +4. 如果你在使用过程中遇到了其他问题，或者有任何好的想法和建议，欢迎提[issue](https://github.com/B1ue1nWh1te/Poseidon/issues)进行反馈。 # 安装 @@ -37,7 +36,7 @@ pip install -U poseidon-python ## Blockchain 模块 -本模块可用于与任何以太坊同构链（即通常所说的 EVM 兼容链）进行交互，支持常用的链上交互操作。 +本模块可用于与任何以太坊同构链（即通常所说的 EVM 链）进行交互，支持常用的链上交互操作。 基于[Web3.py](https://github.com/ethereum/web3.py)实现。 @@ -53,70 +52,85 @@ from Poseidon.Blockchain import * ### Chain 类 -Chain 是区块链实例，作为链上交互的基础。 +Chain 是区块链实例，后续的所有链上交互的操作都将经由该指定节点处理。 -`Chain(RPCUrl: str, RequestParams: dict = None)`： +`Chain(RPCUrl: str, RequestParams: Optional[dict] = None)`： ``` -初始化。当连接失败时会抛出异常。 +初始化。根据给定的节点 RPC 地址进行连接，可通过代理访问。当连接节点失败时会抛出异常。 参数： - RPCUrl (str): 链 RPC 地址 - RequestParams (可选)(dict): 指定连接时使用的 request 参数，默认为 None。 - 例如当需要使用代理进行访问时，则传入 RequestParams={"proxies": {"http": "http://127.0.0.1:","https": "http://127.0.0.1:"}} + RPCUrl (str): 节点 RPC 地址 + RequestParams (Optional[dict]): 连接时使用的 request 参数，默认为 None。 + 例如当需要使用代理进行访问时，则传入 RequestParams={"proxies": {"http": "http://localhost:","https": "http://localhost:"}} 成员变量： - Net (Web3.HTTPProvider): web3.py 原生的链交互器对象 ChainId (int): 链 ID - ClientVersion (str): 链 RPC 的客户端软件版本号 + Node (Web3.HTTPProvider): web3.py 原生的 HTTP 交互器实例 + Eth (Web3.HTTPProvider.eth): HTTP 交互器实例中的 eth 模块 ```
-`GetBasicInformation() -> dict`： +`Chain.GetBasicInformation() -> dict`： ``` -获取链的基本信息。包括链 ID 、区块高度、 GasPrice 、出块间隔、链 RPC 的客户端软件版本号。 +获取区块链基本信息。包括链 ID 、区块高度、 GasPrice 、出块间隔、当前节点的客户端软件版本号。 返回值： - BasicInformation (dict): 链的基本信息构成的字典。 + BasicInformation (dict): 区块链基本信息构成的字典。 {"ChainId"|"BlockNumber"|"GasPrice"|"Timeslot"|"ClientVersion"} ```
-`GetTransactionInformationByHash(self, TransactionHash: str) -> dict`： +`Chain.GetTransactionInformationByHash(TransactionHash: str) -> dict`： ``` -根据交易哈希获取交易信息。包括交易哈希、所在区块号、交易索引号、交易状态、交易类型、交易行为、发送者、接收者、(部署的合约地址)、(GasPrice 或 (MaxFeePerGas 和 MaxPriorityFeePerGas))、GasLimit、GasUsed、Nonce、Value、Logs、InputData。 +根据交易哈希查询该交易的详细回执信息。包括交易哈希、所在区块号、交易索引号、交易状态、交易类型、交易行为、发送者、接收者、(部署的合约地址)、(GasPrice 或 (MaxFeePerGas 和 MaxPriorityFeePerGas))、GasLimit、GasUsed、Nonce、Value、R、S、V、Logs、InputData。 参数： - TransactionHash (str): 要查询的交易哈希 + TransactionHash (str): 要查询的交易的哈希 返回值： TransactionInformation (dict): 交易信息构成的字典。当出现异常时返回 None 。 - {"TransactionHash"|"BlockNumber"|"TransactionIndex"|"Status"|"Type"|"Action"|"From"|"To"|("ContractAddress")|<"GasPrice"|("MaxFeePerGas"&"MaxPriorityFeePerGas")>|"GasLimit"|"GasUsed"|"Nonce"|"Value"|"Logs"|"InputData"} + {"TransactionHash"|"BlockNumber"|"TransactionIndex"|"Status"|"Type"|"Action"|"From"|"To"|("ContractAddress")|<"GasPrice"|"MaxPriorityFeePerGas")>|"GasLimit"|"GasUsed"|"Nonce"|"Value"|"R"|"S"|"V"|"RawTransaction"|"Logs"|"InputData"} ```
-`GetTransactionInformationByBlockIdAndIndex(self, BlockID, TransactionIndex: int) -> dict`： +`Chain.GetTransactionInformationByBlockIdAndIndex(BlockID: Union[str,int], TransactionIndex: int) -> dict`： ``` -根据区块 ID 和交易在块中的索引来获取交易信息。包括交易哈希、所在区块号、交易索引号、交易状态、交易类型、交易行为、发送者、接收者、(部署的合约地址)、(GasPrice 或 (MaxFeePerGas 和 MaxPriorityFeePerGas))、GasLimit、GasUsed、Nonce、Value、Logs、InputData。 +根据区块 ID 和交易在块中的索引来查询该交易的详细回执信息。包括交易哈希、所在区块号、交易索引号、交易状态、交易类型、交易行为、发送者、接收者、(部署的合约地址)、(GasPrice 或 (MaxFeePerGas 和 MaxPriorityFeePerGas))、GasLimit、GasUsed、Nonce、Value、R、S、V、Logs、InputData。 参数： - BlockID (str|int): 区块 ID 。可为区块号或 'latest', 'earliest', 'pending' 。 + BlockID (Union[str,int]): 区块 ID 。可为区块号数值或 'latest', 'earliest', 'pending' 。 TransactionIndex (int): 交易在块中的索引 返回值： TransactionInformation (dict): 交易信息构成的字典。当出现异常时返回 None 。 - {"TransactionHash"|"BlockNumber"|"TransactionIndex"|"Status"|"Type"|"Action"|"From"|"To"|("ContractAddress")|<"GasPrice"|("MaxFeePerGas"&"MaxPriorityFeePerGas")>|"GasLimit"|"GasUsed"|"Nonce"|"Value"|"Logs"|"InputData"} + {"TransactionHash"|"BlockNumber"|"TransactionIndex"|"Status"|"Type"|"Action"|"From"|"To"|("ContractAddress")|<"GasPrice"|("MaxFeePerGas"&"MaxPriorityFeePerGas")>|"GasLimit"|"GasUsed"|"Nonce"|"Value"|"R"|"S"|"V"|"Logs"|"InputData"} +``` + +
+ +`Chain.GetBlockInformation(BlockID: Union[str, int]) -> dict`： + +``` +根据区块 ID 获取该区块的详细信息。包括区块号、区块哈希、矿工、时间戳、GasLimit、GasUsed、块内交易的哈希集合。 + +参数： + BlockID (Union[str,int]): 区块 ID 。可为区块号数值或 'latest', 'earliest', 'pending' 。 + +返回值： + BlockInformation (dict): 区块信息构成的字典。当出现异常时返回 None 。 + {"BlockNumber"|"BlockHash"|"Miner"|"TimeStamp"|"GasLimit"|"GasUsed"|"Transactions"} ```
-`GetBalance(Address: str) -> int`： +`Chain.GetBalance(Address: str) -> int`： ``` 根据账户地址获取其网络原生代币余额。 @@ -130,7 +144,7 @@ Chain 是区块链实例，作为链上交互的基础。
-`GetCode(Address: str) -> str`： +`Chain.GetCode(Address: str) -> str`： ``` 根据合约地址获取其已部署字节码。 @@ -144,7 +158,7 @@ Chain 是区块链实例，作为链上交互的基础。
-`GetStorage(Address: str, Index: int) -> str`： +`Chain.GetStorage(Address: str, Index: int) -> str`： ``` 根据合约地址和存储插槽索引获取存储值。 @@ -159,7 +173,7 @@ Chain 是区块链实例，作为链上交互的基础。
-`DumpStorage(Address: str, Count: int) -> list`： +`Chain.DumpStorage(Address: str, Count: int) -> list`： ``` 根据合约地址和指定插槽数量值，从插槽 0 开始批量遍历存储插槽并获取值。 @@ -174,7 +188,7 @@ Chain 是区块链实例，作为链上交互的基础。
-`GetPublicKeyByTransactionHash(TransactionHash: str) -> tuple`： +`Chain.GetPublicKeyByTransactionHash(TransactionHash: str) -> tuple`： ``` 通过一笔已在链上确认的交易哈希，获取账户公钥。 @@ -190,29 +204,27 @@ Chain 是区块链实例，作为链上交互的基础。 ### Account 类 -Account 是账户实例，作为发起链上调用的基础。 +Account 是账户实例，后续的交易将经由该指定账户发送至链上。 `Account(Chain: Chain, PrivateKey: str)`： ``` -初始化。通过私钥导入账户并与 Chain 实例绑定，后续的所有链上调用都会作用在 Chain 实例化表示的链上。当导入账户失败时将会抛出异常。 +初始化。通过私钥导入账户并与 Chain 实例绑定，后续的交易将经由该指定账户发送至链上。当导入账户失败时将会抛出异常。 参数： Chain (Poseidon.Blockchain.Chain): 区块链实例 PrivateKey (str): 账户私钥。不含 0x 前缀的十六进制形式。 成员变量： - Chain (Poseidon.Blockchain.Chain): 区块链实例 - Address (str): 账户地址 - PrivateKey (str): 账户私钥 + EthAccount (eth_account.Account): eth_account 的原生 Account 对象实例 ```
-`GetSelfBalance() -> int`： +`Account.GetSelfBalance() -> int`： ``` -获取自身账户的网络原生代币余额。当余额为 0 时会触发无法发送交易的警告。 +获取自身账户的网络原生代币余额。 返回值： Balance (int): 自身账户网络原生代币余额。单位为 wei ，当出现异常时返回 None 。 @@ -220,7 +232,7 @@ Account 是账户实例，作为发起链上调用的基础。
-`Transfer(self, To: str, Value: int, GasLimit: int = 100000, Data: str = "0x") -> dict`： +`Account.Transfer(To: str, Value: int, GasLimit: int = 100000, Data: str = "0x") -> dict`： ``` 向指定账户转账指定数量的网络原生代币，可附带信息。若 90 秒内交易未确认则作超时处理。 @@ -237,7 +249,7 @@ Account 是账户实例，作为发起链上调用的基础。
-`SendTransaction(To: str, Data: str, Value: int = 0, GasLimit: int = 1000000) -> dict`： +`Account.SendTransaction(self, To: str, Data: str, Value: int = 0, GasLimit: int = 1000000) -> dict`： ``` 以传统方式发送一笔自定义交易。若 90 秒内交易未确认则作超时处理。 @@ -254,7 +266,7 @@ Account 是账户实例，作为发起链上调用的基础。
-`SendTransactionByEIP1559(To: str, Data: str, Value: int = 0, GasLimit: int = 1000000) -> dict`： +`Account.SendTransactionByEIP1559(self, To: str, Data: str, Value: int = 0, GasLimit: int = 1000000) -> dict`： ``` 以 EIP-1559 方式发送一笔自定义交易。若 90 秒内交易未确认则作超时处理。 @@ -271,7 +283,7 @@ Account 是账户实例，作为发起链上调用的基础。
-`DeployContract(ABI: dict, Bytecode: str, Value: int = 0, *Arguments) -> dict`： +`Account.DeployContract(self, ABI: dict, Bytecode: str, Value: int = 0, *Arguments: Optional[Any]) -> dict`： ``` 部署合约。若 90 秒内交易未确认则作超时处理。 @@ -280,7 +292,7 @@ Account 是账户实例，作为发起链上调用的基础。 ABI (dict): 合约 ABI Bytecode (str): 合约部署字节码。含 0x 前缀的十六进制形式。 Value (可选)(int): 随交易发送给合约的网络原生代币数量。单位为 wei ，默认为 0 wei 。 - *Arguments (可选)(any): 传给合约构造函数的参数，默认为空。 + *Arguments (Optional[Any]): 传给合约构造函数的参数，默认为空。 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 @@ -289,7 +301,7 @@ Account 是账户实例，作为发起链上调用的基础。
-`DeployContractWithoutABI(Bytecode: str, Value: int = 0, GasLimit: int = 10000000) -> dict`： +`Account.DeployContractWithoutABI(self, Bytecode: str, Value: int = 0, GasLimit: int = 10000000) -> dict`： ``` 在没有 ABI 的情况下，仅使用字节码来部署合约。若 90 秒内交易未确认则作超时处理。 @@ -305,7 +317,7 @@ Account 是账户实例，作为发起链上调用的基础。
-`SignMessage(Message: str) -> dict`： +`Account.SignMessage(Message: str) -> dict`： ``` 消息字符串进行签名。 @@ -320,7 +332,7 @@ Account 是账户实例，作为发起链上调用的基础。
-`SignMessageHash(MessageHash: str) -> dict`： +`Account.SignMessageHash(MessageHash: str) -> dict`： ``` 对消息哈希进行签名。 @@ -342,7 +354,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。 `Contract(Account: Account, Address: str, ABI: dict)`： ``` -初始化。通过合约地址与 ABI 来实例化合约，并与 Account 绑定，后续所有对该合约的调用都会由这一账户发起。当实例化失败时会抛出异常。 +初始化。通过合约地址与 ABI 来实例化合约，并与 Account 绑定，后续所有对该合约的调用都会由这一账户发起。当合约实例化失败时会抛出异常。 参数： Account (Poseidon.Blockchain.Account): 账户实例 @@ -350,21 +362,20 @@ Contract 是合约实例，作为与指定合约进行交互的基础。 ABI (str): 合约 ABI 成员变量： - Account (Poseidon.Blockchain.Account): 账户实例 - Address (str): 合约地址 Instance (Web3.eth.Contract): web3.py 原生 contract 对象实例 + Address (str): 合约地址 ```
-`CallFunction(FunctionName: str, *FunctionArguments) -> dict`： +`Contract.CallFunction(FunctionName: str, *FunctionArguments: Optional[Any]) -> dict`： ``` 通过传入函数名及参数来调用该合约内的函数。 参数： FunctionName (str): 函数名称 - *FunctionArguments (可选)(any): 函数参数，默认为空。 + *FunctionArguments (Optional[Any]): 函数参数，默认为空。 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 @@ -372,7 +383,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`CallFunctionWithValueAndGasLimit(Value: int, GasLimit: int, FunctionName: str, *FunctionArguments) -> dict`： +`Contract.CallFunctionWithValueAndGasLimit(Value: int, GasLimit: int, FunctionName: str, *FunctionArguments: Optional[Any]) -> dict`： ``` 通过传入函数名及参数来调用该合约内的函数。支持自定义 Value 和 GasLimit 。 @@ -381,7 +392,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。 Value (int): 随交易发送的网络原生代币数量。单位为 wei 。 GasLimit (int): Gas 最大使用量。单位为 wei 。 FunctionName (str): 函数名称 - *FunctionArguments (可选)(any): 函数参数，默认为空。 + *FunctionArguments (Optional[Any]): 函数参数，默认为空。 返回值： TransactionInformation (dict): 交易信息构成的字典，通过 Chain.GetTransactionInformationByHash 获取。当出现异常时返回 None 。 @@ -389,29 +400,29 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`ReadOnlyCallFunction(FunctionName: str, *FunctionArguments)`： +`Contract.ReadOnlyCallFunction(FunctionName: str, *FunctionArguments: Optional[Any]) -> Any`： ``` 通过传入函数名及参数来调用该合约内的只读函数。 参数： FunctionName (str): 函数名称 - *FunctionArguments (可选)(any): 函数参数，默认为空。 + *FunctionArguments (Optional[Any]): 函数参数，默认为空。 返回值： - Result (any): 调用函数后得到的返回值。当出现异常时返回 None 。 + Result (Any): 调用函数后得到的返回值。当出现异常时返回 None 。 ```
-`EncodeABI(FunctionName: str, *FunctionArguments) -> str`： +`Contract.EncodeABI(FunctionName: str, *FunctionArguments: Optional[Any]) -> str`： ``` 通过传入函数名及参数进行编码，相当于生成调用该函数的 CallData 。 参数： FunctionName (str): 函数名称 - *FunctionArguments (可选)(any): 函数参数，默认为空。 + *FunctionArguments (Optional[Any]): 函数参数，默认为空。 返回值： CallData (str): 调用数据编码。含 0x 前缀的十六进制形式。当出现异常时返回 None 。 @@ -421,12 +432,12 @@ Contract 是合约实例，作为与指定合约进行交互的基础。 ### BlockchainUtils 类 -通用工具，链下使用的功能。 +通用工具集，整合了常用的链下操作。静态类，无需实例化。 -`SwitchSolidityVersion(SolidityVersion: str)`： +`BlockchainUtils.SwitchSolidityVersion(SolidityVersion: str)`： ``` -设置当前使用的 Solidity 版本，若该版本未安装则会自动安装。当设置版本失败时会抛出异常。 +设置当前使用的 Solidity 版本，若该版本未安装则会自动安装。 参数： SolidityVersion (str): Solidity 版本号 @@ -434,7 +445,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`Compile(FileCourse: str, ContractName: str, SolidityVersion: str = None, AllowPaths: str = None, Optimize: bool = False) -> tuple`： +`BlockchainUtils.Compile(FileCourse: str, ContractName: str, SolidityVersion: Optional[str] = None, AllowPaths: Optional[str] = None, Optimize: Optional[bool] = False) -> tuple`： ``` 根据给定的参数使用 py-solc-x 编译合约。当编译失败时会抛出异常。 @@ -442,9 +453,9 @@ Contract 是合约实例，作为与指定合约进行交互的基础。 参数： FileCourse (str): 合约文件完整路径。当合约文件与脚本文件在同一目录下时可直接使用文件名。 ContractName (str): 要编译的合约名称 - SolidityVersion (可选)(str): 指定使用的 Solidity 版本。若不指定则会使用当前已激活的 Solidity 版本进行编译。默认为 None 。 - AllowPaths (可选)(str): 指定许可路径。在编译时可能会出现 AllowPaths 相关错误可在这里解决。默认为 None 。 - Optimize (可选)(str): 是否开启优化器。默认为 False 。 + SolidityVersion (Optional[str]): 指定使用的 Solidity 版本。若不指定则会使用当前已激活的 Solidity 版本进行编译。默认为 None 。 + AllowPaths (Optional[str]): 指定许可路径。在编译时可能会出现 AllowPaths 相关错误可在这里解决。默认为 None 。 + Optimize (Optional[bool]): 是否开启优化器。默认为 False 。 返回值： (ABI, Bytecode) (tuple): 由 ABI 和 Bytecode 组成的元组 @@ -452,7 +463,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`CreateNewAccount() -> tuple`： +`BlockchainUtils.CreateNewAccount() -> tuple`： ``` 创建新账户。 @@ -463,7 +474,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`MnemonicToAddressAndPrivateKey(Mnemonic: str) -> tuple`： +`BlockchainUtils.MnemonicToAddressAndPrivateKey(Mnemonic: str) -> tuple`： ``` 将助记词转换为账户地址与私钥。参考 BIP-39 标准。 @@ -477,7 +488,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`RecoverMessage(Message: str, Signature: str) -> str`： +`BlockchainUtils.RecoverMessage(Message: str, Signature: str) -> str`： ``` 通过消息原文和签名还原出签署者的账户地址。 @@ -492,7 +503,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`RecoverMessageByHash(MessageHash: str, Signature: str) -> str`： +`BlockchainUtils.RecoverMessageByHash(MessageHash: str, Signature: str) -> str`： ``` 通过消息哈希和签名还原出签署者的账户地址。 @@ -507,7 +518,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`RecoverRawTransaction(RawTransactionData: str) -> str`： +`BlockchainUtils.RecoverRawTransaction(RawTransactionData: str) -> str`： ``` 获取签署此交易的账户地址。 @@ -521,7 +532,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`CrackSelector(SourceFunctionName: str, SourceFunctionParameters: list, ToGenerateFunctionParameters: list) -> str`： +`BlockchainUtils.CrackSelector(SourceFunctionName: str, SourceFunctionParameters: List[str], ToGenerateFunctionParameters: List[str]) -> str`： ``` 根据源函数名、参数与想要碰撞生成的函数的参数，碰撞生成出一个函数名，以使得这两个函数的选择器签名相等。 @@ -537,7 +548,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`AssemblyToBytecode(Assembly: str) -> str`： +`BlockchainUtils.AssemblyToBytecode(Assembly: str) -> str`： ``` 将 EVM Assembly 转为 EVM Bytecode 。 @@ -551,7 +562,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`BytecodeToAssembly(Bytecode: str) -> str`： +`BlockchainUtils.BytecodeToAssembly(Bytecode: str) -> str`： ``` 将 EVM Bytecode 转为 EVM Assembly 。 @@ -565,7 +576,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`SignatureToRSV(Signature: str) -> dict`： +`BlockchainUtils.SignatureToRSV(Signature: str) -> dict`： ``` 将签名解析成 R S V 。 @@ -580,7 +591,7 @@ Contract 是合约实例，作为与指定合约进行交互的基础。
-`RSVToSignature(R: str, S: str, V: str) -> dict`： +`BlockchainUtils.RSVToSignature(R: str, S: str, V: str) -> dict`： ``` 将 R S V 合并成签名。 @@ -807,7 +818,7 @@ from Poseidon.Cryptography import ModernCryptoUtils, ClassicalCryptoUtils, MiscU
-`RSA_MultiPrime_Attack(c: int, e: int, n: int, primes: list, powers: list = None) -> str`： +`RSA_MultiPrime_Attack(c: int, e: int, n: int, primes: List[int], powers: Optional[List[int]] = None) -> str`： ``` 用于对长整数形式的 RSA 密文进行多素数攻击并解出原文。 @@ -817,7 +828,7 @@ from Poseidon.Cryptography import ModernCryptoUtils, ClassicalCryptoUtils, MiscU e (int): e 值 n (int): n 值 primes (List[int]): 用于攻击的多素数列表 - powers (可选)(List[int]): 各素数对应的阶数，默认均为 1 次方 + powers (Optional[List[int]]): 各素数对应的阶数，默认均为 1 次方 返回值： m (str): RSA 多素数攻击后得到的原文 @@ -859,7 +870,7 @@ from Poseidon.Cryptography import ModernCryptoUtils, ClassicalCryptoUtils, MiscU
-`RSA_Broadcast_Attack(cs: list, e: int, ns: list) -> str`： +`RSA_Broadcast_Attack(cs: List[int], e: int, ns: List[int]) -> str`： ``` 用于对长整数形式的 RSA 密文列表进行广播攻击并解出原文。 @@ -939,7 +950,7 @@ from Poseidon.Cryptography import ModernCryptoUtils, ClassicalCryptoUtils, MiscU
-`Caesar_Attack(Text: str) -> list`： +`Caesar_Attack(Text: str) -> List[str]`： ``` 用于对 Caesar 密文进行爆破攻击。 @@ -1039,7 +1050,7 @@ from Poseidon.Cryptography import ModernCryptoUtils, ClassicalCryptoUtils, MiscU
-`Fence_Attack(Text: str) -> list`： +`Fence_Attack(Text: str) -> List[tuple]`： ``` 用于对 Fence 密文进行爆破攻击。 @@ -1083,7 +1094,7 @@ from Poseidon.Cryptography import ModernCryptoUtils, ClassicalCryptoUtils, MiscU
-`WFence_Attack(Text: str) -> list`： +`WFence_Attack(Text: str) -> List[tuple]`： ``` 用于对 WFence 密文进行爆破攻击。 diff --git a/setup.py b/setup.py index 1d29e68..3d8ec51 100644 --- a/setup.py +++ b/setup.py @@ -9,7 +9,7 @@ def long_description(): setup( name="poseidon-python", - version="1.0.7", + version="1.1.0", author="B1ue1nWh1te", author_email="b1ue1nwh1te@skiff.com", description="海神波塞冬工具对常用的链上交互操作进行了封装，使得开发者能够便捷地与任何以太坊同构链交互，主要用于在CTF比赛中攻克Blockchain方向的题目。",