add Snyk DataSource

- closes #834

Signed-off-by: Keshav Priyadarshi <git@keshav.space>

Author: keshav-space

vulntotal/datasources/__init__.py

@@ -22,6 +22,10 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
 
-DATASOURCE_REGISTRY = []
+from vulntotal.datasources import snyk
+
+DATASOURCE_REGISTRY = [
+    snyk.SnykDataSource,
+]
 
 DATASOURCE_REGISTRY = {x.__module__.split(".")[-1]: x for x in DATASOURCE_REGISTRY}

vulntotal/datasources/snyk.py

@@ -0,0 +1,178 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# http://nexb.com and https://github.com/nexB/vulnerablecode/
+# The VulnTotal software is licensed under the Apache License version 2.0.
+# Data generated with VulnTotal 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 VulnTotal or any VulnTotal
+# derivative work, you must accompany this data with the following acknowledgment:
+#
+#  Generated with VulnTotal and provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+#  OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+#  VulnTotal should be considered or used as legal advice. Consult an Attorney
+#  for any legal advice.
+#  VulnTotal is a free software tool from nexB Inc. and others.
+#  Visit https://github.com/nexB/vulnerablecode/ for support and download.
+
+import json
+import logging
+from typing import Iterable
+from urllib.parse import quote
+
+import requests
+from bs4 import BeautifulSoup
+from packageurl import PackageURL
+
+from vulntotal.validator import DataSource
+from vulntotal.validator import VendorData
+from vulntotal.vulntotal_utils import snky_constraints_satisfied
+
+logger = logging.getLogger(__name__)
+
+
+class SnykDataSource(DataSource):
+    spdx_license_expression = "TODO"
+    license_url = "TODO"
+
+    def fetch(self, url):
+        response = requests.get(url)
+        if not response.status_code == 200:
+            logger.error(f"Error while fetching {url}")
+            return
+        if response.headers["content-type"] == "application/json, charset=utf-8":
+            return response.json()
+        return response.text
+
+    def datasource_advisory(self, purl) -> Iterable[VendorData]:
+        package_advisory_url = generate_package_advisory_url(purl)
+        package_advisories_list = self.fetch(package_advisory_url)
+        self._raw_dump.append(package_advisories_list)
+        if package_advisories_list:
+            advisories = extract_html_json_advisories(package_advisories_list)
+            for snyk_id, affected in advisories.items():
+                if "*" in affected or is_purl_in_affected(purl.version, affected):
+                    advisory_payload = generate_advisory_payload(snyk_id)
+                    advisory_html = self.fetch(advisory_payload)
+                    self._raw_dump.append(advisory_html)
+                    if advisory_html:
+                        yield parse_html_advisory(advisory_html, snyk_id, affected)
+
+    @classmethod
+    def supported_ecosystem(cls):
+        return {
+            "cocoapods": "cocoapods",
+            "composer": "composer",
+            "golang": "golang",
+            "hex": "hex",
+            "linux": "linux",
+            "maven": "maven",
+            "npm": "npm",
+            "nuget": "nuget",
+            "pypi": "pip",
+            "rubygems": "rubygems",
+            # any purl.type not in supported_ecosystem shall implicitly be treated as unmanaged type
+            "unmanaged": "unmanaged",
+        }
+
+
+def generate_package_advisory_url(purl):
+    url_package_advisories = "https://security.snyk.io/package/{ecosystem}/{package}"
+
+    # Pseudo API, unfortunately gives only 30 vulnerability per package, but this is the best we have for unmanaged packages
+    url_unmanaged_package_advisories = (
+        "https://security.snyk.io/api/listing?search={package}&type=unmanaged"
+    )
+    supported_ecosystem = SnykDataSource.supported_ecosystem()
+
+    if purl.type == "unmanaged" or purl.type not in supported_ecosystem:
+        return url_unmanaged_package_advisories.format(
+            package=purl.name if not purl.namespace else f"{purl.namespace}/{purl.name}",
+        )
+
+    purl_name = purl.name
+    if purl.type == "maven":
+        if not purl.namespace:
+            logger.error(f"Invalid Maven PURL {str(purl)}")
+            return
+        purl_name = quote(f"{purl.namespace}:{purl.name}", safe="")
+
+    elif purl.type in ("golang", "composer"):
+        if purl.namespace:
+            purl_name = quote(f"{purl.namespace}/{purl.name}", safe="")
+
+    elif purl.type == "linux":
+        distro = purl.qualifiers["distro"]
+        purl_name = f"{distro}/{purl.name}"
+
+    return url_package_advisories.format(
+        ecosystem=supported_ecosystem[purl.type],
+        package=purl_name,
+    )
+
+
+def extract_html_json_advisories(package_advisories):
+    vulnerablity = {}
+
+    # If advisories are json and is obtained through pseudo API
+    if isinstance(package_advisories, dict):
+        if package_advisories["status"] == "ok":
+            for vuln in package_advisories["vulnerabilities"]:
+                vulnerablity[vuln["id"]] = vuln["semver"]["vulnerable"]
+    else:
+        soup = BeautifulSoup(package_advisories, "html.parser")
+        vulns_table = soup.find("tbody", class_="vue--table__tbody")
+        if vulns_table:
+            vulns_rows = vulns_table.find_all("tr", class_="vue--table__row")
+            for row in vulns_rows:
+                anchor = row.find(class_="vue--anchor")
+                ranges = row.find_all(
+                    "span", class_="vue--chip vulnerable-versions__chip vue--chip--default"
+                )
+                affected_versions = [vers.text.strip() for vers in ranges]
+                vulnerablity[anchor["href"].rsplit("/", 1)[-1]] = affected_versions
+    return vulnerablity
+
+
+def parse_html_advisory(advisory_html, snyk_id, affected) -> VendorData:
+    aliases = []
+    fixed_versions = []
+
+    advisory_soup = BeautifulSoup(advisory_html, "html.parser")
+    cve_span = advisory_soup.find("span", class_="cve")
+    if cve_span:
+        cve_anchor = cve_span.find("a", class_="vue--anchor")
+        aliases.append(cve_anchor["id"])
+
+    how_to_fix = advisory_soup.find(
+        "div", class_="vue--block vuln-page__instruction-block vue--block--instruction"
+    )
+    if how_to_fix:
+        fixed = how_to_fix.find("p").text.split(" ")
+        if "Upgrade" in fixed:
+            lower = fixed.index("version") if "version" in fixed else fixed.index("versions")
+            upper = fixed.index("or")
+            fixed_versions = "".join(fixed[lower + 1 : upper]).split(",")
+    aliases.append(snyk_id)
+    return VendorData(
+        aliases=aliases,
+        affected_versions=affected,
+        fixed_versions=fixed_versions,
+    )
+
+
+def is_purl_in_affected(version, affected):
+    for affected_range in affected:
+        if snky_constraints_satisfied(affected_range, version):
+            return True
+    return False
+
+
+def generate_advisory_payload(snyk_id):
+    return f"https://security.snyk.io/vuln/{snyk_id}"

