pxe-server/scripts/check-updates.py

#!/usr/bin/env python3
"""check-updates.py - is the offline bundle current? Run before building a USB.

  * offline-packages/*.deb against Ubuntu 24.04 (noble, noble-updates,
    noble-security; main, restricted, universe, multiverse): which have a
    newer version, and which
    of those are security updates.
  * webapp/requirements.txt pins against PyPI (newer release?) and osv.dev
    (known vulnerabilities in the pinned version).

Exit 0 = nothing outstanding; 1 = security updates or vulnerable pins found
(the details are printed); 2 = could not check (no network, bad or missing
input such as a folder with no .debs, an unreadable index or API answer).
Needs internet (Ubuntu archive, pypi.org, api.osv.dev); read-only.

Usage: scripts/check-updates.py [--debs DIR] [--reqs FILE] [--all]
"""
import argparse
import functools
import glob
import gzip
import json
import os
import re
import subprocess
import sys
import urllib.request

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ARCHIVE = "http://archive.ubuntu.com/ubuntu/dists"
SUITES = ("noble", "noble-updates", "noble-security")
COMPONENTS = ("main", "restricted", "universe", "multiverse")


class CheckError(Exception):
    """Could not check (exit 2)."""


def fetch(url, timeout=60):
    with urllib.request.urlopen(url, timeout=timeout) as r:
        return r.read()


try:
    import apt_pkg  # python3-apt: in-process comparison, no fork per pair
    apt_pkg.init_system()
except Exception:  # not installed: fall back to dpkg, cached
    apt_pkg = None


@functools.lru_cache(maxsize=None)
def newer(a, b):
    """True if Debian version a > b."""
    if a == b:
        return False
    if apt_pkg is not None:
        return apt_pkg.version_compare(a, b) > 0
    return subprocess.run(["dpkg", "--compare-versions", a, "gt", b]).returncode == 0


def ubuntu_index(wanted):
    """({package: newest version}, {package: newest security version}),
    for the package names in wanted only."""
    best, sec = {}, {}
    for suite in SUITES:
        for comp in COMPONENTS:
            try:
                text = gzip.decompress(fetch(f"{ARCHIVE}/{suite}/{comp}/binary-amd64/Packages.gz")).decode("utf-8", "replace")
            except (OSError, EOFError) as ex:
                raise CheckError(f"{suite}/{comp} index: {ex}")
            pkg = None
            for line in text.splitlines():
                if line.startswith("Package: "):
                    pkg = line[9:].strip()
                    if pkg not in wanted:
                        pkg = None
                elif line.startswith("Version: ") and pkg:
                    v = line[9:].strip()
                    if pkg not in best or newer(v, best[pkg]):
                        best[pkg] = v
                    if suite.endswith("security") and (pkg not in sec or newer(v, sec[pkg])):
                        sec[pkg] = v
    return best, sec


def check_debs(deb_dir, show_all):
    debs = glob.glob(os.path.join(deb_dir, "*.deb"))
    if not debs:
        raise CheckError(f"no .deb files in {deb_dir}: nothing to check (a failed download?)")
    names = {os.path.basename(d).split("_")[0] for d in debs}
    best, sec = ubuntu_index(names)
    stale, security = [], []
    for d in debs:
        parts = os.path.basename(d)[:-4].split("_")
        name, ver = parts[0], parts[1].replace("%3a", ":")
        if name in best and newer(best[name], ver):
            is_sec = name in sec and newer(sec[name], ver)
            (security if is_sec else stale).append((name, ver, best[name]))
    print(f"Ubuntu packages: {len(debs)} in {deb_dir}; {len(security) + len(stale)} have newer versions, "
          f"{len(security)} of them security updates")
    for n, v, nv in sorted(security):
        print(f"  SECURITY {n:34} {v:30} -> {nv}")
    if show_all:
        for n, v, nv in sorted(stale):
            print(f"  update   {n:34} {v:30} -> {nv}")
    return len(security)


def check_pins(reqs):
    try:
        text = open(reqs).read()
    except OSError as ex:
        raise CheckError(f"{reqs}: {ex}")
    pins = re.findall(r"^([A-Za-z0-9_.-]+)==([^\s\\]+)", text, re.M)
    if not pins:
        print(f"no pinned packages in {reqs}")
        return 0
    q = {"queries": [{"package": {"name": n, "ecosystem": "PyPI"}, "version": v} for n, v in pins]}
    req = urllib.request.Request("https://api.osv.dev/v1/querybatch", data=json.dumps(q).encode(),
                                 headers={"Content-Type": "application/json"})
    try:
        results = json.load(urllib.request.urlopen(req, timeout=60))["results"]
        if not isinstance(results, list) or len(results) != len(pins):
            raise ValueError("osv.dev returned %r results for %d pins" % (type(results).__name__, len(pins)))
    except (ValueError, KeyError, TypeError) as ex:
        raise CheckError(f"osv.dev answer: {ex}")
    bad = 0
    print(f"Python pins in {os.path.relpath(reqs, ROOT)}:")
    for (n, v), r in zip(pins, results):
        try:
            latest = json.loads(fetch(f"https://pypi.org/pypi/{n}/json", 30))["info"]["version"]
        except Exception:  # report, keep going
            latest = "?"
        vulns = [x.get("id", "?") for x in (r or {}).get("vulns", [])]
        bad += bool(vulns)
        flag = "VULNERABLE" if vulns else ("newer     " if latest not in (v, "?") else "ok        ")
        print(f"  {flag} {n:14} {v:10} latest {latest:10} {', '.join(vulns)}")
    return bad


def main():
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--debs", default=os.path.join(ROOT, "offline-packages"))
    ap.add_argument("--reqs", default=os.path.join(ROOT, "webapp", "requirements.txt"))
    ap.add_argument("--all", action="store_true", help="also list non-security updates")
    a = ap.parse_args()
    try:
        problems = check_debs(a.debs, a.all) + check_pins(a.reqs)
    except (OSError, CheckError) as ex:
        print(f"could not check: {ex}", file=sys.stderr)
        return 2
    print("PASSED: nothing outstanding" if not problems else f"ATTENTION: {problems} item(s) need updating")
    return 1 if problems else 0


if __name__ == "__main__":
    sys.exit(main())