import time import json import secrets from web3 import Web3 from web3.middleware import ExtraDataToPOAMiddleware from eth_abi import decode as abi_decode # -------- CHAIN CONFIGS -------- # Each chain entry defines the DEX router, wrapped native token, and block time. # Block time is used to compute a realistic timestamp delta when advancing τ blocks. CHAIN_CONFIGS = { "bsc": { "router": "0x10ed43c718714eb63d5aa57b78b54704e256024e", # PancakeSwap V2 "wnative": "0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c", # WBNB "native": "BNB", "block_time": 3, # seconds per block }, "eth": { "router": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", # Uniswap V2 "wnative": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", # WETH "native": "ETH", "block_time": 12, # seconds per block }, } # -------- CONSTANTS -------- ANVIL_PORT = 8545 ANVIL_URL = f"http://127.0.0.1:{ANVIL_PORT}" CHECKER_PREFIX = "0x000000000000000000000000000000000000" TRANSFER_TOPIC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" # Default number of blocks to advance between buy and sell (τ in the paper). # Set to 0 to execute buy and sell in the same block. # Increase to replicate anti-simulation mechanisms that require a temporal gap. TAU_BLOCKS = 0 # Default amount of native asset used for the buy (in wei). # BALANCE_START is derived automatically as 10x AMOUNT_IN to ensure the trader # has sufficient funds without being unrealistically large. DEFAULT_AMOUNT_IN = Web3.to_wei(0.01, "ether") # -------- ABIs -------- router_abi = json.loads(""" [ { "name": "swapExactETHForTokens", "type": "function", "stateMutability": "payable", "inputs": [ {"name": "amountOutMin", "type": "uint256"}, {"name": "path", "type": "address[]"}, {"name": "to", "type": "address"}, {"name": "deadline", "type": "uint256"} ], "outputs": [ {"name": "amounts", "type": "uint256[]"} ] }, { "name": "swapExactTokensForETH", "type": "function", "stateMutability": "nonpayable", "inputs": [ {"name": "amountIn", "type": "uint256"}, {"name": "amountOutMin", "type": "uint256"}, {"name": "path", "type": "address[]"}, {"name": "to", "type": "address"}, {"name": "deadline", "type": "uint256"} ], "outputs": [ {"name": "amounts", "type": "uint256[]"} ] }, { "name": "getAmountsOut", "type": "function", "stateMutability": "view", "inputs": [ {"name": "amountIn", "type": "uint256"}, {"name": "path", "type": "address[]"} ], "outputs": [ {"name": "amounts", "type": "uint256[]"} ] }, { "name": "swapExactTokensForETHSupportingFeeOnTransferTokens", "type": "function", "stateMutability": "nonpayable", "inputs": [ {"name": "amountIn", "type": "uint256"}, {"name": "amountOutMin", "type": "uint256"}, {"name": "path", "type": "address[]"}, {"name": "to", "type": "address"}, {"name": "deadline", "type": "uint256"} ], "outputs": [] } ] """) token_abi = json.loads(""" [ {"name":"approve","type":"function","stateMutability":"nonpayable", "inputs":[{"name":"spender","type":"address"},{"name":"amount","type":"uint256"}], "outputs":[{"name":"","type":"bool"}]}, {"name":"balanceOf","type":"function","stateMutability":"view", "inputs":[{"name":"account","type":"address"}], "outputs":[{"name":"","type":"uint256"}]} ] """) # -------- HELPERS -------- def get_w3(chain): """Return a Web3 instance connected to the local Anvil fork for the given chain.""" w3 = Web3(Web3.HTTPProvider(ANVIL_URL)) if chain == "bsc": # BSC uses Proof-of-Authority; inject middleware to handle the extraData field w3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0) return w3 def generate_trader_address(w3): """Generate a fresh random wallet address for each simulation.""" private_key = "0x" + secrets.token_hex(32) return w3.eth.account.from_key(private_key).address def build_balance_checker_bytecode(address): """ Minimal EVM bytecode that returns the native asset balance of `address`. Deployed at a throwaway address via stateOverrides to read the trader's final balance after the sell without requiring a real contract deployment. """ addr = address[2:].lower().zfill(40) return "0x" + "73" + addr + "31" + "6000" + "52" + "6020" + "6000" + "f3" def get_checker_address(trader_address): """Derive a deterministic checker address from the trader address suffix.""" suffix = trader_address[-4:].lower() return Web3.to_checksum_address(CHECKER_PREFIX + suffix) def parse_tokens_received(call_result, recipient, token_address): """Sum Transfer events fired by the token contract to the recipient.""" received = 0 for log in call_result.get("logs", []): if log.get("address", "").lower() != token_address.lower(): continue topics = log.get("topics", []) if not topics or topics[0].lower() != TRANSFER_TOPIC.lower(): continue if len(topics) < 3: continue to_addr = "0x" + topics[2][-40:] if to_addr.lower() == recipient.lower(): received += int(log.get("data", "0x0"), 16) return received def run_simulation( w3, token, pair, block, chain_cfg, amount_in=DEFAULT_AMOUNT_IN, tau=TAU_BLOCKS, ): """ Simulate a buy and sell of `token` at `block` to detect honeypot behavior. The simulation: 1. Forks the blockchain state at `block`. 2. Creates a fresh simulated trader funded with 10x `amount_in`. 3. Computes theoretical buy output β_exp via getAmountsOut. 4. Executes the buy and determines actual tokens received β_act. 5. Approves the router to spend β_act. 6. Advances the simulated block by `tau` (or stays in the same block if tau == 0). 7. Computes theoretical sell output σ_exp via getAmountsOut. 8. Executes the sell using swapExactTokensForETHSupportingFeeOnTransferTokens. 9. Reads the trader's final native balance to determine σ_act. 10. Computes buy/sell taxes and returns the verdict. """ ROUTER = Web3.to_checksum_address(chain_cfg["router"]) WNATIVE = Web3.to_checksum_address(chain_cfg["wnative"]) NATIVE = chain_cfg["native"] BLK_TIME = chain_cfg["block_time"] # Give the simulated trader enough native currency to cover the buy. BALANCE_START = amount_in * 10 TOKEN = Web3.to_checksum_address(token) BUY_PATH = [WNATIVE, TOKEN] SELL_PATH = [TOKEN, WNATIVE] # --------------------------------------------------------- # Fresh simulated trader # --------------------------------------------------------- FROM = generate_trader_address(w3) CHECKER_ADDRESS = get_checker_address(FROM) router = w3.eth.contract( address=ROUTER, abi=router_abi, ) token_contract = w3.eth.contract( address=TOKEN, abi=token_abi, ) deadline = int(time.time()) + 600 # Fund the trader and inject the native-balance checker. state_override = { FROM: { "balance": hex(BALANCE_START), }, CHECKER_ADDRESS: { "code": build_balance_checker_bytecode(FROM), }, } # ========================================================= # PHASE 1 # BUY -> BALANCE CHECK # ========================================================= # --------------------------------------------------------- # Theoretical buy output β_exp # --------------------------------------------------------- try: buy_theoretical = ( router.functions.getAmountsOut( amount_in, BUY_PATH, ) .call(block_identifier=block)[-1] ) except Exception as e: return { "token": token, "pair": pair, "block": block, "error": f"getAmountsOut_buy_failed: {str(e)}", } # --------------------------------------------------------- # Buy transaction # --------------------------------------------------------- buy_tx = { "from": FROM, "to": ROUTER, "value": hex(amount_in), "data": router.functions.swapExactETHForTokens( 0, BUY_PATH, FROM, deadline, )._encode_transaction_data(), "gas": hex(10_000_000), "gasPrice": hex(0), "nonce": hex(0), } # --------------------------------------------------------- # balanceOf(FROM) # --------------------------------------------------------- balance_of_data = ( Web3.keccak(text="balanceOf(address)")[:4].hex() + "000000000000000000000000" + FROM[2:].lower() ) token_balance_check_tx = { "from": FROM, "to": TOKEN, "value": hex(0), "data": "0x" + balance_of_data, "gas": hex(100_000), "gasPrice": hex(0), "nonce": hex(0), } # --------------------------------------------------------- # Simulate buy in Block A and balanceOf in Block B. # # Splitting these calls allows balanceOf to observe the # committed post-buy state. # # Block A gets an explicit timestamp so it can never collide # with or exceed Block B's timestamp in Phase 2 below. # --------------------------------------------------------- parent_block = w3.eth.get_block(block) parent_timestamp = int(parent_block["timestamp"]) block_a_timestamp = parent_timestamp + 1 r1 = w3.provider.make_request( "eth_simulateV1", [{ "blockStateCalls": [ { "stateOverrides": state_override, "blockOverrides": {"time": hex(block_a_timestamp)}, "calls": [buy_tx], }, { "blockOverrides": {"time": hex(block_a_timestamp + 1)}, "calls": [token_balance_check_tx], }, ] }, hex(block)] ) if "error" in r1: return { "token": token, "pair": pair, "block": block, "error": r1["error"]["message"], } p1_block_a = r1["result"][0]["calls"] p1_block_b = r1["result"][1]["calls"] # --------------------------------------------------------- # Check buy # --------------------------------------------------------- if p1_block_a[0].get("status") == "0x0": return { "token": token, "pair": pair, "block": block, "error": "buy_reverted", } # --------------------------------------------------------- # Determine β_act. # # Prefer balanceOf. Fall back to Transfer logs for tokens # whose balanceOf behaves unexpectedly in simulation. # --------------------------------------------------------- tokens_from_logs = parse_tokens_received( p1_block_a[0], FROM, TOKEN, ) raw = p1_block_b[0].get( "returnData", "0x0", ) token_balance = ( int(raw, 16) if raw and raw != "0x" else 0 ) if token_balance > 0: beta_act = token_balance elif tokens_from_logs > 0: beta_act = tokens_from_logs else: beta_act = 0 # --------------------------------------------------------- # Buy tax # --------------------------------------------------------- buy_tax = ( (1 - beta_act / buy_theoretical) * 100 if buy_theoretical > 0 else 0 ) # Zero tokens received = 100% buy tax. if beta_act == 0: return { "token": token, "pair": pair, "block": block, "verdict": "HONEYPOT", "buy_tax": 100.0, "sell_tax": None, "error": "no_tokens_received", } # ========================================================= # PHASE 2 # BUY + APPROVE -> SELL + BALANCE CHECK # ========================================================= # --------------------------------------------------------- # Approval # --------------------------------------------------------- approve_tx = { "from": FROM, "to": TOKEN, "value": hex(0), "data": token_contract.functions.approve( ROUTER, 2**256 - 1, )._encode_transaction_data(), "gas": hex(500_000), "gasPrice": hex(0), "nonce": hex(1), } # --------------------------------------------------------- # Theoretical sell output σ_exp # # This is only the router quote. Actual native proceeds are # measured from the trader's balance after the sell. # --------------------------------------------------------- get_amounts_out_tx = { "from": FROM, "to": ROUTER, "value": hex(0), "data": router.encode_abi( "getAmountsOut", [beta_act, SELL_PATH], ), "gas": hex(1_000_000), "gasPrice": hex(0), "nonce": hex(2), } # --------------------------------------------------------- # Sell using the fee-on-transfer-supporting router method. # # IMPORTANT: # This function returns no amount array. Therefore σ_act # must be measured through the native balance checker. # --------------------------------------------------------- sell_tx = { "from": FROM, "to": ROUTER, "value": hex(0), "data": ( router.functions .swapExactTokensForETHSupportingFeeOnTransferTokens( beta_act, 0, SELL_PATH, FROM, deadline, ) ._encode_transaction_data() ), "gas": hex(10_000_000), "gasPrice": hex(0), "nonce": hex(3), } # --------------------------------------------------------- # Native balance checker # --------------------------------------------------------- balance_check_tx = { "from": FROM, "to": CHECKER_ADDRESS, "value": hex(0), "data": "0x", "gas": hex(100_000), "gasPrice": hex(0), "nonce": hex(4), } # --------------------------------------------------------- # Simulate Phase 2. # # tau == 0: # Buy, approve, quote, sell, and balance check must all # land in the SAME simulated block (same-block round trip). # eth_simulateV1 treats every entry in blockStateCalls as a # distinct block with a strictly increasing timestamp, so # "same block" has to be expressed as a single blockStateCalls # entry containing every call, not as two entries with equal # timestamps. # # tau > 0: # Block A: buy + approve # Block B: getAmountsOut + supporting sell + balance check, # timestamped tau blocks ahead of Block A. # --------------------------------------------------------- if tau == 0: r2 = w3.provider.make_request( "eth_simulateV1", [{ "blockStateCalls": [ { "stateOverrides": state_override, "blockOverrides": {"time": hex(block_a_timestamp)}, "calls": [ buy_tx, approve_tx, get_amounts_out_tx, sell_tx, balance_check_tx, ], }, ] }, hex(block)] ) if "error" in r2: return { "token": token, "pair": pair, "block": block, "error": r2["error"]["message"], } ( buy_call, approve_call, quote_call, sell_call, balance_call, ) = r2["result"][0]["calls"] else: future_timestamp = parent_timestamp + (tau * BLK_TIME) if future_timestamp <= block_a_timestamp: raise ValueError( f"tau={tau} with BLK_TIME={BLK_TIME} does not advance " f"past block_a_timestamp={block_a_timestamp}" ) r2 = w3.provider.make_request( "eth_simulateV1", [{ "blockStateCalls": [ { "stateOverrides": state_override, "blockOverrides": {"time": hex(block_a_timestamp)}, "calls": [buy_tx, approve_tx], }, { "blockOverrides": {"time": hex(future_timestamp)}, "calls": [ get_amounts_out_tx, sell_tx, balance_check_tx, ], }, ] }, hex(block)] ) if "error" in r2: return { "token": token, "pair": pair, "block": block, "error": r2["error"]["message"], } buy_call, approve_call = r2["result"][0]["calls"] quote_call, sell_call, balance_call = r2["result"][1]["calls"] # --------------------------------------------------------- # Check buy in Phase 2 # --------------------------------------------------------- if buy_call.get("status") == "0x0": return { "token": token, "pair": pair, "block": block, "error": "buy_reverted_p2", } # --------------------------------------------------------- # Approval reverted -> honeypot # --------------------------------------------------------- if approve_call.get("status") == "0x0": return { "token": token, "pair": pair, "block": block, "verdict": "HONEYPOT", "buy_tax": round(buy_tax, 4), "sell_tax": None, "error": "approve_reverted", } # --------------------------------------------------------- # Sell reverted -> honeypot # # This is now testing the fee-on-transfer-supporting sell, # which is the correct behavior for taxed tokens. # --------------------------------------------------------- if sell_call.get("status") == "0x0": return { "token": token, "pair": pair, "block": block, "verdict": "HONEYPOT", "buy_tax": round(buy_tax, 4), "sell_tax": None, "error": "sell_reverted", } # --------------------------------------------------------- # Decode theoretical sell output σ_exp # --------------------------------------------------------- sell_theoretical = 0 if quote_call.get("status") == "0x1": rd = quote_call.get( "returnData", "0x", ) try: sell_theoretical = abi_decode( ["uint256[]"], bytes.fromhex(rd[2:]), )[0][-1] except Exception: sell_theoretical = 0 # --------------------------------------------------------- # Read actual native balance after sell. # --------------------------------------------------------- final_balance = int( balance_call.get( "returnData", "0x0", ), 16, ) # The trader started with BALANCE_START and spent amount_in # on the buy. Therefore this is the amount returned by the sell. sigma_act = ( final_balance - (BALANCE_START - amount_in) ) # --------------------------------------------------------- # Sell tax # --------------------------------------------------------- sell_tax = ( (1 - sigma_act / sell_theoretical) * 100 if sell_theoretical > 0 else 0 ) # --------------------------------------------------------- # Verdict # --------------------------------------------------------- verdict = ( "HONEYPOT" if sell_tax >= 90 or buy_tax >= 90 else "CLEAN" ) return { "token": token, "pair": pair, "block": block, "chain": NATIVE, "buy_theoretical": buy_theoretical, "beta_act": beta_act, "buy_tax": round(buy_tax, 4), "sell_theoretical": sell_theoretical, "sigma_act": sigma_act, "sell_tax": round(sell_tax, 4), "verdict": verdict, "error": None, } # -------- MAIN -------- if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( formatter_class=argparse.RawDescriptionHelpFormatter, description=( "Honeypot detection tool via transaction simulation on a local Anvil fork.\n\n" "Before running, start Anvil forked at the target block:\n" " anvil --fork-url --fork-block-number --port 8545\n\n" "Examples:\n" " # BSC token\n" " python simulation.py --chain bsc --token 0xABC... --pair 0xDEF... --block 28000000\n\n" " # Ethereum token with 0.1 ETH buy amount\n" " python simulation.py --chain eth --token 0xABC... --pair 0xDEF... --block 18000000 --amount-in 0.1\n" ) ) parser.add_argument("--chain", required=True, choices=CHAIN_CONFIGS.keys(), help="Blockchain to simulate on (bsc or eth)") parser.add_argument("--token", required=True, help="Token contract address to simulate") parser.add_argument("--pair", required=True, help="Liquidity pool address (for logging)") parser.add_argument("--block", required=True, type=int, help="Block height to fork at") parser.add_argument("--amount-in", default=0.01, type=float, help="Amount of native asset to use for the buy, in ETH/BNB units (default: 0.01)") parser.add_argument("--tau", default=TAU_BLOCKS, type=int, help=f"Blocks to advance between buy and sell (default: {TAU_BLOCKS})") args = parser.parse_args() chain_cfg = CHAIN_CONFIGS[args.chain] amount_in = Web3.to_wei(args.amount_in, "ether") w3 = get_w3(args.chain) result = run_simulation( w3, token=args.token, pair=args.pair, block=args.block, chain_cfg=chain_cfg, amount_in=amount_in, tau=args.tau, ) print(json.dumps(result, indent=2))