#!/usr/bin/env python3 """Am I affected? Checks this machine for the issues listed in advisories.json (Oct 2026: PHP, Node.js, BIND, LiteLLM, GitLab, eMudhra; on Linux servers also the kernel, nginx, Apache, MariaDB, MongoDB, WordPress and more). Stdlib only. Exit code 1 if anything is affected. Usage: python3 secinspect.py [dirs ...] [--docker] [--html report.html] [--no-update-check] With no dirs, searches your home folder and system Python paths for envs, lockfiles and CA bundles. With dirs, searches only those. Installed programs and system trust stores are always checked. """ import argparse import base64 import datetime import glob import hashlib import html import json import os import platform import re import shutil import subprocess import sys import tarfile import tempfile import textwrap import urllib.request HOME = os.path.expanduser("~") ADVISORIES = json.load(open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "advisories.json"))) PACKAGES = ADVISORIES["packages"] ROOTS = {r["name"]: base64.b64decode(r["der"]) for r in ADVISORIES["ca"]["roots"]} INTERMEDIATES = {i["name"]: base64.b64decode(i["der"]) for i in ADVISORIES["ca"]["intermediates"]} AAA = "AAA Certificate Services" OPENSSL = shutil.which("openssl") def ver(s): m = re.search(r"(\d+)\.(\d+)\.(\d+)", s or "") return tuple(map(int, m.groups())) if m else None def fmt(v): return ".".join(map(str, v)) def run(*cmd): try: p = subprocess.run(cmd, capture_output=True, text=True, timeout=30) return p.stdout + p.stderr except (OSError, subprocess.TimeoutExpired): return "" def binaries(name, *globs): found = [shutil.which(name)] + [p for g in globs for p in glob.glob(os.path.expanduser(g))] return sorted({os.path.realpath(p) for p in found if p}) def by_branch(v, fixed, width): """fixed: first fixed version per supported branch; width: how many parts make a branch. A branch older than the newest fixed one but missing from the list is end-of-life, which the PHP, Node and GitLab advisories all count as affected with no fix.""" for f in fixed: if f[:width] == v[:width]: return ("AFFECTED", f"upgrade to {fmt(f)}+") if v < f else ("OK", "") if v < max(fixed): return "AFFECTED", f"end-of-life branch, no fix coming: move to {fmt(max(fixed))}+ or another patched branch" return "UNKNOWN", "newer than the advisory, check for a later one" def version_status(adv, v): if "affected" in adv: # explicit inclusive ranges, e.g. BIND return next((("AFFECTED", fix) for lo, hi, fix in adv["affected"] if ver(lo) <= v <= ver(hi)), ("OK", "")) status, note = by_branch(v, [ver(f) for f in adv["fixed"]], adv["branch_width"]) if status == "AFFECTED" and adv.get("after_fix"): note += ", " + adv["after_fix"] return status, note def check_version(adv): name = adv["name"] if "version_files" in adv: # apps in web roots (WordPress, Nextcloud apps): one row per install found = [] for p in sorted({p for g in adv["version_files"] for p in glob.glob(os.path.expanduser(g))}): try: m = re.search(adv["version_regex"], open(p, errors="replace").read(MAX_READ)) except OSError: m = None found.append((os.path.dirname(os.path.dirname(p)), nums(m[1]) if m else None)) elif "version_file" in adv: found = [(adv["version_file"], ver(open(adv["version_file"]).read()))] if os.path.exists(adv["version_file"]) else [] else: found = [(b, ver(run(b, adv["flag"]))) for b in binaries(adv["binary"], *adv.get("globs", []))] if not found: return [(name, "-", "-", "NOT FOUND", adv.get("not_found", ""))] return [(name, p, fmt(v) if v else "?", *(version_status(adv, v) if v else ("UNKNOWN", "couldn't read version"))) for p, v in found] # --- Linux server packages (apt / dnf) ---------------------------------------------------------- # Distros backport fixes without changing the upstream version (a patched Ubuntu nginx still says # 1.24.0), so these compare the installed package against the distro's own fixed build, using # dpkg/rpm themselves to order versions. Builds from vendor repos (nginx.org, PGDG, MongoDB, Docker) # carry a different upstream version and fall back to the upstream fixed versions. DISTRO = ADVISORIES.get("distro", []) def os_release(text): """'ubuntu:24.04', 'debian:12', 'el:9' (RHEL, Alma, Rocky, CentOS Stream, Oracle) or None.""" d = dict(re.findall(r'^(\w+)="?([^"\n]*)"?$', text, re.M)) os_id, like, v = d.get("ID", ""), d.get("ID_LIKE", "").split(), d.get("VERSION_ID", "") if os_id == "ubuntu": return f"ubuntu:{v}" if os_id == "debian" and v: return f"debian:{v.split('.')[0]}" if os_id != "fedora" and {"rhel", "centos"} & {os_id, *like} and v: return f"el:{v.split('.')[0]}" return None def upstream(v): """'1:9.6p1-3ubuntu13.5' -> '9.6p1': drop the epoch and the distro revision.""" v = re.sub(r"^\d+:", "", v) return v.rsplit("-", 1)[0] if "-" in v else v def nums(v): return tuple(int(x) for x in re.findall(r"\d+", v)) def deb_lt(a, b): return subprocess.run(["dpkg", "--compare-versions", a, "lt", b]).returncode == 0 def rpm_lt(a, b): """EVR comparison with rpm's own rpmvercmp, applied to epoch, version and release in turn.""" def evr(s): e, _, vr = s.rpartition(":") if ":" in s else ("0", "", s) v, _, r = vr.partition("-") return [e or "0", v, r] for x, y in zip(evr(a), evr(b)): c = run("rpm", "--eval", f'%{{lua:print(rpm.vercmp("{x}", "{y}"))}}').strip() if c != "0": return c == "-1" return False def installed(fam, names): """[(package, version, source package)] for the installed ones.""" if not names: return [] # dpkg-query/rpm with no names list every package if fam == "deb": out = run("dpkg-query", "-W", "-f", "${db:Status-Abbrev}\t${Package}\t${Version}\t${source:Package}\n", *names) rows = [l.split("\t") for l in out.splitlines()] return [tuple(r[1:]) for r in rows if len(r) == 4 and r[0].startswith("ii")] out = run("rpm", "-q", "--qf", "%{NAME}\t%|EPOCH?{%{EPOCH}:}:{}|%{VERSION}-%{RELEASE}\t%{NAME}\n", *names) return [tuple(l.split("\t")) for l in out.splitlines() if l.count("\t") == 2] def running_kernel(fam): """(package, version, source) of the kernel actually booted: a fixed one only counts after a reboot.""" rel = platform.release() if fam == "deb": owner = run("dpkg-query", "-S", f"/boot/vmlinuz-{rel}").split(": ")[0].split(":")[0].strip() found = installed("deb", [owner]) if owner and " " not in owner else [] # linux-signed-hwe-6.8 / linux-signed-amd64 ship the image; the advisory names linux-hwe-6.8 / linux return [(p, v, re.sub(r"-signed(-(amd64|arm64|i386))?", "", s)) for p, v, s in found] out = run("rpm", "-qf", "--qf", "%{NAME}\t%|EPOCH?{%{EPOCH}:}:{}|%{VERSION}-%{RELEASE}\t%{NAME}\n", f"/lib/modules/{rel}/vmlinuz") return [tuple(l.split("\t")) for l in out.splitlines() if l.count("\t") == 2] def distro_status(adv, release, version, source, lt): """Status for one installed package. adv["fixed"][release] is the first fixed build, a {source package: build} map (kernel flavours), "not-affected" or "unfixed".""" fixed = adv.get("fixed", {}).get(release) if isinstance(fixed, dict): fixed = fixed.get(source) or fixed.get("*") ups = [nums(f) for f in adv.get("upstream_fixed", [])] after = (", " + adv["after_fix"]) if adv.get("after_fix") else "" if fixed == "not-affected": return "OK", f"{release} isn't affected" if fixed and fixed != "unfixed" and (not ups or upstream(version) == upstream(fixed)): return ("AFFECTED", f"update to {fixed}+{after}") if lt(version, fixed) else ("OK", "") unfixed = ("AFFECTED", f"no fixed {release} package yet: {adv.get('mitigation', 'watch for the update')}") if ups: # vendor build, or the distro moved to a new upstream release status, note = by_branch(nums(upstream(version)), ups, adv.get("branch_width", 2)) if status == "AFFECTED" and fixed == "unfixed": return unfixed return status, (note + after if status == "AFFECTED" else note) if fixed == "unfixed": return unfixed return "UNKNOWN", f"no fixed version known for {release or 'this distro'}" def check_distro(os_text=None): if os_text is None: try: os_text = open("/etc/os-release").read() except OSError: return [] # not Linux: nothing to say about server packages fam = "deb" if shutil.which("dpkg-query") else "rpm" if shutil.which("rpm") else None release = os_release(os_text) if not fam: return [("Server packages", "-", "-", "UNKNOWN", "only apt (Ubuntu/Debian) and dnf (RHEL family) systems are checked")] lt = deb_lt if fam == "deb" else rpm_lt rows = [] for adv in DISTRO: if not adv.get("kernel") and not adv.get(f"{fam}_packages"): continue # e.g. mod_http2 is its own package only on RHEL; apache2 covers it on Debian found = running_kernel(fam) if adv.get("kernel") else installed(fam, adv.get(f"{fam}_packages", [])) if not found: rows.append((adv["name"], "-", "-", "NOT FOUND", "")) continue # one row per advisory: nginx + nginx-core or mariadb-server + -core are one thing to fix checked = [(pkg, version, *distro_status(adv, release, version, source, lt)) for pkg, version, source in found] worst = min(checked, key=lambda c: STATUSES.index(c[2])) pkgs = " ".join(sorted({c[0] for c in checked})) rows.append((adv["name"], f"{pkgs} ({'apt' if fam == 'deb' else 'dnf'})", worst[1], worst[2], worst[3])) return rows # --- CA bundles --------------------------------------------------------------------------------- PEM = re.compile(rb"-----BEGIN CERTIFICATE-----(.+?)-----END CERTIFICATE-----", re.S) BUNDLE_FILES = {"cacert.pem", "cert.pem", "ca-bundle.crt", "ca-certificates.crt", "tls-ca-bundle.pem", "cacerts"} # Stores outside the walked folders. Linux system trust lands in the extracted bundles, after blocklists apply. STORES = [ # OpenSSL, curl and whatever links them "/etc/ssl/cert.pem", "/etc/ssl/certs/ca-certificates.crt", "/etc/pki/tls/certs/ca-bundle.crt", "/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem", "/opt/homebrew/etc/ca-certificates/cert.pem", "/opt/homebrew/etc/openssl*/cert.pem", "/usr/local/etc/openssl*/cert.pem", "/usr/local/etc/ca-certificates/cert.pem", # Java "/Library/Java/JavaVirtualMachines/*/Contents/Home/lib/security/cacerts", "/Library/Java/JavaVirtualMachines/*/Contents/Home/jre/lib/security/cacerts", "/usr/lib/jvm/*/lib/security/cacerts", "/etc/ssl/certs/java/cacerts", "/etc/pki/ca-trust/extracted/java/cacerts", "C:/Program Files/Java/*/lib/security/cacerts", # NSS built-in roots (Firefox, Thunderbird) "/Applications/Firefox*.app/Contents/MacOS/libnssckbi.dylib", "/Applications/Thunderbird.app/Contents/MacOS/libnssckbi.dylib", "/usr/lib/firefox*/libnssckbi.so", "/usr/lib/thunderbird/libnssckbi.so", "C:/Program Files/Mozilla Firefox/nssckbi.dll", ] def pem(der): return ("-----BEGIN CERTIFICATE-----\n" + "\n".join(textwrap.wrap(base64.b64encode(der).decode(), 64)) + "\n-----END CERTIFICATE-----\n").encode() def roots_in(data): """Listed roots in a PEM bundle or a binary store (JKS/PKCS12 cacerts, libnssckbi), matched by certificate bytes.""" if b"-----BEGIN CERTIFICATE-----" in data: blocks = {b"".join(m.split()) for m in PEM.findall(data)} return [n for n, der in ROOTS.items() if base64.b64encode(der) in blocks] return [n for n, der in ROOTS.items() if der in data] def chain_accepts(bundle): """Cracked intermediates `openssl verify` accepts against a PEM bundle, or None if it can't run. OpenSSL doesn't check revocation by default, so a bundle trusting the root accepts them.""" if not OPENSSL or b"-----BEGIN CERTIFICATE-----" not in bundle: return None with tempfile.TemporaryDirectory() as d: open(f"{d}/bundle.pem", "wb").write(bundle) accepted = [] for name, der in INTERMEDIATES.items(): open(f"{d}/i.pem", "wb").write(pem(der)) if run(OPENSSL, "verify", "-CAfile", f"{d}/bundle.pem", f"{d}/i.pem").strip().endswith(": OK"): accepted.append(name) return accepted def fix_hint(path): p = path.lower() if "/certifi/" in p: # not "ca-certificates" return "pip install -U certifi once a release drops it" if "site-packages" in p: # e.g. botocore ships its own bundle return f"upgrade the {p.split('site-packages/')[1].split('/')[0]} Python package once it drops the root" if "cacerts" in p: return "update the JDK, or keytool -delete the alias" return "update ca-certificates once it drops the root, or blocklist it (Red Hat KB 7149312)" def store_status(path, names, accepted): """Status for one trust store. accepted: chain_accepts() result (None = couldn't test).""" if not names: return "OK", "" hint = fix_hint(path) if accepted: return "AFFECTED", f"openssl verify accepts the cracked {', '.join(accepted)}: {hint}" if "nss" in path.lower(): return "REVIEW", "Firefox also blocks the intermediates via OneCRL, keep it updated" if accepted is None and AAA in names: return "AFFECTED", f"trusts {AAA}, which cross-signed the cracked intermediates: {hint}" return "REVIEW", f"has a root Red Hat lists, but no cracked chain verified through it: {hint}" def bundle_summary(bundles): """Every venv has its own certifi, so one row per status and fix instead of one per bundle.""" groups = {} for path, names, accepted in bundles: status, _ = store_status(path, names, accepted) groups.setdefault((status, fix_hint(path)), []).append((path, names, accepted)) rows = [] for (status, hint), items in groups.items(): names = sorted({n for _, ns, _ in items for n in ns}) proven = sum(bool(a) for _, _, a in items) note = (f"openssl verify accepts the cracked chain in {proven} of them: " if proven else "") + hint rows.append(("eMudhra CA", f"{len(items)} app CA bundles, e.g. {items[0][0]}", ", ".join(names), status, note)) return rows # --- File checks, shared by the folder walk and Docker layers ----------------------------------- LOCKFILE = re.compile(r"(requirements.*\.txt|pyproject\.toml|uv\.lock|poetry\.lock|Pipfile\.lock)$") SKIP = {"node_modules", ".git", ".Trash"} MAX_READ = 50_000_000 # cap per-file reads (disk and tar members) so an untrusted file can't exhaust memory def inspect_file(path, read): """Checks one file; read() returns its bytes and is only called when needed. Returns (rows, bundles).""" cache = [] def data(): if not cache: cache.append(read()) return cache[0] rows, bundles = [], [] base, parent = os.path.basename(path), os.path.basename(os.path.dirname(path)) for pkg in PACKAGES: name = pkg["name"] m = re.fullmatch(rf"{re.escape(pkg['package'])}-([\w.]+)\.(?:dist|egg)-info", parent) if m and base in ("METADATA", "PKG-INFO"): bad = m[1] in pkg["bad"] rows.append((name, os.path.dirname(path), m[1], "AFFECTED" if bad else "OK", pkg["cleanup"] if bad else "")) if "site-packages" in path and base in pkg["malware_files"]: rows.append((name, path, "malware file", "AFFECTED", pkg["cleanup"])) # Python only auto-runs it from the top of site-packages; Homebrew ships a legit one there elif base in pkg["suspicious_files"] and parent == "site-packages" and b"created by Homebrew" not in data()[:200]: rows.append((name, path, "possible malware", "REVIEW", f"{name} malware drops this; check it if you didn't add it")) # ponytail: proximity regex, not a real lockfile parser; good enough for pinned versions if LOCKFILE.search(base): for bad in pkg["bad"]: if re.search(rf"{re.escape(pkg['package'])}.{{0,80}}?{re.escape(bad)}".encode(), data(), re.S | re.I): rows.append((name, path, bad, "AFFECTED", f"lockfile pins the backdoored build: pin {pkg['safe']}")) if base in BUNDLE_FILES: names = roots_in(data()) if names: bundles.append((path, names, chain_accepts(data()))) return rows, bundles def scan_files(roots): roots = sorted({os.path.realpath(os.path.expanduser(r)) for r in roots}) roots = [r for r in roots if not any(r.startswith(o + os.sep) for o in roots)] # nested roots = duplicate rows rows, bundles = [], [] for root in roots: for d, dirs, files in os.walk(root): dirs[:] = [x for x in dirs if x not in SKIP] if d == f"{HOME}/Library": # huge on macOS; only its Python site-packages matter dirs[:] = [x for x in dirs if x == "Python"] for f in files: p = os.path.join(d, f) def read(p=p): try: return open(p, "rb").read() if os.path.getsize(p) < MAX_READ else b"" except OSError: return b"" r, b = inspect_file(p, read) rows += r bundles += b for pkg in PACKAGES: if os.path.exists(pkg.get("lock_file", "")): rows.append((pkg["name"], pkg["lock_file"], "malware lock", "AFFECTED", "malware daemon may be running: " + pkg["cleanup"])) if not any(r[0] == pkg["name"] for r in rows): rows.append((pkg["name"], "-", "-", "NOT FOUND", "")) return rows + bundle_summary(bundles) def scan_image_tar(label, stream): """Walks a `docker save` stream (a tar of layer tars) without running the image. ponytail: ignores whiteouts, so files deleted in a later layer still count; zstd layers are skipped.""" rows, bundles = [], [] with tarfile.open(fileobj=stream, mode="r|") as outer: for member in outer: if not member.isfile(): continue try: with tarfile.open(fileobj=outer.extractfile(member), mode="r|*") as layer: for f in layer: if f.isfile(): # cap by tar metadata before reading: an untrusted image could ship a huge/bomb member read = (lambda f=f: layer.extractfile(f).read()) if f.size < MAX_READ else (lambda: b"") r, b = inspect_file("/" + f.name, read) rows += [(i, f"{label}:{p}", *rest) for i, p, *rest in r] bundles += [(f"{label}:{p}", n, a) for p, n, a in b] except tarfile.ReadError: # manifest/config json, or a layer format tarfile can't read continue return rows, bundles def check_docker(scan_images): if not shutil.which("docker"): return [("Docker", "-", "-", "NOT FOUND", "docker not installed")] out = run("docker", "images", "--digests", "--format", "{{.ID}} {{.Repository}}:{{.Tag}} {{.Digest}}") # daemon-down / permission errors go to the daemon, not per-image output; match those, not an image named "error" if any(s in out.lower() for s in ("cannot connect", "permission denied", "is the docker daemon running")): return [("Docker", "-", "-", "UNKNOWN", "docker is installed but not reachable (daemon down or no permission)")] rows, bundles, ids = [], [], {} for line in out.splitlines(): parts = line.split() if len(parts) != 3: # unexpected format; don't crash the whole scan continue image_id, name, digest = parts ids.setdefault(image_id, name if not name.startswith("") else image_id) for pkg in PACKAGES: if digest.startswith(tuple(pkg.get("image_digests", []))): rows.append((pkg["name"], f"docker:{name}", digest[:19], "AFFECTED", "backdoored image: remove it, rebuild, " + pkg["cleanup"])) for image_id, label in ids.items() if scan_images else []: print(f" scanning image {label}...", file=sys.stderr) p = subprocess.Popen(["docker", "save", image_id], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) r, b = scan_image_tar(label, p.stdout) p.wait() rows += r or [("Docker", label, "-", "OK", "")] bundles += b return rows + bundle_summary(bundles) # --- System trust stores ------------------------------------------------------------------------ def check_os_store(): if sys.platform == "darwin": # ask macOS itself: it checks revocation, which OpenSSL bundles don't with tempfile.TemporaryDirectory() as d: trusted = [] for name, der in INTERMEDIATES.items(): open(f"{d}/i.pem", "wb").write(pem(der)) if "successful" in run("security", "verify-cert", "-c", f"{d}/i.pem").lower(): trusted.append(name) if trusted: return [("eMudhra CA", "macOS Keychain", ", ".join(trusted), "AFFECTED", "macOS trusts the cracked intermediates: install pending updates")] return [("eMudhra CA", "macOS Keychain", "-", "OK", "macOS rejects the cracked intermediates (revoked or untrusted)")] if sys.platform == "win32": # ponytail: untested on Windows. Root stores plus Disallowed, where Windows Update puts distrusted certs ps = ("Get-ChildItem Cert:\\LocalMachine\\Root,Cert:\\CurrentUser\\Root,Cert:\\LocalMachine\\Disallowed | " "% { $_.PSParentPath.Split('\\')[-1] + ' ' + $_.Thumbprint }") lines = run("powershell", "-NoProfile", "-Command", ps).upper().splitlines() thumb = lambda der: hashlib.sha1(der).hexdigest().upper() if all(f"DISALLOWED {thumb(d)}" in lines for d in INTERMEDIATES.values()): return [("eMudhra CA", "Windows cert store", "-", "OK", "the cracked intermediates are in the Disallowed store")] names = [n for n, d in ROOTS.items() if f"ROOT {thumb(d)}" in lines] status, note = store_status("windows", names, None) return [("eMudhra CA", "Windows cert store", ", ".join(names) or "-", status, note + ("; CryptoAPI accepts the chain unless revocation checking is on" if names else ""))] return [] # Linux system trust is covered by the extracted bundles in STORES def test_bundle(path, label=None): """One eMudhra row for a PEM bundle on disk, or None if it can't be read.""" try: data = open(path, "rb").read() except OSError: return None names = roots_in(data) status, note = store_status(path, names, chain_accepts(data) if names else None) return ("eMudhra CA", label or path, ", ".join(names) or "-", status, note) # CA-bundle overrides: a custom bundle here replaces the system store for that tool, so a machine can # look clean while its real trust anchor is still compromised. Env vars read by curl/OpenSSL, Node, # Python requests, Git, the AWS SDKs; plus config a tool keeps in its own file. CA_ENV_VARS = ["SSL_CERT_FILE", "CURL_CA_BUNDLE", "NODE_EXTRA_CA_CERTS", "REQUESTS_CA_BUNDLE", "AWS_CA_BUNDLE", "GIT_SSL_CAINFO", "PIP_CERT", "DENO_CERT", "CARGO_HTTP_CAINFO"] CA_CONFIG_CMDS = { # tool: argv that prints the configured bundle path (empty/null if unset) "npm": ["npm", "config", "get", "cafile"], "pip": ["pip", "config", "get", "global.cert"], "git": ["git", "config", "--global", "--get", "http.sslCAInfo"], "yarn": ["yarn", "config", "get", "cafile"], } def check_overrides(): found = {} # path -> source label, so one custom bundle shared by several tools is tested once for var in CA_ENV_VARS: if os.environ.get(var): found.setdefault(os.path.realpath(os.path.expanduser(os.environ[var])), []).append(f"${var}") for tool, cmd in CA_CONFIG_CMDS.items(): if not shutil.which(cmd[0]): continue out = run(*cmd).strip() # tools print their own "unset" marker: npm/yarn "null", pip "ERROR: No such key", git nothing if out and out.lower() not in ("null", "undefined", "none", "") and "error" not in out.lower() and "no such key" not in out.lower(): found.setdefault(os.path.realpath(os.path.expanduser(out)), []).append(f"{tool} config") rows = [] for path, sources in found.items(): label = f"{path} (via {', '.join(sources)})" rows.append(test_bundle(path, label) or ("eMudhra CA", label, "-", "REVIEW", "configured CA bundle is missing or unreadable; the tool may fall back to the system store")) return rows def check_stores(): rows = check_os_store() + check_overrides() for p in sorted({os.path.realpath(p) for g in STORES for p in glob.glob(g)}): row = test_bundle(p) if row: rows.append(row) return rows # --- Output ------------------------------------------------------------------------------------- STATUSES = ("AFFECTED", "REVIEW", "UNKNOWN", "OK", "NOT FOUND") def fix_for(item, where, note=""): """One (title, command) for an actionable row, or None. Rows with the same command collapse into one job in the fix plan, so 15 Node installs or 80 venvs become a single step.""" w = where.lower() if item == "Node.js": if ".nvm" in w: return ("Replace end-of-life Node versions under nvm", "nvm install --lts && nvm alias default 'lts/*' # then: nvm uninstall ") if ".volta" in w: return ("Upgrade the Volta-managed Node", "volta install node@22") return ("Upgrade Homebrew Node to a patched release", "brew upgrade node") if item == "eMudhra CA": if "keychain" in w: return ("Install pending macOS updates (system Keychain trust)", "softwareupdate --install --all") if "nss" in w or "firefox" in w or "thunderbird" in w: return ("Update Firefox / Thunderbird (they carry their own roots)", "update Firefox/Thunderbird to the latest version") if "cacerts" in w or "/java/" in w or "jvm" in w: return ("Update the Java trust store", "update the JDK, or: keytool -delete -alias -keystore ") if "certifi" in w: # before site-packages and /vendor/: pip ships certifi under pip/_vendor/certifi return ("Update certifi in affected virtualenvs (after a fixed release ships)", "pip install -U certifi # run in each affected venv") if "/vendor/" in w: # composer convention; note /vendor/ not _vendor/ (python) return ("Refresh vendored CA bundles (e.g. PHP/composer packages)", "composer update # in each project shipping its own cacert.pem") if "site-packages" in w: return ("Update Python packages that ship their own CA bundle", "pip install -U # e.g. botocore, once it drops the root") if "(via" in where: return ("Fix the custom CA bundle pointed to by an env var / tool config", "point the override at a patched bundle, or unset it to use the system store") return ("Update the system CA bundle (OpenSSL / curl / git)", "brew upgrade ca-certificates # Linux: sudo update-ca-certificates (Red Hat: sudo update-ca-trust)") if item == "PHP": return ("Upgrade PHP and restart workers", "upgrade to 8.5.12 / 8.4.27 / 8.3.36 / 8.2.35, then restart PHP-FPM") if item == "BIND": return ("Upgrade BIND", "upgrade to 9.20.30+ / 9.21.27+") if item == "GitLab": return ("Upgrade self-managed GitLab", "upgrade to 19.5.1 (or the latest patch for your branch, back to 18.11.13)") if item == "LiteLLM": return ("Remove the backdoored LiteLLM and rotate credentials", "don't pip install first; delete cook.pth/cook.py/sitecustomize.py, kill the daemon, rebuild pinned to 1.106.2, rotate keys") if item == "Docker": return ("Rebuild the affected Docker image", "remove the image, rebuild from a patched base, rotate any keys it held") adv = next((a for a in DISTRO if a["name"] == item), None) if adv and where.endswith(("(apt)", "(dnf)")): if note.startswith("no fixed"): # upgrading won't help yet: the workaround is the job return (f"Work around {item} until your distro ships a fix", adv.get("mitigation", "watch for the update")) if adv.get("kernel"): cmd = "sudo apt update && sudo apt full-upgrade" if where.endswith("(apt)") else "sudo dnf upgrade 'kernel*'" return (f"Update the kernel and reboot ({item})", cmd + " && sudo reboot") pkg = where.rsplit(" (", 1)[0] cmd = (f"sudo apt update && sudo apt install --only-upgrade {pkg}" if where.endswith("(apt)") else f"sudo dnf upgrade {pkg}") return (f"Update {item}" + (f", then {adv['after_fix']}" if adv.get("after_fix") else ""), cmd) return None def bundle_count(where): m = re.match(r"(\d+) app CA bundles", where) return int(m.group(1)) if m else 1 def fix_plan(rows): """Deduplicated, prioritized actions from AFFECTED/REVIEW rows: [(worst_status, title, command, count, examples)].""" groups = {} for item, where, _, status, note in rows: if status not in ("AFFECTED", "REVIEW"): continue fix = fix_for(item, str(where), note or "") if not fix: continue title, command = fix g = groups.setdefault(command, {"title": title, "statuses": set(), "count": 0, "examples": []}) g["statuses"].add(status) g["count"] += bundle_count(str(where)) if len(g["examples"]) < 3 and str(where) not in g["examples"]: g["examples"].append(str(where)) plan = [("AFFECTED" if "AFFECTED" in g["statuses"] else "REVIEW", g["title"], cmd, g["count"], g["examples"]) for cmd, g in groups.items()] return sorted(plan, key=lambda x: (STATUSES.index(x[0]), -x[3])) def html_report(rows, path): e = html.escape cls = lambda s: s.replace(" ", "") counts = "".join(f'
{sum(r[3] == s for r in rows)}{s.lower()}
' for s in STATUSES) # Fix plan: one card per distinct job, affected first plan = fix_plan(rows) if plan: cards = "".join( f'
{e(status)}' f'{e(title)}{count} item{"s" if count != 1 else ""}
' f'
{e(command)}
' f'
where
    ' + "".join(f'
  • {e(x)}
    • ' for x in examples) + (f'
  • … and more
    • ' if count > len(examples) else '') + '
