Time travel for githubtags, maven version api importers

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

Author: sbs2001

vulnerabilities/importers/alpine_linux.py

@@ -43,40 +43,6 @@
 BASE_URL = "https://secdb.alpinelinux.org/"
 
 
-# def validate_schema(advisory_dict):
-#     scheme = {
-#         "distroversion": Regex(r"v\d.\d*"),
-#         "reponame": str,
-#         "archs": list,
-#         "packages": [
-#             {
-#                 "pkg": {
-#                     "name": str,
-#                     "secfixes": {
-#                         str: Or(
-#                             [
-#                                 Or(
-#                                     Regex(r"CVE.\d+-\d+", flags=IGNORECASE),
-#                                     Regex(r"XSA-\d{3}"),
-#                                     Regex(r"ZBX-\d{4}"),
-#                                     Regex(r"wnpa-sec-\d{4}-\d{2}"),
-#                                     Regex(r"GHSA-.{4}-.{4}-.{4}"),
-#                                 )
-#                             ],
-#                             "",
-#                             # FIXME: Remove the None when below issue gets fixed
-#                             # https://gitlab.alpinelinux.org/alpine/infra/alpine-secdb/-/issues/1
-#                             None,
-#                         ),
-#                     },
-#                 }
-#             }
-#         ],
-#         object: object,
-#     }
-#     Schema(scheme).validate(advisory_dict)
-
-
 class AlpineDataSource(DataSource):
     @staticmethod
     def fetch_advisory_links():

vulnerabilities/importers/elixir_security.py

@@ -83,7 +83,7 @@ def get_versions_for_pkg_from_range_list(self, version_range_list, pkg_name):
 
         safe_pkg_versions = []
         vuln_pkg_versions = []
