aur-shield/aur_shield/ioc_fetcher.py

"""IOC fetcher — pulls known malicious package lists from public sources.

Based on AegisAUR's ioc_fetcher.rs. Checks package names against:
1. HedgeDoc live paste (Arch Linux community — always current)
2. Atomic Arch Gist (GitHub — versioned fallback)
3. Arch Security Tracker (official advisories)
4. AUR RPC (orphan takeover detection)
"""
from __future__ import annotations

import asyncio
import json
import re
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any

import httpx


class ThreatType(str, Enum):
    MALICIOUS_BUILD = "MaliciousBuildScript"
    CREDENTIAL_STEALER = "CredentialStealer"
    ROOTKIT = "Rootkit"
    CRYPTOMINER = "Cryptominer"
    BACKDOOR = "Backdoor"
    TYPOSQUATTING = "Typosquatting"
    ORPHAN_TAKEOVER = "OrphanTakeover"
    UNKNOWN = "Unknown"


class Confidence(str, Enum):
    CRITICAL = "Critical"
    HIGH = "High"
    MEDIUM = "Medium"
    LOW = "Low"


@dataclass
class IOCEntry:
    package_name: str
    threat_type: ThreatType
    source: str
    description: str
    confidence: Confidence
    discovered_date: str = ""


@dataclass
class IOCResult:
    package: str
    matches: list[IOCEntry] = field(default_factory=list)
    sources_checked: list[str] = field(default_factory=list)
    is_known_malicious: bool = False


# Source URLs (from AegisAUR)
HEDGEDOC_URL = "https://md.archlinux.org/s/SxbqukK6IA"
GIST_URL = "https://gist.githubusercontent.com/Kidev/85756c3dcad3623ca5604a8135bafd14/raw"
ARCH_SECURITY_URL = "https://security.archlinux.org/advisory/atomic-arch/json"
AUR_RPC_ORPHAN_URL = "https://aur.archlinux.org/rpc/v5/search?by=maintainer&arg=orphan"

CACHE_TTL = 300  # 5 minutes


def _valid_pkg_name(name: str) -> bool:
    """Validate that a string looks like a valid package name."""
    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
    )


