intelligence/ip_intel.py

from __future__ import annotations

from typing import Dict, List

from shield.ip_intel import get_ip_intelligence
from shield.models.evidence import Evidence


async def collect_ip_evidence(ip: str | None, historical: Dict[str, object] | None = None) -> List[Evidence]:
    evidence: List[Evidence] = []
    if ip:
        result = await get_ip_intelligence(ip)
        score = int(result.get("score", 0))
        reasons = result.get("reasons", [])
        if score and isinstance(reasons, list):
            evidence.extend([
                Evidence(
                    source="ip_intelligence",
                    category="network_risk",
                    risk_score=score,
                    weight=0.72,
                    confidence=0.84,
                    explanation=str(reason),
                )
                for reason in reasons[:5]
            ])

    if historical:
        abuse_count = int(historical.get("abuse_count", 0))
        avg_risk = float(historical.get("avg_risk", 0))
        linked_customers = int(historical.get("linked_customers", 0))
        if abuse_count or avg_risk >= 60:
            risk = min(100, max(int(avg_risk), 40 + (abuse_count * 8)))
            evidence.append(Evidence(
                source="global_memory",
                category="historical_abuse",
                risk_score=risk,
                weight=0.8,
                confidence=0.86,
                explanation="IP has prior abuse history in Shield memory",
                metadata={"abuse_count": abuse_count, "linked_customers": linked_customers},
            ))
    return evidence