added HexVersionAPI and mock test it

Signed-off-by: Tushar912 <tushar.912u@gmail.com>

Author: tushar912

vulnerabilities/importers/elixir_security.py

@@ -20,20 +20,22 @@
 #  VulnerableCode is a free software code scanning tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
+import asyncio
 import yaml
-import re
 import json
 import requests
+import re
 from typing import Set
 from typing import List
+
 from dephell_specifier import RangeSpecifier
 from packageurl import PackageURL
 
 from vulnerabilities.data_source import GitDataSource
 from vulnerabilities.data_source import GitDataSourceConfiguration
 from vulnerabilities.data_source import Advisory
 from vulnerabilities.data_source import Reference
-
+from vulnerabilities.package_managers import HexVersionAPI
 
 class ElixirSecurityDataSource(GitDataSource):
     def __enter__(self):
@@ -43,6 +45,11 @@ def __enter__(self):
             self._added_files, self._updated_files = self.file_changes(
                 recursive=True, file_ext="yml", subdir="./packages"
             )
+        self.pkg_manager_api = HexVersionAPI()
+        self.set_api(self.collect_packages())
+
+    def set_api(self, packages):
+        asyncio.run(self.pkg_manager_api.load_api(packages))
 
     def updated_advisories(self) -> Set[Advisory]:
         files = self._updated_files
@@ -62,17 +69,25 @@ def added_advisories(self) -> Set[Advisory]:
                 advisories.append(processed_data)
         return self.batch_advisories(advisories)
 
-    @staticmethod
-    def generate_all_version_list(pkg_name):
-        resp = requests.get(f"https://hex.pm/api/packages/{pkg_name}")
-        resp = resp.content
-        json_resp = json.loads(resp)
-        version_list = []
-        for release in json_resp["releases"]:
-            version_list.append(release["version"])
+    def collect_packages(self):
+        packages = set()
+        files = self._updated_files.union(self._added_files)
+        for f in files:
+            with open(f) as file:
+                data = yaml.safe_load(file)
+                if data.get("package"):
+                    packages.add(data["package"])
+
+        return packages
+            
+    def generate_all_version_list(self,pkg_name):
+        if not getattr(self, 'pkg_manager_api', None):
+            self.pkg_manager_api = HexVersionAPI()
+        version_list = self.pkg_manager_api.get(
+            pkg_name)
         return version_list
 
-    def get_pkg_from_range(self, version_list, pkg_name):
+    def get_versions_from_range(self, version_list, pkg_name):
         pkg_versions = []
         all_version_list = self.generate_all_version_list(pkg_name)
         if version_list is None:
@@ -89,15 +104,16 @@ def process_file(self, path):
             pkg_name = yaml_file["package"]
             safe_pkg_versions = []
             if yaml_file.get("unaffected_versions"):
-                safe_pkg_versions = self.get_pkg_from_range(
+                safe_pkg_versions = self.get_versions_from_range(
                     yaml_file["patched_versions"] + yaml_file["unaffected_versions"],
                     pkg_name,
                 )
             else:
-                safe_pkg_versions = self.get_pkg_from_range(
+                safe_pkg_versions = self.get_versions_from_range(
                     yaml_file["patched_versions"], pkg_name
                 )
-            cve_id = yaml_file["cve"]
+            
+            cve_id = "CVE-"+str(yaml_file["cve"])
             safe_purls = []
             if safe_pkg_versions is not None:
                 safe_purls = {
@@ -107,6 +123,7 @@ def process_file(self, path):
 
             vuln_reference = [
                 Reference(
+                    reference_id=yaml_file["id"],
                     url=yaml_file["link"],
                 )
             ]

vulnerabilities/package_managers.py

@@ -321,3 +321,24 @@ async def fetch(self, owner_repo: str, session) -> None:
         resp = await resp.json()
         print(resp)
         self.cache[owner_repo] = [release["ref"].split("/")[-1] for release in resp]
+
+class HexVersionAPI(VersionAPI):
+    async def load_api(self, pkg_set):
+        async with ClientSession(raise_for_status=True) as session:
+            await asyncio.gather(
+                *[self.fetch(pkg, session) for pkg in pkg_set if pkg not in self.cache]
+            )
+
+    async def fetch(self, pkg, session):
+        url = f"https://hex.pm/api/packages/{pkg}"
+        versions = set()
+        try:
+            response = await session.request(method="GET", url=url)
+            response = await response.json()
+            for release in response["releases"]:
+                versions.add(release["version"])
+        except (ClientResponseError, JSONDecodeError):
+            pass
+
+        self.cache[pkg] = versions
+        

vulnerabilities/tests/test_elixir_security.py

@@ -22,13 +22,16 @@
 
 import os
 from unittest import TestCase
+from unittest.mock import patch
 from collections import OrderedDict
 
 from vulnerabilities.data_source import Reference
 from packageurl import PackageURL
 
-from vulnerabilities.importers.elixir_security import ElixirSecurityDataSource
 from vulnerabilities.data_source import Advisory
+from vulnerabilities.importers.elixir_security import ElixirSecurityDataSource
+from vulnerabilities.package_managers import HexVersionAPI
+
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 
@@ -37,11 +40,26 @@ class TestElixirSecurityDataSource(TestCase):
     @classmethod
     def setUpClass(cls):
         data_source_cfg = {
-            "repository_url": "https://test.net",
+            "repository_url": 'https://github.com/dependabot/elixir-security-advisories',
         }
         cls.data_src = ElixirSecurityDataSource(1, config=data_source_cfg)
+        cls.data_src.pkg_manager_api = HexVersionAPI()
 
-    def test_generate_all_version_list(self):
+    @patch('vulnerabilities.package_managers.HexVersionAPI.get',
+           return_value=[
+            "0.5.2",
+            "0.5.1",
+            "0.5.0",
+            "0.4.0",
+            "0.3.1",
+            "0.3.0",
+            "0.2.0",
+            "0.1.3",
+            "0.1.2",
+            "0.1.1",
+            "0.1.0",
+        ])
+    def test_generate_all_version_list(self,mock_write):
         package = "coherence"
         actual_list = self.data_src.generate_all_version_list(package)
         expected_list = [
@@ -58,8 +76,21 @@ def test_generate_all_version_list(self):
             "0.1.0",
         ]
         assert actual_list == expected_list
-
-    def test_process_file(self):
+    @patch('vulnerabilities.package_managers.HexVersionAPI.get',
+           return_value=[
+            "0.5.2",
+            "0.5.1",
+            "0.5.0",
+            "0.4.0",
+            "0.3.1",
+            "0.3.0",
+            "0.2.0",
+            "0.1.3",
+            "0.1.2",
+            "0.1.1",
+            "0.1.0",
+        ])
+    def test_process_file(self,mock_write):
 
         path = os.path.join(BASE_DIR, "test_data/elixir_security/test_file.yml")
         expected_data = Advisory(
@@ -75,9 +106,10 @@ def test_process_file(self):
                 ),
             },
             vuln_references=[
-                Reference(url="https://github.com/smpallen99/coherence/issues/270")
+                Reference(reference_id='2aae6e3a-24a3-4d5f-86ff-b964eaf7c6d1',
+                url="https://github.com/smpallen99/coherence/issues/270")
             ],
-            cve_id="2018-20301",
+            cve_id="CVE-2018-20301",
         )
 
         found_data = self.data_src.process_file(path)