arch_agent
f015092b1f
Server: - config: server.scan_only flag (default: false) - /api/build returns scan_only=true without building when enabled - /api/status reports scan_only mode Client (safe-yay): - Detects server scan_only from API response - Scan-only server: builds locally with yay/paru after clean scan - Full server: installs from pacman repo as before - --noinstall flag for scan-only without build (was --scan-only) - Suspicious: prompts for local build or repo install depending on server mode
255 lines
7.2 KiB
Python
255 lines
7.2 KiB
Python
"""FastAPI server for AUR-Shield."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from .config import Config
|
|
from .aur_client import aur_sources, aur_info
|
|
from .scanner import scan_pkgbuild, ScanVerdict
|
|
from .builder import build_package, add_to_repo, BuildResult
|
|
from .cache import ScanCache
|
|
|
|
app = FastAPI(
|
|
title="AUR-Shield",
|
|
description="AI-powered AUR firewall",
|
|
version="0.1.0",
|
|
)
|
|
|
|
# Global config — set by main()
|
|
_config: Config | None = None
|
|
_scan_cache: ScanCache | None = None
|
|
|
|
|
|
def get_config() -> Config:
|
|
if _config is None:
|
|
raise RuntimeError("Server not initialized")
|
|
return _config
|
|
|
|
|
|
def get_cache() -> ScanCache:
|
|
if _scan_cache is None:
|
|
raise RuntimeError("Cache not initialized")
|
|
return _scan_cache
|
|
|
|
|
|
def _hash_pkgbuild(pkgbuild: str) -> str:
|
|
return hashlib.sha256(pkgbuild.encode()).hexdigest()[:16]
|
|
|
|
|
|
@app.get("/api/status")
|
|
async def status() -> dict[str, Any]:
|
|
"""Server status and cache info."""
|
|
cfg = get_config()
|
|
cache = get_cache()
|
|
return {
|
|
"status": "running",
|
|
"model": cfg.ollama.model,
|
|
"ollama_url": cfg.ollama.url,
|
|
"repo_dir": cfg.server.repo_dir,
|
|
"cached_scans": len(cache.list_all()),
|
|
"scan_only": cfg.server.scan_only,
|
|
}
|
|
|
|
|
|
@app.get("/api/scan/{package}")
|
|
async def scan_package(package: str) -> dict[str, Any]:
|
|
"""Scan a package without building."""
|
|
cfg = get_config()
|
|
cache = get_cache()
|
|
|
|
# Fetch from AUR
|
|
source = await aur_sources(package)
|
|
if source is None:
|
|
raise HTTPException(404, f"Package '{package}' not found in AUR")
|
|
|
|
pkgbuild_hash = _hash_pkgbuild(source.pkgbuild)
|
|
|
|
# Check cache
|
|
cached = cache.get(package, pkgbuild_hash)
|
|
if cached:
|
|
return {
|
|
"package": package,
|
|
"cached": True,
|
|
"verdict": cached.verdict,
|
|
"confidence": cached.confidence,
|
|
"findings": cached.findings,
|
|
"reasoning": cached.reasoning,
|
|
"version": source.package.version,
|
|
}
|
|
|
|
# Scan
|
|
result = await scan_pkgbuild(
|
|
package, source.pkgbuild, source.srcinfo,
|
|
cfg.ollama, cfg.security,
|
|
)
|
|
|
|
# Cache the result
|
|
cache.put(result, pkgbuild_hash)
|
|
|
|
return {
|
|
"package": package,
|
|
"cached": False,
|
|
"verdict": result.verdict.value,
|
|
"confidence": result.confidence,
|
|
"findings": result.findings,
|
|
"reasoning": result.reasoning,
|
|
"sources": result.sources_checked,
|
|
"version": source.package.version,
|
|
"ioc_matches": result.ioc_matches,
|
|
"typosquat_matches": result.typosquat_matches,
|
|
}
|
|
|
|
|
|
@app.get("/api/build/{package}")
|
|
async def build_endpoint(package: str) -> dict[str, Any]:
|
|
"""Scan + build + add to repo."""
|
|
cfg = get_config()
|
|
cache = get_cache()
|
|
|
|
# 1. Fetch from AUR
|
|
source = await aur_sources(package)
|
|
if source is None:
|
|
raise HTTPException(404, f"Package '{package}' not found in AUR")
|
|
|
|
pkgbuild_hash = _hash_pkgbuild(source.pkgbuild)
|
|
|
|
# 2. Check cache for scan
|
|
cached = cache.get(package, pkgbuild_hash)
|
|
if cached:
|
|
if cached.verdict == ScanVerdict.MALICIOUS.value:
|
|
raise HTTPException(403, f"Package '{package}' is cached as MALICIOUS")
|
|
verdict = cached.verdict
|
|
findings = cached.findings
|
|
confidence = cached.confidence
|
|
else:
|
|
# 3. Scan
|
|
result = await scan_pkgbuild(
|
|
package, source.pkgbuild, source.srcinfo,
|
|
cfg.ollama, cfg.security,
|
|
)
|
|
cache.put(result, pkgbuild_hash)
|
|
verdict = result.verdict.value
|
|
findings = result.findings
|
|
confidence = result.confidence
|
|
|
|
# 4. Block if malicious
|
|
if verdict == ScanVerdict.MALICIOUS.value:
|
|
raise HTTPException(
|
|
403,
|
|
f"Package '{package}' was flagged as MALICIOUS. Findings: {findings}",
|
|
)
|
|
|
|
if verdict == ScanVerdict.SUSPICIOUS.value:
|
|
# Allow suspicious but warn
|
|
pass
|
|
|
|
# 5. Build (skip if scan_only mode)
|
|
if cfg.server.scan_only:
|
|
return {
|
|
"package": package,
|
|
"verdict": verdict,
|
|
"confidence": confidence,
|
|
"findings": findings,
|
|
"scan_only": True,
|
|
"build_success": False,
|
|
"message": "Scan-only mode — build locally on client",
|
|
}
|
|
|
|
# 6. Build
|
|
build_result = build_package(
|
|
source.pkgbuild, package,
|
|
cfg.server, cfg.build,
|
|
)
|
|
|
|
if not build_result.success:
|
|
return {
|
|
"package": package,
|
|
"verdict": verdict,
|
|
"confidence": confidence,
|
|
"build_success": False,
|
|
"build_error": build_result.error,
|
|
"build_log": build_result.log[-2000:],
|
|
}
|
|
|
|
# 6. Add to repo
|
|
try:
|
|
db_path = add_to_repo(build_result.pkg_files, cfg.server.repo_dir)
|
|
except Exception as e:
|
|
return {
|
|
"package": package,
|
|
"verdict": verdict,
|
|
"confidence": confidence,
|
|
"build_success": True,
|
|
"repo_success": False,
|
|
"repo_error": str(e),
|
|
}
|
|
|
|
return {
|
|
"package": package,
|
|
"verdict": verdict,
|
|
"confidence": confidence,
|
|
"findings": findings,
|
|
"build_success": True,
|
|
"repo_success": True,
|
|
"pkg_files": [Path(f).name for f in build_result.pkg_files],
|
|
"repo_db": Path(db_path).name,
|
|
}
|
|
|
|
|
|
@app.get("/api/report/{package}")
|
|
async def get_report(package: str) -> dict[str, Any]:
|
|
"""Get the last scan report for a package."""
|
|
cache = get_cache()
|
|
entries = cache.list_all()
|
|
for entry in entries:
|
|
if entry["package"] == package:
|
|
return entry
|
|
raise HTTPException(404, f"No scan report for '{package}'")
|
|
|
|
|
|
@app.get("/api/cache")
|
|
async def list_cache() -> dict[str, Any]:
|
|
"""List all cached scan results."""
|
|
cache = get_cache()
|
|
return {"entries": cache.list_all()}
|
|
|
|
|
|
@app.delete("/api/cache/{package}")
|
|
async def clear_cache_entry(package: str) -> dict[str, str]:
|
|
"""Clear a cached scan result."""
|
|
cache = get_cache()
|
|
path = cache._cache_path(package)
|
|
if path.exists():
|
|
path.unlink()
|
|
return {"status": "deleted", "package": package}
|
|
raise HTTPException(404, f"No cache for '{package}'")
|
|
|
|
|
|
# Serve the pacman repo files
|
|
@app.get("/repo/{filename}")
|
|
async def serve_repo_file(filename: str):
|
|
"""Serve repo files for pacman."""
|
|
cfg = get_config()
|
|
repo_path = Path(cfg.server.repo_dir) / filename
|
|
if not repo_path.exists():
|
|
raise HTTPException(404, f"File '{filename}' not in repo")
|
|
return FileResponse(repo_path)
|
|
|
|
|
|
def init_server(config: Config) -> None:
|
|
"""Initialize the server with config."""
|
|
global _config, _scan_cache
|
|
_config = config
|
|
_scan_cache = ScanCache(
|
|
cache_dir=str(Path(config.server.work_dir) / "cache"),
|
|
ttl_hours=config.cache.ttl_hours,
|
|
)
|
|
# Ensure dirs exist
|
|
Path(config.server.repo_dir).mkdir(parents=True, exist_ok=True)
|
|
Path(config.server.work_dir).mkdir(parents=True, exist_ok=True) |