ruby initial config

Reference: #796

Signed-off-by: ziadhany <ziadhany2016@gmail.com>

Author: ziadhany

vulnerabilities/api.py

@@ -12,11 +12,11 @@
 from django.db.models import Prefetch
 from django_filters import rest_framework as filters
 from packageurl import PackageURL
-from rest_framework import serializers
-from rest_framework import viewsets
 from rest_framework.decorators import action
 from rest_framework.response import Response
 
+from rest_framework import serializers
+from rest_framework import viewsets
 from vulnerabilities.models import Alias
 from vulnerabilities.models import Package
 from vulnerabilities.models import Vulnerability

vulnerabilities/importers/ruby.py

@@ -7,132 +7,193 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
-import asyncio
-from typing import List
-from typing import Set
+import logging
+from typing import Iterable
 
 from dateutil.parser import parse
 from packageurl import PackageURL
 from pytz import UTC
-from univers.version_range import VersionRange
-from univers.versions import SemverVersion
+from univers.version_range import GemVersionRange
+from univers.versions import InvalidVersion
+from univers.versions import RubygemsVersion
 
 from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import GitConfig
 from vulnerabilities.importer import GitImporter
 from vulnerabilities.importer import Reference
+from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.package_managers import RubyVersionAPI
+from vulnerabilities.severity_systems import SCORING_SYSTEMS
+from vulnerabilities.utils import build_description
 from vulnerabilities.utils import load_yaml
 from vulnerabilities.utils import nearest_patched_package
 
+logger = logging.getLogger(__name__)
+
 
 class RubyImporter(GitImporter):
-    def __enter__(self):
-        super(RubyImporter, self).__enter__()
+    license_url = "https://github.com/rubysec/ruby-advisory-db/blob/master/LICENSE.txt"
+    spdx_license_expression = "unknow"
+    config = GitConfig(
+        repository_url="https://github.com/rubysec/ruby-advisory-db.git",
+        working_directory="ruby",
+        branch="master",
+    )
+    cutoff_timestamp = 1
+
+    def __init__(self):
+        super().__init__(config=self.config, cutoff_timestamp=self.cutoff_timestamp)
+        self.pkg_manager_api = RubyVersionAPI()
+
+        self._added_files_gems, self._updated_files_gems = self.file_changes(
+            recursive=True,
+            file_ext="yml",
+            subdir="./gems",
+        )
+
+        self._added_files_rubies, self._updated_files_rubies = self.file_changes(
+            recursive=True, file_ext="yml", subdir="./rubies"
+        )
 
