feat: extended IOC sources + archcanary features
Extended IOC sources (from archcanary): - aur-audit.wtako.net black/red API (3rd-party continuous AUR scanner) - Community reports list (community-curated malicious packages) - CHAOS RAT campaign list (backdoor payload) - Russian spam campaign list (.bashrc injection) Client features (archcanary-inspired): - Exit codes: 0=clean, 1=warning, 2=malicious (scriptable) - --doctor health check (server, ollama, repo status) - --scan-only mode (scan without building) - IOC match display in malicious blocks - Suspicious packages: interactive install prompt All IOC fetches run concurrently for speed.
This commit is contained in:
arch_agent
@@ -0,0 +1,195 @@
|
||||
"""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
|
||||
@@ -1,6 +1,7 @@
|
||||
"""LLM scanner — sends PKGBUILD to Ollama for security analysis."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
@@ -16,6 +17,7 @@ from .ioc_fetcher import (
|
||||
fetch_all_iocs, check_package_against_iocs,
|
||||
check_typosquatting, IOCEntry, IOCResult, Confidence,
|
||||
)
|
||||
from .extended_iocs import fetch_extended_iocs
|
||||
|
||||
|
||||
class ScanVerdict(str, Enum):
|
||||
@@ -49,7 +51,12 @@ async def pre_check_iocs(package: str) -> tuple[list[str], list[dict], list[dict
|
||||
typo_dicts = []
|
||||
|
||||
try:
|
||||
iocs = await fetch_all_iocs()
|
||||
# Fetch from both base + extended IOC sources concurrently
|
||||
base_iocs, ext_iocs = await asyncio.gather(
|
||||
fetch_all_iocs(),
|
||||
fetch_extended_iocs(),
|
||||
)
|
||||
iocs = base_iocs + ext_iocs
|
||||
|
||||
# Exact match
|
||||
result = check_package_against_iocs(package, iocs)
|
||||
|
||||
Reference in New Issue
Block a user