ERC-20 Approval Transaction Simulator with UI Spoofing Detection (Educational Example)

simulatingtransaction.py

from web3 import Web3
import json
from eth_abi import decode
from eth_utils import to_checksum_address

# Connect to a local fork of mainnet
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))

# ERC-20 ABI (minimal for approval/allowance checks)
ERC20_ABI = [
    {
        "constant": False,
        "inputs": [
            {"name": "spender", "type": "address"},
            {"name": "value", "type": "uint256"}
        ],
        "name": "approve",
        "outputs": [{"name": "", "type": "bool"}],
        "type": "function"
    },
    {
        "constant": True,
        "inputs": [
            {"name": "owner", "type": "address"},
            {"name": "spender", "type": "address"}
        ],
        "name": "allowance",
        "outputs": [{"name": "", "type": "uint256"}],
        "type": "function"
    },
    {
        "constant": True,
        "inputs": [],
        "name": "decimals",
        "outputs": [{"name": "", "type": "uint8"}],
        "type": "function"
    }
]

def setup_addresses(user_address, token_address, spender_address):
    """Convert addresses to checksum format and create contract instance"""
    user = Web3.to_checksum_address(user_address)
    token = Web3.to_checksum_address(token_address)
    spender = Web3.to_checksum_address(spender_address)
    contract = w3.eth.contract(address=token, abi=ERC20_ABI)
    
    return user, token, spender, contract

def get_token_info(contract, token_address, token_decimals=None):
    """Get token decimals and calculate normalized amount"""
    if token_decimals is None:
        try:
            token_decimals = contract.functions.decimals().call()
        except:
            # Use appropriate fallback based on token address
            if token_address.lower() == "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48".lower():  # USDC
                token_decimals = 6
            else:
                token_decimals = 18  # Default fallback
    
    return token_decimals

def check_initial_state(contract, user, spender, token_decimals):
    """Check the initial allowance"""
    initial_allowance = contract.functions.allowance(user, spender).call()
    initial_allowance_normalized = initial_allowance / (10 ** token_decimals)
    
    return initial_allowance, initial_allowance_normalized

def build_transaction(contract, user, spender, amount):
    """Build the approval transaction"""
    tx = contract.functions.approve(spender, amount).build_transaction({
        'from': user,
        'nonce': w3.eth.get_transaction_count(user),
        'gas': 200000,
        'gasPrice': w3.eth.gas_price,
        'chainId': w3.eth.chain_id
    })
    
    # Get gas estimate
    gas_estimate = w3.eth.estimate_gas(tx)
    tx['gas'] = gas_estimate
    
    return tx, gas_estimate

def compare_calldata(tx_data, wallet_calldata):
    """Compare generated calldata with wallet calldata"""
    if not wallet_calldata:
        return None
        
    # Normalize both calldata strings for comparison
    norm_generated = tx_data.lower()
    norm_wallet = wallet_calldata.lower()
    if norm_wallet.startswith('0x'):
        norm_wallet = norm_wallet[2:]
    if norm_generated.startswith('0x'):
        norm_generated = norm_generated[2:]
        
    calldata_matches = (norm_generated == norm_wallet)
    
    if not calldata_matches:
        print("WARNING: Wallet calldata doesn't match expected calldata!")
        print("This could indicate a malicious transaction or UI spoofing attack.")
    
    return calldata_matches

def execute_transaction(tx, user):
    """Execute the transaction using impersonation with gas safety limits"""
    w3.provider.make_request("anvil_impersonateAccount", [user])
    try:
        # Send transaction with gas cap for security
        safe_tx = {**tx, 'gas': min(tx.get('gas', 0), 300000)}
        tx_hash = w3.eth.send_transaction(safe_tx)
    finally:
        w3.provider.make_request("anvil_stopImpersonatingAccount", [user])
    
    return tx_hash

def check_final_state(contract, user, spender, amount, token_decimals):
    """Check the final allowance and transaction success"""
    final_allowance = contract.functions.allowance(user, spender).call()
    final_allowance_normalized = final_allowance / (10 ** token_decimals)
    
    # Check for infinite approval
    infinite_approval = amount >= (2**256 - 1) or amount >= (2**64 - 1)
    
    return final_allowance, final_allowance_normalized, infinite_approval