-        if not getattr(self, "_added_files", None):
-            self._added_files, self._updated_files = self.file_changes(
-                recursive=True, file_ext="yml", subdir="./gems"
+    @staticmethod
+    def categorize_versions(all_versions, unaffected_version_ranges):
+        try:
+            for id, elem in enumerate(unaffected_version_ranges):
+                unaffected_version_ranges[id] = GemVersionRange.from_native(elem)
+
+            safe_versions = []
+            vulnerable_versions = []
+            for i in all_versions:
+                vobj = RubygemsVersion(i.value)
+                is_vulnerable = False
+                for ver_rng in unaffected_version_ranges:
+                    if vobj in ver_rng:
+                        safe_versions.append(i)
+                        is_vulnerable = True
+                        break
+
+                if not is_vulnerable:
+                    vulnerable_versions.append(i)
+            return safe_versions, vulnerable_versions
+        except:
+            logger.error(f"Invalid Version affected_pkg Error ")
+            return [], []
+
+    def parse_ruby_advisory(self, record, schema_type):
+        """
+        Parse a ruby advisory file and return an AdvisoryData or None.
+        Each advisory file contains the advisory information in YAML format.
+        Schema: https://github.com/rubysec/ruby-advisory-db/tree/master/spec/schemas
+        """
+        if schema_type == "gem":
+            package_name = record.get("gem")
+            library = record.get("library")  # not used
+            framework = record.get("framework")  # not used
+            platform = record.get("platform")  # not used, but I think we can make a purl from this
+            safe_version_ranges = record.get("patched_versions", [])
+            # this case happens when the advisory contain only 'patched_versions' field
+            # and it has value None(i.e it is empty :( ).
+            if not safe_version_ranges:
+                safe_version_ranges = []
+            safe_version_ranges += record.get("unaffected_versions", [])
+            safe_version_ranges = [i for i in safe_version_ranges if i]
+
+            all_vers = self.pkg_manager_api.fetch(package_name)
+            safe_versions, affected_versions = self.categorize_versions(
+                all_vers, safe_version_ranges
             )
 
-        self.pkg_manager_api = RubyVersionAPI()
-        self.set_api(self.collect_packages())
-
-    def set_api(self, packages):
-        asyncio.run(self.pkg_manager_api.load_api(packages))
-
-    def updated_advisories(self) -> Set[AdvisoryData]:
-        files = self._updated_files.union(self._added_files)
-        advisories = []
-        for f in files:
-            processed_data = self.process_file(f)
-            if processed_data:
-                advisories.append(processed_data)
-        return self.batch_advisories(advisories)
-
-    def collect_packages(self):
-        packages = set()
-        files = self._updated_files.union(self._added_files)
-        for f in files:
-            data = load_yaml(f)
-            if data.get("gem"):
-                packages.add(data["gem"])
-
-        return packages
-
-    def process_file(self, path) -> List[AdvisoryData]:
-        record = load_yaml(path)
-        package_name = record.get("gem")
-        if not package_name:
-            return
-
-        if "cve" in record:
-            cve_id = "CVE-{}".format(record["cve"])
-        else:
-            return
-
-        publish_time = parse(record["date"]).replace(tzinfo=UTC)
-        safe_version_ranges = record.get("patched_versions", [])
-        # this case happens when the advisory contain only 'patched_versions' field
-        # and it has value None(i.e it is empty :( ).
-        if not safe_version_ranges:
-            safe_version_ranges = []
-        safe_version_ranges += record.get("unaffected_versions", [])
-        safe_version_ranges = [i for i in safe_version_ranges if i]
-
-        if not getattr(self, "pkg_manager_api", None):
-            self.pkg_manager_api = RubyVersionAPI()
-        all_vers = self.pkg_manager_api.get(package_name, until=publish_time).valid_versions
-        safe_versions, affected_versions = self.categorize_versions(all_vers, safe_version_ranges)
-
-        impacted_purls = [
-            PackageURL(
-                name=package_name,
-                type="gem",
-                version=version,
+            impacted_purls = [
+                PackageURL(
+                    name=package_name,
+                    type="gem",
+                    version=version.value,
+                )
+                for version in affected_versions
+            ]
+
+            resolved_purls = [
+                PackageURL(
+                    name=package_name,
+                    type="gem",
+                    version=version.value,
+                )
+                for version in safe_versions
+            ]
+
+            affected_packages = nearest_patched_package(impacted_purls, resolved_purls)
+            return AdvisoryData(
+                aliases=get_aliases(record),
+                summary=get_summary(record),
+                affected_packages=affected_packages,
+                references=get_references(record),
+                date_published=get_publish_time(record),
             )
-            for version in affected_versions
-        ]
-
-        resolved_purls = [
-            PackageURL(
-                name=package_name,
-                type="gem",
-                version=version,
+        elif schema_type == "ruby":
+            return AdvisoryData(
+                aliases=get_aliases(record),
+                summary=get_summary(record),
+                references=get_references(record),
+                date_published=get_publish_time(record),
             )
-            for version in safe_versions
-        ]
 
-        references = []
-        if record.get("url"):
-            references.append(Reference(url=record.get("url")))
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        files = self._updated_files_gems.union(self._added_files_gems)
+        for file in files:
+            raw_data = load_yaml(file)
+            yield self.parse_ruby_advisory(raw_data, schema_type="gem")
 
-        return AdvisoryData(
-            summary=record.get("description", ""),
-            affected_packages=nearest_patched_package(impacted_purls, resolved_purls),
-            references=references,
-            vulnerability_id=cve_id,
-        )
+        files = self._added_files_rubies.union(self._updated_files_rubies)
+        for file in files:
+            raw_data = load_yaml(file)
+            yield self.parse_ruby_advisory(raw_data, schema_type="ruby")
 
-    @staticmethod
-    def categorize_versions(all_versions, unaffected_version_ranges):
 
-        for id, elem in enumerate(unaffected_version_ranges):
-            unaffected_version_ranges[id] = VersionRange.from_scheme_version_spec_string(
-                "semver", elem
+def get_aliases(record) -> [str]:
+    aliases = []
+    if record.get("cve"):
+        aliases.append("CVE-{}".format(record.get("cve")))
+    if record.get("osvdb"):
+        aliases.append("OSV-{}".format(record.get("osvdb")))
+    if record.get("ghsa"):
+        aliases.append("GHSA-{}".format(record.get("ghsa")))
+    return aliases
+
+
+def get_references(record) -> [Reference]:
+    references = []
+    cvss_v2 = record.get("cvss_v2")
+    cvss_v3 = record.get("cvss_v3")
+
+    if record.get("url"):
+        if not (cvss_v2 or cvss_v3):
+            references.append(Reference(url=record.get("url")))
+        if cvss_v2:
+            references.append(
+                Reference(
+                    url=record.get("url"),
+                    severities=[
+                        VulnerabilitySeverity(system=SCORING_SYSTEMS["cvssv2"], value=cvss_v2)
+                    ],
+                )
+            )
+        if cvss_v3:
+            references.append(
+                Reference(
+                    url=record.get("url"),
+                    severities=[
+                        VulnerabilitySeverity(system=SCORING_SYSTEMS["cvssv3"], value=cvss_v3)
+                    ],
+                )
             )
+    return references
+
+
+def get_publish_time(record):
+    return parse(record["date"]).replace(tzinfo=UTC)
+
 
-        safe_versions = []
-        vulnerable_versions = []
-        for i in all_versions:
-            vobj = SemverVersion(i)
-            is_vulnerable = False
-            for ver_rng in unaffected_version_ranges:
-                if vobj in ver_rng:
-                    safe_versions.append(i)
-                    is_vulnerable = True
-                    break
-
-            if not is_vulnerable:
-                vulnerable_versions.append(i)
-
-        return safe_versions, vulnerable_versions
+def get_summary(record):
+    title = record.get("title")
+    description = record.get("description", "")
+    return build_description(summary=title, description=description)