Add gitlab importer

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

Author: TG1999

requirements.txt

@@ -107,9 +107,10 @@ toml==0.10.2
 tomli==2.0.1
 traitlets==5.1.1
 typing_extensions==4.1.1
-univers==30.4.0
+univers==30.5.1
 urllib3==1.26.9
 wcwidth==0.2.5
 websocket-client==0.59.0
 yarl==1.7.2
 zipp==3.8.0
+fetchcode==0.1.0

setup.cfg

@@ -80,6 +80,7 @@ install_requires =
     GitPython>=3.1.17
     aiohttp>=3.7.4.post0
     requests>=2.25.1
+    fetchcode>=0.1.0
 
 [options.extras_require]
 dev =

vulnerabilities/importers/__init__.py

@@ -21,6 +21,7 @@
 #  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
 from vulnerabilities.importers import nvd
 from vulnerabilities.importers import openssl
@@ -31,6 +32,7 @@
     github.GitHubAPIImporter,
     nvd.NVDImporter,
     openssl.OpensslImporter,
+    gitlab.GitLabAPIImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/gitlab.py

@@ -0,0 +1,190 @@
+# 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
+
+
+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)
+
+
+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:
+            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,
+    version_class: Version,
+) -> Iterable[AffectedPackage]:
+    """
+    Yield a list of AffectedPackage objects
+    """
+    for fixed_version in fixed_versions or []:
+        yield AffectedPackage(
+            package=purl,
+            fixed_version=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
+    version_class = vrc.version_class
+    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, version_class)
+    )
+
+    return AdvisoryData(
+        aliases=aliases,
+        summary=summary,
+        references=references,
+        date_published=date_published,
+        affected_packages=affected_packages,
+    )