#!/usr/bin/env python3 """ Ingest free scam data sources into RAG known_scams collection. Sources: 1. MetaMask eth-phishing-detect blacklist (~198K domains) 2. CryptoScamDB API (if available - currently 502) 3. ChainAbuse API (needs API key) The MetaMask list is the largest accessible source of confirmed scam domains. """ import asyncio import hashlib import logging import os import sys import httpx sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") logger = logging.getLogger("ingest_free_scam_sources") RAG_API = "http://localhost:8000/api/v1/rag/ingest" COLLECTION = "known_scams" METAMASK_LIST_URL = "https://raw.githubusercontent.com/MetaMask/eth-phishing-detect/master/src/config.json" CRYPTOSCAMDB_URLS = [ "https://api.cryptoscamdb.org/v1/all", "https://api.cryptoscamdb.org/v1/scams", ] def make_doc_id(content: str, source: str) -> str: return hashlib.sha256(f"known_scams:{content.lower()}:{source}".encode()).hexdigest()[:16] async def ingest_via_api(content: str, metadata: dict) -> bool: try: async with httpx.AsyncClient(timeout=30) as client: resp = await client.post( RAG_API, json={ "collection": COLLECTION, "content": content, "metadata": metadata, }, ) return resp.status_code == 200 except Exception as e: logger.debug(f"Ingest API error: {e}") return False async def ingest_metamask_blacklist(): """Ingest MetaMask phishing domain blacklist (~198K domains).""" logger.info("Fetching MetaMask phishing blacklist...") async with httpx.AsyncClient(timeout=60) as client: try: resp = await client.get(METAMASK_LIST_URL) if resp.status_code != 200: logger.error(f"Failed to fetch MetaMask list: {resp.status_code}") return 0, 0 data = resp.json() except Exception as e: logger.error(f"MetaMask list fetch error: {e}") return 0, 0 blacklist = data.get("blacklist", []) logger.info(f"MetaMask blacklist: {len(blacklist)} entries") # Batch into groups by domain pattern for efficiency # Instead of 198K individual entries, group by TLD + category ingested = 0 errors = 0 # Group in batches of 50 domains per RAG doc batch_size = 50 total_batches = (len(blacklist) + batch_size - 1) // batch_size for batch_idx in range(total_batches): batch = blacklist[batch_idx * batch_size : (batch_idx + 1) * batch_size] domain_list = "\n".join(batch) content = ( f"MetaMask Confirmed Phishing/Scam Domains (Batch {batch_idx + 1}/{total_batches}):\n" f"{domain_list}\n" f"Source: MetaMask eth-phishing-detect project. These domains are confirmed " f"phishing, scam, or malicious websites targeting crypto users." ) metadata = { "source": "metamask_phishing_detect", "batch_index": batch_idx, "total_batches": total_batches, "domain_count": len(batch), "category": "phishing_domain", "severity": "high", "doc_id": make_doc_id(f"metamask_batch_{batch_idx}", "metamask"), } if await ingest_via_api(content, metadata): ingested += 1 else: errors += 1 if (batch_idx + 1) % 100 == 0: logger.info(f"Progress: {batch_idx + 1}/{total_batches} batches | ingested: {ingested}, errors: {errors}") await asyncio.sleep(0.05) # Rate limit: small delay every 20 batches if (batch_idx + 1) % 20 == 0: await asyncio.sleep(0.1) return ingested, errors async def try_cryptoscamdb(): """Try CryptoScamDB API if accessible.""" for url in CRYPTOSCAMDB_URLS: try: async with httpx.AsyncClient(timeout=30) as client: resp = await client.get(url) if resp.status_code == 200: data = resp.json() scams = data if isinstance(data, list) else data.get("result", data.get("scams", [])) logger.info(f"CryptoScamDB returned {len(scams)} entries from {url}") return scams except Exception as e: logger.debug(f"CryptoScamDB {url} failed: {e}") logger.warning("CryptoScamDB API unavailable (502)") return None async def ingest_cryptoscamdb(scams: list): """Ingest CryptoScamDB entries.""" ingested = 0 errors = 0 for i, scam in enumerate(scams): if not isinstance(scam, dict): continue name = scam.get("name", "") url = scam.get("url", "") category = scam.get("category", "unknown") status = scam.get("status", "unknown") description = scam.get("description", "") addresses = scam.get("addresses", []) if addresses: for addr in addresses: if not addr or len(addr) < 10: continue content = f"CryptoScamDB flagged address: {addr}. Project: {name}. Category: {category}. Status: {status}. Description: {description}" metadata = { "source": "cryptoscamdb", "address": addr.lower(), "category": category, "status": status, "name": name, "severity": "high" if status == "Active" else "medium", } if await ingest_via_api(content, metadata): ingested += 1 else: errors += 1 else: content = f"CryptoScamDB entry: {name or url}. Category: {category}. Status: {status}. URL: {url}. Description: {description}" metadata = { "source": "cryptoscamdb", "category": category, "status": status, "name": name, "url": url, "severity": "high" if status == "Active" else "medium", } if await ingest_via_api(content, metadata): ingested += 1 else: errors += 1 if (i + 1) % 100 == 0: logger.info(f"CryptoScamDB progress: {i + 1}/{len(scams)} | ingested: {ingested}, errors: {errors}") return ingested, errors async def main(): total_ingested = 0 total_errors = 0 # 1. MetaMask blacklist (always accessible) mm_ingested, mm_errors = await ingest_metamask_blacklist() total_ingested += mm_ingested total_errors += mm_errors logger.info(f"MetaMask: {mm_ingested} docs ingested, {mm_errors} errors") # 2. CryptoScamDB (may be down) scams = await try_cryptoscamdb() if scams: csdb_ingested, csdb_errors = await ingest_cryptoscamdb(scams) total_ingested += csdb_ingested total_errors += csdb_errors logger.info(f"CryptoScamDB: {csdb_ingested} docs ingested, {csdb_errors} errors") else: logger.warning("Skipping CryptoScamDB - API unavailable") logger.info(f"TOTAL: {total_ingested} docs ingested, {total_errors} errors") if __name__ == "__main__": asyncio.run(main())