diff --git a/scripts/scan_sbom.py b/scripts/scan_sbom.py new file mode 100644 index 00000000..c6b7b300 --- /dev/null +++ b/scripts/scan_sbom.py @@ -0,0 +1,357 @@ +#!/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 upgrade_plan(findings): + """[(component, version, count)] - one row per component that has a fix. + + The highest fix version across that component's advisories, because + upgrading once to the top of the list clears the ones below it. + """ + by_component = {} + for f in findings: + if f['allowlisted'] or not f['fixedin']: + continue + name = f['component'] + best, count, current = by_component.get(name, (None, 0, f['version'])) + for version in f['fixedin']: + if best is None or version_key(version) > version_key(best): + best = version + 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, highest fix seen') + 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()) diff --git a/scripts/vulnerability-allowlist.json b/scripts/vulnerability-allowlist.json new file mode 100644 index 00000000..8781ba7b --- /dev/null +++ b/scripts/vulnerability-allowlist.json @@ -0,0 +1,27 @@ +{ + "_comment": [ + "Advisories reviewed and accepted, with the reason and the date.", + "This is a RECORD, not a mute button: an entry says somebody looked at", + "the advisory and worked out why it does not apply here. Anyone can", + "re-read the reason and disagree.", + "", + "An entry needs an id, a reason that names the precondition that is", + "absent, who accepted it, and when. 'reviewuntil' is when the reasoning", + "should be checked again - a reason stops being true when the code that", + "made it true changes.", + "", + "Do NOT add an entry to make a build pass. Upgrading the dependency is", + "almost always cheaper than defending a decision to skip it." + ], + "accepted": [ + { + "id": "GHSA-xgmm-8j9v-c9wx", + "component": "pyjwt", + "reason": "Needs a verifier configured with BOTH symmetric and asymmetric algorithms and a raw JSON JWK passed as key=. Neither is present: auth goes through flask-jwt-extended with no JWT_ALGORITHM or JWT_DECODE_ALGORITHMS override anywhere in the tree, so it is HS256-only, and the key is the JWT_SECRET_KEY string. Upstream rates the attack complexity High for exactly this reason. Upgrade to pyjwt 2.13.0 anyway at the next dependency bump; this entry only says it is not urgent.", + "acceptedby": "DT team, reference site", + "accepted": "2026-08-29", + "reviewuntil": "2026-11-29", + "invalidatedby": "Setting JWT_ALGORITHM or JWT_DECODE_ALGORITHMS to anything asymmetric, or moving to SSO-issued tokens verified against a JWK." + } + ] +} diff --git a/tests/test_scan_sbom.py b/tests/test_scan_sbom.py new file mode 100644 index 00000000..de81b0bb --- /dev/null +++ b/tests/test_scan_sbom.py @@ -0,0 +1,194 @@ +"""The vulnerability scanner's judgement, without touching the network. + +scan_sbom.py decides three things that matter: whether a finding can stop a +release, which version clears it, and whether somebody already accepted it. +Each is pure given an OSV entry, so each is tested against a hand-written entry +rather than a live query. The network call itself is not tested - if OSV is +down the script says so and exits 2, which is the only sensible behaviour and +not worth a mock. +""" +import importlib.util +import json +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parents[1] +SCANNER = REPO / 'scripts' / 'scan_sbom.py' + + +def load(): + """Import the script by path - scripts/ is not a package.""" + spec = importlib.util.spec_from_file_location('scan_sbom', SCANNER) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture(scope='module') +def scan(): + return load() + + +class TestSeverity: + def test_word_is_taken_from_the_database(self, scan): + entry = {'database_specific': {'severity': 'HIGH'}} + assert scan.severity_of(entry) == 'HIGH' + + def test_lowercase_is_accepted(self, scan): + assert scan.severity_of( + {'database_specific': {'severity': 'critical'}}) == 'CRITICAL' + + def test_a_cvss_vector_alone_is_unknown_not_guessed(self, scan): + # Deliberate: parsing a vector into a score we might get wrong is worse + # than an honest gap, because somebody would act on the wrong number. + entry = {'severity': [{'type': 'CVSS_V3', + 'score': 'CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/' + 'C:H/I:H/A:H'}]} + assert scan.severity_of(entry) == 'UNKNOWN' + + def test_missing_severity_is_unknown(self, scan): + assert scan.severity_of({}) == 'UNKNOWN' + + def test_unknown_never_reaches_the_gate(self, scan): + assert not scan.at_or_above('UNKNOWN', scan.GATE_AT) + + def test_gate_floor_and_above(self, scan): + assert scan.at_or_above('HIGH', 'HIGH') + assert scan.at_or_above('CRITICAL', 'HIGH') + assert not scan.at_or_above('MODERATE', 'HIGH') + assert not scan.at_or_above('LOW', 'HIGH') + + +class TestPurl: + def test_npm(self, scan): + assert scan.parse_purl('pkg:npm/axios@1.13.2') == ('npm', 'axios') + + def test_pypi_maps_to_the_osv_name(self, scan): + assert scan.parse_purl('pkg:pypi/pyjwt@2.12.1') == ('PyPI', 'pyjwt') + + def test_scoped_npm_name_survives(self, scan): + assert scan.parse_purl('pkg:npm/%40vue/compiler-sfc@3.5.0') == ( + 'npm', '@vue/compiler-sfc') + + def test_no_version_is_fine(self, scan): + assert scan.parse_purl('pkg:npm/axios') == ('npm', 'axios') + + +class TestFixedVersions: + def test_reads_the_fixed_event(self, scan): + entry = {'affected': [{ + 'package': {'ecosystem': 'PyPI', 'name': 'pyjwt'}, + 'ranges': [{'events': [{'introduced': '0'}, {'fixed': '2.13.0'}]}], + }]} + assert scan.fixed_versions(entry, 'pkg:pypi/pyjwt@2.12.1') == ['2.13.0'] + + def test_another_packages_fix_is_not_attributed_here(self, scan): + # An advisory covering several packages must not lend one package's + # fix version to another - that would send somebody to a wrong version. + entry = {'affected': [ + {'package': {'ecosystem': 'npm', 'name': 'other'}, + 'ranges': [{'events': [{'fixed': '9.9.9'}]}]}, + {'package': {'ecosystem': 'npm', 'name': 'axios'}, + 'ranges': [{'events': [{'fixed': '1.16.0'}]}]}, + ]} + assert scan.fixed_versions(entry, 'pkg:npm/axios@1.13.2') == ['1.16.0'] + + def test_same_name_in_another_ecosystem_is_ignored(self, scan): + entry = {'affected': [{ + 'package': {'ecosystem': 'npm', 'name': 'pyjwt'}, + 'ranges': [{'events': [{'fixed': '9.9.9'}]}], + }]} + assert scan.fixed_versions(entry, 'pkg:pypi/pyjwt@2.12.1') == [] + + def test_no_fix_published_is_empty(self, scan): + entry = {'affected': [{ + 'package': {'ecosystem': 'npm', 'name': 'axios'}, + 'ranges': [{'events': [{'introduced': '0'}]}], + }]} + assert scan.fixed_versions(entry, 'pkg:npm/axios@1.13.2') == [] + + +class TestVersionOrder: + def test_ten_is_above_nine(self, scan): + # A string compare puts 1.9.0 above 1.10.0, which would recommend + # downgrading. + assert scan.version_key('1.10.0') > scan.version_key('1.9.0') + + def test_prerelease_does_not_displace_a_release(self, scan): + assert scan.version_key('2.0.0') > scan.version_key('2.0.0-rc1') + + def test_versions_with_different_segment_counts_compare(self, scan): + # OSV fix versions are not normalised, so a two-segment and a + # three-segment version land in the same sort. + assert scan.version_key('2.0.1') > scan.version_key('2.0') + assert scan.version_key('1.16.0') > scan.version_key('1.9') + + +class TestUpgradePlan: + def _finding(self, component, version, fixedin, allowlisted=False): + return {'component': component, 'version': version, 'fixedin': fixedin, + 'allowlisted': allowlisted} + + def test_highest_fix_wins_and_counts_what_it_clears(self, scan): + findings = [ + self._finding('axios 1.13.2', '1.13.2', ['1.13.5']), + self._finding('axios 1.13.2', '1.13.2', ['1.16.0']), + self._finding('axios 1.13.2', '1.13.2', ['1.15.1']), + ] + assert scan.upgrade_plan(findings) == [('axios 1.13.2', '1.16.0', 3, False)] + + def test_a_major_jump_is_flagged(self, scan): + findings = [self._finding('nanoid 3.3.11', '3.3.11', ['5.1.16'])] + assert scan.upgrade_plan(findings)[0][3] is True + + def test_a_patch_bump_is_not_flagged(self, scan): + findings = [self._finding('pyjwt 2.12.1', '2.12.1', ['2.13.0'])] + assert scan.upgrade_plan(findings)[0][3] is False + + def test_allowlisted_findings_do_not_ask_for_an_upgrade(self, scan): + findings = [self._finding('pyjwt 2.12.1', '2.12.1', ['2.13.0'], + allowlisted=True)] + assert scan.upgrade_plan(findings) == [] + + def test_a_finding_with_no_fix_is_omitted(self, scan): + findings = [self._finding('thing 1.0', '1.0', [])] + assert scan.upgrade_plan(findings) == [] + + +class TestAllowlist: + def test_the_shipped_allowlist_parses_and_is_complete(self, scan): + """Every entry must carry the reasoning, or it is just a mute.""" + data = json.loads(scan.ALLOWLIST.read_text(encoding='utf8')) + for entry in data.get('accepted', []): + for field in ('id', 'reason', 'acceptedby', 'accepted'): + assert entry.get(field), '%s missing %s' % (entry, field) + assert len(entry['reason']) > 40, ( + 'reason for %s is too short to be a reason' % entry['id']) + + def test_allowlist_is_keyed_by_advisory_id(self, scan): + allowed = scan.load_allowlist() + for key, entry in allowed.items(): + assert key == entry['id'] + + +class TestScopeSeparation: + def test_build_only_components_are_read_as_not_shipped(self, scan, tmp_path): + sbom = tmp_path / 'sbom.json' + sbom.write_text(json.dumps({'components': [ + {'name': 'axios', 'version': '1.0', 'purl': 'pkg:npm/axios@1.0'}, + {'name': 'vite', 'version': '5.0', 'purl': 'pkg:npm/vite@5.0', + 'scope': 'excluded'}, + ]}), encoding='utf8') + components = scan.load_components(sbom) + shipped = {name: ship for _p, name, _v, ship in components} + assert shipped == {'axios': True, 'vite': False} + + def test_a_component_without_a_purl_is_skipped(self, scan, tmp_path): + # OSV is queried by purl; a component without one cannot be looked up, + # and inventing an identifier would produce confident wrong answers. + sbom = tmp_path / 'sbom.json' + sbom.write_text(json.dumps({'components': [ + {'name': 'local-thing', 'version': '1.0'}, + ]}), encoding='utf8') + assert scan.load_components(sbom) == []