From 7f46bc8f9ac2bb555b50afe1cc6d880e1db00aeb Mon Sep 17 00:00:00 2001 From: arch_agent Date: Tue, 4 Aug 2026 09:39:53 +0200 Subject: [PATCH] feat: IOC pre-check via public threat intel (AegisAUR integration) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ioc_fetcher.py: fetches from HedgeDoc, Atomic Arch Gist, Arch Security Tracker, AUR RPC orphan detection (concurrent) - scanner.py: IOC pre-check before LLM scan — known malicious packages get instant MALICIOUS verdict without LLM cost - typosquatting check with Levenshtein distance - server.py: API returns ioc_matches + typosquat_matches - README: threat intel sources documented Sources ported from AegisAUR (Rust) to Python. --- README.md | 22 ++- aur_shield/ioc_fetcher.py | 275 ++++++++++++++++++++++++++++++++++++++ aur_shield/scanner.py | 92 ++++++++++++- aur_shield/server.py | 2 + 4 files changed, 383 insertions(+), 8 deletions(-) create mode 100644 aur_shield/ioc_fetcher.py diff --git a/README.md b/README.md index de8cc75..9cbe64b 100644 --- a/README.md +++ b/README.md @@ -92,18 +92,30 @@ cache: ## How It Works -1. **Request:** Client asks for `aur-shield/` +1. **IOC Pre-Check:** Package name checked against public threat lists (HedgeDoc, Atomic Arch Gist, Arch Security Tracker, AUR Orphan detection). Known malicious → instant block, no LLM needed. 2. **Fetch:** Server pulls PKGBUILD + .SRCINFO from AUR API -3. **Scan:** LLM analyzes the PKGBUILD for: +3. **Regex Pre-Scan:** Fast pattern matching for `curl|bash`, `eval|base64`, `/dev/tcp`, etc. +4. **LLM Scan:** Ollama analyzes the PKGBUILD for: - Suspicious `source=()` URLs (npm, tor, raw IPs) - Obfuscated bash (`eval`, `base64 -d`, hex encoding) - Reverse shells, `nc`, `/dev/tcp` - `post_install` hooks creating services/cronjobs - Typosquatting package names - Unusual `depends` for the package type -4. **Build:** If clean, `makepkg` builds the package -5. **Serve:** `repo-add` adds it to the local pacman repo -6. **Cache:** Approved packages stay cached until upstream update +5. **Build:** If clean, `makepkg` builds the package +6. **Serve:** `repo-add` adds it to the local pacman repo +7. **Cache:** Approved packages stay cached until upstream update + +## Threat Intelligence Sources + +Based on [AegisAUR](https://gitea.die-heimatlosen.eu/arch_agent/aegisaur) IOC fetcher: + +| Source | Type | Freshness | URL | +|--------|------|-----------|-----| +| HedgeDoc | Live paste | Always current | `md.archlinux.org/s/SxbqukK6IA` | +| Atomic Arch Gist | GitHub Gist | Versioned | `gist.githubusercontent.com/Kidev/...` | +| Arch Security | Official advisory | Slow but authoritative | `security.archlinux.org` | +| AUR RPC | API | Real-time | `aur.archlinux.org/rpc/v5` | ## API diff --git a/aur_shield/ioc_fetcher.py b/aur_shield/ioc_fetcher.py new file mode 100644 index 0000000..5b528da --- /dev/null +++ b/aur_shield/ioc_fetcher.py @@ -0,0 +1,275 @@ +"""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 \ No newline at end of file diff --git a/aur_shield/scanner.py b/aur_shield/scanner.py index ab35b27..5c85c8e 100644 --- a/aur_shield/scanner.py +++ b/aur_shield/scanner.py @@ -12,6 +12,10 @@ import httpx from .config import OllamaConfig, SecurityConfig from .aur_client import extract_sources, extract_install_hooks +from .ioc_fetcher import ( + fetch_all_iocs, check_package_against_iocs, + check_typosquatting, IOCEntry, IOCResult, Confidence, +) class ScanVerdict(str, Enum): @@ -30,6 +34,61 @@ class ScanResult: reasoning: str = "" sources_checked: list[str] = field(default_factory=list) raw_response: str = "" + ioc_matches: list[dict] = field(default_factory=list) + typosquat_matches: list[dict] = field(default_factory=list) + + +async def pre_check_iocs(package: str) -> tuple[list[str], list[dict], list[dict]]: + """Check package against public IOC lists before LLM scan. + + Returns (findings, ioc_matches, typosquat_matches). + If ioc_matches contains critical entries, the package is known malicious. + """ + findings = [] + ioc_dicts = [] + typo_dicts = [] + + try: + iocs = await fetch_all_iocs() + + # Exact match + result = check_package_against_iocs(package, iocs) + if result.matches: + for m in result.matches: + ioc_dicts.append({ + "package": m.package_name, + "threat": m.threat_type.value, + "source": m.source, + "confidence": m.confidence.value, + "description": m.description, + }) + if m.confidence == Confidence.CRITICAL: + findings.append( + f"KNOWN MALICIOUS: '{package}' found in {m.source} " + f"({m.threat_type.value}: {m.description})" + ) + else: + findings.append( + f"Suspicious: '{package}' flagged in {m.source} " + f"({m.threat_type.value})" + ) + + # Typosquatting + typo_matches = check_typosquatting(package, iocs) + for m in typo_matches: + typo_dicts.append({ + "package": m.package_name, + "threat": m.threat_type.value, + "source": m.source, + "confidence": m.confidence.value, + "description": m.description, + }) + findings.append(f"Typosquatting: {m.description}") + + except Exception as e: + findings.append(f"IOC check error (non-blocking): {e}") + + return findings, ioc_dicts, typo_dicts SYSTEM_PROMPT = """\ @@ -76,7 +135,27 @@ async def scan_pkgbuild( ollama_cfg: OllamaConfig, sec_cfg: SecurityConfig, ) -> ScanResult: - """Scan a PKGBUILD with regex pre-check + LLM analysis.""" + """Scan a PKGBUILD with IOC pre-check + regex pre-scan + LLM analysis.""" + + # 0. IOC pre-check — public threat intelligence + ioc_findings, ioc_matches, typo_matches = await pre_check_iocs(package_name) + + # If package is known malicious from public lists, skip LLM scan + has_critical_ioc = any( + m.get("confidence") == "Critical" for m in ioc_matches + ) + if has_critical_ioc: + return ScanResult( + package=package_name, + verdict=ScanVerdict.MALICIOUS, + confidence=1.0, + findings=ioc_findings, + reasoning="Package found in public IOC lists (known malicious). " + "LLM scan skipped — package is confirmed malicious by " + "Arch Linux security advisories.", + ioc_matches=ioc_matches, + typosquat_matches=typo_matches, + ) # 1. Fast regex pre-scan regex_findings = _pre_scan(pkgbuild, sec_cfg) @@ -124,8 +203,8 @@ async def scan_pkgbuild( raw = data.get("message", {}).get("content", "") result = _parse_llm_response(raw) - # Combine with regex findings - all_findings = result.get("findings", []) + regex_findings + # Combine with regex + IOC findings + all_findings = result.get("findings", []) + regex_findings + ioc_findings verdict = result.get("verdict", "suspicious") confidence = result.get("confidence", 0.5) @@ -134,6 +213,11 @@ async def scan_pkgbuild( verdict = "suspicious" confidence = max(confidence, 0.7) + # If IOC found non-critical matches, escalate + if ioc_matches and verdict == "clean": + verdict = "suspicious" + confidence = max(confidence, 0.8) + return ScanResult( package=package_name, verdict=ScanVerdict(verdict), @@ -142,6 +226,8 @@ async def scan_pkgbuild( reasoning=result.get("reasoning", ""), sources_checked=sources, raw_response=raw, + ioc_matches=ioc_matches, + typosquat_matches=typo_matches, ) except Exception as e: diff --git a/aur_shield/server.py b/aur_shield/server.py index 3e6ed3a..86e3452 100644 --- a/aur_shield/server.py +++ b/aur_shield/server.py @@ -100,6 +100,8 @@ async def scan_package(package: str) -> dict[str, Any]: "reasoning": result.reasoning, "sources": result.sources_checked, "version": source.package.version, + "ioc_matches": result.ioc_matches, + "typosquat_matches": result.typosquat_matches, }