' for status, title, command, count, examples in plan) fixhtml = f'

Fix plan

{len(plan)} distinct actions, most urgent first.

{cards}' else: fixhtml = '

Fix plan

Nothing affected. No action needed.

' # Issue tree: item → status → rows tree = "" for item in sorted({r[0] for r in rows}, key=lambda i: min(STATUSES.index(r[3]) for r in rows if r[0] == i)): irows = [r for r in rows if r[0] == item] worst = min(irows, key=lambda r: STATUSES.index(r[3]))[3] tally = " ".join(f'{sum(r[3] == s for r in irows)} {s.lower()}' for s in STATUSES if any(r[3] == s for r in irows)) leaves = "".join( f'
  • {e(r[3])} ' f'{e(str(r[2]))} {e(str(r[1]))}' + (f'
    {e(r[4])}
    ' if r[4] else '') + '
    • ' for r in sorted(irows, key=lambda r: STATUSES.index(r[3]))) openattr = " open" if worst in ("AFFECTED", "REVIEW") else "" tree += (f'
    {e(item)} {tally}' f'
      {leaves}
    ') sources = "".join(f'
  • {e(a["name"])}: {e(a["about"])}
    • ' for a in ADVISORIES["versions"] + DISTRO + PACKAGES + [{"name": "eMudhra CA", **ADVISORIES["ca"]}]) when = datetime.datetime.now().strftime("%d %b %Y, %H:%M") open(path, "w").write(f""" Am I affected? — report

    Am I affected?

    {e(platform.node())} · {e(platform.platform())} · {when}

    {counts}
    {fixhtml}

    Issues by component

    {tree}

    Advisories checked

      {sources}
    """) FEED_URL = "https://cve.now/feed?tech={}&format=json" def _dt(s): return datetime.datetime.fromisoformat(s.replace("Z", "").split("+")[0]) def new_cves(items, checked, known=()): """CVE ids in a cve.now JSON-feed changed after the 'checked' watermark that the advisory doesn't already cover. The feed re-lists a CVE whenever its record changes (KEV listing, CVSS rescore), so known ids and ids from before last year are updates, not new issues. ponytail: the year cut can hide a genuinely late old-year CVE; cve.now has no published date to do better.""" cut = _dt(checked) out = [] for it in items: try: cve = it["id"].split("#")[0] if _dt(it["date_modified"]) > cut and cve not in known and int(cve.split("-")[1]) >= cut.year - 1 \ and cve not in out: out.append(cve) except (KeyError, ValueError, IndexError): continue # ponytail: skip malformed items, the feed shape is not ours to trust return out def check_feed_updates(advs, timeout=6): """Best-effort: one 'data may be stale' line per tech with CVEs newer than its watermark. Read-only GET of the public cve.now feed; nothing about this machine is sent. Any network/parse error is swallowed so the offline scan never depends on it.""" lines = [] for adv in advs: slug, checked = adv.get("feed"), adv.get("checked") if not slug or not checked: continue # no feed slug (e.g. BIND) -> not feed-tracked try: with urllib.request.urlopen(FEED_URL.format(slug), timeout=timeout) as r: items = json.load(r).get("items", []) except Exception: continue fresh = new_cves(items, checked, adv.get("cves", ())) if fresh: shown = ", ".join(fresh[:5]) + (f" +{len(fresh) - 5} more" if len(fresh) > 5 else "") lines.append(f"{adv['name']}: {len(fresh)} CVE(s) on cve.now newer than {checked[:10]} " f"({shown}) -- secinspect data may be stale, see cve.now") return lines def main(argv): ap = argparse.ArgumentParser(description="Am I affected by the issues in advisories.json?") ap.add_argument("dirs", nargs="*", help="search only these folders for Python envs, lockfiles and CA bundles") ap.add_argument("--docker", action="store_true", help="also look inside every local Docker image (slow)") ap.add_argument("--html", metavar="FILE", help="also write an HTML report") ap.add_argument("--no-update-check", action="store_true", help="skip the cve.now advisory-freshness check (offline/CI; also $SECINSPECT_NO_UPDATE_CHECK)") args = ap.parse_args(argv) roots = args.dirs or [HOME, "/usr/local/lib", "/opt/homebrew/lib", *glob.glob("/usr/lib/python3*")] print(f"Scanning {', '.join(roots)} for Python envs, lockfiles and CA bundles, give it a minute...", file=sys.stderr) if not (args.no_update_check or os.environ.get("SECINSPECT_NO_UPDATE_CHECK")): for line in check_feed_updates(ADVISORIES["versions"]): print("update: " + line, file=sys.stderr) if not OPENSSL: print("openssl not found: CA bundles are judged by which roots they hold, not by a chain test", file=sys.stderr) rows = [r for adv in ADVISORIES["versions"] for r in check_version(adv)] + check_distro() rows += scan_files(roots) + check_docker(args.docker) + check_stores() widths = [max(len(str(r[i])) for r in rows) for i in range(4)] for r in rows: print(" ".join(str(c).ljust(w) for c, w in zip(r, widths)), r[4]) if args.html: html_report(rows, args.html) print(f"Report written to {os.path.abspath(args.html)}", file=sys.stderr) return 1 if any(r[3] == "AFFECTED" for r in rows) else 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))