home-network-k3s-impl/platform/vuln-scanner/src/trivy.py

"""Run Trivy and normalise its output. Everything Trivy emits is untrusted upstream text."""
import json
import logging
import re
import subprocess
from urllib.parse import urlsplit

from state import parse_ts

log = logging.getLogger("vuln-scanner.trivy")

SEVERITIES = ("CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN")
_CREDENTIAL = re.compile(
    r"unauthorized|authentication required|\bdenied:|access to the resource is denied|forbidden"
    r"|status(?: code)?:? ?(?:401|403)\b|\b40[13] (?:unauthorized|forbidden)",
    re.IGNORECASE,
)
# Lone surrogates too: json.loads accepts them, and they cannot be encoded as UTF-8 for the page.
_CONTROL = re.compile(r"[\x00-\x1f\x7f\ud800-\udfff]")
# Exact and lowercase, matching report.schema.json, whose pattern is case-sensitive: an `HTTPS://x`
# accepted here would make the schema refuse the entire report.
_URL_SCHEME = re.compile(r"https?://")


class ScanFailure(Exception):
    """A scan did not produce a result. kind is `failed` or `credential-required`."""

    def __init__(self, kind, reason):
        super().__init__(reason)
        self.kind = kind
        self.reason = reason


def text(value, limit=512):
    """A bounded, control-character-free string; non-strings become empty."""
    if not isinstance(value, str):
        return ""
    return _CONTROL.sub(" ", value).strip()[:limit]


def safe_url(value):
    """value if it is a plain http(s) URL, else the empty string."""
    url = text(value, 2048)
    if not url or url != value or re.search(r"\s", url) or not _URL_SCHEME.match(url):
        return ""
    try:
        parts = urlsplit(url)
    except ValueError:
        return ""
    return url if parts.netloc else ""


def registry_of(reference):
    """The lowercase registry host (and port) of a normalised registry/repository reference."""
    return reference.split("/", 1)[0].lower()


def _cvss(vuln):
    best = None
    for vendor in (vuln.get("CVSS") or {}).values() if isinstance(vuln.get("CVSS"), dict) else ():
        if not isinstance(vendor, dict):
            continue
        for key in ("V40Score", "V3Score", "V2Score"):
            score = vendor.get(key)
            if isinstance(score, (int, float)) and not isinstance(score, bool) and 0 <= score <= 10:
                best = max(best, float(score)) if best is not None else float(score)
                break
    return best


def parse_results(document):
    """Findings (id, severity, cvss, package, installedVersion, fixedVersion, title, url), deduped."""
    if not isinstance(document, dict) or "SchemaVersion" not in document:
        raise ScanFailure("failed", "trivy produced an unrecognised report")
    results = document.get("Results")
    if results is None:
        return []
    if not isinstance(results, list):
        raise ScanFailure("failed", "trivy report has a malformed Results field")
    seen = {}
    for result in results:
        vulnerabilities = result.get("Vulnerabilities") if isinstance(result, dict) else None
        if vulnerabilities is None:
            continue
        if not isinstance(vulnerabilities, list):
            raise ScanFailure("failed", "trivy report has a malformed Vulnerabilities field")
        for vuln in vulnerabilities:
            if not isinstance(vuln, dict) or not text(vuln.get("VulnerabilityID")):
                raise ScanFailure("failed", "trivy report has a finding with no identifier")
            severity = text(vuln.get("Severity")).upper()
            fixed = text(vuln.get("FixedVersion"))
            finding = {
                "id": text(vuln.get("VulnerabilityID"), 256),
                "severity": severity if severity in SEVERITIES else "UNKNOWN",
                "cvss": _cvss(vuln),
                "package": text(vuln.get("PkgName"), 256),
                "installedVersion": text(vuln.get("InstalledVersion"), 256),
                "fixedVersion": fixed or None,
                "title": text(vuln.get("Title") or vuln.get("Description"), 512),
                "url": safe_url(vuln.get("PrimaryURL")),
            }
            seen[(finding["id"], finding["package"], finding["installedVersion"])] = finding
    return sorted(seen.values(), key=lambda f: (f["id"], f["package"], f["installedVersion"]))


def classify_failure(registry, stderr):
    """Sort a non-zero Trivy exit into credential-required or failed. Reasons name the registry."""
    if _CREDENTIAL.search(stderr or ""):
        return ScanFailure("credential-required", "registry " + text(registry, 253) + " refused anonymous access")
    lines = [line.strip() for line in (stderr or "").splitlines() if line.strip()]
    return ScanFailure("failed", text(lines[-1], 400) if lines else "trivy exited with an error and no message")


class Runner:
    """The real Trivy binary. Tests inject a runner with the same four methods."""

    def __init__(self, trivy_bin, cache_dir, timeout=600):
        self.bin = trivy_bin
        self.cache_dir = cache_dir
        self.timeout = timeout

    def _run(self, args, timeout):
        return subprocess.run([self.bin] + args, capture_output=True, text=True, timeout=timeout, check=False)

    def download_db(self):
        result = self._run(["image", "--download-db-only", "--no-progress", "--cache-dir", self.cache_dir], 900)
        if result.returncode != 0:
            raise ScanFailure("failed", "database update failed: " + text(result.stderr, 300))

    def db_info(self):
        """{updatedAt, version} of the DB on disk, or None when no DB is present."""
        result = self._run(["version", "--format", "json", "--cache-dir", self.cache_dir], 60)
        if result.returncode != 0:
            return None
        try:
            db = json.loads(result.stdout).get("VulnerabilityDB")
        except (ValueError, AttributeError):
            return None
        if not isinstance(db, dict) or parse_ts(db.get("UpdatedAt")) is None:
            return None  # an unparseable timestamp is no usable database, and must not reach the report
        return {"updatedAt": db["UpdatedAt"], "version": text(str(db.get("Version", "unknown")), 128) or "unknown"}

    def scan(self, reference):
        args = [
            "image", "--skip-db-update", "--image-src", "remote", "--scanners", "vuln", "--format", "json",
            "--no-progress", "--cache-dir", self.cache_dir, "--timeout", str(self.timeout) + "s", reference,
        ]
        try:
            result = self._run(args, self.timeout + 30)
        except subprocess.TimeoutExpired:
            raise ScanFailure("failed", "timed out after " + str(self.timeout) + "s") from None
        except OSError as exc:
            raise ScanFailure("failed", "could not start trivy: " + type(exc).__name__) from None
        if result.returncode < 0:
            raise ScanFailure("failed", "trivy killed by signal " + str(-result.returncode) + " (out of memory?)")
        if result.returncode != 0:
            raise classify_failure(registry_of(reference), result.stderr)
        try:
            return json.loads(result.stdout)
        except ValueError:
            raise ScanFailure("failed", "trivy output was not valid JSON") from None

    def clean_cache(self):
        self._run(["clean", "--scan-cache", "--cache-dir", self.cache_dir], 120)