ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add vulnerability scanning on top of the SBOM

Browse Source 
scripts/scan_sbom.py checks the generated SBOM against OSV.dev by purl. The SBOM
shipped to every server and nothing read it, so 'are we exposed to this advisory'
was still a person going and looking.

Shipped and build-only findings are both reported, but only shipped components
can fail a build - the highest-severity match in the tree today is against the
test runner, and a gate that fails a release over that gets bypassed and then
catches nothing. Severity is taken from the advisory rather than derived from a
CVSS vector, and an advisory with no recorded word is reported UNKNOWN and never
gates.

scripts/vulnerability-allowlist.json records reviewed advisories with the reason,
the accepter, the date, and what would invalidate the reasoning.

Documented in docs/DEPENDENCY-SECURITY.md.
This commit is contained in:
Cameron Proudlock
2026-08-29 17:19:27 -04:00
parent 6beb2afc09
commit d49b421914
3 changed files with 578 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

357
scripts/scan_sbom.py Normal file
View 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
View 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."
}
]
}

194
tests/test_scan_sbom.py Normal file
View File

@@ -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) == []