Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """contract-guard engine — pre-interaction risk signals for an EVM contract/token. | |
| Pure JSON-RPC (no signing, no upstream cost). Given a contract address it reports: | |
| - is it a contract (eth_getCode) or an EOA / self-destructed | |
| - upgradeable proxy detection (EIP-1967 implementation + admin slots) | |
| - ERC20 metadata (name / symbol / decimals / totalSupply via eth_call) | |
| - a risk score + human-readable flags an agent can act on before interacting | |
| Why agents pay for this: the on-chain-data category is the busiest on the x402 | |
| discovery layer, but it's all raw RPC proxies. This adds a security verdict | |
| (mutable-logic proxy = rug vector, non-standard token, dead address) on top. | |
| """ | |
| import json | |
| import urllib.request | |
| RPCS = { | |
| "base": ["https://mainnet.base.org", "https://base-rpc.publicnode.com", "https://base.drpc.org"], | |
| "ethereum": ["https://ethereum-rpc.publicnode.com", "https://eth.drpc.org"], | |
| } | |
| CHAIN_IDS = {"base": 8453, "ethereum": 1} | |
| # EIP-1967 storage slots (+ legacy zeppelinos slot used by older proxies e.g. USDC) | |
| SLOT_IMPL = "0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc" | |
| SLOT_ADMIN = "0xb53127684a568b3173ae13b9f8a6016e243e63b6e8ee1178d6a717850b5d6103" | |
| SLOT_LEGACY = "0x7050c9e0f4ca769c69bd3a8ef740bc37934f8e2c036e5a723fd8ee048ed3f8c3" | |
| # keccak256("eip1967.proxy.beacon") - 1. Every constant in this file was re-derived | |
| # 2026-08-05 (see verify_constants below) after three mislabelled selectors were found in | |
| # the sibling tokenguard engine. | |
| SLOT_BEACON = "0xa3f0ad74e5423aebfd80d3ef4346578335a9a72aeaee59ff6cb3582b35133d50" | |
| # Common ERC20 view selectors | |
| SEL = { | |
| "name": "0x06fdde03", | |
| "symbol": "0x95d89b41", | |
| "decimals": "0x313ce567", | |
| "totalSupply": "0x18160ddd", | |
| } | |
| SEL_IMPLEMENTATION = "0x5c60da1b" # implementation() — asked of a beacon contract | |
| def _rpc(rpc_urls, method, params): | |
| if isinstance(rpc_urls, str): | |
| rpc_urls = [rpc_urls] | |
| body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params}).encode() | |
| last = None | |
| for url in rpc_urls: | |
| try: | |
| req = urllib.request.Request(url, data=body, headers={ | |
| "Content-Type": "application/json", | |
| "User-Agent": "Mozilla/5.0 (contract-guard x402)", | |
| }, method="POST") | |
| with urllib.request.urlopen(req, timeout=12) as r: | |
| d = json.loads(r.read().decode()) | |
| if "error" in d: | |
| last = RuntimeError(d["error"].get("message", "rpc error")) | |
| continue | |
| return d.get("result") | |
| except Exception as e: | |
| last = e | |
| continue | |
| if last: | |
| raise last | |
| return None | |
| def _eth_call(rpc_url, to, data): | |
| try: | |
| return _rpc(rpc_url, "eth_call", [{"to": to, "data": data}, "latest"]) | |
| except Exception: | |
| return None | |
| def _dec_uint(hexstr): | |
| if not hexstr or hexstr == "0x": | |
| return None | |
| try: | |
| return int(hexstr, 16) | |
| except Exception: | |
| return None | |
| def _dec_string(hexstr): | |
| """Decode an ABI string return, falling back to bytes32-style tokens.""" | |
| if not hexstr or hexstr == "0x": | |
| return None | |
| raw = bytes.fromhex(hexstr[2:]) | |
| # dynamic string: [offset(32)][length(32)][data] | |
| if len(raw) >= 64: | |
| try: | |
| length = int.from_bytes(raw[32:64], "big") | |
| if 0 < length <= len(raw) - 64: | |
| s = raw[64:64 + length].decode("utf-8", "replace").strip("\x00") | |
| if s: | |
| return s | |
| except Exception: | |
| pass | |
| # bytes32 fallback (non-standard tokens) | |
| s = raw.rstrip(b"\x00").decode("utf-8", "replace").strip("\x00") | |
| return s or None | |
| def _addr_from_slot(hexstr): | |
| if not hexstr: | |
| return None | |
| h = hexstr[2:].rjust(64, "0") | |
| addr = "0x" + h[-40:] | |
| return None if int(addr, 16) == 0 else addr.lower() | |
| def analyze(address, chain="base"): | |
| chain = (chain or "base").lower() | |
| if chain not in RPCS: | |
| return {"error": "unsupported chain '%s' (use: %s)" % (chain, ", ".join(RPCS))} | |
| a = (address or "").strip() | |
| if not (a.startswith("0x") and len(a) == 42): | |
| try: | |
| int(a, 16) | |
| except Exception: | |
| return {"error": "invalid EVM address"} | |
| return {"error": "invalid EVM address (expected 0x + 40 hex)"} | |
| rpc = RPCS[chain] | |
| flags = [] | |
| score = 0 | |
| code = _rpc(rpc, "eth_getCode", [a, "latest"]) | |
| is_contract = bool(code) and code != "0x" | |
| bytecode_len = (len(code) - 2) // 2 if is_contract else 0 | |
| if not is_contract: | |
| return { | |
| "address": a, "chain": chain, "chain_id": CHAIN_IDS[chain], | |
| "is_contract": False, | |
| "risk_level": "NOT_A_CONTRACT", | |
| "risk_score": 0, | |
| "flags": ["Address has no bytecode — it is an EOA (wallet) or a self-destructed contract, not a token/contract."], | |
| "summary": "Not a contract.", | |
| } | |
| # EIP-7702 delegated EOA: bytecode is exactly 0xef0100 + 20-byte delegate address | |
| if code[:8].lower() == "0xef0100": | |
| delegate = "0x" + code[8:48].lower() | |
| return { | |
| "address": a, "chain": chain, "chain_id": CHAIN_IDS[chain], | |
| "is_contract": False, | |
| "eip7702_delegated": True, | |
| "delegate": delegate, | |
| "risk_level": "HIGH", | |
| "risk_score": 45, | |
| "flags": ["EIP-7702 delegated EOA — this wallet has set code delegating control to contract %s. The delegate contract can move assets per the wallet's authorization; treat as smart-account, not a plain wallet." % delegate], | |
| "summary": "HIGH — EIP-7702 delegated EOA -> %s" % delegate, | |
| } | |
| # Proxy detection: EIP-1967 (impl + admin + beacon) and the legacy zeppelinos slot. | |
| impl = _addr_from_slot(_rpc(rpc, "eth_getStorageAt", [a, SLOT_IMPL, "latest"])) | |
| admin = _addr_from_slot(_rpc(rpc, "eth_getStorageAt", [a, SLOT_ADMIN, "latest"])) | |
| if impl is None: | |
| impl = _addr_from_slot(_rpc(rpc, "eth_getStorageAt", [a, SLOT_LEGACY, "latest"])) | |
| # Beacon proxies were previously invisible here: only the implementation and legacy slots | |
| # were read, so a BeaconProxy — whose implementation address lives on a separate beacon | |
| # contract, not in its own storage — was reported is_proxy=False. Confirmed on three | |
| # verified BeaconProxy contracts on Base (0xDc470dC0…, 0x210E7481…, 0x644435a9…), all of | |
| # which have a populated beacon slot and were being called "not a proxy". That is a false | |
| # negative on the single risk signal this route exists to report. | |
| beacon = None | |
| if impl is None: | |
| beacon = _addr_from_slot(_rpc(rpc, "eth_getStorageAt", [a, SLOT_BEACON, "latest"])) | |
| if beacon is not None: | |
| # The beacon holds the current implementation behind implementation() (0x5c60da1b). | |
| # Resolving it is best-effort: a beacon that does not answer is still a proxy. | |
| impl = _addr_from_slot(_eth_call(rpc, beacon, SEL_IMPLEMENTATION)) | |
| is_proxy = impl is not None or beacon is not None | |
| if is_proxy: | |
| score += 40 | |
| if beacon is not None: | |
| flags.append( | |
| "Upgradeable proxy (EIP-1967 beacon): the beacon owner can swap the " | |
| "implementation for EVERY proxy pointed at it, so this contract's logic can " | |
| "change without any transaction touching this address. beacon=%s%s" | |
| % (beacon, (" impl=%s" % impl) if impl else " (implementation unresolved)") | |
| ) | |
| else: | |
| flags.append( | |
| "Upgradeable proxy (EIP-1967): the admin can swap the implementation, so " | |
| "token/transfer logic can change after you interact. impl=%s%s" | |
| % (impl, (" admin=%s" % admin) if admin else "") | |
| ) | |
| # ERC20 metadata | |
| name = _dec_string(_eth_call(rpc, a, SEL["name"])) | |
| symbol = _dec_string(_eth_call(rpc, a, SEL["symbol"])) | |
| decimals = _dec_uint(_eth_call(rpc, a, SEL["decimals"])) | |
| supply = _dec_uint(_eth_call(rpc, a, SEL["totalSupply"])) | |
| looks_erc20 = symbol is not None and decimals is not None | |
| if not looks_erc20: | |
| score += 15 | |
| flags.append("No standard ERC20 metadata (symbol/decimals) — non-standard token or a non-token contract; verify intent before approving/swapping.") | |
| if bytecode_len < 200: | |
| score += 10 | |
| flags.append("Very small bytecode (%d bytes) — minimal/forwarder contract; confirm it does what you expect." % bytecode_len) | |
| score = min(100, score) | |
| level = "CRITICAL" if score >= 50 else "HIGH" if score >= 40 else "MEDIUM" if score >= 15 else "LOW" if score > 0 else "OK" | |
| return { | |
| "address": a, "chain": chain, "chain_id": CHAIN_IDS[chain], | |
| "is_contract": True, | |
| "bytecode_bytes": bytecode_len, | |
| "is_proxy": is_proxy, | |
| "proxy_kind": ("beacon" if beacon else "eip1967" if is_proxy else None), | |
| "implementation": impl, | |
| "beacon": beacon, | |
| "admin": admin, | |
| "token": {"name": name, "symbol": symbol, "decimals": decimals, "total_supply": str(supply) if supply is not None else None}, | |
| "looks_erc20": looks_erc20, | |
| "risk_level": level, | |
| "risk_score": score, | |
| "flags": flags or ["No elevated risk signals from on-chain checks. (Not a substitute for a full audit.)"], | |
| "summary": "%s — %s%s" % ( | |
| level, | |
| ("proxy " if is_proxy else "") + (("%s (%s)" % (name, symbol)) if symbol else "contract"), | |
| " | %d risk flag(s)" % len(flags) if flags else "", | |
| ), | |
| } | |
| def verify_constants() -> int: | |
| """Re-derive every magic constant in this file and check it. | |
| The sibling tokenguard engine shipped three mislabelled keccak selectors for months | |
| because nothing recomputed them, and they made its headline check report both false | |
| positives and false negatives. Same author, same class of risk, and this route is the | |
| busiest thing we serve — so the constants here are checkable by machine, not by eye. | |
| keccak is imported lazily so the deployed Space keeps its dependency-free engine. | |
| Run: python engine_server.py verify-constants | |
| """ | |
| from eth_utils import keccak | |
| def slot(text, minus_one=True): | |
| v = int.from_bytes(keccak(text=text), "big") | |
| if minus_one: | |
| v -= 1 | |
| return "0x" + v.to_bytes(32, "big").hex() | |
| expected = [ | |
| ("SLOT_IMPL", SLOT_IMPL, slot("eip1967.proxy.implementation")), | |
| ("SLOT_ADMIN", SLOT_ADMIN, slot("eip1967.proxy.admin")), | |
| ("SLOT_BEACON", SLOT_BEACON, slot("eip1967.proxy.beacon")), | |
| ("SLOT_LEGACY", SLOT_LEGACY, slot("org.zeppelinos.proxy.implementation", minus_one=False)), | |
| ("SEL_IMPLEMENTATION", SEL_IMPLEMENTATION, "0x" + keccak(text="implementation()")[:4].hex()), | |
| ] | |
| for name, sig in (("name", "name()"), ("symbol", "symbol()"), | |
| ("decimals", "decimals()"), ("totalSupply", "totalSupply()")): | |
| expected.append((f"SEL[{name}]", SEL[name], "0x" + keccak(text=sig)[:4].hex())) | |
| bad = 0 | |
| for label, have, real in expected: | |
| if have.lower() == real.lower(): | |
| print(f"ok {label:20s} {have}") | |
| else: | |
| bad += 1 | |
| print(f"WRONG {label:20s} have {have}\n {'':20s} real {real}") | |
| print(f"\n{len(expected) - bad}/{len(expected)} constants verify" | |
| + ("" if bad == 0 else f" — {bad} WRONG")) | |
| return 1 if bad else 0 | |
| if __name__ == "__main__": | |
| import sys | |
| if len(sys.argv) > 1 and sys.argv[1] == "verify-constants": | |
| raise SystemExit(verify_constants()) | |
| addr = sys.argv[1] if len(sys.argv) > 1 else "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" | |
| chain = sys.argv[2] if len(sys.argv) > 2 else "base" | |
| print(json.dumps(analyze(addr, chain), indent=2)) | |