ip_intel.py

"""
IP Intelligence Module
Queries AbuseIPDB, IPQualityScore, ipinfo.io, StopForumSpam, and VirusTotal
concurrently and returns a unified risk assessment.
"""

import asyncio
import os
from typing import Any, Dict, List, Tuple

import httpx

ABUSEIPDB_KEY     = os.getenv("ABUSEIPDB_KEY", "")
IPQS_KEY          = os.getenv("IPQUALITYSCORE_KEY", "")
IPINFO_TOKEN      = os.getenv("IPINFO_TOKEN", "")
VIRUSTOTAL_KEY    = os.getenv("VIRUSTOTAL_KEY", "")

# Datacenter / hosting ASN keywords (ipinfo org field)
_DC_KEYWORDS = {
    "amazon", "aws", "google", "microsoft", "azure", "digitalocean", "linode",
    "vultr", "hetzner", "ovh", "cloudflare", "fastly", "akamai", "choopa",
    "quadranet", "datacamp", "psychz", "leaseweb", "serverius", "zayo",
    "hurricane electric", "he net", "tzulo", "path network",
}


# ──────────────────────────────────────────────
# Individual source fetchers
# ──────────────────────────────────────────────

async def _abuseipdb(ip: str, client: httpx.AsyncClient) -> Dict[str, Any]:
    if not ABUSEIPDB_KEY:
        return {}
    try:
        r = await client.get(
            "https://api.abuseipdb.com/api/v2/check",
            params={"ipAddress": ip, "maxAgeInDays": 90, "verbose": True},
            headers={"Key": ABUSEIPDB_KEY, "Accept": "application/json"},
            timeout=8,
        )
        if r.status_code == 200:
            d = r.json().get("data", {})
            return {
                "abuse_confidence":  d.get("abuseConfidenceScore", 0),
                "total_reports":     d.get("totalReports", 0),
                "usage_type":        d.get("usageType", ""),
                "country":           d.get("countryCode", ""),
                "isp":               d.get("isp", ""),
                "is_whitelisted":    d.get("isWhitelisted", False),
                "last_reported":     d.get("lastReportedAt"),
                "num_distinct_users": d.get("numDistinctUsers", 0),
            }
    except Exception:
        pass
    return {}


async def _ipqualityscore(ip: str, client: httpx.AsyncClient) -> Dict[str, Any]:
    if not IPQS_KEY:
        return {}
    try:
        r = await client.get(
            f"https://ipqualityscore.com/api/json/ip/{IPQS_KEY}/{ip}",
            params={"strictness": 1, "allow_public_access_points": True, "fast": False},
            timeout=8,
        )
        if r.status_code == 200:
            d = r.json()
            if d.get("success"):
                return {
                    "fraud_score":    d.get("fraud_score", 0),
                    "proxy":          d.get("proxy", False),
                    "vpn":            d.get("vpn", False),
                    "tor":            d.get("tor", False),
                    "recent_abuse":   d.get("recent_abuse", False),
                    "bot_status":     d.get("bot_status", False),
                    "country":        d.get("country_code", ""),
                    "isp":            d.get("ISP", ""),
                    "organization":   d.get("organization", ""),
                    "asn":            d.get("ASN", 0),
                    "connection_type": d.get("connection_type", ""),
                    "abuse_velocity": d.get("abuse_velocity", ""),
                }
    except Exception:
        pass
    return {}


async def _ipinfo(ip: str, client: httpx.AsyncClient) -> Dict[str, Any]:
    if not IPINFO_TOKEN:
        return {}
    try:
        r = await client.get(
            f"https://ipinfo.io/{ip}",
            params={"token": IPINFO_TOKEN},
            timeout=8,
        )
        if r.status_code == 200:
            d = r.json()
            org = d.get("org", "").lower()
            is_datacenter = any(kw in org for kw in _DC_KEYWORDS)
            # privacy field available on paid plans
            privacy = d.get("privacy", {})
            return {
                "org":          d.get("org", ""),
                "country":      d.get("country", ""),
                "city":         d.get("city", ""),
                "hostname":     d.get("hostname", ""),
                "is_datacenter": is_datacenter,
                "is_vpn":       privacy.get("vpn", False),
                "is_proxy":     privacy.get("proxy", False),
                "is_tor":       privacy.get("tor", False),
                "is_relay":     privacy.get("relay", False),
                "is_hosting":   privacy.get("hosting", is_datacenter),
            }
    except Exception:
        pass
    return {}


async def _stopforumspam(ip: str, client: httpx.AsyncClient) -> Dict[str, Any]:
    """Free API — no key needed."""
    try:
        r = await client.get(
            "https://api.stopforumspam.org/api",
            params={"ip": ip, "json": 1},
            timeout=6,
        )
        if r.status_code == 200:
            ip_data = r.json().get("ip", {})
            return {
                "appears":    ip_data.get("appears", 0) == 1,
                "frequency":  ip_data.get("frequency", 0),
                "confidence": float(ip_data.get("confidence", 0)),
                "lastseen":   ip_data.get("lastseen", ""),
            }
    except Exception:
        pass
    return {}


