Add gitlab importer

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/importers/__init__.py

@@ -21,8 +21,14 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 from vulnerabilities.importers import alpine_linux
 from vulnerabilities.importers import github
+from vulnerabilities.importers import gitlab
 from vulnerabilities.importers import nginx
 
-IMPORTERS_REGISTRY = [nginx.NginxImporter, alpine_linux.AlpineImporter, github.GitHubAPIImporter]
+IMPORTERS_REGISTRY = [
+    nginx.NginxImporter,
+    alpine_linux.AlpineImporter,
+    github.GitHubAPIImporter,
+    gitlab.GitLabAPIImporter,
+]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/gitlab.py

@@ -1,53 +1,187 @@
-from dataclasses import dataclass
-import dataclasses
+# 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.
+#
+# 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  from nexB Inc. and others.
+#  Visit https://github.com/nexB/vulnerablecode/ for support and download.
+
+import logging
 import os
+from typing import Iterable
 from typing import List
+
+import yaml
+from dateutil import parser as dateparser
 from fetchcode.vcs import fetch_via_vcs
+from packageurl import PackageURL
+from univers.version_range import RANGE_CLASS_BY_SCHEMES
+from univers.version_range import VersionRange
+from univers.versions import Version
+
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import Importer
+from vulnerabilities.importer import Reference
+from vulnerabilities.importers.github import get_reference_id
+
+LOGGER = logging.getLogger(__name__)
+
+
+PURL_TYPE_BY_ECOSYSTEM = {
+    "gem": "gem",
+    "go": "golang",
+    "maven": "maven",
+    "npm": "npm",
+    "nuget": "nuget",
+    "pypi": "pypi",
+    "packagist": "composer",
+}
+
 
 def fork_and_get_dir(url):
     return fetch_via_vcs(url).dest_dir
 
-def walk_in_directory(dir):
-    for root, dirs, files in os.walk(dir):
+
+class ForkError(Exception):
+    pass
+
+
+class GitLabAPIImporter(Importer):
+    spdx_license_expression = "MIT"
+    license_url = "https://gitlab.com/gitlab-org/advisories-community/-/blob/main/LICENSE"
+    gitlab_url = "git+https://gitlab.com/gitlab-org/advisories-community/"
+
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        """
+        Return a list of AdvisoryData objects
+        """
+        try:
+            fork_directory = fork_and_get_dir(self.gitlab_url)
+        except Exception as e:
+            LOGGER.error(f"Can't fetch url {self.gitlab_url}")
+            raise ForkError(e)
+        ecosystems = ["nuget", "maven", "gem", "npm", "go", "packagist", "pypi"]
+        for ecosystem in ecosystems:
+            for file in get_files(os.path.join(fork_directory, ecosystem)):
+                yield parse_yaml_file(file)
+
+
+def get_files(dir):
+    for root, _, files in os.walk(dir):
         for file in files:
-            yield os.path.join(root, file)
-
-# binary search
-def binary_search(lst, item):
-    low = 0
-    high = len(lst) - 1
-
-    while low <= high:
-        mid = (low + high) // 2
-        guess = lst[mid]
-        if guess == item:
-            return mid
-        if guess > item:
-            high = mid - 1
+            if "git" in root:
+                yield os.path.join(root, file)
+
+
+def extract_references(urls: List[str]) -> Iterable[Reference]:
+    """
+    Yield `reference` by iterating over `reference_data`
+    >>> list(extract_references(["https://github.com/advisories/GHSA-c9hw-wf7x-jp9j"]))
+    [Reference(reference_id='GHSA-c9hw-wf7x-jp9j', url='https://github.com/advisories/GHSA-c9hw-wf7x-jp9j', severities=[])]
+    >>> list(extract_references(["https://github.com/advisories/c9hw-wf7x-jp9j"]))
+    [Reference(reference_id='', url='https://github.com/advisories/c9hw-wf7x-jp9j', severities=[])]
+    """
+    for url in urls:
+        if not isinstance(url, str):
+            LOGGER.error(f"extract_references: url is not of type `str`: {url}")
+            continue
+        if "GHSA-" in url.upper():
+            reference = Reference(url=url, reference_id=get_reference_id(url))
         else:
