new --audit option for sbom output

query OSV API for vulnerabilities and (just) print a summary

nxdbom/__init__.py

@@ -37,6 +37,8 @@ import json
 import sys, configparser, re, codecs
 import uuid
 
+import requests
+
 
 # PkgInfo represents information about a package
 PkgInfo = namedtuple('PkgInfo', [
@@ -546,7 +548,7 @@ def fmt_bom(bom): # -> str
 
     return ''.join(outv)
 
-def fmt_bom_cyclonedx_json(bom, software_path):
+def fmt_bom_cyclonedx_json(bom, software_path, audit=False):
 
     # possible future extensions:
     # - describe patches applied to components (using components[*].pedigree.patches )
@@ -595,40 +597,54 @@ def fmt_bom_cyclonedx_json(bom, software_path):
         }
     }
     components = bom_json["components"] = []
-    for _, pkginfo in sorted(bom.items()):
-        cpe = None
-        externalReferences = []
-        if pkginfo.url:
-            externalReferences.append(
-                {
-                    'url': pkginfo.url,
-                    'type': (
-                        'vcs'
-                        if pkginfo.kind == 'git'
-                        else 'distribution'
-                    ),
-                }
-            )
-        purl_type = 'generic'
-        if pkginfo.kind == 'egg':
-            purl_type = 'pypi'
-        elif pkginfo.kind == 'gem':
-            purl_type = 'gem'
-        else:
-            cpe = f'cpe:2.3:*:*:{pkginfo.name}:{pkginfo.version}:*:*:*:*:*:*:*'
-        purl = f'pkg:{purl_type}/{pkginfo.name}@{pkginfo.version}'
-        component = {
-            'name': pkginfo.name,
-            'purl': purl,
-            'type': 'library',
-            'version': pkginfo.version,
-        }
-        if cpe:
-            component['cpe'] = cpe
-        if externalReferences:
-            component['externalReferences'] = externalReferences
-        components.append(component)
-
+    with requests.Session() as session:
+        for _, pkginfo in sorted(bom.items()):
+            cpe = None
+            externalReferences = []
+            if pkginfo.url:
+                externalReferences.append(
+                    {
+                        'url': pkginfo.url,
+                        'type': (
+                            'vcs'
+                            if pkginfo.kind == 'git'
+                            else 'distribution'
+                        ),
+                    }
+                )
+            purl_type = 'generic'
+            if pkginfo.kind == 'egg':
+                purl_type = 'pypi'
+            elif pkginfo.kind == 'gem':
+                purl_type = 'gem'
+            else:
+                cpe = f'cpe:2.3:*:*:{pkginfo.name}:{pkginfo.version}:*:*:*:*:*:*:*'
+            purl = f'pkg:{purl_type}/{pkginfo.name}@{pkginfo.version}'
+            component = {
+                'name': pkginfo.name,
+                'purl': purl,
+                'type': 'library',
+                'version': pkginfo.version,
+            }
+            if cpe:
+                component['cpe'] = cpe
+            if externalReferences:
+                component['externalReferences'] = externalReferences
+            components.append(component)
+            if audit:
+                resp = session.post(
+                    'https://api.osv.dev/v1/query',
+                    json={"package": {"purl": purl}})
+                for vuln in resp.json().get('vulns', []):
+                    reference = sorted(
+                        vuln['references'] + [{'url': 'N/A'}],
+                        key=lambda r: not r.get('type') == 'ADVISORY')[0]['url']
+                    print("WARNING: {} {} {} {}".format(
+                        pkginfo.name,
+                        reference,
+                        vuln['id'],
+                        vuln.get('summary', vuln.get('details', '')),
+                    ))
     return bom_json
 
 
@@ -638,6 +654,7 @@ def main():
         description=__doc__
     )
     parser.add_argument('-f', '--format', choices=['text', 'cyclonedx-json'], default='text')
+    parser.add_argument('--audit', action="store_true", help='Query OSV database for known vulnerabilities')
     parser.add_argument('-o', '--output',
         type=argparse.FileType('w', encoding='UTF-8'),
         default=sys.stdout)
@@ -659,7 +676,7 @@ def main():
         print(fmt_bom(bom), file=args.output)
     else:
         assert args.format == 'cyclonedx-json'
-        json.dump(fmt_bom_cyclonedx_json(bom, args.software_path), args.output, indent=True)
+        json.dump(fmt_bom_cyclonedx_json(bom, args.software_path, args.audit), args.output, indent=True)
 
 
 if __name__ == '__main__':

setup.py

@@ -13,6 +13,7 @@ setup(
     keywords    = 'Nexedi software build BOM',
 
     packages    = find_packages(),
+    requires    = [ 'requests', ],
     extras_require = {
                    'test': ['pytest'],
     },