async def _virustotal(ip: str, client: httpx.AsyncClient) -> Dict[str, Any]:
    if not VIRUSTOTAL_KEY:
        return {}
    try:
        r = await client.get(
            f"https://www.virustotal.com/api/v3/ip_addresses/{ip}",
            headers={"x-apikey": VIRUSTOTAL_KEY},
            timeout=10,
        )
        if r.status_code == 200:
            attrs = r.json().get("data", {}).get("attributes", {})
            stats = attrs.get("last_analysis_stats", {})
            return {
                "malicious":  stats.get("malicious", 0),
                "suspicious": stats.get("suspicious", 0),
                "harmless":   stats.get("harmless", 0),
                "country":    attrs.get("country", ""),
                "as_owner":   attrs.get("as_owner", ""),
            }
    except Exception:
        pass
    return {}


# ──────────────────────────────────────────────
# Aggregator
# ──────────────────────────────────────────────

async def get_ip_intelligence(ip: str) -> Dict[str, Any]:
    """
    Query all IP intelligence sources concurrently.
    Returns { score: int, reasons: list[str], raw: dict }
    """
    if not ip:
        return {"score": 0, "reasons": [], "raw": {}}

    async with httpx.AsyncClient() as client:
        raw_results = await asyncio.gather(
            _abuseipdb(ip, client),
            _ipqualityscore(ip, client),
            _ipinfo(ip, client),
            _stopforumspam(ip, client),
            _virustotal(ip, client),
            return_exceptions=True,
        )

    abusedb = raw_results[0] if isinstance(raw_results[0], dict) else {}
    ipqs    = raw_results[1] if isinstance(raw_results[1], dict) else {}
    ipinfo  = raw_results[2] if isinstance(raw_results[2], dict) else {}
    sfs     = raw_results[3] if isinstance(raw_results[3], dict) else {}
    vt      = raw_results[4] if isinstance(raw_results[4], dict) else {}

    score: int = 0
    reasons: List[str] = []

    # ── AbuseIPDB ──────────────────────────────
    ac = abusedb.get("abuse_confidence", 0)
    if ac >= 80:
        score += 45
        reasons.append(
            f"IP confidence {ac}% on AbuseIPDB "
            f"({abusedb.get('total_reports', 0)} reports, "
            f"{abusedb.get('num_distinct_users', 0)} distinct users)"
        )
    elif ac >= 50:
        score += 28
        reasons.append(f"IP flagged on AbuseIPDB (confidence {ac}%)")
    elif ac >= 20:
        score += 12
        reasons.append(f"IP has low-level AbuseIPDB reports (confidence {ac}%)")

    # ── IPQualityScore ─────────────────────────
    if ipqs.get("tor"):
        score += 40
        reasons.append("IP is a Tor exit node (IPQualityScore)")
    else:
        if ipqs.get("proxy"):
            score += 22
            reasons.append("IP identified as proxy (IPQualityScore)")
        if ipqs.get("vpn"):
            score += 15
            reasons.append("IP identified as VPN (IPQualityScore)")

    if ipqs.get("recent_abuse"):
        score += 28
        reasons.append("IP has recent abuse history (IPQualityScore)")
    if ipqs.get("bot_status"):
        score += 32
        reasons.append("IP associated with automated bot traffic (IPQualityScore)")

    ipqs_fraud = ipqs.get("fraud_score", 0)
    if ipqs_fraud >= 85:
        score += 20
        reasons.append(f"Very high IP fraud score {ipqs_fraud}/100 (IPQualityScore)")
    elif ipqs_fraud >= 60:
        score += 10
        reasons.append(f"Elevated IP fraud score {ipqs_fraud}/100 (IPQualityScore)")

    av = ipqs.get("abuse_velocity", "")
    if av in ("high", "medium"):
        score += 15
        reasons.append(f"IP abuse velocity: {av} (IPQualityScore)")

    # ── ipinfo ─────────────────────────────────
    if ipinfo.get("is_tor"):
        score += 35  # may not already be counted above
        reasons.append("IP is a Tor exit node (ipinfo)")
    if ipinfo.get("is_datacenter") or ipinfo.get("is_hosting"):
        score += 12
        reasons.append(f"IP belongs to datacenter/hosting provider: {ipinfo.get('org', '')}")
    if ipinfo.get("is_relay"):
        score += 18
        reasons.append("IP is an Apple iCloud Private Relay or similar anonymous relay")

    # ── StopForumSpam ──────────────────────────
    if sfs.get("appears"):
        conf = sfs.get("confidence", 0.0)
        freq = sfs.get("frequency", 0)
        score += min(35, max(10, int(conf / 3)))
        reasons.append(
            f"IP in StopForumSpam database (confidence {conf:.1f}%, "
            f"seen {freq} times, last: {sfs.get('lastseen', 'unknown')})"
        )

    # ── VirusTotal ─────────────────────────────
    vt_mal = vt.get("malicious", 0)
    vt_sus = vt.get("suspicious", 0)
    if vt_mal >= 3:
        score += 30
        reasons.append(f"IP flagged malicious by {vt_mal} VirusTotal vendors")
    elif vt_mal >= 1:
        score += 12
        reasons.append(f"IP flagged malicious by {vt_mal} VirusTotal vendor(s)")
    elif vt_sus >= 3:
        score += 10
        reasons.append(f"IP flagged suspicious by {vt_sus} VirusTotal vendors")

    return {
        "score": min(score, 100),
        "reasons": reasons,
        "raw": {
            "abuseipdb":      abusedb,
            "ipqualityscore": ipqs,
            "ipinfo":         ipinfo,
            "stopforumspam":  sfs,
            "virustotal":     vt,
        },
    }