graph/campaign_detection.py

from __future__ import annotations

import re
from typing import Any, Dict, List

from shield.memory.campaign_store import CampaignStore
from shield.models.evidence import Evidence
from shield.utils.hashing import hash_entity, sha256_hex


def _username_pattern(username: str | None) -> str | None:
    if not username:
        return None
    pattern = re.sub(r"[a-z]", "a", username.lower())
    pattern = re.sub(r"\d", "0", pattern)
    return pattern[:32]


def campaign_fingerprint(signals: Dict[str, Any], entities: Dict[str, str]) -> str | None:
    pieces: List[str] = []
    if entities.get("ip"):
        ip = entities["ip"]
        if "." in ip:
            pieces.append(f"ip_prefix:{hash_entity('ip_prefix', '.'.join(ip.split('.')[:3]))}")
        else:
            pieces.append(f"ip:{hash_entity('ip', ip)}")
    if entities.get("device"):
        pieces.append(f"device:{hash_entity('device', entities['device'])}")
    pattern = _username_pattern(entities.get("username"))
    if pattern:
        pieces.append(f"username_pattern:{sha256_hex(pattern)}")
    metadata = signals.get("metadata") if isinstance(signals.get("metadata"), dict) else {}
    timezone = metadata.get("timezone") if isinstance(metadata, dict) else None
    if isinstance(timezone, str) and timezone.strip():
        pieces.append(f"timezone:{sha256_hex(timezone.strip().lower())}")
    if len(pieces) < 2:
        return None
    return sha256_hex("|".join(sorted(pieces)))


def detect_campaign(
    signals: Dict[str, Any],
    entities: Dict[str, str],
    *,
    account_hash: str,
    store: CampaignStore | None = None,
) -> Dict[str, Any]:
    store = store or CampaignStore()
    campaign_id = campaign_fingerprint(signals, entities)
    if not campaign_id:
        return {
            "campaign_id": None,
            "campaign_risk_score": 0,
            "confidence": 0.0,
            "reasons": [],
            "evidence": [],
        }

    existing = store.get(campaign_id)
    account_count = int(existing.get("account_count", 0))
    device_count = int(existing.get("device_count", 0))
    ip_count = int(existing.get("ip_count", 0))
    risk = 0
    reasons: List[str] = []

    if account_count >= 25:
        risk = 85
        reasons.append("Large cluster of accounts shares campaign-like traits")
    elif account_count >= 10:
        risk = 70
        reasons.append("Multiple accounts share campaign-like traits")
    elif account_count >= 3:
        risk = 45
        reasons.append("Small cluster of accounts shares campaign-like traits")

    if device_count >= 5 and ip_count >= 3:
        risk = max(risk, 75)
        reasons.append("Campaign spans multiple devices and IPs")

    evidence: List[Evidence] = []
    if risk:
        evidence.append(Evidence(
            source="campaign_detection",
            category="campaign_abuse",
            risk_score=risk,
            weight=0.75,
            confidence=0.78,
            explanation=reasons[0],
            metadata={
                "campaign_id": campaign_id,
                "account_count": account_count,
                "device_count": device_count,
                "ip_count": ip_count,
            },
        ))

    return {
        "campaign_id": campaign_id,
        "campaign_risk_score": risk,
        "confidence": 0.78 if risk else 0.0,
        "reasons": reasons,
        "evidence": evidence,
    }