415 lines
16 KiB
Python
415 lines
16 KiB
Python
"""
|
|
RAG Historical Scam Ingestion — Rekt DB, Chainabuse, TRM, Elliptic
|
|
==================================================================
|
|
Ingests historical crypto scam and hack data into RAG collections.
|
|
Sources: Rekt DB (3K+ DeFi hacks), Chainabuse (scam reports),
|
|
TRM Labs (annual crime report), Elliptic (scam report),
|
|
SlowMist (hacked archive), Immunefi (bug bounties), CertiK (audits).
|
|
|
|
Collections fed:
|
|
- defi_hacks: Rekt DB, SlowMist, Rekt News
|
|
- rug_timeline: Chainabuse, SENTINEL
|
|
- vuln_patterns: Immunefi, CertiK
|
|
- crime_reports: TRM, Elliptic, Chainalysis
|
|
- forensic_reports: All exploit post-mortems
|
|
|
|
Runs: weekly (Sunday 2am) for historical sources.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from datetime import UTC, datetime
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger("rag.historical")
|
|
|
|
# ── Config ──
|
|
BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:8000")
|
|
RAG_INGEST_URL = f"{BACKEND_URL}/api/v1/rag/ingest"
|
|
REQUEST_TIMEOUT = 30
|
|
|
|
# ── Source Definitions ──
|
|
SOURCES = {
|
|
"rekt_db": {
|
|
"name": "Rekt Database",
|
|
"url": "https://de.fi/rekt-database",
|
|
"collection": "defi_hacks",
|
|
"content_type": "scam_report",
|
|
"description": "3,000+ DeFi hacks since 2020 — the most comprehensive exploit database",
|
|
"cadence": "weekly",
|
|
},
|
|
"chainabuse": {
|
|
"name": "Chainabuse",
|
|
"url": "https://chainabuse.com",
|
|
"collection": "rug_timeline",
|
|
"content_type": "scam_report",
|
|
"description": "Community-reported scam addresses with narratives",
|
|
"cadence": "weekly",
|
|
},
|
|
"slowmist_hacked": {
|
|
"name": "SlowMist Hacked Archive",
|
|
"url": "https://slowmist.com/en/#hacked",
|
|
"collection": "defi_hacks",
|
|
"content_type": "scam_report",
|
|
"description": "Detailed exploit analysis from SlowMist security team",
|
|
"cadence": "weekly",
|
|
},
|
|
"rekt_news": {
|
|
"name": "Rekt News",
|
|
"url": "https://rekt.news",
|
|
"collection": "defi_hacks",
|
|
"content_type": "scam_report",
|
|
"description": "In-depth DeFi exploit post-mortems",
|
|
"cadence": "weekly",
|
|
},
|
|
"immunefi": {
|
|
"name": "Immunefi Bug Bounties",
|
|
"url": "https://immunefi.com",
|
|
"collection": "vuln_patterns",
|
|
"content_type": "contract_code",
|
|
"description": "Bug bounty reports — real vulnerability patterns",
|
|
"cadence": "weekly",
|
|
},
|
|
"certik": {
|
|
"name": "CertiK Audit Findings",
|
|
"url": "https://certik.com",
|
|
"collection": "vuln_patterns",
|
|
"content_type": "contract_code",
|
|
"description": "Professional smart contract audit findings",
|
|
"cadence": "weekly",
|
|
},
|
|
"trm_crime_report": {
|
|
"name": "TRM Labs Crypto Crime Report",
|
|
"url": "https://www.trmlabs.com/reports-and-whitepapers/2026-crypto-crime-report",
|
|
"collection": "crime_reports",
|
|
"content_type": "annual_report",
|
|
"description": "Annual crypto crime typologies and trends — $158B illicit volume in 2025",
|
|
"cadence": "yearly",
|
|
},
|
|
"elliptic_scams": {
|
|
"name": "Elliptic State of Crypto Scams",
|
|
"url": "https://info.elliptic.co/hubfs/The%20state%20of%20crypto%20scams%202025/",
|
|
"collection": "crime_reports",
|
|
"content_type": "annual_report",
|
|
"description": "Annual crypto scam typologies and case studies",
|
|
"cadence": "yearly",
|
|
},
|
|
}
|
|
|
|
|
|
# ── Scraper Functions ──
|
|
|
|
|
|
async def scrape_rekt_db() -> list[dict]:
|
|
"""
|
|
Scrape Rekt DB for historical DeFi hacks.
|
|
de.fi/rekt-database lists 3,000+ exploits with:
|
|
- Project name, date, amount lost, chain, vulnerability type
|
|
- Links to post-mortems on rekt.news
|
|
"""
|
|
docs = []
|
|
try:
|
|
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
|
|
# Try the API endpoint first
|
|
resp = await client.get("https://de.fi/api/rekt")
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
for item in data[:500]: # Limit per run
|
|
docs.append(
|
|
{
|
|
"text": f"DeFi Hack: {item.get('name', 'Unknown')} — "
|
|
f"${item.get('amount', 0):,.0f} lost on {item.get('chain', 'Unknown')}. "
|
|
f"Vulnerability: {item.get('vulnerability', 'Unknown')}. "
|
|
f"Date: {item.get('date', 'Unknown')}. "
|
|
f"Details: {item.get('description', '')}",
|
|
"source": "rekt_db",
|
|
"url": item.get("url", "https://de.fi/rekt-database"),
|
|
"category": "defi_hack",
|
|
"date": item.get("date", ""),
|
|
"chain": item.get("chain", ""),
|
|
"tags": [
|
|
item.get("vulnerability", ""),
|
|
item.get("chain", ""),
|
|
"defi_hack",
|
|
],
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Rekt DB scrape failed: {e}")
|
|
|
|
logger.info(f"Rekt DB: {len(docs)} hacks scraped")
|
|
return docs
|
|
|
|
|
|
async def scrape_chainabuse() -> list[dict]:
|
|
"""
|
|
Scrape Chainabuse for scam reports with addresses.
|
|
chainabuse.com has community-reported scams with:
|
|
- Address, chain, scam type, description, reporter
|
|
"""
|
|
docs = []
|
|
try:
|
|
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
|
|
# Chainabuse has a public reports page
|
|
resp = await client.get(
|
|
"https://chainabuse.com/api/reports",
|
|
params={"limit": 200, "sort": "newest"},
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
for item in data.get("reports", []):
|
|
address = item.get("address", "")
|
|
docs.append(
|
|
{
|
|
"text": f"Scam Report: {item.get('title', 'Unknown')} — "
|
|
f"Address {address} on {item.get('chain', 'Unknown')}. "
|
|
f"Type: {item.get('scam_type', 'Unknown')}. "
|
|
f"Description: {item.get('description', '')}. "
|
|
f"Reported by: {item.get('reporter', 'Anonymous')}",
|
|
"source": "chainabuse",
|
|
"url": f"https://chainabuse.com/report/{item.get('id', '')}",
|
|
"category": "scam_report",
|
|
"date": item.get("created_at", ""),
|
|
"chain": item.get("chain", ""),
|
|
"tags": [
|
|
item.get("scam_type", ""),
|
|
item.get("chain", ""),
|
|
"scam_report",
|
|
],
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Chainabuse scrape failed: {e}")
|
|
|
|
logger.info(f"Chainabuse: {len(docs)} reports scraped")
|
|
return docs
|
|
|
|
|
|
async def scrape_slowmist() -> list[dict]:
|
|
"""Scrape SlowMist Hacked Archive for detailed exploit analysis."""
|
|
docs = []
|
|
try:
|
|
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
|
|
resp = await client.get("https://slowmist.com/en/api/hacked")
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
for item in data[:200]:
|
|
docs.append(
|
|
{
|
|
"text": f"SlowMist Analysis: {item.get('title', 'Unknown')} — "
|
|
f"${item.get('amount', 0):,.0f} lost. "
|
|
f"Root cause: {item.get('root_cause', 'Unknown')}. "
|
|
f"Attack vector: {item.get('attack_vector', 'Unknown')}. "
|
|
f"Recommendations: {item.get('recommendations', '')}",
|
|
"source": "slowmist",
|
|
"url": item.get("url", "https://slowmist.com"),
|
|
"category": "defi_hack",
|
|
"date": item.get("date", ""),
|
|
"chain": item.get("chain", ""),
|
|
"tags": [
|
|
item.get("root_cause", ""),
|
|
item.get("attack_vector", ""),
|
|
"defi_hack",
|
|
"slowmist",
|
|
],
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"SlowMist scrape failed: {e}")
|
|
|
|
logger.info(f"SlowMist: {len(docs)} analyses scraped")
|
|
return docs
|
|
|
|
|
|
async def scrape_rekt_news() -> list[dict]:
|
|
"""Scrape Rekt News for in-depth exploit post-mortems."""
|
|
docs = []
|
|
try:
|
|
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
|
|
# Rekt News has an RSS feed
|
|
resp = await client.get("https://rekt.news/feed.xml")
|
|
if resp.status_code == 200:
|
|
import defusedxml.ElementTree as ET
|
|
|
|
root = ET.fromstring(resp.text)
|
|
for item in root.findall(".//item")[:50]:
|
|
title = item.findtext("title", "")
|
|
description = item.findtext("description", "")
|
|
link = item.findtext("link", "")
|
|
pub_date = item.findtext("pubDate", "")
|
|
|
|
# Extract amount from title (e.g., "$10M Rekt")
|
|
amount_match = re.search(r"\$(\d+(?:\.\d+)?[MB]?)", title)
|
|
amount = amount_match.group(0) if amount_match else "Unknown"
|
|
|
|
docs.append(
|
|
{
|
|
"text": f"Rekt News: {title} — {amount} lost. {description}",
|
|
"source": "rekt_news",
|
|
"url": link,
|
|
"category": "defi_hack",
|
|
"date": pub_date,
|
|
"tags": ["defi_hack", "rekt_news", "post_mortem"],
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Rekt News scrape failed: {e}")
|
|
|
|
logger.info(f"Rekt News: {len(docs)} articles scraped")
|
|
return docs
|
|
|
|
|
|
async def ingest_trm_report() -> list[dict]:
|
|
"""
|
|
Ingest TRM Labs 2026 Crypto Crime Report.
|
|
Key findings: $158B illicit volume, $2.87B stolen in 150 hacks,
|
|
Russia-linked sanctions evasion via A7A5 stablecoin.
|
|
"""
|
|
docs = []
|
|
try:
|
|
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
|
|
resp = await client.get("https://www.trmlabs.com/reports-and-whitepapers/2026-crypto-crime-report")
|
|
if resp.status_code == 200:
|
|
# Extract key findings from the page
|
|
text = resp.text
|
|
|
|
# Parse key sections
|
|
sections = re.split(r"<h[23][^>]*>", text)
|
|
for section in sections:
|
|
# Strip HTML tags
|
|
clean = re.sub(r"<[^>]+>", " ", section)
|
|
clean = re.sub(r"\s+", " ", clean).strip()
|
|
if len(clean) > 100:
|
|
docs.append(
|
|
{
|
|
"text": f"TRM 2026 Crypto Crime Report: {clean[:2000]}",
|
|
"source": "trm_labs",
|
|
"url": "https://www.trmlabs.com/reports-and-whitepapers/2026-crypto-crime-report",
|
|
"category": "crime_report",
|
|
"date": "2026-01-28",
|
|
"tags": [
|
|
"crime_report",
|
|
"trm_labs",
|
|
"annual_2026",
|
|
"sanctions",
|
|
"hacks",
|
|
],
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"TRM report ingest failed: {e}")
|
|
|
|
logger.info(f"TRM Report: {len(docs)} sections ingested")
|
|
return docs
|
|
|
|
|
|
# ── Master Ingestion Orchestrator ──
|
|
|
|
|
|
async def ingest_historical_source(source_id: str) -> dict:
|
|
"""Ingest a single historical source into RAG."""
|
|
source = SOURCES.get(source_id)
|
|
if not source:
|
|
return {"error": f"Unknown source: {source_id}"}
|
|
|
|
logger.info(f"Ingesting {source['name']} → {source['collection']}")
|
|
|
|
# Scrape
|
|
scrapers = {
|
|
"rekt_db": scrape_rekt_db,
|
|
"chainabuse": scrape_chainabuse,
|
|
"slowmist_hacked": scrape_slowmist,
|
|
"rekt_news": scrape_rekt_news,
|
|
"trm_crime_report": ingest_trm_report,
|
|
}
|
|
|
|
scraper = scrapers.get(source_id)
|
|
if not scraper:
|
|
return {"error": f"No scraper for {source_id}", "source": source_id}
|
|
|
|
try:
|
|
docs = await scraper()
|
|
except Exception as e:
|
|
return {"error": str(e), "source": source_id, "docs_scraped": 0}
|
|
|
|
if not docs:
|
|
return {"source": source_id, "docs_scraped": 0, "status": "empty"}
|
|
|
|
# Chunk and dedup
|
|
from app.rag_chunking import chunk_batch
|
|
|
|
chunks = chunk_batch(docs, source["content_type"])
|
|
|
|
# Ingest into backend
|
|
ingested = 0
|
|
try:
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
# Batch in groups of 50
|
|
for i in range(0, len(chunks), 50):
|
|
batch = chunks[i : i + 50]
|
|
resp = await client.post(
|
|
RAG_INGEST_URL,
|
|
json={
|
|
"documents": batch,
|
|
"collection": source["collection"],
|
|
"source": source_id,
|
|
"chunking": source["content_type"],
|
|
},
|
|
)
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
ingested += result.get("ingested", len(batch))
|
|
else:
|
|
logger.warning(f"Ingest batch failed for {source_id}: HTTP {resp.status_code}")
|
|
except Exception as e:
|
|
logger.error(f"Ingest HTTP failed for {source_id}: {e}")
|
|
|
|
return {
|
|
"source": source_id,
|
|
"name": source["name"],
|
|
"collection": source["collection"],
|
|
"docs_scraped": len(docs),
|
|
"chunks_created": len(chunks),
|
|
"ingested": ingested,
|
|
"status": "ok" if ingested > 0 else "partial",
|
|
}
|
|
|
|
|
|
async def ingest_all_historical() -> dict:
|
|
"""Run full historical ingestion across all sources."""
|
|
results = {}
|
|
total_ingested = 0
|
|
|
|
for source_id in SOURCES:
|
|
logger.info(f"Starting historical ingest: {source_id}")
|
|
try:
|
|
result = await ingest_historical_source(source_id)
|
|
results[source_id] = result
|
|
total_ingested += result.get("ingested", 0)
|
|
except Exception as e:
|
|
results[source_id] = {"error": str(e), "source": source_id}
|
|
logger.error(f"Historical ingest failed for {source_id}: {e}")
|
|
|
|
return {
|
|
"status": "complete",
|
|
"total_ingested": total_ingested,
|
|
"sources": results,
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
}
|
|
|
|
|
|
# ── CLI Entry Point ──
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
if len(sys.argv) > 1:
|
|
source = sys.argv[1]
|
|
result = asyncio.run(ingest_historical_source(source))
|
|
print(json.dumps(result, indent=2))
|
|
else:
|
|
result = asyncio.run(ingest_all_historical())
|
|
print(json.dumps(result, indent=2))
|