def simulate_approval(
    user_address: str,
    token_address: str,
    spender_address: str,
    amount: int,
    token_decimals: int = None,
    wallet_calldata: str = None
) -> dict:
    """
    Simulate an ERC-20 approval transaction and verify state changes.
    """
    try:
        # Setup addresses and contract
        user, token, spender, contract = setup_addresses(user_address, token_address, spender_address)
        
        # Get token information
        token_decimals = get_token_info(contract, token, token_decimals)
        normalized_amount = amount / (10 ** token_decimals)

        # reset allowances before every simulation
        reset_tx = contract.functions.approve(spender, 0).build_transaction({
            'from': user,
            'nonce': w3.eth.get_transaction_count(user)
        })
        execute_transaction(reset_tx, user)
        
        # Check initial state
        initial_allowance, initial_allowance_normalized = check_initial_state(
            contract, user, spender, token_decimals
        )
        
        # Build intended transaction
        tx, gas_estimate = build_transaction(contract, user, spender, amount)
        
        # Compare calldata and check for infinite approval in wallet data
        calldata_matches = compare_calldata(tx['data'], wallet_calldata)
        
        # Additional security check for wallet calldata
        wallet_amount = None
        wallet_infinite = False
        if wallet_calldata:
            try:
                clean_hex = wallet_calldata[2:] if wallet_calldata.startswith('0x') else wallet_calldata
                if len(clean_hex) >= 136:
                    wallet_spender = to_checksum_address('0x' + clean_hex[32:72])
                    wallet_amount = int(clean_hex[72:136], 16)
                    wallet_infinite = wallet_amount >= (2**64 - 1)
                    
                    if wallet_spender.lower() != spender.lower():
                        print("CRITICAL: Spender address in wallet calldata doesn't match expected spender!")
                    
                    if wallet_infinite:
                        print("WARNING: Wallet calldata contains infinite approval!")
                else:
                    print("WARNING: Wallet calldata is too short to decode properly")
            except Exception as e:
                print(f"Warning: Could not decode wallet calldata - {str(e)}")
        
        # Determine which amount to execute with
        execution_amount = wallet_amount if (wallet_calldata and wallet_amount is not None) else amount
        execution_tx, gas_estimate = build_transaction(contract, user, spender, execution_amount)
        
        # Execute transaction
        tx_hash = execute_transaction(execution_tx, user)
        
        # Check final state
        final_allowance, final_allowance_normalized, infinite_approval = check_final_state(
            contract, user, spender, execution_amount, token_decimals
        )

        return {
            "success": True,
            "calldata": tx['data'],
            "wallet_calldata": wallet_calldata,
            "calldata_matches": calldata_matches,
            "wallet_infinite_approval": wallet_infinite,
            "spender": spender,
            "amount_raw": amount,
            "amount_normalized": normalized_amount,
            "executed_amount_raw": execution_amount,
            "executed_amount_normalized": execution_amount / (10 ** token_decimals),
            "token_decimals": token_decimals,
            "gas_estimate": gas_estimate,
            "infinite_approval": infinite_approval,
            "initial_allowance_raw": initial_allowance,
            "initial_allowance_normalized": initial_allowance_normalized,
            "final_allowance_raw": final_allowance,
            "final_allowance_normalized": final_allowance_normalized,
            "transaction_successful": final_allowance == execution_amount,
            "tx_hash": tx_hash.hex()
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

# Example usage
if __name__ == "__main__":
    result = simulate_approval(
        user_address="0x9965507D1a55bcC2695C58ba16FB37d819B0A4dc",  # Anvil Account 0
        token_address="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",  # USDC
        spender_address="0x6A000F20005980200259B80c5102003040001068",  # Example spender
        amount=25 * 10**6,  # 25 USDC (6 decimals)
        token_decimals=6,  # USDC has 6 decimals
        wallet_calldata="0x095ea7b30000000000000000000000006a000f20005980200259b80c510200304000106800000000000000000000000000000000000000000000000000000000017d7840"  # Optional
    )

    print("\nSimulation Result:")
    print(f"Success: {result['success']}")
    if result['success']:
        print(f"Generated Calldata: {result['calldata']}")
        if result['wallet_calldata']:
            print(f"Wallet Calldata Match: {result['calldata_matches']}")
        
        print(f"\nTransaction Details:")
        print(f"Spender: {result['spender']}")
        print(f"Approval Requested: {result['amount_normalized']} USDC ({result['amount_raw']} raw)")  # What you thought you were approving
        print(f"Actual Approval: {result['executed_amount_normalized']} USDC")  # What would really execute
        print(f"Gas Estimate: {result['gas_estimate']}")
        print(f"Infinite Approval: {result['infinite_approval']}")
        
        print(f"\nState Changes:")
        print(f"Initial Allowance: {result['initial_allowance_normalized']} USDC")
        print(f"Final Allowance: {result['final_allowance_normalized']} USDC")
        print(f"Transaction Successful: {result['transaction_successful']}")
        print(f"Transaction Hash: {result['tx_hash']}")
    else:
        print(f"Error: {result['error']}")