#!/usr/bin/env python3 """Check the SBOM against OSV, and say what actually needs doing. python3 scripts/scan_sbom.py scan, report, exit 0 python3 scripts/scan_sbom.py --gate exit 1 on a shipped finding python3 scripts/scan_sbom.py --json out.json machine-readable too python3 scripts/scan_sbom.py --sbom path.json scan an existing SBOM generate_sbom.py already answers "what are we running": 345 components, each with a purl and a hash. Nothing consumed that file, so "are we exposed" was still a person going and looking. This closes that. WHY OSV, AND NOT A SCANNER OSV.dev takes a purl and answers with the advisories that apply. No account, no API key, no vendor agent, no database to keep current on an air-gapped box. The SBOM is already the input it wants, so this is a lookup rather than another scan of the tree - which also means it cannot disagree with what shipped. SHIPPED VERSUS DEV-ONLY IS THE WHOLE POINT Two thirds of the components here are build tooling that never reaches a server, and the SBOM already marks them scope 'excluded'. A scanner that fails a release over a vite transitive dependency teaches everybody to ignore it within a month, so dev-only findings are REPORTED and never gate. Only what ships can stop a build. SEVERITY IS TAKEN, NOT COMPUTED An OSV entry carries whatever the upstream database recorded. Where that is a plain word (GHSA gives LOW/MODERATE/HIGH/CRITICAL) it is used as-is. Where there is only a CVSS vector, this reports UNKNOWN rather than parsing the vector into a score it might get wrong. UNKNOWN is shown and never gates: a guessed score is worse than an honest gap, because somebody would act on it. TRIAGE IS A FILE, NOT A MEMORY scripts/vulnerability-allowlist.json records advisories reviewed and accepted, each with a reason and a date. That is what turns a recurring report into a record: "we looked at this on the 5th, here is why it does not apply" outlives the person who worked it out. """ import argparse import json import pathlib import sys import urllib.error import urllib.request REPO = pathlib.Path(__file__).resolve().parent.parent DEFAULT_SBOM = REPO / 'sbom.cdx.json' ALLOWLIST = REPO / 'scripts' / 'vulnerability-allowlist.json' OSV_BATCH = 'https://api.osv.dev/v1/querybatch' OSV_VULN = 'https://api.osv.dev/v1/vulns/' # Order matters: the gate compares by index, so adding a level means deciding # where it sits rather than hoping a string comparison does the right thing. SEVERITY_ORDER = ['UNKNOWN', 'LOW', 'MODERATE', 'MEDIUM', 'HIGH', 'CRITICAL'] GATE_AT = 'HIGH' # OSV batches, but not without limit, and a 345-component POST is rude. BATCH_SIZE = 100 TIMEOUT = 40 def post(url, payload): body = json.dumps(payload).encode('utf8') request = urllib.request.Request( url, data=body, headers={'Content-Type': 'application/json'}) with urllib.request.urlopen(request, timeout=TIMEOUT) as response: return json.loads(response.read().decode('utf8')) def get(url): with urllib.request.urlopen(url, timeout=TIMEOUT) as response: return json.loads(response.read().decode('utf8')) def load_components(path): """[(purl, name, version, shipped)] from a CycloneDX document.""" document = json.loads(path.read_text(encoding='utf8')) out = [] for component in document.get('components', []): purl = component.get('purl') if not purl: continue out.append(( purl, component.get('name', purl), component.get('version', ''), component.get('scope', 'required') != 'excluded', )) return out def severity_of(entry): """The severity the upstream database recorded, or UNKNOWN. Deliberately not derived from a CVSS vector. See the module docstring. """ specific = entry.get('database_specific') or {} word = str(specific.get('severity') or '').strip().upper() if word in SEVERITY_ORDER: return word return 'UNKNOWN' # OSV names ecosystems in its own vocabulary; a purl uses another. Only the two # this SBOM produces are mapped, because a wrong guess here silently attributes # some other package's fix version to this component. PURL_ECOSYSTEM = {'npm': 'npm', 'pypi': 'PyPI'} def parse_purl(purl): """('npm', 'axios') from 'pkg:npm/axios@1.13.2'. Scopes survive.""" body = purl[4:] if purl.startswith('pkg:') else purl body = body.split('?')[0].split('#')[0] kind, _, rest = body.partition('/') name = rest.rsplit('@', 1)[0] if '@' in rest[1:] else rest from urllib.parse import unquote return PURL_ECOSYSTEM.get(kind.lower(), kind), unquote(name) def fixed_versions(entry, purl): """Versions upstream says carry the fix, for THIS package only. Reported as a list rather than a single 'upgrade to X' because an advisory that was patched on several release branches has several right answers, and picking one for the reader would be guessing at their constraints. """ ecosystem, name = parse_purl(purl) out = [] for affected in entry.get('affected', []): package = affected.get('package') or {} if package.get('ecosystem') != ecosystem: continue if str(package.get('name', '')).lower() != name.lower(): continue for range_ in affected.get('ranges', []): for event in range_.get('events', []): version = event.get('fixed') if version and version not in out: out.append(version) return out def at_or_above(severity, floor): try: return SEVERITY_ORDER.index(severity) >= SEVERITY_ORDER.index(floor) except ValueError: return False def load_allowlist(): if not ALLOWLIST.is_file(): return {} data = json.loads(ALLOWLIST.read_text(encoding='utf8')) return {entry['id']: entry for entry in data.get('accepted', [])} def query(components): """{purl: [vuln ids]} from OSV, in batches.""" found = {} for start in range(0, len(components), BATCH_SIZE): chunk = components[start:start + BATCH_SIZE] payload = {'queries': [{'package': {'purl': purl}} for purl, _n, _v, _s in chunk]} result = post(OSV_BATCH, payload) for (purl, _n, _v, _s), row in zip(chunk, result.get('results', [])): ids = [v['id'] for v in (row.get('vulns') or [])] if ids: found[purl] = ids return found def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--sbom', type=pathlib.Path, default=None) parser.add_argument('--json', type=pathlib.Path, default=None, help='also write the findings as JSON') parser.add_argument('--gate', action='store_true', help='exit 1 when a SHIPPED component has a finding at ' 'or above %s that is not allowlisted' % GATE_AT) args = parser.parse_args() sbom = args.sbom or DEFAULT_SBOM if not sbom.is_file(): # Generate rather than fail: the input is derived from files in the # repo, so there is no reason to make somebody run two commands. sys.stderr.write('No SBOM at %s - generating one.\n' % sbom) sbom = REPO / 'sbom.cdx.json' import subprocess subprocess.run([sys.executable, str(REPO / 'scripts' / 'generate_sbom.py'), str(REPO), '-o', str(sbom)], check=True) components = load_components(sbom) shipped = {purl: ship for purl, _n, _v, ship in components} names = {purl: '%s %s' % (n, v) for purl, n, v, _s in components} versions = {purl: v for purl, _n, v, _s in components} print('Scanning %d components (%d shipped) against OSV' % (len(components), sum(1 for _p, _n, _v, s in components if s))) try: hits = query(components) except (urllib.error.URLError, TimeoutError) as error: sys.stderr.write( 'Could not reach OSV: %s\n' 'This needs outbound HTTPS. Run it where the repository lives, not ' 'on an air-gapped server.\n' % error) return 2 allowed = load_allowlist() findings = [] for purl, ids in sorted(hits.items()): for vuln_id in ids: try: entry = get(OSV_VULN + vuln_id) except (urllib.error.URLError, TimeoutError): entry = {'id': vuln_id, 'summary': '(details unavailable)'} findings.append({ 'id': vuln_id, 'purl': purl, 'component': names.get(purl, purl), 'shipped': shipped.get(purl, True), 'severity': severity_of(entry), 'summary': (entry.get('summary') or '').strip(), 'version': versions.get(purl, ''), 'fixedin': fixed_versions(entry, purl), 'allowlisted': vuln_id in allowed, 'reason': (allowed.get(vuln_id) or {}).get('reason'), }) findings.sort(key=lambda f: (not f['shipped'], -SEVERITY_ORDER.index(f['severity']), f['component'])) blocking = [f for f in findings if f['shipped'] and not f['allowlisted'] and at_or_above(f['severity'], GATE_AT)] report(findings, blocking) if args.json: args.json.write_text(json.dumps( {'findings': findings, 'blocking': [f['id'] for f in blocking]}, indent=2), encoding='utf8') print('\nWrote %s' % args.json) if args.gate and blocking: return 1 return 0 def version_key(version): """Sort dotted versions numerically, so 1.10 beats 1.9. A prerelease sorts BELOW the release it precedes, so 2.0.0 beats 2.0.0-rc1. Treating '-rc1' as another dotted segment gets this backwards - the longer key wins a list comparison - and the rollup would then send somebody to a release candidate. The release marker is the last element for exactly that reason. """ text = str(version) base, _, pre = text.partition('-') parts = [] for chunk in base.split('.'): parts.append((1, int(chunk)) if chunk.isdigit() else (0, 0)) # 1 = a real release, 0 = a prerelease of it. parts.append((0,) if pre else (1,)) return parts def minimal_fix(fixedin, current): """The SMALLEST published fix above the version in use, or None. An advisory patched on several release branches lists a fix per branch: nanoid's is fixed in both 3.3.18 and 5.1.16. Taking the highest recommends a major upgrade when a patch on the current branch would do - and where the package is transitive, that major may not even be reachable, because the parent's semver range forbids it. The smallest fix above the current version is the one that actually clears the advisory. Fixes at or below the current version belong to older branches and are ignored; recommending one would be a downgrade. """ above = [v for v in fixedin if version_key(v) > version_key(current)] if not above: return None return min(above, key=version_key) def upgrade_plan(findings): """[(component, version, count, major)] - one row per component with a fix. Per advisory, the minimal upgrade that clears it; per component, the highest of those, since one upgrade has to satisfy every advisory against it. """ by_component = {} for f in findings: if f['allowlisted'] or not f['fixedin']: continue target = minimal_fix(f['fixedin'], f['version']) if target is None: continue name = f['component'] best, count, current = by_component.get(name, (None, 0, f['version'])) if best is None or version_key(target) > version_key(best): best = target by_component[name] = (best, count + 1, current) rows = [] for name, (best, count, current) in by_component.items(): # A major bump is a different job from a patch bump - it can be # ESM-only, or drop an API - so it is labelled rather than listed # beside patch upgrades as though the two cost the same. rows.append((name, best, count, major_of(best) != major_of(current))) return sorted(rows) def major_of(version): head = str(version).split('.')[0] return int(head) if head.isdigit() else None def report(findings, blocking): if not findings: print('\nNo advisories match any component. Nothing to do.') return ship = [f for f in findings if f['shipped']] dev = [f for f in findings if not f['shipped']] print('\n%d advisory match(es): %d against shipped components, %d against ' 'build-only tooling.' % (len(findings), len(ship), len(dev))) if ship: print('\nSHIPPED - these reach a server') for f in ship: mark = 'allowed' if f['allowlisted'] else '' fix = ('fix: ' + ', '.join(f['fixedin'][:3])) if f['fixedin'] else 'no fix published' print(' [%-8s] %-30s %-20s %-16s %s' % (f['severity'], f['component'][:30], f['id'], fix[:16], mark)) if f['reason']: print(' accepted: %s' % f['reason']) elif f['summary']: print(' %s' % f['summary'][:92]) if dev: print('\nBUILD ONLY - never reaches a server, never gates a release') for f in dev: print(' [%-8s] %-34s %s' % (f['severity'], f['component'][:34], f['id'])) if ship: upgrades = upgrade_plan(ship) if upgrades: print('\nWHAT TO UPGRADE - shipped components, smallest upgrade that clears them') for component, version, count, major in upgrades: note = ' MAJOR - check for breaking changes' if major else '' print(' %-30s -> %-12s clears %-3d%s' % (component[:30], version, count, note)) print('') if blocking: print('%d shipped finding(s) at or above %s and not allowlisted.' % (len(blocking), GATE_AT)) print('Fix the dependency, or record a decision in %s with a reason ' 'and a date.' % ALLOWLIST.relative_to(REPO)) else: print('Nothing shipped is at or above %s outside the allowlist.' % GATE_AT) if __name__ == '__main__': sys.exit(main())