Merge pull request #227 from sbs2001/stick_ref_ids_with_urls

Stick ref ids with urls

Author: pombredanne

vulnerabilities/data_source.py

@@ -43,6 +43,17 @@
 from vulnerabilities.oval_parser import OvalParser
 
 
+@dataclasses.dataclass
+class Reference:
+
+    url: str = ''
+    reference_id: str = ''
+
+    def __post_init__(self):
+        if not any([self.url, self.reference_id]):
+            raise TypeError
+
+
 @dataclasses.dataclass
 class Advisory:
     """
@@ -57,16 +68,14 @@ class Advisory:
     summary: str
     impacted_package_urls: Iterable[PackageURL]
     resolved_package_urls: Iterable[PackageURL] = dataclasses.field(default_factory=list)
-    reference_urls: Sequence[str] = dataclasses.field(default_factory=list)
-    reference_ids: Sequence[str] = dataclasses.field(default_factory=list)
+    vuln_references: List[Reference] = dataclasses.field(default_factory=list)
     cve_id: Optional[str] = None
 
     def __hash__(self):
-        s = '{}{}{}{}{}'.format(
+        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])),
-            ''.join(sorted(self.reference_urls)),
             self.cve_id,
         )
         return hash(s)
@@ -472,7 +481,8 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
             description = definition_data['description']
             affected_purls = set()
             safe_purls = set()
-            urls = definition_data['reference_urls']
+            references = [Reference(url=url)
+                          for url in definition_data['reference_urls']]
 
             for test_data in definition_data['test_data']:
                 for package in test_data['package_list']:
@@ -511,5 +521,5 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
                     impacted_package_urls=affected_purls,
                     resolved_package_urls=safe_purls,
                     cve_id=vuln_id,
-                    reference_urls=urls))
+                    vuln_references=references))
         return all_adv

vulnerabilities/import_runner.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2017 nexB Inc. and others. All rights reserved.
+# Copyright (c) nexB Inc. and others. All rights reserved.
 # http://nexb.com and https://github.com/nexB/vulnerablecode/
 # The VulnerableCode software is licensed under the Apache License version 2.0.
 # Data generated with VulnerableCode require an acknowledgment.
@@ -18,7 +18,7 @@
 #  OR CONDITIONS OF ANY KIND, either express or implied. No content created from
 #  VulnerableCode should be considered or used as legal advice. Consult an Attorney
 #  for any legal advice.
-#  VulnerableCode is a free software code scanning tool from nexB Inc. and others.
+#  VulnerableCode is a free software tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
 import dataclasses
@@ -51,7 +51,7 @@
 class VulnerabilityReferenceInserter:
     vulnerability: models.Vulnerability
     reference_id: Optional[str] = ''
-    url:  Optional[str] = ''
+    url: Optional[str] = ''
 
     def __post_init__(self):
         if not any([self.reference_id, self.url]):
@@ -165,45 +165,68 @@ def _process_added_advisories(data_source: DataSource) -> None:
 
 def _create_vulnerability_and_references(advisory: Advisory):
     vuln, vuln_created = _get_or_create_vulnerability(advisory)
-    vuln_references = set()
+    new_vuln_references = set()
 
     if vuln_created:
         # This means vulnerability didn't previously exist in the DB, so add
         # the references to bulk create queue  without any hesitation
-        for id_ in advisory.reference_ids:
-            vuln_references.add(VulnerabilityReferenceInserter(vulnerability=vuln, reference_id=id_))  # nopep8
-
-        for url in advisory.reference_urls:
-            vuln_references.add(VulnerabilityReferenceInserter(vulnerability=vuln, url=url))
+        for ref in advisory.vuln_references:
+            new_vuln_references.add(
+                VulnerabilityReferenceInserter(
+                    vulnerability=vuln,
+                    **dataclasses.asdict(ref)))
 
     else:
-        vuln_refs_qs = models.VulnerabilityReference.objects.filter(vulnerability=vuln)
-        vuln_ids = {ref.reference_id for ref in vuln_refs_qs}
-        vuln_urls = {ref.url for ref in vuln_refs_qs}
+        # build indexes of existing references to either update or create new ones
+        # and avoid creating duplicates
+
+        existing_vuln_references = models.VulnerabilityReference.objects.filter(vulnerability=vuln)
+        refs_by_id_url = {
+            (ref.id, ref.url): ref for ref in existing_vuln_references if ref.id and ref.url}
+        refs_by_id = {ref.id: ref for ref in existing_vuln_references if ref.id}
+        refs_by_url = {ref.url: ref for ref in existing_vuln_references if ref.url}
+
+        for adv_ref in advisory.vuln_references:
+            if adv_ref.reference_id and adv_ref.url:
+                existing_ref = refs_by_id_url.get((adv_ref.reference_id, adv_ref.url))
+
+            elif adv_ref.reference_id:
+                existing_ref = refs_by_id.get(adv_ref.reference_id)
+
+            elif adv_ref.url:
+                existing_ref = refs_by_url.get(adv_ref.url)
 
-        for id_ in advisory.reference_ids:
-            # Add the item preventing duplicates pass to through.
-            if id_ not in vuln_ids:
-                vuln_ids.add(id_)
-                vuln_references.add(VulnerabilityReferenceInserter(
-                    vulnerability=vuln, reference_id=id_))
+            if existing_ref:
+                # update ref as needed
+                if not adv_ref.reference_id or not adv_ref.url:
+                    continue
 
-        for url in advisory.reference_urls:
-            # Add the item preventing duplicates pass to through.
-            if url not in vuln_urls:
-                vuln_urls.add(url)
-                vuln_references.add(VulnerabilityReferenceInserter(
-                    vulnerability=vuln, url=url))
+                if adv_ref.reference_id and not existing_ref.reference_id:
+                    existing_ref.reference_id = adv_ref.reference_id
 
-    return vuln, vuln_created, vuln_references
+                if adv_ref.url and not existing_ref.url:
+                    existing_ref.url = adv_ref.url
+
+                existing_ref.save()
+
+            else:
+                new_vuln_references.add(
+                    VulnerabilityReferenceInserter(
+                        vulnerability=vuln,
+                        reference_id=adv_ref.reference_id,
+                        url=adv_ref.url,
+                    )
+                )
+
+    return vuln, vuln_created, new_vuln_references
 
 
 def _create_pkg_vuln_refs(vuln: models.Vulnerability, vuln_created: bool, purls: Sequence[PackageURL], is_vulnerable: bool):  # nopep8
     new_refs = set()
     for purl in purls:
         pkg, pkg_created = _get_or_create_package(purl)
         vuln_pkg_ref = PackageRelatedVulnerabilityInserter(
-                            package=pkg, vulnerability=vuln, is_vulnerable=is_vulnerable)
+            package=pkg, vulnerability=vuln, is_vulnerable=is_vulnerable)
 
         if pkg_created or vuln_created:
             new_refs.add(vuln_pkg_ref)
@@ -371,13 +394,12 @@ def _insert_vulnerabilities_and_references(batch: Set[Advisory]) -> Set[models.V
 
         vulnerabilities.add(vuln)
 
-        for id_ in advisory.reference_ids:
-            models.VulnerabilityReference.objects.get_or_create(
-                vulnerability=vuln, reference_id=id_)
+        # TODO:  Refactor this method to be atleast as efficient as
+        # one used by _process_updated_advisories
 
-        for url in advisory.reference_urls:
+        for ref in advisory.vuln_references:
             models.VulnerabilityReference.objects.get_or_create(
-                vulnerability=vuln, url=url)
+                vulnerability=vuln, reference_id=ref.reference_id, url=ref.url)
 
     return vulnerabilities
 

vulnerabilities/importers/alpine_linux.py

@@ -34,6 +34,7 @@
 
 from vulnerabilities.data_source import Advisory
 from vulnerabilities.data_source import GitDataSource
+from vulnerabilities.data_source import Reference
 
 
 def validate_schema(advisory_dict):
@@ -53,14 +54,12 @@ def validate_schema(advisory_dict):
                                     Regex(r"XSA-\d{3}"),
                                     Regex(r"ZBX-\d{4}"),
                                     Regex(r"wnpa-sec-\d{4}-\d{2}"),
-
-
                                 )
                             ],
                             "",
                             # FIXME: Remove the None when below issue gets fixed
                             # https://gitlab.alpinelinux.org/alpine/infra/alpine-secdb/-/issues/1
-                            None
+                            None,
                         ),
                     },
                 }
@@ -72,13 +71,13 @@ def validate_schema(advisory_dict):
 
 
 class AlpineDataSource(GitDataSource):
-
     def __enter__(self):
         super(AlpineDataSource, self).__enter__()
 
-        if not getattr(self, '_added_files', None):
+        if not getattr(self, "_added_files", None):
             self._added_files, self._updated_files = self.file_changes(
-                recursive=True, file_ext='yaml')
+                recursive=True, file_ext="yaml"
+            )
 
     def updated_advisories(self) -> Set[Advisory]:
         files = self._updated_files.union(self._added_files)
@@ -94,55 +93,80 @@ def _process_file(self, path) -> List[Advisory]:
         with open(path) as f:
             record = yaml.safe_load(f)
 
-            if record['packages'] is None:
+            if record["packages"] is None:
                 return advisories
             validate_schema(record)
 
-            for p in record['packages']:
-                advisories.extend(self._load_advisories(
-                    p['pkg'],
-                    record['distroversion'],
-                    record['reponame'],
-                    record['archs'],
-                ))
+            for p in record["packages"]:
+                advisories.extend(
+                    self._load_advisories(
+                        p["pkg"], record["distroversion"], record["reponame"], record["archs"],
+                    )
+                )
 
         return advisories
 
     def _load_advisories(
-            self,
-            pkg_infos: Mapping[str, Any],
-            distroversion: str,
-            reponame: str,
-            archs: Iterable[str],
+        self, pkg_infos: Mapping[str, Any], distroversion: str, reponame: str, archs: Iterable[str],
     ) -> List[Advisory]:
 
         advisories = []
 
-        for version, fixed_vulns in pkg_infos['secfixes'].items():
+        for version, fixed_vulns in pkg_infos["secfixes"].items():
 
             if fixed_vulns is None:
                 continue
 
             resolved_purls = {
                 PackageURL(
-                    name=pkg_infos['name'],
-                    type='alpine',
+                    name=pkg_infos["name"],
+                    type="alpine",
                     version=version,
-                    qualifiers={
-                        'arch': arch, 'distroversion': distroversion, 'reponame': reponame},
+                    qualifiers={"arch": arch, "distroversion": distroversion, "reponame": reponame},
                 )
                 for arch in archs
             }
 
             for vuln_ids in fixed_vulns:
                 vuln_ids = vuln_ids.split()
-
-                advisories.append(Advisory(
-                    summary='',
-                    impacted_package_urls=[],
-                    resolved_package_urls=resolved_purls,
-                    reference_ids=vuln_ids[1:],
-                    cve_id=vuln_ids[0] if vuln_ids[0] != 'CVE-????-?????' else None,
-                ))
+                references = []
+                for reference_id in vuln_ids[1:]:
+
+                    if reference_id.startswith("XSA"):
+                        xsa_id = reference_id.split("-")[-1]
+                        references.append(
+                            Reference(
+                                reference_id=reference_id,
+                                url="https://xenbits.xen.org/xsa/advisory-{}.html".format(xsa_id),
+                            )
+                        )
+
+                    elif reference_id.startswith("ZBX"):
+                        references.append(
+                            Reference(
+                                reference_id=reference_id,
+                                url="https://support.zabbix.com/browse/{}".format(reference_id),
+                            )
+                        )
+
+                    elif reference_id.startswith("wnpa-sec"):
+                        references.append(
+                            Reference(
+                                reference_id=reference_id,
+                                url="https://www.wireshark.org/security/{}.html".format(
+                                    reference_id
+                                ),
+                            )
+                        )
+
+                advisories.append(
+                    Advisory(
+                        summary="",
+                        impacted_package_urls=[],
+                        resolved_package_urls=resolved_purls,
+                        vuln_references=references,
+                        cve_id=vuln_ids[0] if vuln_ids[0] != "CVE-????-?????" else None,
+                    )
+                )
 
         return advisories