Refactor codebase and tests to treat Advisory class mutable

Signed-off-by: Shivam Sandbhor <shivam.sandbhor@gmail.com>

Author: sbs2001

vulnerabilities/data_source.py

@@ -78,19 +78,10 @@ class Advisory:
     """
 
     summary: str
-    impacted_package_urls: Iterable[PackageURL]
+    vulnerability_id: Optional[str] = None
+    impacted_package_urls: Iterable[PackageURL] = dataclasses.field(default_factory=list)
     resolved_package_urls: Iterable[PackageURL] = dataclasses.field(default_factory=list)
     vuln_references: List[Reference] = dataclasses.field(default_factory=list)
-    vulnerability_id: Optional[str] = None
-
-    def __hash__(self):
-        s = "{}{}{}{}".format(
-            self.summary,
-            ''.join(sorted([str(p) for p in self.impacted_package_urls])),
-            ''.join(sorted([str(p) for p in self.resolved_package_urls])),
-            self.vulnerability_id,
-        )
-        return hash(s)
 
 
 class InvalidConfigurationError(Exception):
@@ -209,7 +200,7 @@ def batch_advisories(self, advisories: List[Advisory]) -> Set[Advisory]:
 
         while advisories:
             b, advisories = advisories[: self.batch_size], advisories[self.batch_size:]
-            yield set(b)
+            yield b
 
 
 @dataclasses.dataclass

vulnerabilities/importers/alpine_linux.py

@@ -187,7 +187,7 @@ def _load_advisories(
                         impacted_package_urls=[],
                         resolved_package_urls=resolved_purls,
                         vuln_references=references,
-                        vulnerability_id=vuln_ids[0] if vuln_ids[0] != "CVE-????-?????" else None,
+                        vulnerability_id=vuln_ids[0] if vuln_ids[0] != "CVE-????-?????" else "",
                     )
                 )
 

vulnerabilities/importers/rust.py

@@ -75,11 +75,11 @@ def _load_advisories(self, files) -> Set[Advisory]:
 
         while files:
             batch, files = files[: self.batch_size], files[self.batch_size:]
-            advisories = set()
+            advisories = []
             for path in batch:
                 advisory = self._load_advisory(path)
                 if advisory:
-                    advisories.add(advisory)
+                    advisories.append(advisory)
             yield advisories
 
     def collect_packages(self, paths):

vulnerabilities/tests/test_alpine.py

@@ -30,6 +30,7 @@
 
 from vulnerabilities.data_source import Advisory, Reference
 from vulnerabilities.importers.alpine_linux import AlpineDataSource
+from vulnerabilities.tests.utils import advisories_are_equal
 
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -141,7 +142,7 @@ def test__process_link(self):
                         url="https://xenbits.xen.org/xsa/advisory-295.html", reference_id="XSA-295"
                     )
                 ],
-                vulnerability_id=None,
+                vulnerability_id="",
             ),
         ]
         mock_requests = MagicMock()
@@ -151,4 +152,4 @@ def test__process_link(self):
             mock_content.content = f
             with patch("vulnerabilities.importers.alpine_linux.requests", new=mock_requests):
                 found_advisories = self.data_source._process_link("does not matter")
-                assert expected_advisories == found_advisories
+                assert advisories_are_equal(expected_advisories, found_advisories)

vulnerabilities/tests/test_apache_kafka.py

@@ -31,6 +31,7 @@
 from vulnerabilities.package_managers import GitHubTagsAPI
 from vulnerabilities.importers.apache_kafka import ApacheKafkaDataSource
 from vulnerabilities.importers.apache_kafka import to_version_ranges
+from vulnerabilities.tests.utils import advisories_are_equal
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 TEST_DATA = os.path.join(BASE_DIR, "test_data", "apache_kafka", "cve-list.html")
@@ -59,7 +60,7 @@ def test_to_advisory(self):
         data_source.version_api = GitHubTagsAPI(
             cache={"apache/kafka": ["2.1.2", "0.10.2.2"]}
         )
-        expected_data = [
+        expected_advisories = [
             Advisory(
                 summary="In Apache Kafka versions between 0.11.0.0 and 2.1.0, it is possible to "
                 "manually\n    craft a Produce request which bypasses transaction/idempotent ACL "
@@ -97,6 +98,6 @@ def test_to_advisory(self):
             )
         ]
         with open(TEST_DATA) as f:
-            found_data = data_source.to_advisory(f)
+            found_advisories = data_source.to_advisory(f)
 
-        assert found_data == expected_data
+        assert advisories_are_equal(found_advisories, expected_advisories)

vulnerabilities/tests/test_apache_tomcat.py

@@ -30,6 +30,7 @@
 from vulnerabilities.data_source import Advisory
 from vulnerabilities.data_source import Reference
 from vulnerabilities.importers.apache_tomcat import ApacheTomcatDataSource
+from vulnerabilities.tests.utils import advisories_are_equal
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 TEST_DATA = os.path.join(BASE_DIR, "test_data", "apache_tomcat", "security-9.html")
@@ -47,8 +48,7 @@ def setUpClass(cls):
         cls.data_src.version_api = mock_api
 
     def test_to_advisories(self):
-        expected_advisories = sorted(
-            [
+        expected_advisories = [
                 Advisory(
                     summary="",
                     impacted_package_urls=[
@@ -209,19 +209,8 @@ def test_to_advisories(self):
                     ],
                     vulnerability_id="CVE-2016-0714",
                 ),
-            ],
-            key=lambda x: x.vulnerability_id,
-        )
+            ]
 
         with open(TEST_DATA) as f:
             found_advisories = self.data_src.to_advisories(f)
-
-        found_advisories.sort(key=lambda x: x.vulnerability_id)
-
-        for i in range(len(found_advisories)):
-            found_advisories[i].vuln_references.sort(key=lambda x: x.url)
-
-        for i in range(len(expected_advisories)):
-            expected_advisories[i].vuln_references.sort(key=lambda x: x.url)
-
-        assert expected_advisories == found_advisories
+        assert advisories_are_equal(expected_advisories, found_advisories)

vulnerabilities/tests/test_debian_oval.py

@@ -12,6 +12,7 @@
 from vulnerabilities.oval_parser import OvalParser
 from vulnerabilities.importers.debian_oval import DebianOvalDataSource
 from vulnerabilities.data_source import Advisory
+from vulnerabilities.tests.utils import advisories_are_equal
 
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -43,7 +44,7 @@ def setUpClass(cls):
             '2.3.9'})
     @patch('vulnerabilities.importers.debian_oval.DebianVersionAPI.load_api', new=mock)
     def test_get_data_from_xml_doc(self, mock_write):
-        expected_data = {
+        expected_advisories = [
             Advisory(
                 summary='denial of service',
                 impacted_package_urls={
@@ -97,18 +98,17 @@ def test_get_data_from_xml_doc(self, mock_write):
                                qualifiers=OrderedDict([('distro', 'wheezy')]),
                                subpath=None)},
                 vulnerability_id='CVE-2001-1593')
-
-        }
+        ]
 
         xml_doc = ET.parse(os.path.join(TEST_DATA, "debian_oval_data.xml"))
         # Dirty quick patch to mock batch_advisories
         with patch('vulnerabilities.importers.debian_oval.DebianOvalDataSource.batch_advisories',
                    new=return_adv):
-            data = {i for i in self.debian_oval_data_src.get_data_from_xml_doc(
+            found_advisories = [i for i in self.debian_oval_data_src.get_data_from_xml_doc(
                 xml_doc,
                 {
                     "type": "deb",
                     "qualifiers": {"distro": "wheezy"}
                 })
-            }
-        assert expected_data == data
+            ]
+        assert advisories_are_equal(expected_advisories, found_advisories)

vulnerabilities/tests/test_elixir_security.py

@@ -30,6 +30,7 @@
 from vulnerabilities.data_source import Reference
 from vulnerabilities.importers.elixir_security import ElixirSecurityDataSource
 from vulnerabilities.package_managers import HexVersionAPI
+from vulnerabilities.tests.utils import advisories_are_equal
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 
@@ -62,7 +63,7 @@ def setUpClass(cls):
     def test_process_file(self):
 
         path = os.path.join(BASE_DIR, "test_data/elixir_security/test_file.yml")
-        expected_data = Advisory(
+        expected_advisory = Advisory(
             summary=('The Coherence library has "Mass Assignment"-like vulnerabilities.\n'),
             impacted_package_urls={
                 PackageURL(
@@ -132,6 +133,6 @@ def test_process_file(self):
             vulnerability_id="CVE-2018-20301",
         )
 
-        found_data = self.data_src.process_file(path)
+        found_advisory = self.data_src.process_file(path)
 
-        assert expected_data == found_data
+        assert advisories_are_equal([expected_advisory], [found_advisory])

vulnerabilities/tests/test_gentoo.py

@@ -31,6 +31,7 @@
 from vulnerabilities.importers.gentoo import GentooDataSource
 from vulnerabilities.data_source import Advisory
 from vulnerabilities.data_source import Reference
+from vulnerabilities.tests.utils import advisories_are_equal
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 TEST_DATA = os.path.join(BASE_DIR, "test_data/gentoo/glsa-201709-09.xml")
@@ -87,7 +88,7 @@ def test_cves_from_reference(self):
 
     def test_process_file(self):
 
-        exp_data = [
+        expected_advisories = [
             Advisory(
                 summary=(
                     'A command injection vulnerability in '
@@ -115,5 +116,5 @@ def test_process_file(self):
                         reference_id='GLSA-201709-09')],
                 vulnerability_id='CVE-2017-9800')]
 
-        found_data = self.data_src.process_file(TEST_DATA)
-        assert exp_data == found_data
+        found_advisories = self.data_src.process_file(TEST_DATA)
+        assert advisories_are_equal(expected_advisories, found_advisories)

vulnerabilities/tests/test_github.py

@@ -42,6 +42,7 @@
 from vulnerabilities.severity_systems import ScoringSystem
 from vulnerabilities.importers.github import GitHubTokenError
 from vulnerabilities.importers.github import query
+from vulnerabilities.tests.utils import advisories_are_equal
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 TEST_DATA = os.path.join(BASE_DIR, "test_data")
@@ -163,7 +164,7 @@ def test_process_response(self):
             resp = json.load(f)
             self.data_src.advisories = resp
 
-        expected_result = [
+        expected_advisories = [
             Advisory(
                 summary="Denial of Service in Tomcat",
                 impacted_package_urls=set(),
@@ -371,10 +372,8 @@ def test_process_response(self):
         mock_version_api = MagicMock()
         mock_version_api.package_type = "maven"
         mock_version_api.get = lambda x: {"1.2.0", "9.0.2"}
-        with patch(
-            "vulnerabilities.importers.github.MavenVersionAPI", return_value=mock_version_api
-        ):  # nopep8
+        with patch("vulnerabilities.importers.github.MavenVersionAPI", return_value=mock_version_api):  # nopep8
             with patch("vulnerabilities.importers.github.GitHubAPIDataSource.set_api"):
-                found_result = self.data_src.process_response()
+                found_advisories = self.data_src.process_response()
 
-        assert expected_result == found_result
+        assert advisories_are_equal(expected_advisories, found_advisories)