""" title: Etherscan Tool author: naoufel version: 1.0 license: MIT requirements: web3, requests """ import os import json import requests from typing import Dict, List, Optional from pydantic import BaseModel, Field from datetime import datetime import logging from web3 import Web3 class EventEmitter: def __init__(self, emitter=None): self.emitter = emitter async def emit( self, description="Unknown State", status="in_progress", done=False, step_number=None, source=None ): if self.emitter: message = { "type": "status", "data": { "status": status, "description": description, "done": done, } } if step_number: message["data"]["step"] = step_number await self.emitter(message) if source: await self.emitter({ "type": "source", "data": source }) class Tools: class Valves(BaseModel): """Configuration for Etherscan API interactions.""" # API Configuration API_KEY: str = Field(default=os.getenv("ETHERSCAN_API_KEY", "")) API_BASE: str = Field(default="https://api.etherscan.io/api") # Rate Limiting and Retries RATE_LIMIT_DELAY: float = Field(default=0.2) # 200ms between requests MAX_RETRIES: int = Field(default=3) # Error messages ERROR_MESSAGES: Dict = Field(default={ "invalid_address": "Invalid Ethereum address provided", "api_error": "Error accessing Etherscan API", "rate_limit": "Rate limit exceeded", "contract_not_verified": "Contract source code not verified" }) # Security Audit Configuration SECURITY_CHECKS: Dict = Field(default={ "reentrancy": { "pattern": ["call.value", "send", "transfer"], "risk": "High", "description": "Potential reentrancy vulnerability" }, "tx_origin": { "pattern": ["tx.origin"], "risk": "High", "description": "Usage of tx.origin for authentication" }, "assembly": { "pattern": ["assembly"], "risk": "Medium", "description": "Usage of inline assembly" }, "deprecated": { "pattern": ["sha3(", "suicide(", "block.timestamp"], "risk": "Medium", "description": "Usage of deprecated functions" }, "unchecked_math": { "pattern": ["SafeMath"], "risk": "Medium", "description": "No SafeMath usage detected" } }) def __init__(self): self.valves = self.Valves() self.w3 = Web3() logging.basicConfig(level=logging.INFO) self.logger = logging.getLogger(__name__) async def get_contract_source( self, contract_address: str, __event_emitter__=None ) -> str: """Fetches verified contract source code from Etherscan.""" emitter = EventEmitter(__event_emitter__) await emitter.emit(f"Fetching source code for contract {contract_address}...") if not self.w3.is_address(contract_address): await emitter.emit(self.valves.ERROR_MESSAGES["invalid_address"], done=True) return self.valves.ERROR_MESSAGES["invalid_address"] params = { "module": "contract", "action": "getsourcecode", "address": contract_address, "apikey": self.valves.API_KEY } try: response = requests.get(self.valves.API_BASE, params=params) response.raise_for_status() result = response.json() if result["status"] == "1" and result["message"] == "OK": source_info = result["result"][0] if source_info["SourceCode"] == "": await emitter.emit(self.valves.ERROR_MESSAGES["contract_not_verified"], done=True) return self.valves.ERROR_MESSAGES["contract_not_verified"] contract_data = { "contract_name": source_info["ContractName"], "compiler_version": source_info["CompilerVersion"], "source_code": source_info["SourceCode"], "abi": source_info["ABI"], "constructor_arguments": source_info["ConstructorArguments"] } await emitter.emit( "Contract source code retrieved successfully", done=True, source={ "document": [json.dumps(contract_data)], "metadata": [{ "source": "Etherscan API", "date_accessed": datetime.now().isoformat(), "name": f"Contract Source for {contract_address}" }], "source": { "name": "Etherscan", "url": f"https://etherscan.io/address/{contract_address}" } } ) return json.dumps({ "contract_name": source_info["ContractName"], "compiler_version": source_info["CompilerVersion"], "source_code": source_info["SourceCode"], "abi": source_info["ABI"], "constructor_arguments": source_info["ConstructorArguments"] }, indent=2) except Exception as e: error_msg = f"Error fetching contract source: {str(e)}" self.logger.error(error_msg) return error_msg async def basic_contract_audit( self, contract_address: str, __event_emitter__=None ) -> str: """Performs a basic security audit of the contract code.""" source_result = await self.get_contract_source(contract_address) if source_result.startswith("Error"): return source_result try: contract_data = json.loads(source_result) source_code = contract_data["source_code"] # Define security patterns to check security_checks = { "reentrancy": { "pattern": ["call.value", "send", "transfer"], "risk": "High", "description": "Potential reentrancy vulnerability" }, "tx_origin": { "pattern": ["tx.origin"], "risk": "High", "description": "Usage of tx.origin for authentication" }, "assembly": { "pattern": ["assembly"], "risk": "Medium", "description": "Usage of inline assembly" }, "deprecated": { "pattern": ["sha3(", "suicide(", "block.timestamp"], "risk": "Medium", "description": "Usage of deprecated functions" }, "unchecked_math": { "pattern": ["SafeMath"], "risk": "Medium", "description": "No SafeMath usage detected" } } findings = [] for check_name, check_info in security_checks.items(): for pattern in check_info["pattern"]: if pattern in source_code: findings.append({ "check": check_name, "risk_level": check_info["risk"], "description": check_info["description"], "pattern_found": pattern }) return json.dumps({ "contract_address": contract_address, "compiler_version": contract_data["compiler_version"], "findings": findings, "audit_timestamp": datetime.now().isoformat() }, indent=2) except Exception as e: error_msg = f"Error during contract audit: {str(e)}" self.logger.error(error_msg) return error_msg async def get_contract_transactions( self, contract_address: str, page: int = 1, offset: int = 100, __event_emitter__=None ) -> str: """Fetches contract transactions from Etherscan.""" if not self.w3.is_address(contract_address): return self.valves.ERROR_MESSAGES["invalid_address"] params = { "module": "account", "action": "txlist", "address": contract_address, "page": page, "offset": offset, "sort": "desc", "apikey": self.valves.API_KEY } try: response = requests.get(self.valves.API_BASE, params=params) response.raise_for_status() result = response.json() if result["status"] == "1" and result["message"] == "OK": return json.dumps({ "contract_address": contract_address, "transactions": result["result"], "page": page, "offset": offset }, indent=2) except Exception as e: error_msg = f"Error fetching contract transactions: {str(e)}" self.logger.error(error_msg) return error_msg