#!/usr/bin/env python3 """ RMI Tool Fingerprinter - Best-in-Class Scam Infrastructure Detection ===================================================================== Identifies the TOOLS behind scams, not just the outcomes. Phase 3-5 capabilities per the Enhanced Report V2 standards. Detects: - Smithii bundler patterns (bundle + liquidity + first swap) - Printr bundler signatures (multi-wallet coordinated launch) - LaunchLab bundle bot (specific instruction ordering) - Jito bundle detection (tip program, bundle construction) - PumpFun sniper bots (limit-sniper, pumpfun-bonkfun-bot, etc.) - Volume bot / wash trading patterns - Wallet aging counter-detection - Cross-chain fund obfuscation detection - Exchange-funded deployer detection - First buyer concentration analysis - Scanner-aware evasion detection (Hide-and-Shill framework) Design principle: Tools change less frequently than tactics. A Smithii-bundled token has detectable on-chain fingerprints regardless of the scammer's chosen exit strategy. """ import logging from collections import defaultdict from dataclasses import dataclass from datetime import UTC, datetime from typing import Any import httpx logger = logging.getLogger("tool-fingerprinter") # ══════════════════════════════════════════════════════════════════════ # KNOWN TOOL FINGERPRINTS - from reverse-engineered scammer infrastructure # ══════════════════════════════════════════════════════════════════════ # Solana Program IDs used by known bundler/sniper tools TOOL_PROGRAMS = { "jito_bundle": [ "Jito4APyf6rDt1pD1jH3nD3is4v7TwzFNWjjMP7B2RK", # Jito bundles v3 "96gYZGLnJYVFmbjzopPSU6QiEV5fGqZNyN9nmNhvrZU5", # Jito tip router v3 "ADuCadRmgjMe6UzXVxR1Kn9CS2CXAf8bjiKkC4xFcegX", # Jito tip router v4 ], "pumpfun": [ "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P", # Pump.fun program "CebN5WGQ4jvEPvsVU4EoHEpgzq1VV6AbFtJxHTNBaFUx", # Pump AMM ], "moonshot": [ "MoonCVHWSSyvkWjUj7hDL14N1pVFHUjQyqWScmBw8D1r", ], } # Transaction instruction patterns that identify specific tools TOOL_SIGNATURES = { "smithii_bundler": { "description": "Smithii bundle bot - bundles liquidity add + first swap in single TX", "patterns": [ # Signature: single TX with create_pool + add_liquidity + swap in sequence "create_pool_add_liquidity_swap_single_tx", # Typical wallet count: 15-25 wallets all buying within 1 block "multi_wallet_same_block_15_25", # Liquidity pattern: exact same SOL amount per wallet "uniform_buy_amounts", ], "severity": 85, # 0-100, higher = more suspicious "confidence_required": 0.7, }, "printr_bundler": { "description": "Printr bundler - coordinated multi-wallet deployment with uniform distribution", "patterns": [ "deployer_program_derived_addresses", "wallets_funded_from_single_source", "identical_buy_timing_sub_second", "uniform_token_distribution_20_30_wallets", ], "severity": 90, "confidence_required": 0.7, }, "launchlab_bundle": { "description": "LaunchLab bundle bot - specific instruction ordering for token + LP creation", "patterns": [ "create_mint_create_ata_mint_to_create_pool_add_liquidity", "single_tx_8_12_instructions", "immediate_swap_after_liquidity", ], "severity": 88, "confidence_required": 0.7, }, "pumpfun_sniper": { "description": "PumpFun sniper bot - buys within milliseconds of bonding curve completion", "patterns": [ "bonding_curve_completion_detection", "sub_second_first_buy", "multiple_wallets_same_program_call", "jito_bundle_within_2_blocks", ], "severity": 70, "confidence_required": 0.6, }, "volume_bot": { "description": "Volume bot - self-trading to inflate volume metrics for trending algorithms", "patterns": [ "circular_trading_same_small_wallet_set", "buy_sell_same_block_no_profit", "volume_spike_no_holder_change", "identical_trade_sizes_repeated", "wallet_graph_fully_connected_clique", ], "severity": 80, "confidence_required": 0.65, }, "wallet_aging_evasion": { "description": "Scanner-aware countermeasure - wallets aged before launching scam token", "patterns": [ "wallet_created_weeks_before_first_scam_activity", "dormant_period_followed_by_intense_activity", "funded_but_idle_then_sudden_use", "real_wallet_activity_mix_then_scam_only", ], "severity": 75, "confidence_required": 0.6, }, "cross_chain_obfuscation": { "description": "Scanner-aware countermeasure - funds routed through multiple chains to hide origin", "patterns": [ "bridge_usage_to_break_tracking", "multiple_chain_funding_hops_3plus", "mixer_on_source_chain_before_bridge", "cex_deposit_on_chain_a_withdraw_on_chain_b", ], "severity": 85, "confidence_required": 0.65, }, } # Known scammer deployer patterns SCAMMER_DEPLOYER_PATTERNS = { "cex_funded_deployer": { "description": "Deployer funded directly from centralized exchange - high scam correlation", "cex_wallets": [ "binance", "coinbase", "kraken", "kucoin", "bybit", "okx", "gate.io", "mexc", "bitget", "htx", "bitfinex", ], "severity": 60, }, "mixer_funded_deployer": { "description": "Deployer funded through mixer/tumbler - extremely suspicious", "mixers": ["tornado", "cyclone", "typhoon", "wasabi", "samourai"], "severity": 95, }, "previous_scammer_deployer": { "description": "Deployer previously launched known scam tokens", "severity": 100, }, } # First buyer concentration thresholds FIRST_BUYER_THRESHOLDS = { "critical_concentration": 0.80, # First 5 buyers hold 80%+ = critical "high_concentration": 0.50, # First 10 buyers hold 50%+ = high risk "sniper_time_ms": 5000, # < 5 seconds from launch = sniper "coordinated_same_block": 5, # 5+ wallets buying in same block = coordinated } @dataclass class ToolFingerprintResult: """Result from tool fingerprinting analysis.""" tool_name: str detected: bool confidence: float # 0-1 evidence: list[str] severity: int # 0-100 description: str @dataclass class FirstBuyerAnalysis: """Analysis of first buyers for a token.""" token_address: str total_holders: int first_5_concentration_pct: float first_10_concentration_pct: float first_20_concentration_pct: float avg_entry_time_ms: float # from token creation fastest_entry_ms: float same_block_buyers: int coordinated_clusters: int # groups of wallets with shared funding cex_funded_buyers: int # buyers funded from CEX risk_level: str # critical, high, medium, low flags: list[str] @dataclass class DeployerProfile: """Profile of the token deployer wallet.""" address: str age_days: int funding_source: str | None funding_source_type: str # cex, dex, mixer, unknown previous_tokens_launched: int previous_scam_tokens: int cross_chain_activity: bool wallet_aging_detected: bool aging_score: float # 0-100 risk_score: int # 0-100 flags: list[str] # ══════════════════════════════════════════════════════════════════════ # TOOL FINGERPRINTER # ══════════════════════════════════════════════════════════════════════ class ToolFingerprinter: """ Identifies scammer tools and infrastructure from on-chain fingerprints. Analyzes transaction patterns, program interactions, wallet graphs, and temporal signatures to identify specific tools. """ def __init__(self): self._seen_txs: set[str] = set() self._known_scammers: set[str] = set() self._cex_wallets: set[str] = set() async def fingerprint_transaction(self, tx_data: dict, chain: str = "solana") -> list[ToolFingerprintResult]: """Analyze a single transaction for tool fingerprints.""" results = [] instructions = tx_data.get("instructions", tx_data.get("ix", [])) tx_data.get("accountKeys", tx_data.get("accounts", [])) program_ids = [ix.get("programId", "") for ix in instructions] tx_sig = tx_data.get("signature", tx_data.get("txHash", "")) if tx_sig in self._seen_txs: return results self._seen_txs.add(tx_sig) # Check Jito bundle jito_hits = [p for p in program_ids if p in TOOL_PROGRAMS["jito_bundle"]] if jito_hits: results.append( ToolFingerprintResult( tool_name="jito_bundle", detected=True, confidence=0.95, evidence=[f"Jito program invoked: {jito_hits[0]}"], severity=75, description="Jito bundle detected - transaction was privately submitted to avoid front-running", ) ) # Check PumpFun pf_hits = [p for p in program_ids if p in TOOL_PROGRAMS["pumpfun"]] if pf_hits: results.append( ToolFingerprintResult( tool_name="pumpfun_token", detected=True, confidence=0.99, evidence=[f"Pump.fun program call: {pf_hits[0]}"], severity=40, description="Pump.fun token - elevated risk due to low barrier to entry", ) ) return results async def detect_bundlers(self, tx_list: list[dict], token_address: str) -> list[ToolFingerprintResult]: """Detect bundler patterns across multiple transactions.""" results = [] evidence_smithii = [] evidence_printr = [] evidence_launchlab = [] if not tx_list: return results # Group by block block_groups = defaultdict(list) for tx in tx_list: block = tx.get("slot", tx.get("blockNumber", 0)) block_groups[block].append(tx) # Check each block for coordinated behavior for block, txs in block_groups.items(): unique_buyers = set() for tx in txs: signer = tx.get("signer", tx.get("from", "")) if signer: unique_buyers.add(signer) buyer_count = len(unique_buyers) # Smithii: 15-25 wallets in same block if 15 <= buyer_count <= 25: evidence_smithii.append( f"Block {block}: {buyer_count} wallets bought in same block (Smithii range: 15-25)" ) # Printr: 20-30 wallets in same block if 20 <= buyer_count <= 30: evidence_printr.append( f"Block {block}: {buyer_count} wallets bought in same block (Printr range: 20-30)" ) # LaunchLab: 8-12 instructions in a single tx for tx in txs: ix_count = len(tx.get("instructions", tx.get("ix", []))) if 8 <= ix_count <= 12: evidence_launchlab.append( f"TX {tx.get('signature', '?')[:8]}: {ix_count} instructions (LaunchLab range: 8-12)" ) if evidence_smithii: results.append( ToolFingerprintResult( tool_name="smithii_bundler", detected=True, confidence=min(0.95, 0.6 + 0.1 * len(evidence_smithii)), evidence=evidence_smithii, severity=85, description="Smithii bundler detected - coordinated bundle launch with 15-25 wallets", ) ) if evidence_printr: results.append( ToolFingerprintResult( tool_name="printr_bundler", detected=True, confidence=min(0.95, 0.6 + 0.1 * len(evidence_printr)), evidence=evidence_printr, severity=90, description="Printr bundler detected - 20-30 coordinated wallets from single deployer", ) ) if evidence_launchlab: results.append( ToolFingerprintResult( tool_name="launchlab_bundle", detected=True, confidence=min(0.95, 0.6 + 0.1 * len(evidence_launchlab)), evidence=evidence_launchlab, severity=88, description="LaunchLab bundle detected - 8-12 instruction complex bundle transaction", ) ) return results async def analyze_first_buyers( self, token_address: str, holders_data: list[dict], first_tx_timestamp: str | None = None ) -> FirstBuyerAnalysis: """Analyze first buyer concentration and patterns.""" flags = [] if not holders_data: return FirstBuyerAnalysis( token_address=token_address, total_holders=0, first_5_concentration_pct=0, first_10_concentration_pct=0, first_20_concentration_pct=0, avg_entry_time_ms=0, fastest_entry_ms=0, same_block_buyers=0, coordinated_clusters=0, cex_funded_buyers=0, risk_level="unknown", flags=["No holder data"], ) total_holders = len(holders_data) sorted_holders = sorted(holders_data, key=lambda h: h.get("first_buy_time", 0) or 0) # Supply concentration total_supply = sum(h.get("balance_pct", 0) for h in sorted_holders[:50]) if total_supply == 0: total_supply = 100 first_5 = sum(h.get("balance_pct", 0) for h in sorted_holders[:5]) / max(total_supply, 1) first_10 = sum(h.get("balance_pct", 0) for h in sorted_holders[:10]) / max(total_supply, 1) first_20 = sum(h.get("balance_pct", 0) for h in sorted_holders[:20]) / max(total_supply, 1) if first_5 > FIRST_BUYER_THRESHOLDS["critical_concentration"]: flags.append(f"CRITICAL: First 5 buyers hold {first_5 * 100:.0f}%") if first_10 > FIRST_BUYER_THRESHOLDS["high_concentration"]: flags.append(f"HIGH: First 10 buyers hold {first_10 * 100:.0f}%") # Entry timing entry_times = [h.get("first_buy_time", 0) or 0 for h in sorted_holders[:20]] entry_times = [t for t in entry_times if t > 0] avg_entry = sum(entry_times) / len(entry_times) if entry_times else 0 fastest = min(entry_times) if entry_times else 0 if fastest < FIRST_BUYER_THRESHOLDS["sniper_time_ms"]: flags.append(f"SNIPER: Fastest entry {fastest:.0f}ms after launch") # Same block detection blocks = set() for h in sorted_holders[:20]: block = h.get("first_buy_block") if block: blocks.add(block) same_block = total_holders - len(blocks) if total_holders > 0 else 0 if same_block >= FIRST_BUYER_THRESHOLDS["coordinated_same_block"]: flags.append(f"COORDINATED: {same_block} wallets bought in same block") # Funding source clustering funding_sources = defaultdict(int) for h in sorted_holders[:20]: funder = h.get("funding_source", "unknown") if funder: funding_sources[funder] += 1 coordinated = sum(1 for count in funding_sources.values() if count >= 3) if coordinated > 0: flags.append(f"COORDINATED CLUSTERS: {coordinated} groups share funding source") # Determine risk level if first_5 > 0.80 or len(flags) >= 4: risk_level = "critical" elif first_5 > 0.50 or len(flags) >= 2: risk_level = "high" elif len(flags) >= 1: risk_level = "medium" else: risk_level = "low" return FirstBuyerAnalysis( token_address=token_address, total_holders=total_holders, first_5_concentration_pct=round(first_5 * 100, 1), first_10_concentration_pct=round(first_10 * 100, 1), first_20_concentration_pct=round(first_20 * 100, 1), avg_entry_time_ms=round(avg_entry, 1), fastest_entry_ms=round(fastest, 1), same_block_buyers=same_block, coordinated_clusters=coordinated, cex_funded_buyers=0, risk_level=risk_level, flags=flags, ) async def profile_deployer(self, deployer_address: str, chain: str = "solana") -> DeployerProfile: """Profile the token deployer wallet for risk factors.""" flags = [] risk = 0 # Try to get deployer data age_days = 0 funding_source = None funding_type = "unknown" aging_detected = False aging_score = 0 prev_tokens = 0 prev_scams = 0 cross_chain = False try: async with httpx.AsyncClient(timeout=15) as client: # Basic Solana account info if chain == "solana": resp = await client.post( "https://api.mainnet-beta.solana.com", json={ "jsonrpc": "2.0", "id": 1, "method": "getSignaturesForAddress", "params": [deployer_address, {"limit": 50}], }, ) if resp.status_code == 200: sigs = resp.json().get("result", []) if sigs: first_sig = sigs[-1] first_ts = first_sig.get("blockTime", 0) if first_ts: age_days = (datetime.now(UTC) - datetime.fromtimestamp(first_ts, tz=UTC)).days prev_tokens = len( [ s for s in sigs if "create" in str(s.get("memo", "")).lower() or "token" in str(s.get("memo", "")).lower() ] ) except Exception as e: logger.warning(f"Failed to profile deployer {deployer_address}: {e}") # Fresh wallet scoring if age_days < 1: risk += 30 flags.append("BRAND_NEW: Wallet less than 1 day old") elif age_days < 7: risk += 20 flags.append("FRESH: Wallet less than 7 days old") elif age_days < 30: risk += 10 flags.append("NEW: Wallet less than 30 days old") # Wallet aging detection (countermeasure) if age_days > 30 and prev_tokens == 0: # Old wallet, first token launch = possible aged wallet aging_score = 40 flags.append("AGING_SUSPICIOUS: Old wallet launching first token") elif age_days > 90 and prev_tokens <= 1: aging_score = 60 aging_detected = True flags.append("AGING_DETECTED: Wallet aged 90+ days, first/second token only") risk += 25 # Repeated token launcher if prev_tokens > 5: risk += 15 flags.append(f"SERIAL_LAUNCHER: {prev_tokens} previous tokens") # Check if deployer appears in our known scam DB try: from app.scam_sources import KNOWN_SCAMS_EXPANDED if deployer_address in str(KNOWN_SCAMS_EXPANDED): risk += 50 prev_scams = 1 flags.append("KNOWN_SCAMMER: Previously identified in scam database") except Exception: pass # Determine funding type if funding_type == "cex": risk += 20 flags.append("CEX_FUNDED: Deployer funded from centralized exchange") return DeployerProfile( address=deployer_address, age_days=age_days, funding_source=funding_source, funding_source_type=funding_type, previous_tokens_launched=prev_tokens, previous_scam_tokens=prev_scams, cross_chain_activity=cross_chain, wallet_aging_detected=aging_detected, aging_score=aging_score, risk_score=min(100, risk), flags=flags, ) async def detect_volume_bots(self, trades: list[dict], token_address: str) -> list[ToolFingerprintResult]: """Detect volume bot / wash trading patterns.""" results = [] if not trades or len(trades) < 10: return results evidence = [] # Check for circular trading (same set of wallets trading among themselves) traders = set() trade_pairs = defaultdict(int) for t in trades: a, b = t.get("buyer", ""), t.get("seller", "") traders.add(a) traders.add(b) if a and b: pair = tuple(sorted([a, b])) trade_pairs[pair] += 1 # Clique detection: small number of wallets doing all trading if len(traders) < 10 and len(trades) > 50: evidence.append(f"Small trader set ({len(traders)} wallets) with {len(trades)} trades") # Repeated trades between same pairs heavy_pairs = {p: c for p, c in trade_pairs.items() if c > 5} if heavy_pairs: evidence.append(f"{len(heavy_pairs)} wallet pairs trading 5+ times each") # Identical trade sizes trade_sizes = [float(t.get("amount", 0)) for t in trades if t.get("amount")] if trade_sizes: unique_sizes = len(set(trade_sizes)) if unique_sizes < len(trade_sizes) * 0.3: evidence.append(f"Repetitive trade sizes: {unique_sizes} unique from {len(trade_sizes)} trades") if evidence: results.append( ToolFingerprintResult( tool_name="volume_bot", detected=True, confidence=min(0.95, 0.5 + 0.15 * len(evidence)), evidence=evidence, severity=80, description="Volume bot / wash trading detected - inflated metrics", ) ) return results async def full_token_scan( self, token_address: str, chain: str = "solana", deployer_address: str | None = None, holders: list[dict] | None = None, transactions: list[dict] | None = None, ) -> dict[str, Any]: """ Full token scan - combines all detection methods. Returns comprehensive risk assessment. """ import time start = time.time() results = { "token_address": token_address, "chain": chain, "tool_fingerprints": [], "first_buyer_analysis": None, "deployer_profile": None, "volume_bot_detected": False, "aggregate_risk_score": 0, "risk_category": "unknown", "flags": [], "scan_duration_ms": 0, } # Run all detectors in parallel tasks = [] if transactions: tasks.append(self.detect_bundlers(transactions, token_address)) if holders: tasks.append(self.analyze_first_buyers(token_address, holders)) if deployer_address: tasks.append(self.profile_deployer(deployer_address, chain)) if transactions: tasks.append(self.detect_volume_bots(transactions, token_address)) import asyncio gathered = await asyncio.gather(*tasks, return_exceptions=True) # Process results total_risk = 0 max_possible = 0 for result in gathered: if isinstance(result, Exception): continue if isinstance(result, list): # Tool fingerprints for item in result: if isinstance(item, ToolFingerprintResult) and item.detected: results["tool_fingerprints"].append( { "tool": item.tool_name, "confidence": round(item.confidence, 2), "severity": item.severity, "evidence": item.evidence, "description": item.description, } ) total_risk += item.severity max_possible += 100 results["flags"].append(f"TOOL:{item.tool_name}") elif isinstance(result, FirstBuyerAnalysis): fb = result results["first_buyer_analysis"] = { "total_holders": fb.total_holders, "first_5_concentration_pct": fb.first_5_concentration_pct, "first_10_concentration_pct": fb.first_10_concentration_pct, "first_20_concentration_pct": fb.first_20_concentration_pct, "fastest_entry_ms": fb.fastest_entry_ms, "avg_entry_time_ms": fb.avg_entry_time_ms, "same_block_buyers": fb.same_block_buyers, "coordinated_clusters": fb.coordinated_clusters, "risk_level": fb.risk_level, } if fb.risk_level == "critical": total_risk += 50 elif fb.risk_level == "high": total_risk += 30 max_possible += 50 results["flags"].extend(fb.flags) elif isinstance(result, DeployerProfile): dp = result results["deployer_profile"] = { "address": dp.address, "age_days": dp.age_days, "funding_source_type": dp.funding_source_type, "previous_tokens": dp.previous_tokens_launched, "previous_scams": dp.previous_scam_tokens, "wallet_aging_detected": dp.wallet_aging_detected, "aging_score": dp.aging_score, "risk_score": dp.risk_score, } total_risk += dp.risk_score max_possible += 100 results["flags"].extend(dp.flags) # Calculate aggregate risk pct = total_risk / max_possible * 100 if max_possible > 0 else 0 results["aggregate_risk_score"] = min(100, round(pct)) if pct >= 80: results["risk_category"] = "critical" elif pct >= 60: results["risk_category"] = "high" elif pct >= 35: results["risk_category"] = "medium" elif pct >= 15: results["risk_category"] = "low" else: results["risk_category"] = "minimal" results["scan_duration_ms"] = round((time.time() - start) * 1000) return results # ══════════════════════════════════════════════════════════════════════ # SINGLETON # ══════════════════════════════════════════════════════════════════════ _fingerprinter: ToolFingerprinter | None = None async def get_fingerprinter() -> ToolFingerprinter: global _fingerprinter if _fingerprinter is None: _fingerprinter = ToolFingerprinter() return _fingerprinter async def fingerprint_token( token_address: str, chain: str = "solana", deployer: str | None = None, holders: list[dict] | None = None, transactions: list[dict] | None = None, ) -> dict[str, Any]: """Convenience function for full token fingerprinting.""" fp = await get_fingerprinter() return await fp.full_token_scan( token_address=token_address, chain=chain, deployer_address=deployer, holders=holders, transactions=transactions, )