Migrate ubuntu usn importer

Signed-off-by: Tushar Goel <[email protected]>

Author: TG1999

vulnerabilities/importers/__init__.py

@@ -23,6 +23,7 @@
 from vulnerabilities.importers import redhat
 from vulnerabilities.importers import retiredotnet
 from vulnerabilities.importers import ubuntu
+from vulnerabilities.importers import ubuntu_usn
 
 IMPORTERS_REGISTRY = [
     nginx.NginxImporter,
@@ -41,6 +42,7 @@
     debian_oval.DebianOvalImporter,
     npm.NpmImporter,
     retiredotnet.RetireDotnetImporter,
+    ubuntu_usn.UbuntuUSNImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/ubuntu_usn.py

@@ -15,55 +15,42 @@
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
-from vulnerabilities.utils import create_etag
 from vulnerabilities.utils import is_cve
 
 
 class UbuntuUSNImporter(Importer):
-    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)))
+    db_url = "https://usn.ubuntu.com/usn-db/database-all.json.bz2"
+    spdx_license_expression = "GPL"
 
-        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
+    def advisory_data(self):
+        usn_db = fetch(self.db_url)
+        yield from self.to_advisories(usn_db=usn_db)
 
     @staticmethod
     def to_advisories(usn_db):
-        advisories = []
         for usn in usn_db:
-            reference = get_usn_references(usn_db[usn]["id"])
-            for cve in usn_db[usn].get("cves", [""]):
+            usn_data = usn_db[usn]
+            references = get_usn_references(usn_data.get("id"))
+            for cve in usn_data.get("cves", []):
                 # The db sometimes contains entries like
                 # {'cves': ['python-pgsql vulnerabilities', 'CVE-2006-2313', 'CVE-2006-2314']}
                 # This `if` filters entries like 'python-pgsql vulnerabilities'
                 if not is_cve(cve):
                     cve = ""
 
-                advisories.append(
-                    AdvisoryData(
-                        vulnerability_id=cve,
-                        summary="",
-                        references=[reference],
-                    )
+                if not cve:
+                    continue
+                yield AdvisoryData(
+                    aliases=[cve],
+                    summary="",
+                    references=references,
                 )
 
-        return advisories
-
 
 def get_usn_references(usn_id):
-    return Reference(reference_id="USN-" + usn_id, url="https://usn.ubuntu.com/{}/".format(usn_id))
+    if not usn_id:
+        return []
+    return [Reference(reference_id=f"USN-{usn_id}", url=f"https://usn.ubuntu.com/{usn_id}/")]
 
 
 def fetch(url):

vulnerabilities/tests/conftest.py

@@ -42,6 +42,5 @@ def no_rmtree(monkeypatch):
     "test_suse_backports.py",
     "test_suse.py",
     "test_suse_scores.py",
-    "test_ubuntu_usn.py",
     "test_upstream.py",
 ]

vulnerabilities/tests/test_data/ubuntu_usn_db/ubuntu-usn-expected.json

@@ -0,0 +1,32 @@
+[
+  {
+    "aliases": [
+      "CVE-2009-0698"
+    ],
+    "summary": "",
+    "affected_packages": [],
+    "references": [
+      {
+        "reference_id": "USN-763-1",
+        "url": "https://usn.ubuntu.com/763-1/",
+        "severities": []
+      }
+    ],
+    "date_published": null
+  },
+  {
+    "aliases": [
+      "CVE-2009-1274"
+    ],
+    "summary": "",
+    "affected_packages": [],
+    "references": [
+      {
+        "reference_id": "USN-763-1",
+        "url": "https://usn.ubuntu.com/763-1/",
+        "severities": []
+      }
+    ],
+    "date_published": null
+  }
+]

vulnerabilities/tests/test_ubuntu_usn.py

@@ -17,71 +17,21 @@
 
 from packageurl import PackageURL
 
-import vulnerabilities.importers.ubuntu_usn as ubuntu_usn
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import Reference
+from vulnerabilities.importers.ubuntu_usn import UbuntuUSNImporter
+from vulnerabilities.tests import util_tests
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-TEST_DATA = os.path.join(BASE_DIR, "test_data/", "ubuntu_usn_db", "database-all.json.bz2")
-
-
-class TestUbuntuUSNImporter(TestCase):
-    @classmethod
-    def setUpClass(cls):
-        data_src_cfg = {"etags": {}, "db_url": "http://exampledb.com"}
-        cls.data_src = ubuntu_usn.UbuntuUSNImporter(batch_size=1, config=data_src_cfg)
-        with open(TEST_DATA, "rb") as f:
-            cls.raw_data = f.read()
-            cls.db = json.loads(bz2.decompress(cls.raw_data))
-
-    def test_get_usn_references(self):
-
-        eg_usn = "435-1"
-        expected_references = Reference(
-            reference_id="USN-435-1", url="https://usn.ubuntu.com/435-1/"
-        )
-
-        found_references = ubuntu_usn.get_usn_references(eg_usn)
-        assert found_references == expected_references
-
-    def test_fetch(self):
-
-        mock_response = MagicMock()
-        mock_response.content = self.raw_data
-        with patch("vulnerabilities.importers.ubuntu_usn.requests.get", return_value=mock_response):
-            assert ubuntu_usn.fetch("www.db.com") == self.db
-
-    def test_to_advisories(self):
-
-        expected_advisories = [
-            Advisory(
-                summary="",
-                references=[
-                    Reference(url="https://usn.ubuntu.com/763-1/", reference_id="USN-763-1")
-                ],
-                vulnerability_id="CVE-2009-0698",
-            ),
-            Advisory(
-                summary="",
-                references=[
-                    Reference(url="https://usn.ubuntu.com/763-1/", reference_id="USN-763-1")
-                ],
-                vulnerability_id="CVE-2009-1274",
-            ),
-        ]
-        found_advisories = self.data_src.to_advisories(self.db)
-
-        found_advisories = list(map(Advisory.normalized, found_advisories))
-        expected_advisories = list(map(Advisory.normalized, expected_advisories))
-        assert sorted(found_advisories) == sorted(expected_advisories)
-
-    def test_create_etag(self):
-        assert self.data_src.config.etags == {}
-
-        mock_response = MagicMock()
-        mock_response.headers = {"etag": "2131151243&2191"}
-
-        with patch("vulnerabilities.importers.ubuntu.requests.head", return_value=mock_response):
-            assert self.data_src.create_etag("https://example.org")
-            assert self.data_src.config.etags == {"https://example.org": "2131151243&2191"}
-            assert not self.data_src.create_etag("https://example.org")
+TEST_DIR = os.path.join(BASE_DIR, "test_data", "ubuntu_usn_db")
+
+
+def test_ubuntu_usn():
+    database = os.path.join(TEST_DIR, "database-all.json.bz2")
+    with open(database, "rb") as f:
+        raw_data = f.read()
+        db = json.loads(bz2.decompress(raw_data))
+        advisories = UbuntuUSNImporter().to_advisories(db)
+        expected_file = os.path.join(TEST_DIR, f"ubuntu-usn-expected.json")
+        result = [data.to_dict() for data in list(advisories)]
+        util_tests.check_results_against_json(result, expected_file)