-        all_version_list = self.pkg_manager_api.get(pkg_name)
+        all_version_list = self.pkg_manager_api.get(pkg_name)["valid"]
         if not version_range_list:
             return [], all_version_list
         version_ranges = [

vulnerabilities/importers/istio.py

@@ -20,7 +20,9 @@
 #  VulnerableCode is a free software tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 import asyncio
+import pytz
 import re
+from dateutil import parser
 from typing import Set
 
 import saneyaml
@@ -60,11 +62,11 @@ def updated_advisories(self) -> Set[Advisory]:
                 advisories.extend(processed_data)
         return self.batch_advisories(advisories)
 
-    def get_pkg_versions_from_ranges(self, version_range_list):
+    def get_pkg_versions_from_ranges(self, version_range_list, release_date):
         """Takes a list of version ranges(affected) of a package
         as parameter and returns a tuple of safe package versions and
         vulnerable package versions"""
-        all_version = self.version_api.get("istio/istio")
+        all_version = self.version_api.get("istio/istio", release_date)["valid"]
         safe_pkg_versions = []
         vuln_pkg_versions = []
         version_ranges = [
@@ -123,6 +125,7 @@ def process_file(self, path):
         advisories = []
 
         data = self.get_data_from_md(path)
+        release_date = parser.parse(data["publishdate"]).replace(tzinfo=pytz.UTC)
 
         releases = []
         if data.get("releases"):
@@ -165,7 +168,7 @@ def process_file(self, path):
                 data["release_ranges"] = []
 
             safe_pkg_versions, vuln_pkg_versions = self.get_pkg_versions_from_ranges(
-                data["release_ranges"]
+                data["release_ranges"], release_date
             )
 
             affected_packages = []

vulnerabilities/importers/nginx.py

@@ -53,8 +53,8 @@ def set_api(self):
 
         # For some reason nginx tags it's releases are in the form of `release-1.2.3`
         # Chop off the `release-` part here.
-        for index, version in enumerate(self.version_api.cache["nginx/nginx"]):
-            self.version_api.cache["nginx/nginx"][index] = version.replace("release-", "")
+        for index, version in enumerate(self.version_api.cache["nginx/nginx"]["valid"]):
+            self.version_api.cache["nginx/nginx"]["valid"][index] = version.replace("release-", "")
 
     def updated_advisories(self):
         advisories = []
@@ -134,7 +134,9 @@ def extract_fixed_pkgs(self, vuln_info):
                 VersionSpecifier.from_scheme_version_spec_string("semver", "^" + rng[:-1])
             )
 
-        valid_versions = find_valid_versions(self.version_api.get("nginx/nginx"), version_ranges)
+        valid_versions = find_valid_versions(
+            self.version_api.get("nginx/nginx")["valid"], version_ranges
+        )
 
         return [
             PackageURL(type="generic", name="nginx", version=version) for version in valid_versions

vulnerabilities/importers/suse_backports.py

@@ -1,95 +1,95 @@
-# Copyright (c) 2017 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.
-#
-# You may not use this software except in compliance with the License.
-# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
-# Unless required by applicable law or agreed to in writing, software distributed
-# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
-# CONDITIONS OF ANY KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations under the License.
-#
-# When you publish or redistribute any data created with VulnerableCode or any VulnerableCode
-# derivative work, you must accompany this data with the following acknowledgment:
-#
-#  Generated with VulnerableCode and provided on an "AS IS" BASIS, WITHOUT WARRANTIES
-#  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.
-#  Visit https://github.com/nexB/vulnerablecode/ for support and download.
-import dataclasses
+# # Copyright (c) 2017 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.
+# #
+# # You may not use this software except in compliance with the License.
+# # You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# # Unless required by applicable law or agreed to in writing, software distributed
+# # under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# # CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# # specific language governing permissions and limitations under the License.
+# #
+# # When you publish or redistribute any data created with VulnerableCode or any VulnerableCode
+# # derivative work, you must accompany this data with the following acknowledgment:
+# #
+# #  Generated with VulnerableCode and provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# #  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.
+# #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
+# import dataclasses
 
-import requests
-import saneyaml
-from bs4 import BeautifulSoup
-from packageurl import PackageURL
+# import requests
+# import saneyaml
+# from bs4 import BeautifulSoup
+# from packageurl import PackageURL
 
-from vulnerabilities.data_source import Advisory
-from vulnerabilities.data_source import DataSource
-from vulnerabilities.data_source import DataSourceConfiguration
-from vulnerabilities.helpers import create_etag
+# from vulnerabilities.data_source import Advisory
+# from vulnerabilities.data_source import DataSource
+# from vulnerabilities.data_source import DataSourceConfiguration
+# from vulnerabilities.helpers import create_etag
 
 
-@dataclasses.dataclass
-class SUSEBackportsConfiguration(DataSourceConfiguration):
-    url: str
-    etags: dict
+# @dataclasses.dataclass
+# class SUSEBackportsConfiguration(DataSourceConfiguration):
+#     url: str
+#     etags: dict
 
 
-class SUSEBackportsDataSource(DataSource):
+# class SUSEBackportsDataSource(DataSource):
 
-    CONFIG_CLASS = SUSEBackportsConfiguration
+#     CONFIG_CLASS = SUSEBackportsConfiguration
 
-    @staticmethod
-    def get_all_urls_of_backports(url):
-        r = requests.get(url)
-        soup = BeautifulSoup(r.content, "lxml")
-        for a_tag in soup.find_all("a", href=True):
-            if a_tag["href"].endswith(".yaml") and a_tag["href"].startswith("backports"):
-                yield url + a_tag["href"]
+#     @staticmethod
+#     def get_all_urls_of_backports(url):
+#         r = requests.get(url)
+#         soup = BeautifulSoup(r.content, "lxml")
+#         for a_tag in soup.find_all("a", href=True):
+#             if a_tag["href"].endswith(".yaml") and a_tag["href"].startswith("backports"):
+#                 yield url + a_tag["href"]
 
-    def updated_advisories(self):
-        advisories = []
-        all_urls = self.get_all_urls_of_backports(self.config.url)
-        for url in all_urls:
-            if not create_etag(data_src=self, url=url, etag_key="ETag"):
-                continue
-            advisories.extend(self.process_file(self._fetch_yaml(url)))
-        return self.batch_advisories(advisories)
+#     def updated_advisories(self):
+#         advisories = []
+#         all_urls = self.get_all_urls_of_backports(self.config.url)
+#         for url in all_urls:
+#             if not create_etag(data_src=self, url=url, etag_key="ETag"):
+#                 continue
+#             advisories.extend(self.process_file(self._fetch_yaml(url)))
+#         return self.batch_advisories(advisories)
 
-    def _fetch_yaml(self, url):
+#     def _fetch_yaml(self, url):
 
-        try:
-            resp = requests.get(url)
-            resp.raise_for_status()
-            return saneyaml.load(resp.content)
+#         try:
+#             resp = requests.get(url)
+#             resp.raise_for_status()
+#             return saneyaml.load(resp.content)
 
-        except requests.HTTPError:
-            return {}
+#         except requests.HTTPError:
+#             return {}
 
-    @staticmethod
-    def process_file(yaml_file):
-        advisories = []
-        try:
-            for pkg in yaml_file[0]["packages"]:
-                for version in yaml_file[0]["packages"][pkg]["fixed"]:
-                    for vuln in yaml_file[0]["packages"][pkg]["fixed"][version]:
-                        # yaml_file specific data can be added
-                        purl = [
-                            PackageURL(name=pkg, type="rpm", version=version, namespace="opensuse")
-                        ]
-                        advisories.append(
-                            Advisory(
-                                vulnerability_id=vuln,
-                                resolved_package_urls=purl,
-                                summary="",
-                                impacted_package_urls=[],
-                            )
-                        )
-        except TypeError:
-            # could've used pass
-            return advisories
+#     @staticmethod
+#     def process_file(yaml_file):
+#         advisories = []
+#         try:
+#             for pkg in yaml_file[0]["packages"]:
+#                 for version in yaml_file[0]["packages"][pkg]["fixed"]:
+#                     for vuln in yaml_file[0]["packages"][pkg]["fixed"][version]:
+#                         # yaml_file specific data can be added
+#                         purl = [
+#                             PackageURL(name=pkg, type="rpm", version=version, namespace="opensuse")
+#                         ]
+#                         advisories.append(
+#                             Advisory(
+#                                 vulnerability_id=vuln,
+#                                 resolved_package_urls=purl,
+#                                 summary="",
+#                                 impacted_package_urls=[],
+#                             )
+#                         )
+#         except TypeError:
+#             # could've used pass
+#             return advisories
 
-        return advisories
+#         return advisories

vulnerabilities/package_managers.py

@@ -22,19 +22,16 @@
 
 import asyncio
 from collections import namedtuple
-import aiohttp
+import pytz
 from bs4 import BeautifulSoup
 from dateutil import parser
 from json import JSONDecodeError
 from typing import Mapping
 from typing import Set
-from typing import List
-import xml.etree.ElementTree as ET
 
 from aiohttp import ClientSession
 from aiohttp.client_exceptions import ClientResponseError
 from aiohttp.client_exceptions import ServerDisconnectedError
-from requests.sessions import session
 
 
 Version = namedtuple("Version", field_names=["value", "release_date"])
@@ -255,12 +252,15 @@ class MavenVersionAPI(VersionAPI):
     async def load_api(self, pkg_set):
         async with client_session() as session:
             await asyncio.gather(
-                *[self.fetch(pkg, session) for pkg in pkg_set if pkg not in self.cache]
+                *[
+                    self.fetch(pkg, session)
+                    for pkg in pkg_set
+                    if pkg not in self.cache and "camel" not in pkg
+                ]
             )
 
     async def fetch(self, pkg, session) -> None:
-        artifact_comps = pkg.split(":")
-        endpoint = self.artifact_url(artifact_comps)
+        endpoint = self.artifact_url(pkg)
         try:
             resp = await session.request(method="GET", url=endpoint)
             resp = await resp.read()
@@ -269,11 +269,15 @@ async def fetch(self, pkg, session) -> None:
             self.cache[pkg] = set()
             return
 
-        xml_resp = ET.ElementTree(ET.fromstring(resp.decode("utf-8")))
-        self.cache[pkg] = self.extract_versions(xml_resp)
+        soup = BeautifulSoup(resp, features="lxml")
+        try:
+            self.cache[pkg] = self.extract_versions(soup)
+        except:
+            raise
 
     @staticmethod
-    def artifact_url(artifact_comps: List[str]) -> str:
+    def artifact_url(pkg: str) -> str:
+        artifact_comps = pkg.split(":")
         base_url = "https://repo1.maven.org/maven2/{}"
         try:
             group_id, artifact_id = artifact_comps
@@ -289,19 +293,35 @@ def artifact_url(artifact_comps: List[str]) -> str:
                 raise
 
         group_url = group_id.replace(".", "/")
-        suffix = group_url + "/" + artifact_id + "/" + "maven-metadata.xml"
+        suffix = group_url + "/" + artifact_id + "/"
         endpoint = base_url.format(suffix)
 
         return endpoint
 
     @staticmethod
-    def extract_versions(xml_response: ET.ElementTree) -> Set[str]:
-        all_versions = set()
-        for child in xml_response.getroot().iter():
-            if child.tag == "version":
-                all_versions.add(child.text)
+    def extract_versions(soup: BeautifulSoup) -> Set[Version]:
+        pre_tag = soup.find("pre")
+        prev_tag = None
+        versions = set()
+        for atag in pre_tag:
+            if atag.name == "a" and atag["href"] != "../":
+                prev_tag = atag
+            elif prev_tag:
+                text_groups = atag.split()
+                if text_groups[-1] != "-":
+                    break
+                date = " ".join(text_groups[:-1])
+                if date != "-":
+                    versions.add(
+                        Version(
+                            value=prev_tag.text[:-1],
+                            release_date=parser.parse(date).replace(tzinfo=pytz.UTC),
+                        )
+                    )
+                else:
+                    versions.add(Version(value=prev_tag.text[:-1], release_date=None))
 
-        return all_versions
+        return versions
 
 
 class NugetVersionAPI(VersionAPI):