|
| 1 | +import argparse |
| 2 | +import os |
| 3 | +from typing import Optional |
| 4 | + |
| 5 | +import opengradient as og |
| 6 | +from x402v2.mechanisms.evm.constants import PERMIT2_ADDRESS |
| 7 | +from web3 import Web3 |
| 8 | + |
| 9 | +BASE_OPG_ADDRESS = "0x240b09731D96979f50B2C649C9CE10FcF9C7987F" |
| 10 | +BASE_SEPOLIA_RPC = "https://sepolia.base.org" |
| 11 | +BASE_TESTNET = "base-testnet" |
| 12 | +MAX_UINT256 = (1 << 256) - 1 |
| 13 | + |
| 14 | +ERC20_ABI = [ |
| 15 | + { |
| 16 | + "inputs": [ |
| 17 | + {"name": "owner", "type": "address"}, |
| 18 | + {"name": "spender", "type": "address"}, |
| 19 | + ], |
| 20 | + "name": "allowance", |
| 21 | + "outputs": [{"name": "", "type": "uint256"}], |
| 22 | + "stateMutability": "view", |
| 23 | + "type": "function", |
| 24 | + }, |
| 25 | + { |
| 26 | + "inputs": [ |
| 27 | + {"name": "spender", "type": "address"}, |
| 28 | + {"name": "amount", "type": "uint256"}, |
| 29 | + ], |
| 30 | + "name": "approve", |
| 31 | + "outputs": [{"name": "", "type": "bool"}], |
| 32 | + "stateMutability": "nonpayable", |
| 33 | + "type": "function", |
| 34 | + }, |
| 35 | +] |
| 36 | + |
| 37 | + |
| 38 | +def _get_base_sepolia_web3() -> Web3: |
| 39 | + return Web3(Web3.HTTPProvider(BASE_SEPOLIA_RPC)) |
| 40 | + |
| 41 | + |
| 42 | +def get_permit2_allowance(client_address: str) -> int: |
| 43 | + w3 = _get_base_sepolia_web3() |
| 44 | + token = w3.eth.contract(address=Web3.to_checksum_address(BASE_OPG_ADDRESS), abi=ERC20_ABI) |
| 45 | + return token.functions.allowance( |
| 46 | + Web3.to_checksum_address(client_address), |
| 47 | + Web3.to_checksum_address(PERMIT2_ADDRESS), |
| 48 | + ).call() |
| 49 | + |
| 50 | + |
| 51 | +def check_permit2_approval(client_address: str, network: str) -> None: |
| 52 | + """Raise an error if Permit2 approval is missing for base-testnet.""" |
| 53 | + if network != BASE_TESTNET: |
| 54 | + return |
| 55 | + |
| 56 | + allowance = get_permit2_allowance(client_address) |
| 57 | + print(f"Current OPG Permit2 allowance: {allowance}") |
| 58 | + |
| 59 | + if allowance == 0: |
| 60 | + raise RuntimeError( |
| 61 | + f"ERROR: No Permit2 approval found for address {client_address}. " |
| 62 | + f"Approve Permit2 ({PERMIT2_ADDRESS}) to spend OPG ({BASE_OPG_ADDRESS}) " |
| 63 | + "on Base Sepolia before using x402 payments." |
| 64 | + ) |
| 65 | + |
| 66 | + |
| 67 | +def grant_permit2_approval( |
| 68 | + private_key: str, |
| 69 | + amount: int = MAX_UINT256, |
| 70 | + gas_multiplier: float = 1.2, |
| 71 | + nonce: Optional[int] = None, |
| 72 | +) -> str: |
| 73 | + """Send ERC-20 approve(spender=Permit2, amount=amount) for OPG on Base Sepolia.""" |
| 74 | + w3 = _get_base_sepolia_web3() |
| 75 | + |
| 76 | + account = w3.eth.account.from_key(private_key) |
| 77 | + token = w3.eth.contract(address=Web3.to_checksum_address(BASE_OPG_ADDRESS), abi=ERC20_ABI) |
| 78 | + |
| 79 | + tx_nonce = nonce |
| 80 | + if tx_nonce is None: |
| 81 | + tx_nonce = w3.eth.get_transaction_count(account.address, "pending") |
| 82 | + |
| 83 | + approve_fn = token.functions.approve(Web3.to_checksum_address(PERMIT2_ADDRESS), amount) |
| 84 | + estimated_gas = approve_fn.estimate_gas({"from": account.address}) |
| 85 | + |
| 86 | + tx = approve_fn.build_transaction( |
| 87 | + { |
| 88 | + "from": account.address, |
| 89 | + "nonce": tx_nonce, |
| 90 | + "gas": int(estimated_gas * gas_multiplier), |
| 91 | + "gasPrice": w3.eth.gas_price, |
| 92 | + "chainId": w3.eth.chain_id, |
| 93 | + } |
| 94 | + ) |
| 95 | + |
| 96 | + signed = account.sign_transaction(tx) |
| 97 | + tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction) |
| 98 | + receipt = w3.eth.wait_for_transaction_receipt(tx_hash) |
| 99 | + |
| 100 | + if receipt.status != 1: |
| 101 | + raise RuntimeError(f"Permit2 approval transaction failed: {tx_hash.hex()}") |
| 102 | + |
| 103 | + return tx_hash.hex() |
| 104 | + |
| 105 | + |
| 106 | +def _parse_args() -> argparse.Namespace: |
| 107 | + parser = argparse.ArgumentParser(description="Grant Permit2 approval to spend OPG on Base Sepolia.") |
| 108 | + parser.add_argument( |
| 109 | + "--amount", |
| 110 | + type=int, |
| 111 | + default=MAX_UINT256, |
| 112 | + help="Approval amount in base units (default: max uint256).", |
| 113 | + ) |
| 114 | + return parser.parse_args() |
| 115 | + |
| 116 | + |
| 117 | +def main() -> None: |
| 118 | + args = _parse_args() |
| 119 | + |
| 120 | + private_key = os.environ.get("OG_PRIVATE_KEY") |
| 121 | + if not private_key: |
| 122 | + raise RuntimeError("OG_PRIVATE_KEY is not set.") |
| 123 | + |
| 124 | + client = og.Client(private_key=private_key) |
| 125 | + wallet_address = client.wallet_address |
| 126 | + |
| 127 | + before = get_permit2_allowance(wallet_address) |
| 128 | + print(f"Wallet: {wallet_address}") |
| 129 | + print(f"Permit2: {PERMIT2_ADDRESS}") |
| 130 | + print(f"OPG Token: {BASE_OPG_ADDRESS}") |
| 131 | + print(f"Allowance before: {before}") |
| 132 | + |
| 133 | + tx_hash = grant_permit2_approval(private_key=private_key, amount=args.amount) |
| 134 | + after = get_permit2_allowance(wallet_address) |
| 135 | + |
| 136 | + print(f"Approval transaction hash: {tx_hash}") |
| 137 | + print(f"Allowance after: {after}") |
| 138 | + |
| 139 | + |
| 140 | +if __name__ == "__main__": |
| 141 | + main() |
0 commit comments