cryptorugmunch
628c1d2a10
Some checks failed
CI / build (push) Failing after 3s
PHASE 2.3 (AUDIT-2026-Q3.md):
Task 1 — Wire-in Wave 3 (1 router mounted, 2 deferred):
- app.routers.unified_scanner_router mounted at /api/v2/scanner/* (2 routes:
POST /api/v2/scanner/token/scan, POST /api/v2/scanner/wallet/scan).
Refactored prefix from /api/v2 -> /api/v2/scanner to avoid future conflicts
with the v1 /api/v1/scanner/ stub.
- app.routers.unified_wallet_scanner DEFERRED (no router APIRouter attribute;
library module consumed by unified_scanner_router via get_wallet_scanner()).
- app.routers.admin_extensions DEFERRED (DORMANT per audit; 25 routes at
/api/v1/admin/* would shadow /api/v1/admin/alerts_webhook).
Task 2 — Archive 136 dead-code files to app/_archive/legacy_2026_07/:
- 73 routers in app/routers/ (reach graph showed zero reach into mount.py).
- 63 flat app/*.py (domain modules never imported by live code).
- 1 file RESTORED post-archive: app/routers/x402_bridge_health.py (caught by
tests/unit/test_bridge_health.py which directly imports it; reach graph
considered tests/ only as transitive reach — to be patched in next cycle).
Forced-LIVE (NOT archived per user directive):
- app/ai_pipeline_v3.py (3 importers in audit window, importers themselves DEAD)
- app/splade_bm25.py (LIVE via app.rag_service)
- app/wallet_manager_v2.py (LIVE via x402_enforcement, x402_tools, sweep_all, sweep_now)
- app/crypto_embeddings.py (NOT in audit ARCHIVE list; heavy import graph)
Verification (forward-import closure from mount.py + main.py + factory.py + lifespan.py):
- imports = 348 app.* modules
- reached = 194 files reachable from roots
- archive set = audit_dead (186) - reached - forced_live (4) - test_live (1) = 136
- Net delta: 136 files moved, 44,932 LOC reduction, 293->295 active routes (+2 from Wave 3)
pyproject.toml updates:
- setuptools.packages.find: added exclude for app._archive*
- ruff.extend-exclude: added "app/_archive/"
- mypy.exclude: added "app/_archive/"
Smoke test: pytest tests/ — 817 passed, 3 pre-existing failures unchanged
(0 new failures; 0 routes lost; all 4 forced-LIVE files still importable).
Restoration: git mv app/_archive/legacy_2026_07/<name>.py <original-path>
and add the import to app/mount.py ROUTER_MODULES.
Refs: AUDIT-2026-Q3.md /home/dev/pry/rmi-final-deadcode-2026-07-06.md
311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""
|
|
Webhook Receivers - Helius (Solana) + Moralis Streams (EVM).
|
|
Processes real-time blockchain events: transfers, swaps, mints, whale activity.
|
|
INTELLIGENT PROCESSING: Whale detection, cluster correlation, scam pattern matching.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/v1/webhooks", tags=["webhooks"])
|
|
|
|
# ── Config ──────────────────────────────────────────────────
|
|
|
|
HELIUS_WEBHOOK_AUTH = os.getenv("HELIUS_WEBHOOK_AUTH", "")
|
|
HELIUS_WEBHOOK_SECRET = os.getenv("HELIUS_WEBHOOK_SECRET", "")
|
|
|
|
# ── Intelligent Processor ────────────────────────────────────
|
|
|
|
|
|
async def _process_with_intelligence(event: dict, source: str) -> dict | None:
|
|
"""Process event with intelligent analysis."""
|
|
try:
|
|
from app.intelligent_webhooks import get_intelligent_processor
|
|
|
|
processor = get_intelligent_processor()
|
|
|
|
if source == "helius":
|
|
alert = await processor.process_helius_event(event)
|
|
elif source == "moralis":
|
|
alert = await processor.process_moralis_event(event)
|
|
else:
|
|
return None
|
|
|
|
if alert:
|
|
# Log critical/high alerts
|
|
if alert.severity in ("critical", "high"):
|
|
logger.warning(
|
|
f"🚨 {alert.severity.upper()}: {alert.alert_type} "
|
|
f"wallet={alert.wallet[:16]}... amount=${alert.amount_usd:,.0f} "
|
|
f"risk={alert.risk_score:.1f}"
|
|
)
|
|
|
|
return {
|
|
"alert_id": alert.alert_id,
|
|
"severity": alert.severity,
|
|
"type": alert.alert_type,
|
|
"wallet": alert.wallet,
|
|
"chain": alert.chain,
|
|
"amount_usd": alert.amount_usd,
|
|
"risk_score": alert.risk_score,
|
|
"description": alert.description,
|
|
"enriched": alert.enriched_data,
|
|
"actions": alert.actions,
|
|
}
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"Intelligent processing failed: {e}")
|
|
return None
|
|
|
|
|
|
def _process_helius_event(event: dict) -> dict:
|
|
"""Process a single Helius webhook event."""
|
|
result = {
|
|
"signature": event.get("signature", ""),
|
|
"type": event.get("type", "unknown"),
|
|
"source": "helius",
|
|
"timestamp": event.get("timestamp", 0),
|
|
"wallet": "",
|
|
"description": "",
|
|
"processed": False,
|
|
}
|
|
|
|
try:
|
|
# Extract key data from Helius enriched transaction
|
|
events = event.get("events", [])
|
|
event.get("accountData", [])
|
|
fee_payer = event.get("feePayer", "")
|
|
description = event.get("description", "")
|
|
|
|
result["wallet"] = fee_payer
|
|
result["description"] = description[:200]
|
|
|
|
# Classify event type
|
|
event_types = [e.get("type", "") for e in events] if isinstance(events, list) else []
|
|
if any("swap" in str(e).lower() for e in [*event_types, description]):
|
|
result["type"] = "swap"
|
|
elif any("transfer" in str(e).lower() for e in [*event_types, description]):
|
|
result["type"] = "transfer"
|
|
elif any("mint" in str(e).lower() for e in [*event_types, description]):
|
|
result["type"] = "mint"
|
|
elif any("nft" in str(e).lower() for e in [*event_types, description]):
|
|
result["type"] = "nft"
|
|
|
|
# Track amounts from account data
|
|
native_transfers = event.get("nativeTransfers", [])
|
|
if native_transfers:
|
|
max_transfer = max(native_transfers, key=lambda t: t.get("amount", 0), default={})
|
|
result["largest_transfer"] = max_transfer.get("amount", 0) / 1e9 # lamports to SOL
|
|
|
|
# Token transfers
|
|
token_transfers = event.get("tokenTransfers", [])
|
|
if token_transfers:
|
|
for tt in token_transfers[:3]: # Top 3 token transfers
|
|
mint = tt.get("mint", "")
|
|
amount = tt.get("tokenAmount", {}).get("uiAmount", 0)
|
|
if amount and float(amount) > 0:
|
|
result.setdefault("token_moves", []).append(
|
|
{
|
|
"mint": mint,
|
|
"amount": float(amount),
|
|
"from": tt.get("fromUserAccount", ""),
|
|
"to": tt.get("toUserAccount", ""),
|
|
}
|
|
)
|
|
|
|
result["processed"] = True
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Event processing error: {e}")
|
|
result["error"] = str(e)[:200]
|
|
|
|
return result
|
|
|
|
|
|
def _process_moralis_event(event: dict) -> dict:
|
|
"""Process a single Moralis Stream (EVM) event."""
|
|
result = {
|
|
"hash": event.get("transactionHash", event.get("hash", "")),
|
|
"chain": event.get("chainId", "0x1"),
|
|
"source": "moralis",
|
|
"type": "unknown",
|
|
"from": event.get("fromAddress", ""),
|
|
"to": event.get("toAddress", ""),
|
|
"value": "0",
|
|
"timestamp": event.get("blockTimestamp", ""),
|
|
"processed": False,
|
|
}
|
|
|
|
try:
|
|
# Classify by log events or description
|
|
logs = event.get("logs", [])
|
|
erc20_transfers = event.get("erc20Transfers", [])
|
|
|
|
if erc20_transfers:
|
|
result["type"] = "erc20_transfer"
|
|
for t in erc20_transfers[:3]:
|
|
result.setdefault("token_transfers", []).append(
|
|
{
|
|
"from": t.get("from", ""),
|
|
"to": t.get("to", ""),
|
|
"value": t.get("value", "0"),
|
|
"symbol": t.get("symbol", ""),
|
|
"token": t.get("address", ""),
|
|
}
|
|
)
|
|
elif logs:
|
|
# Check for swap events (Uniswap, etc.)
|
|
result["type"] = "contract_interaction"
|
|
else:
|
|
native_value = event.get("value", "0")
|
|
if native_value and float(native_value) > 0:
|
|
result["type"] = "native_transfer"
|
|
result["value"] = str(float(native_value) / 1e18)
|
|
|
|
# Whale detection - flag transfers > 10 ETH equivalent
|
|
try:
|
|
val = float(event.get("value", "0")) / 1e18 if event.get("value") else 0
|
|
if val > 10:
|
|
result["whale_alert"] = True
|
|
result["whale_value_eth"] = val
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
result["processed"] = True
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Moralis event processing error: {e}")
|
|
result["error"] = str(e)[:200]
|
|
|
|
return result
|
|
|
|
|
|
# ── Endpoints ───────────────────────────────────────────────
|
|
|
|
|
|
@router.post("/helius")
|
|
async def helius_webhook(request: Request):
|
|
"""Receive Helius webhook events (Solana transfers, swaps, mints).
|
|
INTELLIGENT: Whale detection, cluster correlation, scam pattern matching."""
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid JSON") from None
|
|
|
|
# Helius can send single events or arrays
|
|
events = body if isinstance(body, list) else [body]
|
|
processed = []
|
|
alerts = []
|
|
|
|
for event in events:
|
|
# Basic processing
|
|
result = _process_helius_event(event)
|
|
processed.append(result)
|
|
|
|
# Intelligent processing
|
|
alert = await _process_with_intelligence(event, "helius")
|
|
if alert:
|
|
alerts.append(alert)
|
|
|
|
# Store in cache for retrieval
|
|
try:
|
|
from app.chain_cache import get_chain_cache
|
|
|
|
cache = get_chain_cache()
|
|
await cache.set("helius_events", processed, "latest", ttl=3600)
|
|
if alerts:
|
|
await cache.set("helius_alerts", alerts, "latest", ttl=3600)
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"status": "ok",
|
|
"events_processed": len(processed),
|
|
"alerts_generated": len(alerts),
|
|
"critical_alerts": len([a for a in alerts if a.get("severity") == "critical"]),
|
|
}
|
|
|
|
|
|
@router.post("/moralis")
|
|
async def moralis_webhook(request: Request):
|
|
"""Receive Moralis Stream events (EVM whale tracking, token transfers).
|
|
INTELLIGENT: Whale detection, cluster correlation, scam pattern matching."""
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid JSON") from None
|
|
|
|
# Moralis streams send arrays of confirmed/tx events
|
|
confirmed = body.get("confirmed", body.get("logs", []))
|
|
if isinstance(body, list):
|
|
confirmed = body
|
|
|
|
processed = []
|
|
alerts = []
|
|
for event in confirmed if isinstance(confirmed, list) else [confirmed]:
|
|
# Basic processing
|
|
result = _process_moralis_event(event)
|
|
processed.append(result)
|
|
|
|
# Intelligent processing
|
|
alert = await _process_with_intelligence(event, "moralis")
|
|
if alert:
|
|
alerts.append(alert)
|
|
|
|
# Cache latest events
|
|
try:
|
|
from app.chain_cache import get_chain_cache
|
|
|
|
cache = get_chain_cache()
|
|
await cache.set("moralis_events", processed, "latest", ttl=3600)
|
|
if alerts:
|
|
await cache.set("moralis_alerts", alerts, "latest", ttl=3600)
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"status": "ok",
|
|
"events_processed": len(processed),
|
|
"alerts_generated": len(alerts),
|
|
"critical_alerts": len([a for a in alerts if a.get("severity") == "critical"]),
|
|
}
|
|
|
|
|
|
@router.get("/events")
|
|
async def get_recent_events(source: str | None = None):
|
|
"""Get recent webhook events from cache."""
|
|
try:
|
|
from app.chain_cache import get_chain_cache
|
|
|
|
cache = get_chain_cache()
|
|
|
|
events = []
|
|
helius = await cache.get("helius_events", "latest")
|
|
if helius:
|
|
events.extend(helius if source != "moralis" else [])
|
|
|
|
moralis = await cache.get("moralis_events", "latest")
|
|
if moralis:
|
|
events.extend(moralis if source != "helius" else [])
|
|
|
|
return {"total": len(events), "events": events[:50]}
|
|
except Exception:
|
|
return {"total": 0, "events": []}
|
|
|
|
|
|
@router.get("/health")
|
|
async def webhooks_health():
|
|
"""Webhook health check."""
|
|
return {
|
|
"status": "ok",
|
|
"service": "webhook-receiver",
|
|
"endpoints": {
|
|
"helius": "/api/v1/webhooks/helius",
|
|
"moralis": "/api/v1/webhooks/moralis",
|
|
"events": "/api/v1/webhooks/events",
|
|
},
|
|
}
|