Cam Proudlock
7ddfe4d3ef
The rollup took the highest fix version across a component's advisories, which is wrong whenever an advisory is patched on more than one release branch. nanoid is fixed on both 3.3.x and 5.x, and it is transitive under a postcss range of ^3.3.11 that no 5.x release can satisfy - so the recommendation was unreachable rather than merely large. Per advisory the target is now the lowest fix above the version in use; per component, the highest of those. Fixes on older branches are skipped rather than offered as a downgrade.
380 lines
14 KiB
Python
380 lines
14 KiB
Python
#!/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())
|