325 lines
11 KiB
Python
325 lines
11 KiB
Python
"""
|
|
RMI Intelligent Webhook System
|
|
===============================
|
|
Databus-powered webhook receiver and dispatcher.
|
|
Handles inbound webhooks from Arkham, Helius, Moralis, and any service.
|
|
Auto-caches, indexes in RAG, and triggers premium scanner analysis.
|
|
|
|
Architecture:
|
|
Webhook → Validate → Cache(DataBus) → RAG Index → Premium Scanner → Alert Pipeline
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import os
|
|
from datetime import datetime
|
|
|
|
import httpx
|
|
import redis
|
|
|
|
logger = logging.getLogger("webhooks")
|
|
|
|
REDIS_HOST = os.getenv("REDIS_HOST", "rmi-redis")
|
|
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
|
|
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
|
|
|
|
# Webhook configurations per service
|
|
WEBHOOK_CONFIGS = {
|
|
"arkham": {
|
|
"secret_env": "ARKHAM_WEBHOOK_SECRET",
|
|
"verify_header": "X-Arkham-Signature",
|
|
"events": ["entity_updated", "transaction_detected", "label_added"],
|
|
"cache_ttl": 3600,
|
|
},
|
|
"helius": {
|
|
"secret_env": "HELIUS_WEBHOOK_SECRET",
|
|
"verify_header": "x-helius-signature",
|
|
"events": ["transaction", "nft_event", "token_transfer", "account_update"],
|
|
"cache_ttl": 600,
|
|
},
|
|
"moralis": {
|
|
"secret_env": "MORALIS_WEBHOOK_SECRET",
|
|
"verify_header": "x-signature",
|
|
"events": ["txs", "logs", "nft_transfers", "erc20_transfers"],
|
|
"cache_ttl": 900,
|
|
},
|
|
"alchemy": {
|
|
"secret_env": "ALCHEMY_WEBHOOK_SECRET",
|
|
"verify_header": "X-Alchemy-Signature",
|
|
"events": ["address_activity", "nft_activity", "tx_status"],
|
|
"cache_ttl": 600,
|
|
},
|
|
}
|
|
|
|
|
|
def _redis():
|
|
return redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
password=REDIS_PASSWORD,
|
|
decode_responses=True,
|
|
socket_connect_timeout=2,
|
|
)
|
|
|
|
|
|
async def handle_webhook(service: str, payload: dict, headers: dict, raw_body: bytes = b"") -> dict:
|
|
"""Universal webhook handler.
|
|
|
|
1. Validate signature
|
|
2. Store in DataBus cache
|
|
3. Index in RAG for intelligence
|
|
4. Trigger premium scanner
|
|
5. Push to alert pipeline if high risk
|
|
"""
|
|
config = WEBHOOK_CONFIGS.get(service)
|
|
if not config:
|
|
return {"error": f"Unknown service: {service}", "supported": list(WEBHOOK_CONFIGS.keys())}
|
|
|
|
# ── 1. Validate signature ──
|
|
secret = os.getenv(config["secret_env"], "")
|
|
if secret:
|
|
sig_header = headers.get(config["verify_header"], "")
|
|
if not _verify_signature(secret, raw_body, sig_header):
|
|
return {"error": "Invalid signature", "status": "rejected"}
|
|
|
|
# ── 2. Deduplicate (prevent replay) ──
|
|
event_id = (
|
|
payload.get("id")
|
|
or payload.get("eventId")
|
|
or hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()[:16]
|
|
)
|
|
try:
|
|
r = _redis()
|
|
if r.get(f"webhook:dedup:{service}:{event_id}"):
|
|
r.close()
|
|
return {"status": "duplicate", "event_id": event_id}
|
|
r.setex(f"webhook:dedup:{service}:{event_id}", 3600, "1")
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ── 3. Store in DataBus cache ──
|
|
event_type = payload.get("type") or payload.get("event") or "unknown"
|
|
cache_key = f"webhook:{service}:{event_type}:{event_id}"
|
|
try:
|
|
r = _redis()
|
|
r.setex(
|
|
cache_key,
|
|
config["cache_ttl"],
|
|
json.dumps(
|
|
{
|
|
"service": service,
|
|
"event_type": event_type,
|
|
"payload": payload,
|
|
"received_at": datetime.utcnow().isoformat(),
|
|
"headers": {k: v for k, v in headers.items() if k.lower() not in ("authorization", "x-api-key")},
|
|
}
|
|
),
|
|
)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ── 4. Index in RAG for intelligence ──
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as c:
|
|
await c.post(
|
|
"http://localhost:8000/api/v1/rag/permanence/index",
|
|
json={
|
|
"collection": f"webhooks_{service}",
|
|
"documents": [
|
|
{
|
|
"id": f"{service}:{event_id}",
|
|
"text": json.dumps(payload, default=str)[:2000],
|
|
"metadata": {
|
|
"service": service,
|
|
"event_type": event_type,
|
|
"received_at": datetime.utcnow().isoformat(),
|
|
},
|
|
}
|
|
],
|
|
},
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# ── 5. Trigger premium scanner based on event type ──
|
|
scan_triggers = _get_scan_triggers(service, event_type, payload)
|
|
|
|
# ── 6. Push to alert pipeline if high risk ──
|
|
risk = _assess_webhook_risk(service, event_type, payload)
|
|
if risk["level"] in ("HIGH", "CRITICAL"):
|
|
try:
|
|
from app.alert_pipeline import push_alert
|
|
|
|
await push_alert(
|
|
title=f"[{service.upper()}] {risk['title']}",
|
|
severity=risk["level"].lower(),
|
|
description=risk["description"],
|
|
chain=payload.get("chain", "unknown"),
|
|
metadata={"webhook_service": service, "event_id": event_id},
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"status": "processed",
|
|
"service": service,
|
|
"event_id": event_id,
|
|
"event_type": event_type,
|
|
"risk_level": risk["level"],
|
|
"scan_triggers": scan_triggers,
|
|
"cached": True,
|
|
}
|
|
|
|
|
|
def _verify_signature(secret: str, body: bytes, signature: str) -> bool:
|
|
"""HMAC signature verification."""
|
|
if not secret or not signature:
|
|
return True # No secret configured, skip verification
|
|
try:
|
|
expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
|
|
return hmac.compare_digest(expected, signature)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _get_scan_triggers(service: str, event_type: str, payload: dict) -> list[str]:
|
|
"""Determine which premium scans to trigger based on webhook event."""
|
|
triggers = []
|
|
payload.get("address") or payload.get("tokenAddress") or payload.get("wallet", "")
|
|
|
|
if service == "arkham":
|
|
if event_type in ("entity_updated", "label_added"):
|
|
triggers.append("cluster_map")
|
|
triggers.append("bundle_scan")
|
|
if event_type == "transaction_detected":
|
|
triggers.append("mev_sandwich")
|
|
triggers.append("copy_trading")
|
|
|
|
elif service == "helius":
|
|
if event_type == "transaction":
|
|
triggers.append("sniper_detect")
|
|
triggers.append("mev_sandwich")
|
|
triggers.append("wash_trading")
|
|
if event_type == "token_transfer":
|
|
triggers.append("fresh_wallets")
|
|
triggers.append("insider_signals")
|
|
if event_type == "nft_event":
|
|
triggers.append("bot_farm")
|
|
|
|
elif service == "moralis":
|
|
if event_type in ("txs", "erc20_transfers"):
|
|
triggers.append("copy_trading")
|
|
triggers.append("bundle_scan")
|
|
if event_type == "nft_transfers":
|
|
triggers.append("bot_farm")
|
|
|
|
elif service == "alchemy":
|
|
if event_type == "address_activity":
|
|
triggers.append("cluster_map")
|
|
triggers.append("sniper_detect")
|
|
|
|
return triggers
|
|
|
|
|
|
def _assess_webhook_risk(service: str, event_type: str, payload: dict) -> dict:
|
|
"""Quick risk assessment from webhook data."""
|
|
value_usd = float(payload.get("valueUsd") or payload.get("value", 0))
|
|
|
|
if value_usd > 100000:
|
|
return {
|
|
"level": "HIGH",
|
|
"title": f"Large transfer: ${value_usd:,.0f}",
|
|
"description": f"{service} detected a large transaction of ${value_usd:,.0f}",
|
|
}
|
|
|
|
if event_type in ("entity_updated", "label_added") and payload.get("label", "").lower() in (
|
|
"scam",
|
|
"fraud",
|
|
"hack",
|
|
"phishing",
|
|
):
|
|
return {
|
|
"level": "CRITICAL",
|
|
"title": "Scam entity flagged",
|
|
"description": f"{service} flagged {payload.get('address', '?')} as {payload.get('label', 'scam')}",
|
|
}
|
|
|
|
return {"level": "LOW", "title": "Routine event", "description": f"{service} {event_type}"}
|
|
|
|
|
|
async def setup_webhook(service: str, webhook_url: str, events: list[str] | None = None, **kw) -> dict:
|
|
"""Programmatically set up a webhook subscription with a service.
|
|
|
|
For services that support API-based webhook registration (Helius, Moralis),
|
|
this auto-configures the webhook URL and event filters.
|
|
"""
|
|
api_key = kw.get("api_key", "")
|
|
|
|
if service == "helius":
|
|
if not api_key:
|
|
api_key = os.getenv("HELIUS_API_KEY", "")
|
|
if not api_key:
|
|
return {"error": "HELIUS_API_KEY required"}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
resp = await c.post(
|
|
f"https://api.helius.xyz/v0/webhooks?api-key={api_key}",
|
|
json={
|
|
"webhookURL": webhook_url,
|
|
"transactionTypes": events or ["ANY"],
|
|
"accountAddresses": kw.get("addresses", []),
|
|
"webhookType": "enhanced",
|
|
},
|
|
)
|
|
return resp.json() if resp.status_code == 200 else {"error": resp.text[:200]}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
elif service == "moralis":
|
|
if not api_key:
|
|
api_key = os.getenv("MORALIS_API_KEY", "")
|
|
if not api_key:
|
|
return {"error": "MORALIS_API_KEY required"}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
resp = await c.post(
|
|
"https://authapi.moralis.io/streams/evm",
|
|
headers={"X-API-Key": api_key, "Content-Type": "application/json"},
|
|
json={
|
|
"webhookUrl": webhook_url,
|
|
"description": f"RMI DataBus webhook for {service}",
|
|
"tag": "rmi-databus",
|
|
"chains": kw.get("chains", ["eth", "bsc", "polygon"]),
|
|
"includeNativeTxs": True,
|
|
},
|
|
)
|
|
return resp.json() if resp.status_code in (200, 201) else {"error": resp.text[:200]}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
return {
|
|
"status": "manual_setup_required",
|
|
"service": service,
|
|
"webhook_url": webhook_url,
|
|
"instructions": f"Configure webhook at {service} dashboard to point to {webhook_url}",
|
|
}
|
|
|
|
|
|
async def list_webhooks() -> dict:
|
|
"""List all configured webhook subscriptions."""
|
|
try:
|
|
r = _redis()
|
|
keys = r.keys("webhook:*:*")
|
|
webhooks = []
|
|
for key in keys[:50]:
|
|
data = r.get(key)
|
|
if data:
|
|
webhooks.append(json.loads(data))
|
|
r.close()
|
|
return {"webhooks": webhooks, "total": len(webhooks)}
|
|
except Exception:
|
|
return {"webhooks": [], "total": 0}
|