608 lines
23 KiB
Python
608 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
RMI INTELLIGENCE FEED PIPELINE
|
|
==============================
|
|
Pulls crypto threat intelligence from RSS feeds and open APIs.
|
|
Runs continuously, indexing scam/hack/exploit data into RAG.
|
|
|
|
Feeds (verified working):
|
|
- Web3IsGoingGreat (Molly White) — incident tracker RSS
|
|
- SlowMist — blockchain security firm (Medium)
|
|
- PeckShield — on-chain security monitor (via Nitter)
|
|
- Blockworks — crypto news
|
|
- Cointelegraph Security — tagged articles
|
|
|
|
Additional sources:
|
|
- Helius webhooks — new Solana tokens
|
|
- GoPlus API — real-time token security checks
|
|
- On-chain scanning — new token → quick risk assessment
|
|
|
|
Architecture:
|
|
Prometheus → every N minutes pull RSS → extract entities → embed → store
|
|
New token event → quick keyword scan → if suspicious → deep embed → alert
|
|
"""
|
|
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import feedparser
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# CONFIGURED FEEDS
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
FEEDS = [
|
|
{
|
|
"name": "web3isgoinggreat",
|
|
"url": "https://www.web3isgoinggreat.com/feed",
|
|
"category": "incident",
|
|
"severity": "high",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "slowmist",
|
|
"url": "https://slowmist.medium.com/feed",
|
|
"category": "hack_report",
|
|
"severity": "critical",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "peckshield",
|
|
"url": "https://peckshield.medium.com/feed",
|
|
"category": "security_alert",
|
|
"severity": "high",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "certik",
|
|
"url": "https://certik.medium.com/feed",
|
|
"category": "audit_report",
|
|
"severity": "high",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "immunefi",
|
|
"url": "https://immunefi.medium.com/feed",
|
|
"category": "bounty_report",
|
|
"severity": "critical",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "trailofbits",
|
|
"url": "https://blog.trailofbits.com/feed/",
|
|
"category": "security_research",
|
|
"severity": "high",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "cryptosecurity",
|
|
"url": "https://cryptosecurity.substack.com/feed",
|
|
"category": "security_news",
|
|
"severity": "medium",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "blockworks",
|
|
"url": "https://blockworks.co/feed",
|
|
"category": "crypto_news",
|
|
"severity": "medium",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
"keywords": ["hack", "exploit", "scam", "rug", "drain", "attack", "breach", "stolen"],
|
|
},
|
|
{
|
|
"name": "cointelegraph_security",
|
|
"url": "https://cointelegraph.com/rss/tag/security",
|
|
"category": "security_news",
|
|
"severity": "medium",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "chainalysis",
|
|
"url": "https://blog.chainalysis.com/feed/",
|
|
"category": "forensic_report",
|
|
"severity": "high",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
"keywords": ["scam", "ransomware", "sanction", "illicit", "money laundering"],
|
|
},
|
|
{
|
|
"name": "cointelegraph",
|
|
"url": "https://cointelegraph.com/rss",
|
|
"category": "crypto_news",
|
|
"severity": "low",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
"keywords": [
|
|
"hack",
|
|
"exploit",
|
|
"scam",
|
|
"rug",
|
|
"drain",
|
|
"attack",
|
|
"breach",
|
|
"stolen",
|
|
"phishing",
|
|
"theft",
|
|
],
|
|
},
|
|
{
|
|
"name": "decrypt",
|
|
"url": "https://decrypt.co/feed",
|
|
"category": "crypto_news",
|
|
"severity": "low",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
"keywords": [
|
|
"hack",
|
|
"exploit",
|
|
"scam",
|
|
"rug",
|
|
"drain",
|
|
"stolen",
|
|
"theft",
|
|
"honeypot",
|
|
"phishing",
|
|
],
|
|
},
|
|
{
|
|
"name": "theblock",
|
|
"url": "https://www.theblock.co/rss.xml",
|
|
"category": "crypto_news",
|
|
"severity": "low",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
"keywords": ["hack", "exploit", "scam", "rug", "drain", "stolen"],
|
|
},
|
|
{
|
|
"name": "bankless",
|
|
"url": "https://www.bankless.com/feed",
|
|
"category": "crypto_news",
|
|
"severity": "low",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
"keywords": ["hack", "exploit", "scam", "rug", "drain", "attack"],
|
|
},
|
|
{
|
|
"name": "thedefiant",
|
|
"url": "https://thedefiant.io/feed",
|
|
"category": "defi_news",
|
|
"severity": "medium",
|
|
"extractor": "rss",
|
|
"enabled": True,
|
|
"keywords": ["hack", "exploit", "scam", "rug", "drain", "attack", "breach"],
|
|
},
|
|
]
|
|
|
|
# Entities to extract from articles
|
|
ENTITY_PATTERNS = {
|
|
"eth_address": r"0x[a-fA-F0-9]{40}",
|
|
"sol_address": r"[1-9A-HJ-NP-Za-km-z]{32,44}",
|
|
"btc_address": r"[13][a-km-zA-HJ-NP-Z1-9]{25,34}",
|
|
"tornado_cash": r"(?i)tornado\s*cash",
|
|
"amount_usd": r"\$[\d,]+(?:\s*(?:million|billion|M|B|K))?",
|
|
"protocol_name": r"(?<=the\s)(?:Uniswap|Aave|Compound|Curve|Balancer|Sushi|Pancake|Raydium|Orca|Jupiter|Marinade)(?=\s)",
|
|
}
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# FEED PROCESSORS
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@dataclass
|
|
class IntelItem:
|
|
id: str
|
|
title: str
|
|
content: str
|
|
source: str
|
|
url: str
|
|
published: str
|
|
category: str
|
|
severity: str
|
|
entities: dict[str, list[str]] = field(default_factory=dict)
|
|
raw: dict = field(default_factory=dict)
|
|
|
|
|
|
class RSSProcessor:
|
|
"""Process standard RSS/Atom feeds into IntelItems."""
|
|
|
|
@staticmethod
|
|
async def fetch(feed_config: dict) -> list[IntelItem]:
|
|
items = []
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
|
|
resp = await client.get(
|
|
feed_config["url"],
|
|
headers={"User-Agent": "RMI-ThreatIntel/1.0 (crypto security research)"},
|
|
)
|
|
resp.raise_for_status()
|
|
feed = feedparser.parse(resp.text)
|
|
|
|
keywords = feed_config.get("keywords", [])
|
|
|
|
for entry in feed.entries[:20]: # Max 20 per feed per run
|
|
title = entry.get("title", "")
|
|
summary = entry.get("summary", entry.get("description", ""))
|
|
content = f"{title}\n\n{html.escape(summary)[:2000]}"
|
|
|
|
# Keyword filter
|
|
if keywords:
|
|
text_lower = (title + " " + summary).lower()
|
|
if not any(k.lower() in text_lower for k in keywords):
|
|
continue
|
|
|
|
# Extract entities
|
|
entities = RSSProcessor._extract_entities(content)
|
|
|
|
item_id = hashlib.sha256(f"{feed_config['name']}:{entry.get('link', title)}".encode()).hexdigest()[:16]
|
|
|
|
items.append(
|
|
IntelItem(
|
|
id=item_id,
|
|
title=title[:300],
|
|
content=content[:3000],
|
|
source=feed_config["name"],
|
|
url=entry.get("link", ""),
|
|
published=entry.get("published", entry.get("updated", "")),
|
|
category=feed_config["category"],
|
|
severity=feed_config["severity"],
|
|
entities=entities,
|
|
raw={"feed_title": feed.get("feed", {}).get("title", "")},
|
|
)
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"RSS fetch failed for {feed_config['name']}: {e}")
|
|
|
|
return items
|
|
|
|
@staticmethod
|
|
def _extract_entities(text: str) -> dict[str, list[str]]:
|
|
found = {}
|
|
for entity_type, pattern in ENTITY_PATTERNS.items():
|
|
matches = re.findall(pattern, text)
|
|
if matches:
|
|
found[entity_type] = list(set(matches))[:10]
|
|
return found
|
|
|
|
|
|
class NitterProcessor(RSSProcessor):
|
|
"""Process Nitter (Twitter mirror) RSS feeds."""
|
|
|
|
@staticmethod
|
|
async def fetch(feed_config: dict) -> list[IntelItem]:
|
|
items = await RSSProcessor.fetch(feed_config)
|
|
# Clean up Nitter-specific formatting
|
|
for item in items:
|
|
# Extract tweet content from title (format: "user: tweet text")
|
|
if ":" in item.title:
|
|
parts = item.title.split(":", 1)
|
|
item.title = parts[1].strip()[:300]
|
|
# Clean HTML entities
|
|
item.content = html.unescape(item.content)
|
|
return items
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# ON-CHAIN SCANNER
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class OnChainScanner:
|
|
"""Scan new tokens on Solana/Ethereum for immediate risk assessment."""
|
|
|
|
@staticmethod
|
|
async def scan_new_solana_tokens(limit: int = 20) -> list[IntelItem]:
|
|
"""Scan recently deployed Solana tokens via Helius/Birdeye."""
|
|
items = []
|
|
try:
|
|
# Use Birdeye new listing API or Helius webhook data
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
# Birdeye trending/new tokens
|
|
resp = await client.get(
|
|
"https://public-api.birdeye.so/defi/tokenlist",
|
|
params={
|
|
"sort_by": "v24hChangePercent",
|
|
"sort_type": "desc",
|
|
"limit": str(limit),
|
|
},
|
|
headers={
|
|
"X-API-KEY": os.getenv("BIRDEYE_API_KEY", ""),
|
|
"User-Agent": "RMI-Scanner/1.0",
|
|
},
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
tokens = data.get("data", {}).get("tokens", [])
|
|
for token in tokens:
|
|
addr = token.get("address", "")
|
|
name = token.get("name", "")
|
|
symbol = token.get("symbol", "")
|
|
change = token.get("v24hChangePercent", 0)
|
|
|
|
# Quick risk heuristics
|
|
risk_signals = []
|
|
if abs(change) > 500:
|
|
risk_signals.append("extreme_volatility")
|
|
if not name or len(name) < 3:
|
|
risk_signals.append("suspicious_name")
|
|
if float(token.get("liquidity", 0)) < 5000:
|
|
risk_signals.append("low_liquidity")
|
|
|
|
if risk_signals:
|
|
item_id = hashlib.sha256(f"solana_new:{addr}".encode()).hexdigest()[:16]
|
|
items.append(
|
|
IntelItem(
|
|
id=item_id,
|
|
title=f"New token alert: {name} ({symbol})",
|
|
content=f"Token {name} ({symbol}) on Solana. Address: {addr}. "
|
|
f"24h change: {change}%. Risk signals: {', '.join(risk_signals)}. "
|
|
f"Liquidity: ${token.get('liquidity', 0):,.0f}. Volume 24h: ${token.get('v24hUSD', 0):,.0f}.",
|
|
source="onchain_scanner",
|
|
url=f"https://birdeye.so/token/{addr}?chain=solana",
|
|
published=datetime.now(UTC).isoformat(),
|
|
category="new_token",
|
|
severity="medium" if len(risk_signals) < 2 else "high",
|
|
entities={"sol_address": [addr]},
|
|
)
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Solana scan failed: {e}")
|
|
|
|
return items
|
|
|
|
@staticmethod
|
|
async def check_token_security(address: str, chain: str = "solana") -> dict[str, Any]:
|
|
"""Check a token via GoPlus security API (free, no key needed)."""
|
|
try:
|
|
chain_id = {"solana": "solana", "ethereum": "1", "bsc": "56", "base": "8453"}.get(chain, chain)
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(
|
|
f"https://api.gopluslabs.io/api/v1/token_security/{chain_id}",
|
|
params={"contract_addresses": address},
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
return data.get("result", {}).get(address.lower(), {})
|
|
except Exception:
|
|
pass
|
|
return {}
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# INTELLIGENCE PIPELINE ORCHESTRATOR
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class IntelPipeline:
|
|
"""
|
|
Continuous intelligence ingestion pipeline.
|
|
|
|
Usage:
|
|
pipeline = IntelPipeline()
|
|
stats = await pipeline.run_cycle()
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._last_run: datetime | None = None
|
|
self._total_ingested = 0
|
|
self._seen_ids: set[str] = set()
|
|
self._feed_processors = {
|
|
"rss": RSSProcessor,
|
|
"nitter_rss": NitterProcessor,
|
|
}
|
|
|
|
async def run_cycle(self) -> dict[str, Any]:
|
|
"""Run one full ingestion cycle across all sources."""
|
|
from app.crypto_embeddings import get_embedder
|
|
from app.rag_service import _get_redis
|
|
|
|
embedder = await get_embedder()
|
|
r = await _get_redis()
|
|
now = datetime.now(UTC)
|
|
stats = {"feeds": {}, "onchain": {}, "total": 0, "skipped": 0, "errors": 0}
|
|
|
|
# Phase 1: RSS feeds
|
|
for feed_config in FEEDS:
|
|
if not feed_config.get("enabled", True):
|
|
continue
|
|
|
|
try:
|
|
processor_class = self._feed_processors.get(feed_config["extractor"], RSSProcessor)
|
|
items = await processor_class.fetch(feed_config)
|
|
|
|
ingested = 0
|
|
for item in items:
|
|
if item.id in self._seen_ids:
|
|
stats["skipped"] += 1
|
|
continue
|
|
|
|
try:
|
|
# Embed the content
|
|
result = await embedder.embed_scam_pattern(
|
|
pattern_name=item.title,
|
|
description=item.content,
|
|
severity=item.severity,
|
|
)
|
|
|
|
# Store in Redis
|
|
doc = {
|
|
"id": item.id,
|
|
"collection": "market_intel",
|
|
"vector": result.vector,
|
|
"dims": result.dims,
|
|
"model": result.model,
|
|
"metadata": {
|
|
"source": item.source,
|
|
"url": item.url,
|
|
"published": item.published,
|
|
"category": item.category,
|
|
"severity": item.severity,
|
|
"entities": item.entities,
|
|
},
|
|
"content": item.content,
|
|
"source": item.source,
|
|
"severity": item.severity,
|
|
"stored_at": now.isoformat(),
|
|
}
|
|
await r.setex(
|
|
f"rag:market_intel:{item.id}",
|
|
86400 * 7, # 7-day TTL for news
|
|
json.dumps(doc),
|
|
)
|
|
await r.sadd("rag:idx:market_intel", item.id)
|
|
self._seen_ids.add(item.id)
|
|
ingested += 1
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to ingest {item.id}: {e}")
|
|
|
|
stats["feeds"][feed_config["name"]] = ingested
|
|
stats["total"] += ingested
|
|
|
|
except Exception as e:
|
|
logger.error(f"Feed {feed_config['name']} failed: {e}")
|
|
stats["errors"] += 1
|
|
stats["feeds"][feed_config["name"]] = 0
|
|
|
|
# Phase 2: On-chain scanning
|
|
try:
|
|
new_tokens = await OnChainScanner.scan_new_solana_tokens(limit=10)
|
|
ingested = 0
|
|
for item in new_tokens:
|
|
if item.id in self._seen_ids:
|
|
continue
|
|
try:
|
|
result = await embedder.embed_scam_pattern(
|
|
pattern_name=item.title,
|
|
description=item.content,
|
|
severity=item.severity,
|
|
)
|
|
doc = {
|
|
"id": item.id,
|
|
"collection": "token_analysis",
|
|
"vector": result.vector,
|
|
"dims": result.dims,
|
|
"metadata": {
|
|
"source": item.source,
|
|
"category": item.category,
|
|
"severity": item.severity,
|
|
"entities": item.entities,
|
|
},
|
|
"content": item.content,
|
|
"source": item.source,
|
|
"severity": item.severity,
|
|
"stored_at": now.isoformat(),
|
|
}
|
|
await r.setex(f"rag:token_analysis:{item.id}", 86400 * 7, json.dumps(doc))
|
|
await r.sadd("rag:idx:token_analysis", item.id)
|
|
self._seen_ids.add(item.id)
|
|
ingested += 1
|
|
except Exception as e:
|
|
logger.error(f"Failed to ingest onchain {item.id}: {e}")
|
|
|
|
stats["onchain"]["new_tokens"] = ingested
|
|
stats["total"] += ingested
|
|
except Exception as e:
|
|
logger.warning(f"On-chain scan failed: {e}")
|
|
|
|
self._last_run = now
|
|
self._total_ingested += stats["total"]
|
|
|
|
# Cleanup old entries (older than 7 days)
|
|
try:
|
|
old_cutoff = (now - timedelta(days=7)).isoformat()
|
|
for coll in ["market_intel", "token_analysis"]:
|
|
all_ids = await r.smembers(f"rag:idx:{coll}")
|
|
for did in all_ids:
|
|
doc_raw = await r.get(f"rag:{coll}:{did}")
|
|
if doc_raw:
|
|
try:
|
|
doc = json.loads(doc_raw)
|
|
if doc.get("stored_at", "") < old_cutoff:
|
|
await r.delete(f"rag:{coll}:{did}")
|
|
await r.srem(f"rag:idx:{coll}", did)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"status": "completed",
|
|
"total_ingested": stats["total"],
|
|
"cumulative": self._total_ingested,
|
|
"feeds": stats["feeds"],
|
|
"onchain": stats["onchain"],
|
|
"skipped": stats["skipped"],
|
|
"errors": stats["errors"],
|
|
"seen_cache_size": len(self._seen_ids),
|
|
"timestamp": now.isoformat(),
|
|
"next_run_in": "30 minutes",
|
|
}
|
|
|
|
async def get_latest_intel(self, limit: int = 20) -> list[dict[str, Any]]:
|
|
"""Get latest threat intelligence for frontend display."""
|
|
from app.rag_service import _get_redis
|
|
|
|
r = await _get_redis()
|
|
|
|
results = []
|
|
for coll in ["market_intel", "token_analysis", "known_scams"]:
|
|
doc_ids = await r.smembers(f"rag:idx:{coll}")
|
|
for did in list(doc_ids)[:limit]:
|
|
raw = await r.get(f"rag:{coll}:{did}")
|
|
if raw:
|
|
try:
|
|
doc = json.loads(raw)
|
|
results.append(
|
|
{
|
|
"id": doc["id"],
|
|
"title": doc.get("metadata", {}).get("name", doc.get("content", "")[:100]),
|
|
"content": doc.get("content", "")[:500],
|
|
"source": doc.get("source", coll),
|
|
"severity": doc.get("severity", "medium"),
|
|
"category": doc.get("metadata", {}).get("category", ""),
|
|
"url": doc.get("metadata", {}).get("url", ""),
|
|
"published": doc.get("metadata", {}).get("published", ""),
|
|
"entities": doc.get("metadata", {}).get("entities", {}),
|
|
"stored_at": doc.get("stored_at", ""),
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
results.sort(key=lambda x: x.get("stored_at", ""), reverse=True)
|
|
return results[:limit]
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# SINGLETON
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
_pipeline: IntelPipeline | None = None
|
|
|
|
|
|
async def get_pipeline() -> IntelPipeline:
|
|
global _pipeline
|
|
if _pipeline is None:
|
|
_pipeline = IntelPipeline()
|
|
return _pipeline
|