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 <ARCHIVE_RPC_URL> --fork-block-number <BLOCK> --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))