fix: curl datasource (#2984)

* fixes #2977 
* fixes #2969

Author: b31ngd3v

cve_bin_tool/data_sources/curl_source.py

@@ -3,26 +3,19 @@
 
 from __future__ import annotations
 
-import glob
 import json
 import logging
-import re
 from pathlib import Path
 
 import aiohttp
-from bs4 import BeautifulSoup, NavigableString, ResultSet
 
 from cve_bin_tool.async_utils import FileIO, RateLimiter
 from cve_bin_tool.data_sources import (
     DISK_LOCATION_BACKUP,
     DISK_LOCATION_DEFAULT,
     Data_Source,
 )
-from cve_bin_tool.error_handler import (
-    CVEDataForCurlVersionNotInCache,
-    ErrorHandler,
-    ErrorMode,
-)
+from cve_bin_tool.error_handler import ErrorMode
 from cve_bin_tool.log import LOGGER
 
 logging.basicConfig(level=logging.DEBUG)
@@ -33,7 +26,7 @@ class Curl_Source(Data_Source):
     CACHEDIR = DISK_LOCATION_DEFAULT
     BACKUPCACHEDIR = DISK_LOCATION_BACKUP
     LOGGER = LOGGER.getChild("CVEDB")
-    CURL_CVE_FILENAME_TEMPLATE = "curlcve-{}.json"
+    DATA_SOURCE_LINK = "https://curl.se/docs/vuln.json"
 
     def __init__(self, error_mode=ErrorMode.TruncTrace):
         self.cve_list = None
@@ -43,6 +36,7 @@ def __init__(self, error_mode=ErrorMode.TruncTrace):
         self.session = None
         self.affected_data = None
         self.source_name = self.SOURCE
+        self.vulnerbility_data = []
 
     async def get_cve_data(self):
         await self.fetch_cves()
@@ -56,101 +50,34 @@ async def fetch_cves(self):
             self.session = RateLimiter(
                 aiohttp.ClientSession(connector=connector, trust_env=True)
             )
-
-        versions = await self.get_curl_versions(self.session)
-
-        for version in versions:
-            await self.download_curl_version(self.session, version)
-
+        await self.download_curl_vulnerabilities(self.session)
         await self.session.close()
 
-    @staticmethod
-    async def get_curl_versions(session: RateLimiter) -> list[str]:
-        regex = re.compile(r"vuln-(\d+.\d+.\d+)\.html")
-        async with await session.get(
-            "https://curl.haxx.se/docs/vulnerabilities.html"
-        ) as response:
-            response.raise_for_status()
-            html = await response.text()
-        matches = regex.finditer(html)
-        return [match.group(1) for match in matches]
-
-    async def download_curl_version(self, session: RateLimiter, version: str) -> None:
-        async with await session.get(
-            f"https://curl.haxx.se/docs/vuln-{version}.html"
-        ) as response:
+    async def download_curl_vulnerabilities(self, session: RateLimiter) -> None:
+        async with await session.get(self.DATA_SOURCE_LINK) as response:
             response.raise_for_status()
-            html = await response.text()
-        soup = BeautifulSoup(html, "html.parser")
-        table = soup.find("table")
-        if not table or isinstance(table, NavigableString):
-            return
-        headers: ResultSet | list = table.find_all("th")
-        headers = list(map(lambda x: x.text.strip().lower(), headers))
-        self.LOGGER.debug(headers)
-        rows = table.find_all("tr")
-        json_data = []
-        for row in rows:
-            cols = row.find_all("td")
-            values = (ele.text.strip() for ele in cols)
-            data = dict(zip(headers, values))
-            if data:
-                json_data.append(data)
-        path = Path(str(Path(self.cachedir) / f"curlcve-{version}.json"))
+            self.vulnerbility_data = await response.json()
+        path = Path(str(Path(self.cachedir) / "vuln.json"))
         filepath = path.resolve()
         async with FileIO(filepath, "w") as f:
-            await f.write(json.dumps(json_data, indent=4))
-
-    def load_curl_version(self, version: str) -> list[dict[str, str]]:
-        """
-        Return the dict of CVE data for the given curl version.
-        """
-        filename = Path(
-            str(Path(self.cachedir) / self.CURL_CVE_FILENAME_TEMPLATE.format(version))
-        )
-        # Check if file exists
-        if not filename.is_file():
-            with ErrorHandler(mode=self.error_mode, logger=self.LOGGER):
-                raise CVEDataForCurlVersionNotInCache(version)
-        # Open the file and load the JSON data, log the number of CVEs loaded
-        with open(filename, "rb") as fileobj:
-            cves_for_version = json.load(fileobj)
-            self.LOGGER.debug(
-                f"Curl Version {version} has {len(cves_for_version)} CVEs in dataset"
-            )
-            return cves_for_version
-
-    def curl_versions(self) -> list[str]:
-        """
-        Return the versions we have Curl data for.
-        """
-        regex = re.compile(r"curlcve-(\d+.\d+.\d).json")
-        versions = []
-        for filename in glob.glob(str(Path(self.cachedir) / "curlcve-*.json")):
-            match = regex.search(filename)
-            if match:
-                version = match.group(1)
-                versions.append(version)
-        return versions
+            await f.write(json.dumps(self.vulnerbility_data, indent=4))
 
     def get_cve_list(self):
         self.affected_data = []
 
-        for version in self.curl_versions():
-            cve_list = self.load_curl_version(version)
-
-            for cve in cve_list:
-                affected = {
-                    "cve_id": cve["cve"],
-                    "vendor": "haxx",
-                    "product": "curl",
-                    "version": version,
-                    "versionStartIncluding": cve["from version"],
-                    "versionStartExcluding": "",
-                    "versionEndIncluding": cve["to and including"],
-                    "versionEndExcluding": "",
-                }
-
-                self.affected_data.append(affected)
+        for cve in self.vulnerbility_data:
+            affected = {
+                "cve_id": cve["aliases"][0],
+                "vendor": "haxx",
+                "product": "curl",
+                "version": "*",
+                "versionStartIncluding": cve["affected"][0]["ranges"][0]["events"][0][
+                    "introduced"
+                ],
+                "versionStartExcluding": "",
+                "versionEndIncluding": cve["affected"][0]["versions"][0],
+                "versionEndExcluding": "",
+            }
+            self.affected_data.append(affected)
 
         return self.affected_data