Change API design

Signed-off-by: Tushar Goel <[email protected]>

Author: TG1999

vulnerabilities/api_v2.py

@@ -20,6 +20,7 @@
 from vulnerabilities.pipelines import flag_ghost_packages
 from vulnerabilities.pipelines import populate_vulnerability_summary_pipeline
 from vulnerabilities.pipelines import remove_duplicate_advisories
+from vulnerabilities.pipelines.v2_improvers import collect_commits as collect_commits_v2
 from vulnerabilities.pipelines.v2_improvers import compute_package_risk as compute_package_risk_v2
 from vulnerabilities.pipelines.v2_improvers import (
     computer_package_version_rank as compute_version_rank_v2,
@@ -65,6 +66,7 @@
     enhance_with_metasploit_v2.MetasploitImproverPipeline,
     compute_package_risk_v2.ComputePackageRiskPipeline,
     compute_version_rank_v2.ComputeVersionRankPipeline,
+    collect_commits_v2.CollectFixCommitsPipeline,
 ]
 
 IMPROVERS_REGISTRY = {

vulnerabilities/improvers/__init__.py

@@ -1323,7 +1323,7 @@ def url(self):
             return f"https://github.com/nodejs/security-wg/blob/main/vuln/npm/{id}.json"
 
 
-class AdvisoryQuerySet(BaseQuerySet):
+class AdvisoryV2QuerySet(BaseQuerySet):
     def search(query):
         """
         This function will take a string as an input, the string could be an alias or an advisory ID or
@@ -1337,6 +1337,20 @@ def search(query):
         ).distinct()
 
 
+class AdvisoryQuerySet(BaseQuerySet):
+    def search(query):
+        """
+        This function will take a string as an input, the string could be an alias or an advisory ID or
+        something in the advisory description.
+        """
+        return Advisory.objects.filter(
+            Q(advisory_id__icontains=query)
+            | Q(aliases__alias__icontains=query)
+            | Q(summary__icontains=query)
+            | Q(references__url__icontains=query)
+        ).distinct()
+
+
 # FIXME: Remove when migration from Vulnerability to Advisory is completed
 class Advisory(models.Model):
     """
@@ -1820,6 +1834,60 @@ class Meta:
         abstract = True
 
 
+class CodeChangeV2(models.Model):
+    """
+    Abstract base model representing a change in code, either introducing or fixing a vulnerability.
+    This includes details about commits, patches, and related metadata.
+
+    We are tracking commits, pulls and downloads as references to the code change. The goal is to
+    keep track and store the actual code patch in the ``patch`` field. When not available the patch
+    will be inferred from these references using improvers.
+    """
+
+    commits = models.JSONField(
+        blank=True,
+        default=list,
+        help_text="List of commit identifiers using VCS URLs associated with the code change.",
+    )
+    pulls = models.JSONField(
+        blank=True,
+        default=list,
+        help_text="List of pull request URLs associated with the code change.",
+    )
+    downloads = models.JSONField(
+        blank=True, default=list, help_text="List of download URLs for the patched code."
+    )
+    patch = models.TextField(
+        blank=True, null=True, help_text="The code change as a patch in unified diff format."
+    )
+    base_package_version = models.ForeignKey(
+        "PackageV2",
+        null=True,
+        blank=True,
+        on_delete=models.SET_NULL,
+        related_name="codechanges_v2",
+        help_text="The base package version to which this code change applies.",
+    )
+    notes = models.TextField(
+        blank=True, null=True, help_text="Notes or instructions about this code change."
+    )
+    references = models.JSONField(
+        blank=True, default=list, help_text="URL references related to this code change."
+    )
+    is_reviewed = models.BooleanField(
+        default=False, help_text="Indicates if this code change has been reviewed."
+    )
+    created_at = models.DateTimeField(
+        auto_now_add=True, help_text="Timestamp indicating when this code change was created."
+    )
+    updated_at = models.DateTimeField(
+        auto_now=True, help_text="Timestamp indicating when this code change was last updated."
+    )
+
+    class Meta:
+        abstract = True
+
+
 class CodeFix(CodeChange):
     """
     A code fix is a code change that addresses a vulnerability and is associated:
@@ -1844,6 +1912,35 @@ class CodeFix(CodeChange):
     )
 
 
+class CodeFixV2(CodeChangeV2):
+    """
+    A code fix is a code change that addresses a vulnerability and is associated:
+    - with a specific advisory
+    - package that has been affected
+    - optionally with a specific fixing package version when it is known
+    """
+
+    advisory = models.ForeignKey(
+        "AdvisoryV2",
+        on_delete=models.CASCADE,
+        related_name="code_fix_v2",
+        help_text="The affected package version to which this code fix applies.",
+    )
+
+    affected_package = models.ForeignKey(
+        "PackageV2", on_delete=models.CASCADE, related_name="code_fix_v2_affected"
+    )
+
+    fixed_package = models.ForeignKey(
+        "PackageV2",
+        null=True,
+        blank=True,
+        on_delete=models.SET_NULL,
+        related_name="code_fix_v2_fixed",
+        help_text="The fixing package version with this code fix",
+    )
+
+
 class PipelineRun(models.Model):
     """The Database representation of a pipeline execution."""
 
@@ -2451,6 +2548,23 @@ class AdvisoryV2(models.Model):
     into structured data
     """
 
+    # This is similar to a type or a namespace
+    datasource_id = models.CharField(
+        max_length=100,
+        blank=False,
+        null=False,
+        help_text="Unique ID for the datasource used for this advisory ." "e.g.: nginx_importer_v2",
+    )
+
+    avid = models.CharField(
+        max_length=500,
+        blank=False,
+        null=False,
+        help_text="Unique ID for the datasource used for this advisory ."
+        "e.g.: pysec_importer_v2/PYSEC-2020-2233",
+    )
+
+    # This is similar to a name
     advisory_id = models.CharField(
         max_length=50,
         blank=False,
@@ -2460,13 +2574,27 @@ class AdvisoryV2(models.Model):
         "such as PYSEC-2020-2233",
     )
 
+    # This is similar to a version
     unique_content_id = models.CharField(
         max_length=64,
         blank=False,
         null=False,
         unique=True,
         help_text="A 64 character unique identifier for the content of the advisory since we use sha256 as hex",
     )
+    url = models.URLField(
+        blank=False,
+        null=False,
+        help_text="Link to the advisory on the upstream website",
+    )
+
+    # TODO: Have a mapping that gives datasource class by datasource ID
+    # Get label from datasource class
+    # Remove this from model
+    # In the UI - Use label
+    # In the API - Use datasource_id
+    # Have an API endpoint for all info for datasources - show license, label
+
     summary = models.TextField(
         blank=True,
     )
@@ -2497,18 +2625,6 @@ class AdvisoryV2(models.Model):
     date_imported = models.DateTimeField(
         blank=True, null=True, help_text="UTC Date on which the advisory was imported"
     )
-    # TODO: Rename to datasource ID
-    datasource_ID = models.CharField(
-        max_length=100,
-        help_text="Fully qualified name of the importer prefixed with the"
-        "module name importing the advisory. Eg:"
-        "nginx_importer_v2",
-    )
-    url = models.URLField(
-        blank=False,
-        null=False,
-        help_text="Link to the advisory on the upstream website",
-    )
 
     affecting_packages = models.ManyToManyField(
         "PackageV2",
@@ -2558,7 +2674,8 @@ def risk_score(self):
     objects = AdvisoryQuerySet.as_manager()
 
     class Meta:
-        ordering = ["date_published", "unique_content_id"]
+        unique_together = ["datasource_id", "advisory_id", "unique_content_id"]
+        ordering = ["datasource_id", "advisory_id", "date_published", "unique_content_id"]
 
     def save(self, *args, **kwargs):
         self.full_clean()

vulnerabilities/migrations/0093_advisoryalias_advisoryreference_advisoryseverity_and_more.py

@@ -273,6 +273,7 @@ class VulnerableCodeBaseImporterPipelineV2(VulnerableCodePipeline):
 
     pipeline_id = None  # Unique Pipeline ID, this should be the name of pipeline module.
     license_url = None
+    label = None
     spdx_license_expression = None
     repo_url = None
     importer_name = None

vulnerabilities/models.py

@@ -140,10 +140,12 @@ def get_weaknesses(cve_data):
 
 class ApacheHTTPDImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     pipeline_id = "apache_httpd_importer_v2"
+    label = "Apache-Httpd"
     spdx_license_expression = "Apache-2.0"
     license_url = "https://www.apache.org/licenses/LICENSE-2.0"
     importer_name = "Apache HTTPD Importer"
     base_url = "https://httpd.apache.org/security/json/"
+    unfurl_version_ranges = True
 
     links = []
 

vulnerabilities/pipelines/__init__.py

@@ -0,0 +1,120 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from pathlib import Path
+from typing import Iterable
+
+from dateutil import parser as dateparser
+from fetchcode.vcs import fetch_via_vcs
+from packageurl import PackageURL
+from univers.version_constraint import VersionConstraint
+from univers.version_range import HexVersionRange
+
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import Reference
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+from vulnerabilities.utils import is_cve
+from vulnerabilities.utils import load_yaml
+
+
+class ElixirSecurityImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
+
+    pipeline_id = "elixir_security_importer_v2"
+    label = "Elixir Security"
+    repo_url = "git+https://github.com/dependabot/elixir-security-advisories"
+    license_url = "https://github.com/dependabot/elixir-security-advisories/blob/master/LICENSE.txt"
+    spdx_license_expression = "CC0-1.0"
+    importer_name = "Elixir Security Importer"
+
+    @classmethod
+    def steps(cls):
+        return (cls.collect_and_store_advisories,)
+
+    def clone(self):
+        self.log(f"Cloning `{self.repo_url}`")
+        self.vcs_response = fetch_via_vcs(self.repo_url)
+
+    def advisories_count(self) -> int:
+        base_path = Path(self.vcs_response.dest_dir)
+        count = len(list((base_path / "packages").glob("**/*.yml")))
+        return count
+
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        try:
+            base_path = Path(self.vcs_response.dest_dir)
+            vuln = base_path / "packages"
+            for file in vuln.glob("**/*.yml"):
+                yield from self.process_file(file, base_path)
+        finally:
+            if self.vcs_response:
+                self.vcs_response.delete()
+
+    def process_file(self, file, base_path) -> Iterable[AdvisoryData]:
+        relative_path = str(file.relative_to(base_path)).strip("/")
+        advisory_url = (
+            f"https://github.com/dependabot/elixir-security-advisories/blob/master/{relative_path}"
+        )
+        yaml_file = load_yaml(str(file))
+
+        summary = yaml_file.get("description") or ""
+        pkg_name = yaml_file.get("package") or ""
+
+        cve_id = ""
+        cve = yaml_file.get("cve") or ""
+        if cve and not cve.startswith("CVE-"):
+            cve_id = f"CVE-{cve}"
+        elif cve:
+            cve_id = cve
+
+        if not cve_id or not is_cve(cve_id):
+            return
+
+        references = []
+        link = yaml_file.get("link") or ""
+        if link:
+            references.append(Reference(url=link))
+
+        constraints = []
+        vrc = HexVersionRange.version_class
+        unaffected_versions = yaml_file.get("unaffected_versions") or []
+        patched_versions = yaml_file.get("patched_versions") or []
+
+        for version in unaffected_versions:
+            constraints.append(VersionConstraint.from_string(version_class=vrc, string=version))
+
+        for version in patched_versions:
+            if version.startswith("~>"):
+                version = version[2:]
+            constraints.append(
+                VersionConstraint.from_string(version_class=vrc, string=version).invert()
+            )
+
+        affected_packages = []
+        if pkg_name:
+            affected_packages.append(
+                AffectedPackage(
+                    package=PackageURL(type="hex", name=pkg_name),
+                    affected_version_range=HexVersionRange(constraints=constraints),
+                )
+            )
+
+        date_published = None
+        if yaml_file.get("disclosure_date"):
+            date_published = dateparser.parse(yaml_file.get("disclosure_date"))
+
+        yield AdvisoryData(
+            advisory_id=cve_id,
+            aliases=[],
+            summary=summary,
+            references_v2=references,
+            affected_packages=affected_packages,
+            url=advisory_url,
+            date_published=date_published,
+        )