aur-shield/aur_shield/extended_iocs.py

"""Extended IOC sources — additional threat intel from archcanary-inspired feeds.

Sources:
1. aur-audit.wtako.net — third-party AUR audit API (black/red lists)
2. Community reports — community-curated malicious package list
3. CHAOS RAT list — additional threat campaign
4. Russian spam campaign — separate list
"""
from __future__ import annotations

import asyncio
import json
from typing import Any

import httpx

from .ioc_fetcher import IOCEntry, ThreatType, Confidence

# aur-audit.wtako.net API
AUR_AUDIT_BLACK_URL = "https://aur-audit.wtako.net/api/black"
AUR_AUDIT_RED_URL = "https://aur-audit.wtako.net/api/red"

# Community reports (archcanary repo)
COMMUNITY_REPORTS_URL = (
    "https://raw.githubusercontent.com/musqz/archcanary/master/lists/"
    "community_reports.txt"
)

# CHAOS RAT campaign
CHAOS_RAT_URL = (
    "https://raw.githubusercontent.com/musqz/archcanary/master/lists/"
    "chaos_rat_packages.txt"
)

# Russian spam campaign
RUSSIAN_SPAM_URL = (
    "https://raw.githubusercontent.com/musqz/archcanary/master/lists/"
    "malicious_russian_spam_packages.txt"
)


def _valid_pkg(name: str) -> bool:
    if not name or len(name) > 100:
        return False
    return all(
        c.isascii() and (c.islower() or c.isdigit() or c in "-_.")
        for c in name
    )


def _parse_line_list(text: str, source: str, threat: ThreatType,
                      confidence: Confidence, description: str) -> list[IOCEntry]:
    """Parse a plaintext one-package-per-line list."""
    threats = []
    for line in text.splitlines():
        pkg = line.strip()
        if pkg.startswith(("- ", "* ", "#")):
            pkg = pkg[2:].strip() if not pkg.startswith("#") else ""
        if _valid_pkg(pkg):
            threats.append(IOCEntry(
                package_name=pkg,
                threat_type=threat,
                source=source,
                description=description,
                confidence=confidence,
            ))
    return threats


async def _fetch_aur_audit_black(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch confirmed-malicious packages from aur-audit.wtako.net."""
    try:
        resp = await client.get(AUR_AUDIT_BLACK_URL)
        if resp.status_code != 200:
            return []
        data = resp.json()
    except Exception:
        return []

    threats = []
    packages = data if isinstance(data, list) else data.get("packages", [])
    for pkg in packages:
        name = pkg if isinstance(pkg, str) else pkg.get("name", "")
        if _valid_pkg(name):
            threats.append(IOCEntry(
                package_name=name,
                threat_type=ThreatType.MALICIOUS_BUILD,
                source="aur-audit:black",
                description="Confirmed malicious by aur-audit.wtako.net",
                confidence=Confidence.CRITICAL,
            ))
    return threats


async def _fetch_aur_audit_red(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch high-risk packages from aur-audit.wtako.net."""
    try:
        resp = await client.get(AUR_AUDIT_RED_URL)
        if resp.status_code != 200:
            return []
        data = resp.json()
    except Exception:
        return []

    threats = []
    packages = data if isinstance(data, list) else data.get("packages", [])
    for pkg in packages:
        name = pkg if isinstance(pkg, str) else pkg.get("name", "")
        if _valid_pkg(name):
            threats.append(IOCEntry(
                package_name=name,
                threat_type=ThreatType.MALICIOUS_BUILD,
                source="aur-audit:red",
                description="High-risk by aur-audit.wtako.net",
                confidence=Confidence.HIGH,
            ))
    return threats


async def _fetch_community_reports(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch community-reported malicious packages."""
    try:
        resp = await client.get(COMMUNITY_REPORTS_URL)
        if resp.status_code != 200:
            return []
        return _parse_line_list(
            resp.text,
            source="community_reports",
            threat=ThreatType.UNKNOWN("CommunityReport"),
            confidence=Confidence.MEDIUM,
            description="Reported by community (unverified)",
        )
    except Exception:
        return []


async def _fetch_chaos_rat(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch CHAOS RAT campaign packages."""
    try:
        resp = await client.get(CHAOS_RAT_URL)
        if resp.status_code != 200:
            return []
        return _parse_line_list(
            resp.text,
            source="chaos_rat",
            threat=ThreatType.BACKDOOR,
            confidence=Confidence.CRITICAL,
            description="CHAOS RAT campaign — backdoor payload",
        )
    except Exception:
        return []


async def _fetch_russian_spam(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch Russian spam campaign packages."""
    try:
        resp = await client.get(RUSSIAN_SPAM_URL)
        if resp.status_code != 200:
            return []
        return _parse_line_list(
            resp.text,
            source="russian_spam",
            threat=ThreatType.UNKNOWN("SpamInjection"),
            confidence=Confidence.MEDIUM,
            description="Russian spam campaign — .bashrc injection",
        )
    except Exception:
        return []


async def fetch_extended_iocs() -> list[IOCEntry]:
    """Fetch IOCs from extended sources (archcanary-inspired).

    Run all fetches concurrently with individual error handling.
    """
    async with httpx.AsyncClient(timeout=30) as client:
        results = await asyncio.gather(
            _fetch_aur_audit_black(client),
            _fetch_aur_audit_red(client),
            _fetch_community_reports(client),
            _fetch_chaos_rat(client),
            _fetch_russian_spam(client),
            return_exceptions=True,
        )

    all_threats = []
    source_names = [
        "aur-audit:black", "aur-audit:red", "community_reports",
        "chaos_rat", "russian_spam",
    ]
    for name, r in zip(source_names, results):
        if isinstance(r, list) and r:
            all_threats.extend(r)

    return all_threats