""" Wallet Drain Scanner ===================== Scans wallet addresses for dangerous token approvals and signatures. Identifies unlimited spending approvals, phishing permit signatures, and drain contract vulnerabilities before users sign malicious transactions. Signals detected: - Unlimited ERC20/SPL spending approvals (max uint256) - Phishing permit signatures (eip-2612, dApprove) - Known drainer contract interactions - Suspicious approve/transferFrom patterns - Infinite NFT approvals (opensea/wyvern patterns) - High-risk delegate calls to untrusted contracts - Suspicious setApprovalForAll calls to known-bad operators Tier: Premium ($0.05) Endpoint: POST /api/v1/x402-tools/wallet_drain """ import logging import re from datetime import UTC, datetime from typing import Any logger = logging.getLogger("wallet_drain_scanner") # ── Free API sources ───────────────────────────────────────────── ETHERSCAN_API = ( "https://api.etherscan.io/api?module=account&action=txlist&address={}&sort=desc&limit=100" ) BASESCAN_API = ( "https://api.basescan.org/api?module=account&action=txlist&address={}&sort=desc&limit=100" ) BSCSCAN_API = ( "https://api.bscscan.com/api?module=account&action=txlist&address={}&sort=desc&limit=100" ) SOLSCAN_API = "https://api.solscan.io/account/tokens?address={}" DEFILLAMA_TOKEN_APPROVALS = "https://coins.llama.fi/approvals/ethereum/{}" # Known drainer contract address patterns (simplified - real DB would be larger) KNOWN_DRAINER_SIGNATURES = { "0x095ea7b3": "approve(address,uint256)", # ERC20 approve "0xa22cb465": "setApprovalForAll(address,bool)", # ERC721 approval "0x7ecebe00": "permit", # EIP-2612 permit "0xdc7f0124": "transferFrom(address,address,uint256)", # Standard transfer "0x23b872dd": "dApprove", # DAI approve variant "0x2e1a7d4d": "withdraw(uint256)", # WETH withdraw "0x791ac947": "drain", # Common drain function "0x6d0f8c6f": "sweep", # Common sweep function "0xb6734e7e": "emergencyWithdraw", # Common rug function "0x4f6ccce7": "requestWithdraw", # Withdraw request "0x304fc1f": "batchTransfer(address[],uint256)", # Batch drain } # Unlimited approval constant MAX_UINT256 = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF MAX_UINT128 = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF # URL safety regex - stricter than basic to prevent SSRF bypasses _URL_SAFE = re.compile( r"^https?://[a-zA-Z0-9](?:[a-zA-Z0-9-]*[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]*[a-zA-Z0-9])?)*(?::\d{1,5})?(?:/[^\s\"<>]*)?$" ) def _validate_url(url: str) -> bool: """Basic URL validation to prevent SSRF / injection.""" return bool(_URL_SAFE.match(url)) # ── Core scoring engine ───────────────────────────────────────── def _compute_drain_score( unlimited_approvals: int, known_drainer_txs: int, suspicious_permit_count: int, max_approval_amount_score: float, nft_unlimited_approvals: int, delegate_call_count: int, ) -> float: """ Compute a 0-100 wallet drain risk score. Factors: - unlimited_approvals (25%): Number of max-uint256 approvals - known_drainer_txs (25%): Transactions interacting with known drainer contracts - suspicious_permit_count (20%): Off-chain signature approvals (permit) - max_approval_amount_score (10%): Magnitude of largest approval - nft_unlimited_approvals (10%): Unlimited operator approvals - delegate_call_count (10%): Suspicious delegate calls """ unlimited_score = min(unlimited_approvals * 8, 25) drainer_score = min(known_drainer_txs * 10, 25) permit_score = min(suspicious_permit_count * 5, 20) amount_score = min(max_approval_amount_score * 10, 10) nft_score = min(nft_unlimited_approvals * 5, 10) delegate_score = min(delegate_call_count * 3, 10) raw = unlimited_score + drainer_score + permit_score + amount_score + nft_score + delegate_score return min(raw, 100.0) def _classify_drain_risk(score: float) -> str: """Classify score into risk category.""" if score >= 70: return "critical" if score >= 50: return "high" if score >= 30: return "moderate" if score >= 10: return "low" return "none" def _generate_recommendation(score: float, dangerous_approval_ratio: float) -> str: """Generate human-readable recommendation based on score and ratio.""" if score >= 70: return ( "CRITICAL: Wallet has multiple unlimited approvals and/or known drainer " "interactions. Revoke ALL approvals immediately. Use revoke.cash or " "similar tools. Consider this wallet compromised." ) if score >= 50: return ( "HIGH: Significant exposure detected. Revoke unlimited approvals and " "review permit signatures. Wallet may be at elevated risk of drain." ) if score >= 30: return ( "MODERATE: Some risky approvals detected. Review and revoke any " "unnecessary token approvals, especially to untrusted contracts." ) if score >= 10: return ( "LOW: Minor approval risks detected. Review approvals and consider " "revoking unused ones as a security best practice." ) return "No drain risks detected. Wallet approval hygiene appears healthy." # ── Main detector class ───────────────────────────────────────── class WalletDrainScanner: """Scans a wallet address for dangerous approvals and drain vulnerabilities.""" def __init__(self, address: str, chain: str = "ethereum"): if not re.match(r"^0x[a-fA-F0-9]{40}$", address) and not re.match( r"^[1-9A-HJ-NP-Za-km-z]{32,44}$", address ): raise ValueError(f"Invalid address format: {address[:20]}") self.address = address.lower() self.chain = chain.lower() self.is_evm = self.chain in ("ethereum", "base", "bsc", "polygon", "arbitrum", "optimism") self.is_solana = self.chain in ("solana",) async def scan(self) -> dict[str, Any]: """ Full wallet drain scan. Returns: dict with drain analysis, risk score, classification, dangerous approvals list, and recommendations. """ result: dict[str, Any] = { "address": self.address, "chain": self.chain, "scanned_at": datetime.now(UTC).isoformat(), "risk_score": 0.0, "risk_level": "none", "total_approvals_scanned": 0, "unlimited_approvals": 0, "known_drainer_interactions": 0, "suspicious_permits": 0, "nft_unlimited_approvals": 0, "dangerous_approvals": [], "recommendation": "", } if self.is_evm: await self._scan_evm(result) elif self.is_solana: await self._scan_solana(result) else: result["error"] = f"Chain {self.chain} not supported for drain scanning" # Compute score regardless of data richness score = _compute_drain_score( unlimited_approvals=result["unlimited_approvals"], known_drainer_txs=result["known_drainer_interactions"], suspicious_permit_count=result["suspicious_permits"], max_approval_amount_score=min(result["unlimited_approvals"] / 5, 1.0) if result["total_approvals_scanned"] > 0 else 0.0, nft_unlimited_approvals=result["nft_unlimited_approvals"], delegate_call_count=0, ) result["risk_score"] = round(score, 1) result["risk_level"] = _classify_drain_risk(score) dangerous_ratio = ( result["unlimited_approvals"] / result["total_approvals_scanned"] if result["total_approvals_scanned"] > 0 else 0.0 ) result["recommendation"] = _generate_recommendation(score, dangerous_ratio) return result async def _scan_evm(self, result: dict[str, Any]) -> None: """Scan EVM wallet for dangerous approvals via on-chain data.""" # Try to get transaction list from block explorer explorer_txs = await self._fetch_explorer_txs() approvals_seen = set() drainer_hits = set() for tx in explorer_txs: input_data = tx.get("input", "") if not input_data or input_data == "0x": continue # Parse function signature (first 4 bytes = selector) sig = input_data[:10] if len(input_data) >= 10 else "" # Check for known drainer/approval patterns if sig in KNOWN_DRAINER_SIGNATURES: KNOWN_DRAINER_SIGNATURES[sig] if sig == "0x095ea7b3": # approve(address,uint256) result["total_approvals_scanned"] += 1 try: # Approve selector + address(32 bytes) + amount(32 bytes) # offset is after sig (10 chars) + 24 bytes padding value_hex = "0x" + input_data[10 + 64 : 10 + 128] # amount in last 32 bytes value = int(value_hex, 16) spender = "0x" + input_data[34:74] # address after padding is_unlimited = value >= MAX_UINT128 if is_unlimited: result["unlimited_approvals"] += 1 if spender not in approvals_seen: approvals_seen.add(spender) result["dangerous_approvals"].append( { "type": "unlimited_approve", "spender": spender, "value": "unlimited" if value >= MAX_UINT256 else f"{value} (near-max)", "tx_hash": tx.get("hash", ""), } ) # Check if spender is a known drainer if self._is_known_drainer(spender): drainer_hits.add(spender) result["known_drainer_interactions"] += 1 result["dangerous_approvals"].append( { "type": "drainer_approval", "spender": spender, "value": "unlimited" if is_unlimited else value, "tx_hash": tx.get("hash", ""), } ) except (ValueError, IndexError): pass elif sig == "0xa22cb465": # setApprovalForAll result["total_approvals_scanned"] += 1 result["nft_unlimited_approvals"] += 1 operator = "0x" + input_data[34:74] if len(input_data) >= 74 else "unknown" result["dangerous_approvals"].append( { "type": "unlimited_nft_approval", "operator": operator, "tx_hash": tx.get("hash", ""), } ) if self._is_known_drainer(operator): drainer_hits.add(operator) result["known_drainer_interactions"] += 1 elif sig == "0x7ecebe00": # permit result["suspicious_permits"] += 1 # Check for known drainer contracts in recent interactions result["drainer_contracts_found"] = list(drainer_hits) if drainer_hits else [] async def _scan_solana(self, result: dict[str, Any]) -> None: """Scan Solana wallet for dangerous token approvals.""" import aiohttp try: # Use Solscan API to get token accounts url = f"https://api.solscan.io/account/tokens?address={self.address}" async with aiohttp.ClientSession() as session, session.get( url, headers={"Accept": "application/json"}, timeout=aiohttp.ClientTimeout(total=10), ) as resp: if resp.status == 200: data = await resp.json() tokens = data.get("data", []) result["total_approvals_scanned"] = len(tokens) # On Solana, look for token accounts with delegate set for token in tokens: delegate = token.get("delegate", "") if delegate and delegate != "": result["unlimited_approvals"] += 1 result["dangerous_approvals"].append( { "type": "delegated_token", "mint": token.get( "tokenAddress", token.get("token_address", "") ), "delegate": delegate, "amount": token.get("amount", "unlimited"), } ) except Exception as e: logger.debug(f"Solana scan failed: {e}") async def _fetch_explorer_txs(self) -> list[dict]: """Fetch recent transactions from block explorer API.""" explorer_urls = { "ethereum": ETHERSCAN_API.format(self.address), "base": BASESCAN_API.format(self.address), "bsc": BSCSCAN_API.format(self.address), } url = explorer_urls.get(self.chain) if not url: return [] if not _validate_url(url): return [] import aiohttp try: async with aiohttp.ClientSession() as session: # noqa: SIM117 async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp: if resp.status == 200: data = await resp.json() if data.get("status") == "1": return data.get("result", []) except Exception as e: logger.debug(f"Explorer fetch failed for {self.chain}: {e}") return [] @staticmethod def _is_known_drainer(address: str) -> bool: """ Check if an address is a known drainer contract. Uses pattern matching on known drainer addresses. In production, this would query a drainer database. Here we use heuristic checks. """ addr_lower = address.lower() # Known drainer prefixes (red flags - in production use a full DB) drainer_prefixes = [ "0x0000", # Null address tricks "0xdead", # Death address used in scams "0xdef1", # Known exploit deployer ] # Check if address is a well-known benign contract benign_contracts = { "0x7a250d5630b4cf539739df2c5dacb4c659f2488d", # Uniswap V2 Router "0x1f9840a85d5af5bf1d1762f925bdaddc4201f984", # Uniswap V3 "0xd9e1ce17f2641f24ae83637ab66a2cca9c378b9f", # SushiSwap Router "0xdef1c0ded9bec7f1a1670819833240f027b25eff", # 0x Exchange Proxy "0x881d40237659c251811cec9c364ef91dc08d300c", # Metamask Swap "0x68b3465833fb72a70ecdf485e0e4c7bd8665fc45", # Uniswap Universal Router "0x3fc91a3afd70395cd496c647d5a6cc9d4b2b7fad", # Uniswap Universal Router V2 } if addr_lower in benign_contracts: return False # Check heuristic patterns return any(addr_lower.startswith(prefix) for prefix in drainer_prefixes)