arch_agent
f6b5ec4031
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.
272 lines
8.7 KiB
Python
272 lines
8.7 KiB
Python
"""LLM scanner — sends PKGBUILD to Ollama for security analysis."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
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,
|
|
)
|
|
from .extended_iocs import fetch_extended_iocs
|
|
|
|
|
|
class ScanVerdict(str, Enum):
|
|
CLEAN = "clean"
|
|
SUSPICIOUS = "suspicious"
|
|
MALICIOUS = "malicious"
|
|
ERROR = "error"
|
|
|
|
|
|
@dataclass
|
|
class ScanResult:
|
|
package: str
|
|
verdict: ScanVerdict
|
|
confidence: float # 0.0 - 1.0
|
|
findings: list[str] = field(default_factory=list)
|
|
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:
|
|
# 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)
|
|
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 = """\
|
|
You are a security scanner for Arch Linux AUR PKGBUILD files.
|
|
Your job is to detect malicious or suspicious code in PKGBUILDs.
|
|
|
|
Check for these threats:
|
|
1. DOWNLOAD-AND-EXEC: curl|bash, wget|sh, downloading and executing scripts
|
|
2. OBFUSCATION: eval $(base64 -d), hex-encoded commands, printf-encoded payloads
|
|
3. REVERSE SHELLS: nc -, /dev/tcp, bash -i >&, socat
|
|
4. SUSPICIOUS SOURCES: npm, tor, raw IP addresses, non-HTTPS with no fallback
|
|
5. PERSISTENCE: post_install creating systemd services, cronjobs, autostart entries
|
|
6. TYPOSQUATTING: package names mimicking popular packages with slight misspellings
|
|
7. DATA EXFIL: curl/wget sending data to external servers with system info
|
|
8. UNUSUAL DEPENDS: packages requesting network tools unrelated to their purpose
|
|
|
|
Respond in JSON ONLY:
|
|
{
|
|
"verdict": "clean" | "suspicious" | "malicious",
|
|
"confidence": 0.0-1.0,
|
|
"findings": ["list of specific issues found"],
|
|
"reasoning": "brief explanation"
|
|
}
|
|
|
|
Be conservative: if unsure, mark as suspicious.
|
|
A clean PKGBUILD that just downloads a tarball from the official upstream
|
|
URL and runs make/cmake is CLEAN.
|
|
"""
|
|
|
|
|
|
def _pre_scan(pkgbuild: str, sec_cfg: SecurityConfig) -> list[str]:
|
|
"""Fast regex pre-scan before LLM. Returns list of findings."""
|
|
findings = []
|
|
for pattern in sec_cfg.block_patterns:
|
|
if re.search(pattern, pkgbuild, re.IGNORECASE):
|
|
findings.append(f"Pattern match: {pattern}")
|
|
return findings
|
|
|
|
|
|
async def scan_pkgbuild(
|
|
package_name: str,
|
|
pkgbuild: str,
|
|
srcinfo: str,
|
|
ollama_cfg: OllamaConfig,
|
|
sec_cfg: SecurityConfig,
|
|
) -> ScanResult:
|
|
"""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)
|
|
|
|
# 2. Extract metadata for context
|
|
sources = extract_sources(pkgbuild)
|
|
hooks = extract_install_hooks(pkgbuild)
|
|
|
|
# 3. Build the prompt
|
|
user_msg = (
|
|
f"Package: {package_name}\n\n"
|
|
f"=== PKGBUILD ===\n{pkgbuild}\n\n"
|
|
)
|
|
if srcinfo:
|
|
user_msg += f"=== .SRCINFO (excerpt) ===\n{srcinfo[:2000]}\n\n"
|
|
if hooks:
|
|
user_msg += f"=== INSTALL HOOKS ===\n{chr(10).join(hooks)}\n\n"
|
|
if sources:
|
|
user_msg += f"=== SOURCE URLS ===\n{chr(10).join(sources)}\n\n"
|
|
|
|
# 4. Call Ollama
|
|
payload = {
|
|
"model": ollama_cfg.model,
|
|
"messages": [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": user_msg},
|
|
],
|
|
"stream": False,
|
|
"format": "json",
|
|
"options": {
|
|
"temperature": 0.1,
|
|
"num_ctx": 8192,
|
|
},
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=ollama_cfg.timeout) as client:
|
|
resp = await client.post(
|
|
f"{ollama_cfg.url}/api/chat",
|
|
json=payload,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
raw = data.get("message", {}).get("content", "")
|
|
result = _parse_llm_response(raw)
|
|
|
|
# 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)
|
|
|
|
# If regex found blocked patterns, escalate
|
|
if regex_findings and verdict == "clean":
|
|
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),
|
|
confidence=confidence,
|
|
findings=all_findings,
|
|
reasoning=result.get("reasoning", ""),
|
|
sources_checked=sources,
|
|
raw_response=raw,
|
|
ioc_matches=ioc_matches,
|
|
typosquat_matches=typo_matches,
|
|
)
|
|
|
|
except Exception as e:
|
|
return ScanResult(
|
|
package=package_name,
|
|
verdict=ScanVerdict.ERROR,
|
|
confidence=0.0,
|
|
findings=[f"Scanner error: {e}"],
|
|
sources_checked=sources,
|
|
)
|
|
|
|
|
|
def _parse_llm_response(raw: str) -> dict[str, Any]:
|
|
"""Parse the LLM JSON response, handling malformed output."""
|
|
# Try direct JSON parse
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Try to extract JSON from text
|
|
match = re.search(r'\{[^{}]*"verdict"[^{}]*\}', raw, re.DOTALL)
|
|
if match:
|
|
try:
|
|
return json.loads(match.group())
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Fallback
|
|
return {
|
|
"verdict": "suspicious",
|
|
"confidence": 0.3,
|
|
"findings": ["Could not parse LLM response"],
|
|
"reasoning": raw[:500],
|
|
} |