Add vulnerability scanning on top of the SBOM
The SBOM was generated and installed onto every server, and nothing read it. "Are we exposed to this advisory" was still a person going and looking, which meant it happened when somebody remembered to. scripts/scan_sbom.py queries OSV.dev by purl - no account, no API key, no agent, no vulnerability database to maintain on a box that cannot reach the internet. The SBOM is already the input OSV wants, so this is a lookup against what shipped rather than a second scan of the tree that could disagree with it. Three decisions worth recording: Shipped and build-only findings are both reported, but only shipped ones can fail a build. Two thirds of the components here are build tooling the SBOM already marks scope 'excluded', and the highest-severity match in the whole tree today is a CRITICAL against the test runner. A gate that fails a release over that trains everybody to bypass it, and then it catches nothing. Severity is taken from the advisory, never derived. Where there is only a CVSS vector the report says UNKNOWN, which is shown and never gates. A guessed score is worse than an honest gap because somebody would act on it. The allowlist is a record rather than a mute button: each entry carries the reason, the accepter, the date, and what would invalidate the reasoning. The first entry is a worked example - a PyJWT algorithm-confusion advisory that needs both mixed algorithm families and a raw JSON JWK as the key, and this application configures neither. The report ends with a per-component upgrade rollup, since that is the part somebody acts on. A target on a different major version is labelled rather than listed beside patch bumps as though the two cost the same. Current state: 8 shipped components have fixes available, none of them exploitable through a path this application opens today.
This commit is contained in:
357
scripts/scan_sbom.py
Normal file
357
scripts/scan_sbom.py
Normal file
@@ -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())
|
||||
27
scripts/vulnerability-allowlist.json
Normal file
27
scripts/vulnerability-allowlist.json
Normal file
@@ -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."
|
||||
}
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user