shopdb-flask/scripts/scan_sbom.py

#!/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())