Dateien nach "importers" hochladen

importers/redhat.py

@@ -0,0 +1,48 @@
+"""Red Hat CSAF 2.0 advisory importer.
+
+Preserve source product status, remediations and product tree in JSON. Do not
+invent RPM ranges or flatten CSAF product relations to independent CPE matches.
+"""
+from .common import CVE, canonical, sha, upsert_vuln
+
+
+def process(db, obj):
+    if not isinstance(obj, dict):
+        raise ValueError('CSAF must be an object')
+    document = obj.get('document')
+    if not isinstance(document, dict) or not isinstance(document.get('tracking'), dict):
+        raise ValueError('Invalid Red Hat CSAF document/tracking')
+    tracking = document['tracking']
+    advisory_id = tracking.get('id')
+    if not isinstance(advisory_id, str) or not advisory_id.startswith(('RHSA-', 'RHBA-', 'RHEA-')):
+        raise ValueError(f'Invalid Red Hat advisory identifier: {advisory_id!r}')
+    tree = obj.get('product_tree') or {}
+    if not isinstance(tree, dict):
+        raise ValueError(f'Invalid Red Hat product tree for {advisory_id}')
+    vulnerabilities = obj.get('vulnerabilities') or []
+    if not isinstance(vulnerabilities, list):
+        raise ValueError(f'Invalid vulnerabilities for {advisory_id}')
+
+    db.execute('''INSERT INTO redhat_advisories
+      (id,title,initial_release_date,current_release_date,product_tree_json)
+      VALUES(?,?,?,?,?)''', (
+        advisory_id, document.get('title'),
+        tracking.get('initial_release_date'), tracking.get('current_release_date'),
+        canonical(tree)))
+
+    for item in vulnerabilities:
+        if not isinstance(item, dict):
+            raise ValueError(f'Unexpected Red Hat vulnerability in {advisory_id}')
+        cve = item.get('cve')
+        if not isinstance(cve, str) or not CVE.fullmatch(cve):
+            # Non-CVE bug/enhancement statements are not indexed as vulnerabilities.
+            continue
+        status = item.get('product_status') or {}
+        remediations = item.get('remediations') or []
+        if not isinstance(status, dict) or not isinstance(remediations, list):
+            raise ValueError(f'Invalid product status/remediation in {advisory_id}, {cve}')
+        upsert_vuln(db, cve)  # Never override CVEProject/NVD vulnerability state.
+        db.execute('''INSERT INTO redhat_advisory_cves
+          (id,advisory_id,cve_id,product_status_json,remediations_json)
+          VALUES(?,?,?,?,?)''', (sha(canonical([advisory_id,cve])), advisory_id, cve, canonical(status), canonical(remediations)))
+    return True