#!/usr/bin/env python3 """ portcve.py - list FreeBSD ports and the CVEs affecting them, based on the OpenCVE data published as git repositories. No API, no rate limits. Everything lives in the cache directory ($XDG_CACHE_HOME/portcve, by default ~/.cache/portcve, change with --cache-dir): opencve-kb/ shallow clones of the OpenCVE repositories, cloned opencve-nvd/ automatically on the first run (several GB) state.sqlite CVE index and ports cache It only holds re-downloadable data: deleting it is safe, the next run clones and indexes again. Upstream updates every 15 minutes; --pull fetches them. opencve-kb which CVEs are linked to a CPE vendor:product - OpenCVE's own linkage (NVD CPEs plus OpenCVE enrichment and other sources), the same data the OpenCVE web app and API serve. opencve-nvd per-CVE version data: NVD CPE configurations and the CNA's "affected" ranges. CNA ranges are used first, NVD as fallback. Which ports: (no arguments) installed packages (CPE from the package annotation) ORIGIN ... the given ports (-f FILE reads origins from a file) -a, --all the whole ports tree -m MAINTAINER only ports of this maintainer (repeatable, * wildcards allowed) For ports, `make -V` runs in parallel and its results are cached; a port is only evaluated again when one of the makefiles it read has changed (a change in Mk/ therefore re-evaluates everything). Exit status: 0 = nothing affected, 1 = affected ports found, 2 = usage error, 3 = runtime or data error, 130 = interrupted. Only the Python standard library is used. """ import argparse import concurrent.futures as cf import datetime import fnmatch import functools import hashlib import json import os import re import shutil import sqlite3 import subprocess import sys import time from collections import defaultdict from pathlib import Path NON_VERSIONS = {"", "0", "*", "-", "n/a", "unspecified", "unknown", "all"} CLAUSE_RE = re.compile(r"(<=|>=|==|<|>|=|before|prior to)\s*([^\s,;&|]+)", re.I) CLAUSE_OPS = {"<": lambda c: c < 0, "<=": lambda c: c <= 0, ">": lambda c: c > 0, ">=": lambda c: c >= 0, "=": lambda c: c == 0, "==": lambda c: c == 0, "before": lambda c: c < 0, "prior to": lambda c: c < 0} PRERELEASE = {"dev": 0, "snapshot": 0, "alpha": 1, "beta": 2, "pre": 3, "rc": 4} PRODUCT_SEPARATOR = "$PRODUCT$" GIT_BASE_URL = os.environ.get("OPENCVE_GIT_BASE", "https://github.com/opencve") CVE_PATH_RE = re.compile(r"^(\d{4})/(CVE-\d{4}-\d+)\.json$") # Exit status, so automation can tell findings from failures. EXIT_OK, EXIT_AFFECTED, EXIT_USAGE, EXIT_ERROR, EXIT_INTERRUPTED = 0, 1, 2, 3, 130 # Environment variables that change what `make -V` reports; part of the global # ports fingerprint, because .MAKE.MAKEFILES cannot show them. MAKE_ENV_KEYS = ("PORTSDIR", "PORT_DBDIR", "PORTS_DBDIR", "LOCALBASE", "PREFIX", "ARCH", "OSVERSION", "__MAKE_CONF", "MAKEFLAGS", "MAKESYSPATH", "DESTDIR", "UNAME_r", "UNAME_v", "UNAME_m") def log(msg): print(msg, file=sys.stderr, flush=True) def die(msg): log(f"error: {msg}") sys.exit(EXIT_ERROR) class Progress: def __init__(self, label, total, every=5.0): self.label, self.total, self.every = label, total, every self.done, self.last = 0, 0.0 def step(self, n=1): self.done += n now = time.monotonic() if now - self.last >= self.every or self.done == self.total: self.last = now log(f"{self.label}: {self.done}/{self.total}") # -------------------------------------------------------------------------- # CPE / version matching # -------------------------------------------------------------------------- def split_cpe(cpe): """Split a CPE 2.3 formatted string on unescaped colons and unescape.""" return [re.sub(r"\\(.)", r"\1", p) for p in re.split(r"(? 1.1.1 PRERELEASE_LETTERS = {"a": "alpha", "b": "beta", "c": "rc"} def _version_tokens(version): """Numbers and words; a single letter followed by a number is a pre-release (3.13.0b1 = 3.13.0beta1), a trailing letter is a post-release (1.1.1w).""" toks = re.findall(r"\d+|[a-z]+", (version or "").lower()) if len(toks) > 1 and toks[0] == "v" and toks[1].isdigit(): toks = toks[1:] # v2.0.0 is 2.0.0 (CNA records and git tags use both) return [PRERELEASE_LETTERS.get(t, t) if i + 1 < len(toks) and toks[i + 1].isdigit() else t for i, t in enumerate(toks)] # bsd.port.mk builds PORTVERSION from DISTVERSION with :C/([a-z])[a-z]+/\1/g, # so "3.15.0rc2" becomes "3.15.0.r2" and "3.3-dev10" becomes "3.3.d10". A lone # letter is then genuinely ambiguous: lang/python315 3.15.0.r2 is an rc, but # audio/libmodplug 0.8.9.1.r461 is a revision after 0.8.9.1. Only letters that # a real port uses both ways are listed. "p" is deliberately absent: the only # lone "p" in the tree is security/openssh-portable 10.5p1, a patch level. MANGLED_PRERELEASE = {"d": "dev", "r": "rc", "s": "snapshot"} def mangled_readings(version): """(pre-release reading, revision reading) for a mangled PORTVERSION, or None when the version has no ambiguous lone letter.""" toks = re.findall(r"\d+|[a-z]+", (version or "").lower()) marks = [i for i, t in enumerate(toks) if t in MANGLED_PRERELEASE and i + 1 < len(toks) and toks[i + 1].isdigit()] if not marks: return None pre = [MANGLED_PRERELEASE[t] if i in marks else t for i, t in enumerate(toks)] return ".".join(pre), version def _wildcard_prefix(text): """'6.1.*' or '2.4.x' -> '6.1' / '2.4'; None for anything else.""" m = re.fullmatch(r"\s*(\d+(?:\.\d+)*)\.[*xX]\s*", text or "") return m.group(1) if m else None def vercmp_bound(version, bound): """vercmp() against a bound that may be a branch wildcard: 6.1.90 vs 6.1.* is 0, 6.0.9 is below and 6.2.0 above it. Other bounds compare normally.""" prefix = _wildcard_prefix(bound) if prefix is None: return vercmp(version, bound) wanted, numbers = prefix.count(".") + 1, [] for tok in _version_tokens(version): if not tok.isdigit() or len(numbers) == wanted: break numbers.append(tok) return vercmp(".".join(numbers) or "0", prefix) def vercmp(a, b): """Loose upstream version comparison. Returns -1, 0 or 1.""" ta, tb = _version_tokens(a), _version_tokens(b) for i in range(max(len(ta), len(tb))): x = ta[i] if i < len(ta) else None y = tb[i] if i < len(tb) else None if x is None and y.isdigit(): # 1.0 == 1.0.0 x = "0" if y is None and x.isdigit(): y = "0" kx, ky = _key(x), _key(y) if kx != ky: return -1 if kx < ky else 1 return 0 def in_range(match, cpe_version, version): checks = ( ("versionStartIncluding", lambda c: c >= 0), ("versionStartExcluding", lambda c: c > 0), ("versionEndIncluding", lambda c: c <= 0), ("versionEndExcluding", lambda c: c < 0), ) bounded = False for field, ok in checks: if match.get(field): bounded = True if not ok(vercmp_bound(version, match[field])): return False if bounded or cpe_version in ("*", "-", ""): return True return vercmp_bound(version, cpe_version) == 0 def _describe_nvd(match, cpe_version): ops = (("versionStartIncluding", ">="), ("versionStartExcluding", ">"), ("versionEndIncluding", "<="), ("versionEndExcluding", "<")) parts = [f"{op} {match[f]}" for f, op in ops if match.get(f)] if not parts: parts = ["all versions"] if cpe_version in ("*", "-", "") else [f"= {cpe_version}"] return ", ".join(parts) NVD_RANGE_FIELDS = ("versionStartIncluding", "versionStartExcluding", "versionEndIncluding", "versionEndExcluding") def _nvd_platform_restriction(config, vendor, product): """For an AND configuration "product AND running on ": the OS names when none of them is FreeBSD, else None (no restriction that rules out FreeBSD).""" if (config.get("operator") or "").upper() != "AND": return None platforms = set() for node in _dicts(config.get("nodes")): fields = [split_cpe(m.get("criteria")) for m in _dicts(node.get("cpeMatch"))] if any(len(f) > 4 and f[3].lower() == vendor and f[4].lower() == product for f in fields): continue # the application side if node.get("negate"): return None for f in fields: if len(f) < 5 or f[2] != "o": return None # not an operating system: can't judge it if f[3].lower() == "freebsd": return None platforms.add(f"{f[3]}:{f[4]}") return ", ".join(sorted(platforms)) or None def evaluate_nvd(configurations, vendor, product, version): """NVD CPE configurations -> (True/False/None, why). A CPE match without any version information ("-", or "*" without a range) doesn't say which versions are affected - NVD uses it mostly for old or unanalysed entries - so it can only make the result unknown, not affected. Matches that only apply together with another operating system (AND configurations such as "... running on Windows") are unknown as well. """ seen = unversioned = update_only = False other_platform = None for config in _dicts(configurations): restriction = _nvd_platform_restriction(config, vendor, product) for node in _dicts(config.get("nodes")): if node.get("negate"): continue for match in _dicts(node.get("cpeMatch")): f = split_cpe(match.get("criteria")) if len(f) < 6 or f[3].lower() != vendor or f[4].lower() != product: continue seen = True if not match.get("vulnerable"): continue ranged = any(match.get(k) for k in NVD_RANGE_FIELDS) if f[5] in ("*", "-", "") and not ranged: unversioned = True continue # A CPE naming an update ("foo:2.0:beta1") is about that # pre-release or patch level, not about the plain 2.0 release. update = f[6] if len(f) > 6 else "*" if update not in ("*", "-", "") and not ranged: if not in_range(match, f[5] + update, version): update_only = update_only or in_range(match, f[5], version) continue why = "NVD: = " + f[5] + update elif in_range(match, f[5], version): why = "NVD: " + _describe_nvd(match, f[5]) else: continue if restriction is None: return True, why other_platform = other_platform or f"{why}, but only on {restriction}" if other_platform: return None, other_platform if unversioned: return None, "NVD: no version range given (all/unspecified versions)" if update_only: return None, "NVD: only specific updates/pre-releases of this version listed" return (False, "NVD: outside CPE ranges") if seen else (None, None) # -- CNA data (CVE JSON 5 "affected") ---------------------------------------- def _norm(text): return re.sub(r"[^a-z0-9]", "", _text(text).lower()) def _versionish(text): t = _text(text).strip().lower() if not re.search(r"\d", t): return False return not (re.fullmatch(r"[0-9a-f]{7,40}", t) and re.search(r"[a-f]", t)) # git hash def _cna_entry_matches(entry, vendor, product): for cpe in entry.get("cpes") or []: f = split_cpe(cpe) if len(f) > 4 and f[3].lower() == vendor and f[4].lower() == product: return True # The product names what is affected; packageName is only used without one. # (Red Hat e.g. lists product "Red Hat Enterprise Linux 9", packageName "php".) if _norm(entry.get("product")) not in ("", "na"): names = {_norm(entry.get("product"))} else: package = entry.get("packageName") or "" names = {_norm(package), _norm(package.split("/")[-1])} - {""} return _norm(product) in names or _norm(vendor + product) in names def _parse_clauses(text): """">= 1.0, < 2.0" -> [(op, version), ...]; [] = no operators; None = malformed. Every part of the string has to be a clause. GitHub advisories e.g. contain ">= 6.2.0, 6.2.6" (the "<" is missing); reading only ">= 6.2.0" from it would mark every later version as affected. """ clauses = CLAUSE_RE.findall(text) if not clauses: return [] if CLAUSE_RE.sub("", text).strip(" ,;&") or not all(_versionish(v) for _, v in clauses): return None return [(op.lower(), bound) for op, bound in clauses] def _cna_in_range(v, version): """Is version inside this versions[] item? True/False, None = unusable.""" if _text(v.get("versionType")).lower() in ("git", "rpm", "deb", "dpkg", "apk"): return None # commit hashes / distro package versions start = _text(v.get("version")).strip() lt, lte = v.get("lessThan"), v.get("lessThanOrEqual") if not isinstance(lt, str): lt = None if not isinstance(lte, str): lte = None if lt and start.lower() not in NON_VERSIONS and _versionish(start) \ and _versionish(lt) and vercmp(start, lt) == 0: # "version X, lessThan X" is an empty range. curl uses it for "fixed in X" # (affected) and "introduced in X" (unaffected), so neither reading as an # empty range nor as exactly X is right: the item is unusable. return None if lt or lte: if start.lower() not in NON_VERSIONS: if not _versionish(start): return None if vercmp_bound(version, start) < 0: return False bound = (lt or lte).strip() if bound in ("*", ""): return True if not _versionish(bound): return None c = vercmp_bound(version, bound) return c < 0 if lt else c <= 0 clauses = _parse_clauses(start) if clauses is None: return None if clauses: return all(CLAUSE_OPS[op](vercmp_bound(version, bound)) for op, bound in clauses) if start.lower() in NON_VERSIONS or not _versionish(start): return None return vercmp_bound(version, start) == 0 def _cna_well_formed_range(v): """A real lower..upper range (curl's CNA records e.g. use version == lessThan).""" start = _text(v.get("version")).strip() lte = _text(v.get("lessThanOrEqual")) bound = (_text(v.get("lessThan")) or lte).strip() if not bound: # "version": ">= 4.0.0, < 4.1.2" style return bool(_parse_clauses(start)) if bound != "*" and not _versionish(bound): return False if bound == "*" or start.lower() in NON_VERSIONS: return True if not _versionish(start): return False c = vercmp_bound(start, bound) return c < 0 or (bool(lte) and c == 0) def _describe_cna(v, status, change_at=None): start = _text(v.get("version")).strip() bound = _text(v.get("lessThan")) or _text(v.get("lessThanOrEqual")) if bound: parts = [] if start.lower() in NON_VERSIONS else [f">= {start}"] if bound.strip() != "*": parts.append(("< " if v.get("lessThan") else "<= ") + bound) text = ", ".join(parts) or "all versions" else: text = start note = f" ({status} since {change_at})" if change_at else "" return f"CNA: {v.get('status')} {text}{note}" def _cna_changes_usable(v): """False when "changes" is present but malformed: it can turn an affected range into an unaffected one, so ignoring it silently would over-report.""" raw = v.get("changes") if raw is None: return True items = _dicts(raw) return (isinstance(raw, list) and len(items) == len(raw) and all(_versionish(c.get("at")) for c in items)) def _cna_status_at(v, version): """Status for version inside the range, honouring "changes".""" status, change_at = v.get("status"), None changes = [c for c in _dicts(v.get("changes")) if _versionish(c.get("at"))] changes.sort(key=functools.cmp_to_key(lambda a, b: vercmp(a["at"], b["at"]))) for change in changes: if vercmp_bound(version, change["at"]) >= 0: status, change_at = change.get("status", status), change["at"] return status, change_at def _branch(text): """Release branch of a version: its leading numbers without the last one (the wildcard bound 6.1.* is itself branch 6.1).""" prefix = _wildcard_prefix(text) if prefix: return tuple(int(n) for n in prefix.split(".")) numbers = [] for tok in re.findall(r"\d+|[a-z]+", _text(text).lower()): if not tok.isdigit(): break numbers.append(int(tok)) return tuple(numbers[:-1]) if len(numbers) >= 2 else None def _cna_is_range(v): """A wildcard version ("2.4.*") covers a whole branch, so it is a range too: a precise unaffected range must still win over it.""" version = _text(v.get("version")).strip() return bool(v.get("lessThan") or v.get("lessThanOrEqual") or _wildcard_prefix(version) or _parse_clauses(version)) def _cna_range_rank(v): """0 = an explicitly bounded range, 1 = only a branch wildcard ("2.4.*").""" return 0 if (v.get("lessThan") or v.get("lessThanOrEqual") or _parse_clauses(_text(v.get("version")).strip())) else 1 def _cna_upper_bound(v): bound = (_text(v.get("lessThan")) or _text(v.get("lessThanOrEqual"))).strip() if bound: return None if bound == "*" else bound uppers = [b for op, b in _parse_clauses(_text(v.get("version")).strip()) or [] if op in ("<", "<=", "before", "prior to")] return uppers[0] if uppers else None def _evaluate_cna_entry(entry, version): usable = True affected_ranges = unaffected_items = False exact_hits, range_hits = [], [] # matching items: (item, status, change_at) branch_ranges = [] # affected ranges ending in the version's branch branch = _branch(version) raw = entry.get("versions") items = _dicts(raw) if raw is not None and (not isinstance(raw, list) or len(items) != len(raw)): usable = False # malformed items: unknown is allowed, a definitive answer is not for v in items: hit = _cna_in_range(v, version) if hit is None or not _cna_changes_usable(v): usable = False continue is_range = _cna_is_range(v) if v.get("status") == "affected" and is_range: if _cna_well_formed_range(v): affected_ranges = True upper = _cna_upper_bound(v) if branch and upper and _branch(upper) == branch: branch_ranges.append(v) if v.get("status") == "unaffected": unaffected_items = True if hit: status, change_at = _cna_status_at(v, version) (range_hits if is_range else exact_hits).append((v, status, change_at)) # 1. A statement about this exact version is more specific than any range. for wanted in ("affected", "unaffected"): for v, status, change_at in exact_hits: if status == wanted: return status == "affected", _describe_cna(v, status, change_at) # 2. Several overlapping ranges for different release branches (Redis lists # "0 <= v < 8.8.2", "0 <= v < 8.10.1", ...): if one range ends in this # version's own branch, the ranges ending in other branches don't apply. if branch_ranges: range_hits = [h for h in range_hits if h[1] != "affected" or h[0] in branch_ranges] # 2b. A range with real bounds is more specific than a bare branch wildcard # ("2.4.10 <= v <= 2.4.*" beats "2.4.*"). if range_hits: best = min(_cna_range_rank(h[0]) for h in range_hits) range_hits = [h for h in range_hits if _cna_range_rank(h[0]) == best] for v, status, change_at in range_hits: if status == "affected": return True, _describe_cna(v, status, change_at) if range_hits: return False, _describe_cna(*range_hits[0]) if branch_ranges: return False, ("CNA: outside the affected range for this branch (" + _describe_cna(branch_ranges[0], "affected")[len("CNA: affected "):] + ")") if not usable: return None, None default = entry.get("defaultStatus") if affected_ranges and (default != "affected" or not unaffected_items): # Versions outside all listed affected ranges are fixed or never affected - # even if the CNA sets defaultStatus "affected" without listing unaffected # versions, as PHP does. return False, "CNA: outside affected ranges" if default == "affected": return True, "CNA: affected (defaultStatus, version not listed as unaffected)" if default == "unaffected": return False, "CNA: not listed as affected (defaultStatus unaffected)" return None, None def evaluate_cna(groups, vendor, product, version): """CNA "affected" groups (CNA first, then ADPs) -> (True/False/None, why).""" for entries in groups if isinstance(groups, list) else []: results = [_evaluate_cna_entry(e, version) for e in _dicts(entries) if _cna_entry_matches(e, vendor, product)] for status, why in results: if status: return True, why for status, why in results: if status is False: return False, why return None, None def _decide(record, vendor, product, version): status, why = evaluate_cna(record.get("cna_groups"), vendor, product, version) if status is None: status, why = evaluate_nvd(record.get("configurations"), vendor, product, version) return status, why def decide(record, vendor, product, version): """Combine CNA and NVD data -> (True/False/None, why). A FreeBSD version whose lone letter could be either a mangled pre-release marker or a revision is evaluated both ways; if the readings disagree the data does not support a definitive answer. """ if not isinstance(record, dict): return None, "malformed CVE record" try: readings = mangled_readings(version) if not readings: return _decide(record, vendor, product, version) pre, rev = (_decide(record, vendor, product, v) for v in readings) if pre[0] is rev[0]: return pre return None, (f"ambiguous version {version}: {_verdict(pre[0])} read as a " f"pre-release ({readings[0]}), {_verdict(rev[0])} as a revision") except Exception as exc: # malformed upstream data must not abort the run return None, f"malformed CVE record ({type(exc).__name__}: {exc})" def _verdict(status): return {True: "affected", False: "not affected", None: "unknown"}[status] # -------------------------------------------------------------------------- # OpenCVE data from the git repositories # -------------------------------------------------------------------------- def record_from_nvd(d): """Version data from an opencve-nvd file (NVD 2.0 format plus CNA "affected").""" if not isinstance(d, dict): return {"configurations": [], "cna_groups": [], "disputed": False, "rejected": False} source = d.get("sourceIdentifier") groups = sorted(_dicts(d.get("affected")), key=lambda g: g.get("source") != source) tags = {t for entry in _dicts(d.get("cveTags")) for t in entry.get("tags") or []} description = next((_text(x.get("value")) for x in _dicts(d.get("descriptions")) if x.get("lang") == "en"), "") return {"configurations": d.get("configurations") or [], "cna_groups": [g.get("affectedData") or [] for g in _dicts(groups)], # CVE tag "disputed", or the legacy "** DISPUTED **" description prefix "disputed": "disputed" in tags or description.lstrip().startswith("** DISPUTED"), # withdrawn by the CNA; the KB still links some of these to products "rejected": _text(d.get("vulnStatus")).lower() in ("rejected", "revoked")} def _kb_summary(repo, rel): """Parse one opencve-kb file -> (cve_id, links, title, score, kev, created).""" cve_id = CVE_PATH_RE.match(rel).group(2) try: with open(os.path.join(repo, rel), "rb") as fh: data = json.loads(fh.read()) except FileNotFoundError: return cve_id, None, None, None, None, None except ValueError: return cve_id, [], None, None, None, None try: kb = data.get("opencve") or {} if isinstance(data, dict) else {} vendors = kb.get("vendors") links = set() for entry in (vendors.get("data") or [] if isinstance(vendors, dict) else []): if isinstance(entry, str) and PRODUCT_SEPARATOR in entry: vendor, product = entry.split(PRODUCT_SEPARATOR, 1) links.add((vendor.lower(), product.lower())) metrics = kb.get("metrics") or {} score = None for key in ("cvssV4_0", "cvssV3_1", "cvssV3_0", "cvssV2_0"): value = ((metrics.get(key) or {}).get("data") or {}).get("score") if isinstance(value, (int, float)): score = value break title = (kb.get("title") or {}).get("data") or \ _text((kb.get("description") or {}).get("data")).strip().split("\n")[0] return (cve_id, sorted(links), title, score, bool((metrics.get("kev") or {}).get("data")), (kb.get("created") or {}).get("data")) except Exception as exc: # one unexpected shape upstream must not break the index on every run log(f"warning: skipping malformed KB record {cve_id} ({type(exc).__name__}: {exc})") return cve_id, [], None, None, None, None def _kb_summaries(repo, rels): return [_kb_summary(repo, rel) for rel in rels] def git(repo, *cmd, check=False): r = subprocess.run(["git", "-C", str(repo), *cmd], capture_output=True, text=True) if check and r.returncode != 0: die(f"git {' '.join(cmd)} in {repo} failed:\n{r.stderr.strip()}") return r class OpenCVEData: def __init__(self, kb_repo, nvd_repo, db, jobs, clone_missing=()): self.kb_repo, self.nvd_repo, self.db, self.jobs = kb_repo, nvd_repo, db, jobs # Repositories we cloned ourselves may be reset; the user's may not. self.managed = set() for name, repo in (("opencve-kb", kb_repo), ("opencve-nvd", nvd_repo)): if name in clone_missing: self.managed.add(str(repo)) if (repo / ".git").exists(): continue url = f"{GIT_BASE_URL}/{name}.git" if name not in clone_missing: die(f"{repo} is not a clone of {name}. Run:\n" f" git clone --depth 1 {url} {repo}") self._clone(url, repo) db.executescript(""" CREATE TABLE IF NOT EXISTS kb_links (vendor TEXT, product TEXT, cve_id TEXT); CREATE INDEX IF NOT EXISTS kb_links_vp ON kb_links (vendor, product); CREATE INDEX IF NOT EXISTS kb_links_cve ON kb_links (cve_id); CREATE TABLE IF NOT EXISTS kb_cves ( cve_id TEXT PRIMARY KEY, title TEXT, score REAL, kev INTEGER, created TEXT); CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT);""") @staticmethod def _clone(url, repo): """Shallow clone into repo.tmp first, so an interrupted clone isn't used.""" if repo.exists(): die(f"{repo} exists but is not a git clone; remove it and run again") tmp = repo.with_name(repo.name + ".tmp") shutil.rmtree(tmp, ignore_errors=True) repo.parent.mkdir(parents=True, exist_ok=True) log(f"cloning {url} into {repo} (several GB, only needed once) ...") if subprocess.run(["git", "clone", "--depth", "1", url, str(tmp)]).returncode != 0: shutil.rmtree(tmp, ignore_errors=True) die(f"cloning {url} failed") tmp.rename(repo) def pull(self): """Update both clones. Only the ones in the cache directory are ours to reset: "fetch --depth 1" + "reset --hard" would discard local commits and make a full clone shallow, so a repository passed with --kb-repo / --nvd-repo is only fast-forwarded.""" for repo in (self.kb_repo, self.nvd_repo): log(f"updating {repo} ...") if str(repo) in self.managed: git(repo, "fetch", "--depth", "1", "origin", "HEAD", check=True) git(repo, "reset", "--hard", "FETCH_HEAD", check=True) continue r = git(repo, "pull", "--ff-only") if r.returncode != 0: die(f"git pull --ff-only in {repo} failed:\n{r.stderr.strip()}\n" f"({repo} was not cloned by portcve, so it is not reset.)") def update_index(self): head = git(self.kb_repo, "rev-parse", "HEAD", check=True).stdout.strip() row = self.db.execute("SELECT value FROM meta WHERE key='kb_commit'").fetchone() old = row[0] if row else None if old == head: return rels = None if old: r = git(self.kb_repo, "diff", "--name-only", old, head) if r.returncode == 0: rels = [p for p in r.stdout.splitlines() if CVE_PATH_RE.match(p)] log(f"opencve-kb: {len(rels)} changed CVE file(s) since last run") if rels is None: log("opencve-kb: building the CVE index (first run, a few minutes) ...") self.db.execute("DELETE FROM kb_links") self.db.execute("DELETE FROM kb_cves") self.db.execute("DELETE FROM meta WHERE key='kb_commit'") rels = sorted(str(p.relative_to(self.kb_repo)) for p in self.kb_repo.glob("[0-9][0-9][0-9][0-9]/CVE-*.json")) full = True else: full = False chunks = [rels[i:i + 2000] for i in range(0, len(rels), 2000)] progress = Progress("opencve-kb index", len(rels)) if len(chunks) > 1 and self.jobs > 1: pool = cf.ProcessPoolExecutor(self.jobs) results = pool.map(_kb_summaries, [str(self.kb_repo)] * len(chunks), chunks) else: pool = None results = (_kb_summaries(str(self.kb_repo), c) for c in chunks) try: for summaries in results: for cve_id, links, title, score, kev, created in summaries: if not full: self.db.execute("DELETE FROM kb_links WHERE cve_id=?", (cve_id,)) self.db.execute("DELETE FROM kb_cves WHERE cve_id=?", (cve_id,)) if links: self.db.executemany("INSERT INTO kb_links VALUES (?,?,?)", [(v, p, cve_id) for v, p in links]) self.db.execute("INSERT INTO kb_cves VALUES (?,?,?,?,?)", (cve_id, title, score, int(kev), created)) progress.step(len(summaries)) finally: if pool: pool.shutdown() self.db.execute("INSERT OR REPLACE INTO meta VALUES ('kb_commit', ?)", (head,)) self.db.commit() def cves_for(self, vendor, product): return [tuple(r) for r in self.db.execute( """SELECT c.cve_id, c.title, c.score, c.kev, c.created FROM kb_links l JOIN kb_cves c ON c.cve_id = l.cve_id WHERE l.vendor=? AND l.product=?""", (vendor, product))] def _match_group(nvd_repo, vendor, product, cves, ports, since, include_disputed): """Worker: decide every (port, CVE) pair for one vendor:product.""" findings = {key: {"affected": [], "unknown": [], "rejected": 0, "disputed_hidden": {"affected": 0, "unknown": 0}} for key, _ in ports} for cve_id, title, score, kev, created in cves: if since and (created or "")[:10] < since: continue path = os.path.join(nvd_repo, cve_id.split("-")[1], f"{cve_id}.json") try: with open(path, "rb") as fh: record = record_from_nvd(json.loads(fh.read())) no_data = "no CNA or NVD version data for this product" except (FileNotFoundError, ValueError): record, no_data = {}, "no opencve-nvd record for this CVE (yet)" except OSError as exc: record, no_data = {}, f"opencve-nvd record unreadable ({exc.strerror})" if record.get("rejected"): # withdrawn by the CNA; the KB still links some of them to products for key, _ in ports: findings[key]["rejected"] += 1 continue for key, version in ports: status, why = decide(record, vendor, product, version) if status is False: continue if record.get("disputed") and not include_disputed: findings[key]["disputed_hidden"]["affected" if status else "unknown"] += 1 continue item = {"cve_id": cve_id, "score": score, "kev": bool(kev), "title": title or "", "disputed": bool(record.get("disputed")), "why": why or no_data} findings[key]["affected" if status else "unknown"].append(item) return findings # -------------------------------------------------------------------------- # FreeBSD ports # -------------------------------------------------------------------------- def installed_packages(): """[(origin[@flavor], pkgname, cpe_or_None, maintainer)] for installed packages. Annotations are keyed by package name: flavors of one port (py311-foo, py312-foo) share the origin but are separate packages. """ def query(fmt): return subprocess.run(["pkg", "query", "-a", fmt], capture_output=True, text=True, check=True).stdout.splitlines() notes = defaultdict(dict) for line in query("%n\t%At\t%Av"): name, tag, value = (line.split("\t", 2) + ["", ""])[:3] if tag in ("cpe", "flavor"): notes[name][tag] = value packages = [] for line in query("%n\t%o\t%n-%v\t%m"): name, origin, pkgname, maintainer = (line.split("\t", 3) + ["", "", ""])[:4] flavor = notes[name].get("flavor") packages.append((f"{origin}@{flavor}" if flavor else origin, pkgname, notes[name].get("cpe"), maintainer)) return packages def tree_origins(portsdir): """All port origins, from the SUBDIR lists of the category Makefiles.""" subdir = re.compile(r"^\s*SUBDIR\s*\+=\s*(\S+)", re.M) origins = [] for category in subdir.findall((portsdir / "Makefile").read_text()): makefile = portsdir / category / "Makefile" if makefile.exists(): origins += [f"{category}/{p}" for p in subdir.findall(makefile.read_text())] return origins def _make_vars(portsdir, origin): """Worker: PKGNAME, CPE_STR, MAINTAINER and the makefiles read, via make -V.""" path, _, flavor = origin.partition("@") portdir = os.path.join(portsdir, path) def portdir_files(): # used when the makefiles read are unknown try: return [str(p) for p in Path(portdir).iterdir() if p.is_file()] except OSError: return [os.path.join(portdir, "Makefile")] cmd = ["make", "-C", portdir, "-V", "PKGNAME", "-V", "CPE_STR", "-V", "MAINTAINER", "-V", "OPTIONS_FILE", "-V", ".MAKE.MAKEFILES"] if flavor: cmd.append(f"FLAVOR={flavor}") try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=300) except subprocess.TimeoutExpired: return origin, None, None, None, portdir_files(), "make -V timed out" except OSError as exc: return origin, None, None, None, portdir_files(), f"running make failed: {exc}" if r.returncode != 0: lines = r.stderr.strip().splitlines() return (origin, None, None, None, portdir_files(), lines[-1] if lines else f"make exited {r.returncode}") pkgname, cpe, maintainer, options, makefiles = (r.stdout.split("\n") + [""] * 5)[:5] files = [os.path.normpath(os.path.join(portdir, f)) for f in makefiles.split()] # .MAKE.MAKEFILES only lists files that existed; the options file has to be # watched even while it is absent, or "make config" goes unnoticed. if options.strip(): files.append(os.path.normpath(os.path.join(portdir, options.strip()))) return origin, pkgname, cpe or None, maintainer, files or portdir_files(), None class PortsScanner: SCHEMA_VERSION = "3" # bump when the cached make -V variables change def __init__(self, portsdir, db, jobs): self.portsdir, self.db, self.jobs = portsdir, db, jobs self._stamps = {} db.execute("CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT)") row = db.execute("SELECT value FROM meta WHERE key='ports_schema'").fetchone() if not row or row[0] != self.SCHEMA_VERSION: db.execute("DROP TABLE IF EXISTS ports") db.execute("INSERT OR REPLACE INTO meta VALUES ('ports_schema', ?)", (self.SCHEMA_VERSION,)) db.executescript(""" CREATE TABLE IF NOT EXISTS ports ( origin TEXT PRIMARY KEY, pkgname TEXT, cpe TEXT, maintainer TEXT, error TEXT, files TEXT, -- makefiles read, except the global ones files_fp TEXT, global_fp TEXT);""") db.commit() self.global_roots = [portsdir / "Mk", Path("/usr/share/mk")] self.global_files = [Path(f) for f in ("/etc/make.conf", "/etc/src.conf", "/etc/src-env.conf")] self._global_file_set = {str(p) for p in self.global_files} self._global_prefixes = tuple(str(root) + os.sep for root in self.global_roots) env = "\0".join(f"{k}={os.environ[k]}" for k in MAKE_ENV_KEYS if k in os.environ) self.global_fp = hashlib.sha1( (self._fingerprint( [str(p) for root in self.global_roots if root.exists() for p in root.rglob("*") if p.is_file()] + [str(p) for p in self.global_files]) + "\0" + env).encode()).hexdigest() def _stamp(self, path): if path not in self._stamps: try: st = os.stat(path) self._stamps[path] = f"{st.st_mtime_ns}:{st.st_size}" except OSError: self._stamps[path] = "-" return self._stamps[path] def _fingerprint(self, paths): h = hashlib.sha1() for p in sorted(paths): h.update(f"{p}\0{self._stamp(p)}\n".encode()) return h.hexdigest() def _is_global(self, path): return path in self._global_file_set or path.startswith(self._global_prefixes) def scan(self, origins): """[(origin, pkgname, cpe_or_None, maintainer, error)] - cached where still valid.""" cached, todo = {}, [] for origin in origins: row = self.db.execute( """SELECT pkgname, cpe, maintainer, error, files, files_fp, global_fp FROM ports WHERE origin=?""", (origin,)).fetchone() if (row and row[6] == self.global_fp and row[5] == self._fingerprint(json.loads(row[4]))): cached[origin] = (origin, *row[:4]) else: todo.append(origin) if todo: log(f"ports: {len(cached)} cached, running make -V for {len(todo)} " f"with {self.jobs} jobs ...") progress = Progress("ports make -V", len(todo), every=15.0) with cf.ThreadPoolExecutor(self.jobs) as pool: futures = [pool.submit(_make_vars, str(self.portsdir), o) for o in todo] try: for n, future in enumerate(cf.as_completed(futures), 1): origin, pkgname, cpe, maintainer, files, error = future.result() local = sorted({f for f in files if not self._is_global(f)}) self.db.execute("INSERT OR REPLACE INTO ports VALUES (?,?,?,?,?,?,?,?)", (origin, pkgname, cpe, maintainer, error, json.dumps(local), self._fingerprint(local), self.global_fp)) cached[origin] = (origin, pkgname, cpe, maintainer, error) if n % 200 == 0: self.db.commit() progress.step() except KeyboardInterrupt: for f in futures: f.cancel() self.db.commit() raise self.db.commit() return [cached[o] for o in origins] # -------------------------------------------------------------------------- # main # -------------------------------------------------------------------------- def _isodate(text): """--since is compared as a string, so it has to be zero-padded ISO: with "2026-6-1" every date in 2026-06..2026-09 would sort below it and vanish.""" try: return datetime.date.fromisoformat(text).isoformat() except ValueError: raise argparse.ArgumentTypeError(f"{text!r} is not a YYYY-MM-DD date") def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("origins", nargs="*", help="port origins, e.g. security/openssl or devel/py-foo@py311") ap.add_argument("-f", "--file", help="read port origins from FILE ('-' for stdin)") ap.add_argument("-a", "--all", action="store_true", help="check the whole ports tree") ap.add_argument("--portsdir", default=os.environ.get("PORTSDIR", "/usr/ports")) ap.add_argument("--cache-dir", default=os.path.join( os.environ.get("XDG_CACHE_HOME", "~/.cache"), "portcve"), help="repositories and indexes (default: $XDG_CACHE_HOME/portcve " "or ~/.cache/portcve)") ap.add_argument("--kb-repo", default=os.environ.get("OPENCVE_KB_REPO"), help="use an existing opencve-kb clone (default: CACHE_DIR/opencve-kb)") ap.add_argument("--nvd-repo", default=os.environ.get("OPENCVE_NVD_REPO"), help="use an existing opencve-nvd clone (default: CACHE_DIR/opencve-nvd)") ap.add_argument("--pull", action="store_true", help="fetch the latest state of both repositories first") ap.add_argument("-j", "--jobs", type=int, default=os.cpu_count() or 4, help="parallel make/parsing jobs (default: number of CPUs)") ap.add_argument("-m", "--maintainer", action="append", metavar="EMAIL", help="only ports with this maintainer; case-insensitive, repeatable, " "shell wildcards allowed (e.g. '*@FreeBSD.org')") ap.add_argument("--since", metavar="YYYY-MM-DD", type=_isodate, help="ignore CVEs published before this date") ap.add_argument("--include-disputed", action="store_true", help="also list CVEs tagged as disputed (hidden by default)") ap.add_argument("--hide-unknown", action="store_true", help="don't list linked CVEs without usable version data") ap.add_argument("-v", "--verbose", action="store_true", help="show which CNA/NVD range each finding matched") ap.add_argument("--json", action="store_true", help="JSON output") args = ap.parse_args() jobs = max(1, args.jobs) cache_dir = Path(args.cache_dir).expanduser() cache_dir.mkdir(parents=True, exist_ok=True) db = sqlite3.connect(cache_dir / "state.sqlite") # Repositories in the cache directory are cloned automatically, custom paths aren't clone_missing = [name for name, custom in (("opencve-kb", args.kb_repo), ("opencve-nvd", args.nvd_repo)) if not custom] data = OpenCVEData(Path(args.kb_repo or cache_dir / "opencve-kb").expanduser(), Path(args.nvd_repo or cache_dir / "opencve-nvd").expanduser(), db, jobs, clone_missing) if args.pull: data.pull() data.update_index() # Which ports, and their CPEs origins = list(args.origins) if args.file: fh = sys.stdin if args.file == "-" else open(args.file) origins += [l.strip() for l in fh if l.strip() and not l.startswith("#")] portsdir = Path(args.portsdir).resolve() if args.all: origins += tree_origins(portsdir) if origins: origins = list(dict.fromkeys(origins)) try: scanned = PortsScanner(portsdir, db, jobs).scan(origins) except KeyboardInterrupt: log("interrupted - finished ports are cached, run again to continue") sys.exit(EXIT_INTERRUPTED) packages = [(o, p, c, m) for o, p, c, m, e in scanned if not e] errors = {o: e for o, p, c, m, e in scanned if e} else: packages, errors = installed_packages(), {} if args.maintainer: patterns = [m.strip().lower() for m in args.maintainer] packages = [pkg for pkg in packages if pkg[3] and any( fnmatch.fnmatchcase(pkg[3].lower(), pattern) for pattern in patterns)] errors = {} # make failed, so the maintainer of these ports is unknown if not packages: log(f"no ports found for maintainer {', '.join(args.maintainer)}") groups, no_cpe = defaultdict(list), [] port_info = {} packages = list({(p[0], p[1]): p for p in packages}.values()) for origin, pkgname, cpe, maintainer in packages: key = (origin, pkgname) # unique even for several packages of one origin f = split_cpe(cpe) if cpe else [] if len(f) < 6: no_cpe.append(origin) continue vendor, product, version = f[3].lower(), f[4].lower(), f[5] groups[(vendor, product)].append((key, version)) port_info[key] = (cpe, maintainer) # Match, one vendor:product group per task, largest first work = sorted(((vp, data.cves_for(*vp), ports) for vp, ports in groups.items()), key=lambda w: -len(w[1]) * len(w[2])) no_cves = [o for vp, cves, ports in work if not cves for o, _ in ports] work = [w for w in work if w[1]] findings = {} progress = Progress("matching products", len(work), every=15.0) with cf.ProcessPoolExecutor(jobs) as pool: futures = [pool.submit(_match_group, str(data.nvd_repo), v, p, cves, ports, args.since, args.include_disputed) for (v, p), cves, ports in work] for future in cf.as_completed(futures): findings.update(future.result()) progress.step() report = [] for key in sorted(findings): res = findings[key] if args.hide_unknown: res["unknown"] = [] if res["affected"] or res["unknown"]: cpe, maintainer = port_info[key] report.append({"origin": key[0], "pkgname": key[1], "maintainer": maintainer, "cpe": cpe, **res}) affected_ports = sum(1 for e in report if e["affected"]) disputed_hidden = sum(res["disputed_hidden"]["affected"] + ( 0 if args.hide_unknown else res["disputed_hidden"]["unknown"]) for res in findings.values()) rejected_hidden = sum(res.get("rejected", 0) for res in findings.values()) summary = {"ports": len(packages) + len(errors), "with_cpe": len(port_info), "affected": affected_ports, "only_unknown": len(report) - affected_ports, "no_cves_known": len(no_cves), "no_cpe": len(no_cpe), "make_errors": len(errors), "disputed_hidden": disputed_hidden, "rejected_hidden": rejected_hidden, "maintainer_filter": args.maintainer or []} if args.json: json.dump({"summary": summary, "ports": report, "make_errors": errors}, sys.stdout, indent=2) print() else: for e in report: print(f"{e['origin']} {e['pkgname']} <{e['maintainer'] or 'unknown'}>") rows = [("!", c) for c in e["affected"]] + [("?", c) for c in e["unknown"]] rows.sort(key=lambda r: (r[0] != "!", -(r[1]["score"] or 0), r[1]["cve_id"])) for mark, c in rows: score = f"{c['score']:4.1f}" if c["score"] is not None else " -" flags = (" KEV" if c["kev"] else "") + (" DISPUTED" if c["disputed"] else "") print(f" {mark} {c['cve_id']:<16} {score}{flags} {c['title'][:90]}") if args.verbose: print(f" {c['why']}") print(f"\n{summary['ports']} port(s): {summary['with_cpe']} with CPE, " f"{summary['affected']} affected, {summary['only_unknown']} with only " f"unconfirmed CVEs, {summary['no_cves_known']} with no known CVEs; " f"{summary['no_cpe']} without CPE info, {summary['make_errors']} make error(s)" + (f" [maintainer: {', '.join(args.maintainer)}]." if args.maintainer else ".")) if disputed_hidden: print(f"{disputed_hidden} finding(s) for disputed CVEs hidden (--include-disputed).") if rejected_hidden: print(f"{rejected_hidden} link(s) to CVEs rejected by their CNA skipped.") print(" ! = affected (CNA ranges, else NVD CPE data) " "? = linked CVE without usable version data") db.close() sys.exit(EXIT_AFFECTED if affected_ports else EXIT_OK) if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit(EXIT_INTERRUPTED) except BrokenPipeError: os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno()) sys.exit(EXIT_INTERRUPTED) except SystemExit: raise except Exception as exc: log(f"error: {type(exc).__name__}: {exc}") sys.exit(EXIT_ERROR)