vulntotal/vulntotal_utils.py

@@ -0,0 +1,134 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# http://nexb.com and https://github.com/nexB/vulnerablecode/
+# The VulnTotal software is licensed under the Apache License version 2.0.
+# Data generated with VulnTotal 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 VulnTotal or any VulnTotal
+# derivative work, you must accompany this data with the following acknowledgment:
+#
+#  Generated with VulnTotal and provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+#  OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+#  VulnTotal should be considered or used as legal advice. Consult an Attorney
+#  for any legal advice.
+#  VulnTotal is a free software tool from nexB Inc. and others.
+#  Visit https://github.com/nexB/vulnerablecode/ for support and download.
+
+import operator
+
+
+class GenericVersion:
+    def __init__(self, version):
+        self.value = version.replace(" ", "").lstrip("v")
+
+        self.decomposed = tuple(
+            [int(com) if com.isnumeric() else com for com in self.value.split(".")]
+        )
+
+    def __str__(self):
+        return str(self.value)
+
+    def __eq__(self, other):
+        if not isinstance(other, self.__class__):
+            return NotImplemented
+        return self.value.__eq__(other.value)
+
+    def __lt__(self, other):
+        if not isinstance(other, self.__class__):
+            return NotImplemented
+        for i, j in zip(self.decomposed, other.decomposed):
+            if not isinstance(i, type(j)):
+                continue
+            if i.__gt__(j):
+                return False
+        return True
+
+    def __le__(self, other):
+        if not isinstance(other, self.__class__):
+            return NotImplemented
+        return self.__lt__(other) or self.__eq__(other)
+
+
+def compare(version, package_comparator, package_version):
+    operator_comparator = {
+        "<": operator.lt,
+        ">": operator.gt,
+        "=": operator.eq,
+        "<=": operator.le,
+        ">=": operator.ge,
+        "==": operator.eq,
+        "!=": operator.ne,
+        ")": operator.lt,
+        "]": operator.le,
+        "(": operator.gt,
+        "[": operator.ge,
+    }
+    compare = operator_comparator[package_comparator]
+    return compare(version, package_version)
+
+
+def parse_constraint(constraint):
+    if constraint.startswith(("<=", ">=", "==", "!=")):
+        return constraint[:2], constraint[2:]
+
+    if constraint.startswith(("<", ">", "=", "[", "]", "(", ")")):
+        return constraint[0], constraint[1:]
+
+    if constraint.endswith(("[", "]", "(", ")")):
+        return constraint[-1], constraint[:-1]
+
+
+def github_constraints_satisfied(github_constrain, version):
+    gh_constraints = github_constrain.strip().replace(" ", "")
+    constraints = gh_constraints.split(",")
+    for constraint in constraints:
+        gh_comparator, gh_version = parse_constraint(constraint)
+        if not gh_version:
+            continue
+        # TODO: Replace the GenericVersion with ecosystem specific from univers
+        if not compare(GenericVersion(version), gh_comparator, GenericVersion(gh_version)):
+            return False
+    return True
+
+
+def snky_constraints_satisfied(snyk_constrain, version):
+    snyk_constraints = snyk_constrain.strip().replace(" ", "")
+    constraints = snyk_constraints.split(",")
+    for constraint in constraints:
+        snyk_comparator, snyk_version = parse_constraint(constraint)
+        if not snyk_version:
+            continue
+        # TODO: Replace the GenericVersion with ecosystem specific from univers or maybe not if snyk is normalizing versions to semver
+        if not compare(GenericVersion(version), snyk_comparator, GenericVersion(snyk_version)):
+            return False
+    return True
+
+
+def gitlab_constraints_satisfied(gitlab_constrain, version):
+    gitlab_constraints = gitlab_constrain.strip()
+    constraints = gitlab_constraints.split("||")
+
+    for constraint in constraints:
+        is_constraint_satisfied = True
+
+        for subcontraint in constraint.strip().split(" "):
+
+            gitlab_comparator, gitlab_version = parse_constraint(subcontraint.strip())
+            if not gitlab_version:
+                continue
+            # TODO: Replace the GenericVersion with ecosystem specific from univers
+            if not compare(
+                GenericVersion(version), gitlab_comparator, GenericVersion(gitlab_version)
+            ):
+                is_constraint_satisfied = False
+                break
+
+        if is_constraint_satisfied:
+            return True