Bot Detection for
Crypto & Web3

# Python — Airdrop eligibility with Sybil detection
import httpx
from collections import defaultdict

# Track IP clustering across claim attempts
ip_cluster_counts = defaultdict(int)  # ASN -> claim count

async def check_airdrop_eligibility(wallet_address, client_ip):
    """Screen airdrop claims for Sybil/farming patterns"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.ipasis.com/v1/lookup/{client_ip}",
            headers={"Authorization": "Bearer YOUR_API_KEY"},
            timeout=0.05
        )
        ip_data = resp.json()
    
    risk_flags = []
    
    # Infrastructure detection
    if ip_data.get("is_datacenter"):
        risk_flags.append("DATACENTER_IP")
    if ip_data.get("is_residential_proxy"):
        risk_flags.append("RESIDENTIAL_PROXY")  # Big red flag for farming
    if ip_data.get("is_proxy"):
        risk_flags.append("PROXY")
    if ip_data.get("is_vpn"):
        risk_flags.append("VPN")
    if ip_data.get("is_tor"):
        risk_flags.append("TOR")
    
    # ASN clustering — too many claims from same network
    asn = ip_data.get("asn", "unknown")
    ip_cluster_counts[asn] += 1
    if ip_cluster_counts[asn] > 10:
        risk_flags.append("ASN_CLUSTER")
    
    # High risk score
    risk_score = ip_data.get("risk_score", 0)
    if risk_score >= 0.7:
        risk_flags.append("HIGH_RISK_SCORE")
    
    # Decision
    if "TOR" in risk_flags or "DATACENTER_IP" in risk_flags:
        return {"eligible": False, "reason": "suspicious_infrastructure"}
    
    if "RESIDENTIAL_PROXY" in risk_flags or "ASN_CLUSTER" in risk_flags:
        return {"eligible": False, "reason": "sybil_pattern_detected"}
    
    if len(risk_flags) >= 2:
        return {"eligible": "pending_review", "flags": risk_flags}
    
    return {"eligible": True, "risk_score": risk_score}

// TypeScript — Exchange login/signup risk assessment
interface ExchangeRiskResult {
  action: 'allow' | 'mfa_required' | 'enhanced_kyc' | 'block';
  riskLevel: 'low' | 'medium' | 'high' | 'critical';
  signals: string[];
  requiresManualReview: boolean;
}

async function assessExchangeRisk(
  clientIP: string,
  action: 'signup' | 'login' | 'withdrawal'
): Promise<ExchangeRiskResult> {
  const ipData = await fetch(
    `https://api.ipasis.com/v1/lookup/${clientIP}`,
    { headers: { Authorization: 'Bearer YOUR_API_KEY' } }
  ).then(r => r.json());

  const signals: string[] = [];
  
  if (ipData.is_datacenter) signals.push('DATACENTER');
  if (ipData.is_proxy) signals.push('PROXY');
  if (ipData.is_residential_proxy) signals.push('RESIDENTIAL_PROXY');
  if (ipData.is_tor) signals.push('TOR');
  if (ipData.is_vpn) signals.push('VPN');
  if (ipData.risk_score >= 0.7) signals.push('HIGH_RISK');

  // Withdrawal = highest security
  if (action === 'withdrawal') {
    if (signals.includes('TOR') || signals.includes('DATACENTER')) {
      return { action: 'block', riskLevel: 'critical', signals, requiresManualReview: true };
    }
    if (signals.length > 0) {
      return { action: 'mfa_required', riskLevel: 'high', signals, requiresManualReview: false };
    }
  }

  // Signup = moderate security, catch Sybils
  if (action === 'signup') {
    if (ipData.is_datacenter || ipData.is_tor) {
      return { action: 'block', riskLevel: 'critical', signals, requiresManualReview: false };
    }
    if (ipData.is_residential_proxy || (ipData.is_vpn && ipData.risk_score >= 0.6)) {
      return { action: 'enhanced_kyc', riskLevel: 'high', signals, requiresManualReview: true };
    }
  }

  // Login = balanced (many legit users use VPNs)
  if (signals.includes('DATACENTER') || signals.includes('TOR')) {
    return { action: 'mfa_required', riskLevel: 'high', signals, requiresManualReview: false };
  }
  
  if (signals.length >= 2) {
    return { action: 'mfa_required', riskLevel: 'medium', signals, requiresManualReview: false };
  }

  return { action: 'allow', riskLevel: 'low', signals, requiresManualReview: false };
}