-            low = mid + 1
-    return None
-
-#bubble sort
-def bubble_sort(lst):
-    for i in range(len(lst) - 1):
-        for j in range(len(lst) - i - 1):
-            if lst[j] > lst[j + 1]:
-                lst[j], lst[j + 1] = lst[j + 1], lst[j]
-    return lst
-
-@dataclasses.dataclass(order=True)
-class Test:
-    name: str
-    number: int
-    grades: List[str] = dataclasses.field(default_factory=list)
-
-    def __repr__(self):
-        return self.name
-    
-    def to_dict(self):
-        return {
-            'name': self.name,
-            'number': self.number,
-            'grades': [grade for grade in self.list_of_grades]
-        }
+            reference = Reference(url=url)
+        yield reference
+
+
+def not_empty(value):
+    return value is not None and value != ""
+
+
+def get_purl(package_slug):
+    """
+    Return a PackageURL object from a package slug
+    """
+    LOGGER.error(package_slug)
+    parts = package_slug.split("/")
+    parts = list(filter(not_empty, parts))
+    # if package slug is of the form:
+    # "nuget/NuGet.Core"
+    if len(parts) == 2:
+        type, name = parts
+        return PackageURL(type=PURL_TYPE_BY_ECOSYSTEM[type], name=name)
+    # if package slug is of the form:
+    # "nuget/github/user/abc/NuGet.Core"
+    if len(parts) >= 3:
+        type = parts[0]
+        name = parts[-1]
+        namespace = "/".join(parts[1:-1])
+        return PackageURL(type=PURL_TYPE_BY_ECOSYSTEM[type], namespace=namespace, name=name)
+    LOGGER.error(f"get_purl: package_slug can not be parsed: {package_slug!r}")
+
+
+def extract_affected_packages(
+    affected_version_range: VersionRange, fixed_versions: List[str], purl: PackageURL
+) -> Iterable[AffectedPackage]:
+    """
+    Yield a list of AffectedPackage objects
+    """
+    for fixed_version in fixed_versions or []:
+        yield AffectedPackage(
+            package=purl,
+            fixed_version=affected_version_range.version_class(fixed_version),
+            affected_version_range=affected_version_range,
+        )
+
+
+def parse_yaml_file(file):
+    with open(file, "r") as f:
+        yaml_file = yaml.safe_load(f)
+    if not isinstance(yaml_file, dict):
+        LOGGER.error(f"parse_yaml_file: yaml_file is not of type `dict`: {yaml_file!r}")
+        return
+
+    # refer to schema here https://gitlab.com/gitlab-org/advisories-community/-/blob/main/ci/schema/schema.json
+    aliases = yaml_file.get("identifiers")
+    summary = yaml_file.get("title")
+    urls = yaml_file.get("urls")
+    references = list(extract_references(urls))
+    date_published = dateparser.parse(yaml_file.get("pubdate"))
+    purl: PackageURL = get_purl(yaml_file.get("package_slug"))
+    fixed_versions = yaml_file.get("fixed_versions")
+    vrc: VersionRange = RANGE_CLASS_BY_SCHEMES[purl.type]
+    affected_range = yaml_file.get("affected_range")
+    affected_version_range = None
+    try:
+        affected_version_range = vrc.from_native(affected_range) if affected_range else None
+    except Exception:
+        LOGGER.error(f"parse_yaml_file: affected_range is not parsable`: {affected_range!r}")
+
+    if affected_version_range == NotImplementedError:
+        LOGGER.error(f"parse_yaml_file: from_native is not implemented yet for {vrc.__name__}`")
+        affected_version_range = None
+
+    affected_packages = list(
+        extract_affected_packages(affected_version_range, fixed_versions, purl)
+    )
+
+    return AdvisoryData(
+        aliases=aliases,
+        summary=summary,
+        references=references,
+        date_published=date_published,
+        affected_packages=affected_packages,
+    )