Migrate xen importer

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/importers/__init__.py

@@ -22,6 +22,7 @@
 from vulnerabilities.importers import pysec
 from vulnerabilities.importers import redhat
 from vulnerabilities.importers import ubuntu
+from vulnerabilities.importers import xen
 
 IMPORTERS_REGISTRY = [
     nginx.NginxImporter,
@@ -39,6 +40,7 @@
     ubuntu.UbuntuImporter,
     debian_oval.DebianOvalImporter,
     npm.NpmImporter,
+    xen.XenImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/xen.py

@@ -13,60 +13,36 @@
 
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import Importer
-from vulnerabilities.importer import Reference
-from vulnerabilities.utils import create_etag
+from vulnerabilities.references import XsaReference
+from vulnerabilities.utils import fetch_response
 from vulnerabilities.utils import is_cve
 
 
 class XenImporter(Importer):
-    # CONFIG_CLASS = XenDBConfiguration
 
-    def updated_advisories(self):
-        advisories = []
-        if create_etag(data_src=self, url=self.config.db_url, etag_key="etag"):
-            advisories.extend(self.to_advisories(fetch(self.config.db_url)))
-
-        return self.batch_advisories(advisories)
-
-    def create_etag(self, url):
-        etag = requests.head(url).headers.get("etag")
-        if not etag:
-            return True
-
-        elif url in self.config.etags:
-            if self.config.etags[url] == etag:
-                return False
-
-        self.config.etags[url] = etag
-        return True
-
-    @staticmethod
-    def to_advisories(xen_db):
-        advisories = []
-        for xsa in xen_db[0]["xsas"]:
-            reference = get_xen_references(xsa["xsa"])
-            title = xsa.get("title", [""])
-            for cve in xsa.get("cve", [""]):
-                if not is_cve(cve):
-                    cve = ""
-
-                advisories.append(
-                    AdvisoryData(
-                        vulnerability_id=cve,
-                        summary=title,
-                        references=[reference],
-                    )
-                )
-        return advisories
-
-
-def get_xen_references(xsa_id):
-    return Reference(
-        reference_id="XSA-" + xsa_id,
-        url="https://xenbits.xen.org/xsa/advisory-{}.html".format(xsa_id),
-    )
-
-
-def fetch(url):
-    response = requests.get(url).content
-    return json.loads(response)
+    url = "https://xenbits.xen.org/xsa/xsa.json"
+    spdx_license_expression = ""
+    license_url = ""
+
+    def advisory_data(self):
+        data = fetch_response(self.url).json()
+        if not len(data):
+            return []
+        xsas = data[0].get("xsas") or []
+        for xsa in xsas:
+            yield from self.to_advisories(xsa)
+
+    def to_advisories(self, xsa):
+        xsa_id = xsa.get("xsa")
+        references = []
+        if xsa_id:
+            references.append(XsaReference.from_num_id(num_id=xsa_id))
+        title = xsa.get("title")
+        for cve in xsa.get("cve") or []:
+            if not is_cve(cve):
+                continue
+            yield AdvisoryData(
+                aliases=[cve],
+                summary=title,
+                references=references,
+            )

vulnerabilities/references.py

@@ -28,6 +28,16 @@ def from_id(cls, xsa_id):
             url=f"https://xenbits.xen.org/xsa/advisory-{numid}.html",
         )
 
+    @classmethod
+    def from_num_id(cls, num_id):
+        """
+        Return a new XsaReference from an XSA-XXXX id.
+        """
+        return cls(
+            reference_id=f"XSA-{num_id}",
+            url=f"https://xenbits.xen.org/xsa/advisory-{num_id}.html",
+        )
+
 
 class ZbxReference(Reference):
     """

vulnerabilities/tests/test_data/parse-advisory-xen-expected.json

@@ -0,0 +1,17 @@
+[
+  {
+    "aliases": [
+      "CVE-2012-5510"
+    ],
+    "summary": "Grant table version switch list corruption vulnerability",
+    "affected_packages": [],
+    "references": [
+      {
+        "reference_id": "XSA-26",
+        "url": "https://xenbits.xen.org/xsa/advisory-26.html",
+        "severities": []
+      }
+    ],
+    "date_published": null
+  }
+]

vulnerabilities/tests/test_data/xen_data.json

@@ -0,0 +1,42 @@
+{
+    "cve": [
+      "CVE-2012-5510"
+    ],
+    "files": [
+      {
+        "name": "xsa26-4.1.patch",
+        "project": "xen",
+        "upstream": "Xen",
+        "versions": [
+          [
+            "4.1.x"
+          ]
+        ]
+      },
+      {
+        "name": "xsa26-4.2.patch",
+        "project": "xen",
+        "upstream": "Xen",
+        "versions": [
+          [
+            "4.2.x"
+          ]
+        ]
+      },
+      {
+        "name": "xsa26-unstable.patch",
+        "project": "xen",
+        "upstream": "Xen",
+        "versions": [
+          [
+            "xen-unstable"
+          ]
+        ]
+      }
+    ],
+    "public_time": "2012-12-03 17:51",
+    "title": "Grant table version switch list corruption vulnerability",
+    "version": 3,
+    "version_time": "2012-12-03 17:51",
+    "xsa": "26"
+}

vulnerabilities/tests/test_xen.py

@@ -0,0 +1,20 @@
+import json
+import os
+
+from vulnerabilities.importers.xen import XenImporter
+from vulnerabilities.tests import util_tests
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+TEST_DATA = os.path.join(
+    BASE_DIR,
+    "test_data",
+)
+
+
+def test_xen_to_advisories():
+    with open(os.path.join(TEST_DATA, "xen_data.json")) as f:
+        raw_data = json.load(f)
+    advisories = XenImporter().to_advisories(raw_data)
+    result = [data.to_dict() for data in advisories]
+    expected_file = os.path.join(TEST_DATA, f"parse-advisory-xen-expected.json")
+    util_tests.check_results_against_json(result, expected_file)