async def _fetch_hedgedoc(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch IOC list from Arch Linux HedgeDoc live paste."""
    try:
        resp = await client.get(HEDGEDOC_URL)
        resp.raise_for_status()
    except Exception:
        return []

    threats = []
    for line in resp.text.splitlines():
        pkg = line.strip()
        if pkg.startswith(("- ", "* ")):
            pkg = pkg[2:].strip()
        # Skip headers, comments, metadata
        if (not pkg or pkg.startswith(("#", "---", "**", "Arch Linux",
             "Liste", "Quelle:", "Aktualisiert:"))):
            continue
        if _valid_pkg_name(pkg):
            threats.append(IOCEntry(
                package_name=pkg,
                threat_type=ThreatType.MALICIOUS_BUILD,
                source="hedgedoc_live",
                description="Atomic Arch Supply Chain Attack — Live HedgeDoc",
                confidence=Confidence.CRITICAL,
            ))
    return threats


async def _fetch_gist(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch IOC list from Atomic Arch GitHub Gist."""
    try:
        resp = await client.get(GIST_URL)
        resp.raise_for_status()
    except Exception:
        return []

    threats = []
    for line in resp.text.splitlines():
        pkg = line.strip()
        if not pkg or pkg.startswith(("#", "echo")):
            continue
        # Try JSON array
        if pkg.startswith("["):
            try:
                for p in json.loads(pkg):
                    if _valid_pkg_name(p):
                        threats.append(IOCEntry(
                            package_name=p,
                            threat_type=ThreatType.MALICIOUS_BUILD,
                            source="atomic_arch_gist",
                            description="Atomic Arch — Gist Fallback",
                            confidence=Confidence.CRITICAL,
                        ))
                continue
            except json.JSONDecodeError:
                pass
        # Plain package name
        if _valid_pkg_name(pkg) and " " not in pkg and "/" not in pkg:
            threats.append(IOCEntry(
                package_name=pkg,
                threat_type=ThreatType.MALICIOUS_BUILD,
                source="atomic_arch_gist",
                description="Atomic Arch — Gist Fallback",
                confidence=Confidence.CRITICAL,
            ))
    return threats


async def _fetch_arch_security(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch official Arch Linux security advisories."""
    try:
        resp = await client.get(ARCH_SECURITY_URL)
        if not resp.status_code == 200:
            return []
        data = resp.json()
    except Exception:
        return []

    threats = []
    packages = data.get("packages", [])
    if isinstance(packages, list):
        for pkg in packages:
            name = pkg if isinstance(pkg, str) else pkg.get("name", "")
            if _valid_pkg_name(name):
                threats.append(IOCEntry(
                    package_name=name,
                    threat_type=ThreatType.MALICIOUS_BUILD,
                    source="arch_security",
                    description="Arch Linux Security Advisory — Atomic Arch",
                    confidence=Confidence.CRITICAL,
                ))
    return threats


async def _fetch_orphan_takeover(client: httpx.AsyncClient) -> list[IOCEntry]:
    """Fetch suspicious orphaned packages from AUR RPC."""
    try:
        resp = await client.get(AUR_RPC_ORPHAN_URL)
        if not resp.status_code == 200:
            return []
        data = resp.json()
    except Exception:
        return []

    threats = []
    results = data.get("results", [])
    for pkg in results[:50]:
        name = pkg.get("Name", "")
        votes = pkg.get("NumVotes", 0)
        if _valid_pkg_name(name) and votes < 10:
            threats.append(IOCEntry(
                package_name=name,
                threat_type=ThreatType.ORPHAN_TAKEOVER,
                source="aur_rpc",
                description="AUR Orphaned Package — low votes, possible takeover",
                confidence=Confidence.MEDIUM,
            ))
    return threats


async def fetch_all_iocs() -> list[IOCEntry]:
    """Fetch IOCs from all sources concurrently."""
    async with httpx.AsyncClient(timeout=30) as client:
        results = await asyncio.gather(
            _fetch_hedgedoc(client),
            _fetch_gist(client),
            _fetch_arch_security(client),
            _fetch_orphan_takeover(client),
            return_exceptions=True,
        )

    all_threats = []
    for r in results:
        if isinstance(r, list):
            all_threats.extend(r)
    return all_threats


def check_package_against_iocs(
    package: str,
    iocs: list[IOCEntry],
) -> IOCResult:
    """Check if a package name appears in known malicious lists."""
    matches = [
        ioc for ioc in iocs
        if ioc.package_name.lower() == package.lower()
    ]

    return IOCResult(
        package=package,
        matches=matches,
        sources_checked=list({ioc.source for ioc in iocs}),
        is_known_malicious=any(
            ioc.confidence == Confidence.CRITICAL
            for ioc in matches
        ),
    )


def check_typosquatting(
    package: str,
    iocs: list[IOCEntry],
    threshold: int = 2,
) -> list[IOCEntry]:
    """Check for typosquatting — package names close to known malicious ones.

    Uses simple Levenshtein distance (no external dep).
    """
    def levenshtein(a: str, b: str) -> int:
        if len(a) < len(b):
            a, b = b, a
        if not b:
            return len(a)
        prev = list(range(len(b) + 1))
        for i, ca in enumerate(a):
            curr = [i + 1]
            for j, cb in enumerate(b):
                curr.append(min(
                    prev[j + 1] + 1,
                    curr[j] + 1,
                    prev[j] + (ca != cb),
                ))
            prev = curr
        return prev[-1]

    matches = []
    for ioc in iocs:
        if ioc.package_name.lower() == package.lower():
            continue  # exact match handled elsewhere
        dist = levenshtein(package.lower(), ioc.package_name.lower())
        if dist <= threshold and dist > 0:
            matches.append(IOCEntry(
                package_name=ioc.package_name,
                threat_type=ThreatType.TYPOSQUATTING,
                source=f"typosquatting_check(dist={dist})",
                description=f"Close match to known malicious '{ioc.package_name}' (distance={dist})",
                confidence=Confidence.HIGH,
            ))
    return matches