Merge branch 'main' into add-cwe-support-in-multiple-importers

Author: TG1999

Makefile

@@ -42,6 +42,10 @@ else
 	SUDO_POSTGRES=
 endif
 
+ifeq ($(UNAME), Darwin)
+	GET_SECRET_KEY=`head /dev/urandom | base64 | head -c50`
+endif
+
 virtualenv:
 	@echo "-> Bootstrap the virtualenv with PYTHON_EXE=${PYTHON_EXE}"
 	@${PYTHON_EXE} ${VIRTUALENV_PYZ} --never-download --no-periodic-update ${VENV}

vulnerabilities/importer.py

@@ -111,7 +111,7 @@ def to_dict(self):
     def from_dict(cls, ref: dict):
         return cls(
             reference_id=ref["reference_id"],
-            reference_type=ref["reference_type"],
+            reference_type=ref.get("reference_type") or "",
             url=ref["url"],
             severities=[
                 VulnerabilitySeverity.from_dict(severity) for severity in ref["severities"]

vulnerabilities/migrations/0088_fix_alpine_purl_type.py

@@ -0,0 +1,103 @@
+from datetime import datetime
+from datetime import timezone
+
+from aboutcode.pipeline import LoopProgress
+from django.db import migrations
+from packageurl import PackageURL
+
+CHUNK_SIZE = 50000
+BATCH_SIZE = 500
+
+
+class Migration(migrations.Migration):
+    def fix_alpine_purl_type(apps, schema_editor):
+        """Use proper apk package type for Alpine"""
+
+        Package = apps.get_model("vulnerabilities", "Package")
+        batch = []
+        alpine_packages_query = Package.objects.filter(type="alpine")
+
+        log(f"\nFixing PURL for {alpine_packages_query.count():,d} alpine packages")
+        progress = LoopProgress(
+            total_iterations=alpine_packages_query.count(),
+            progress_step=10,
+            logger=log,
+        )
+        for package in progress.iter(alpine_packages_query.iterator(chunk_size=CHUNK_SIZE)):
+            package.type = "apk"
+            package.namespace = "alpine"
+
+            package.package_url = update_alpine_purl(package.package_url, "apk", "alpine")
+            package.plain_package_url = update_alpine_purl(
+                package.plain_package_url, "apk", "alpine"
+            )
+
+            batch.append(package)
+            if len(batch) >= BATCH_SIZE:
+                bulk_update_package(Package, batch)
+                batch.clear()
+
+        bulk_update_package(Package, batch)
+
+    def reverse_fix_alpine_purl_type(apps, schema_editor):
+        Package = apps.get_model("vulnerabilities", "Package")
+        batch = []
+        alpine_packages_query = Package.objects.filter(type="apk", namespace="alpine")
+
+        log(f"\nREVERSE: Fix for {alpine_packages_query.count():,d} alpine packages")
+        progress = LoopProgress(
+            total_iterations=alpine_packages_query.count(),
+            progress_step=10,
+            logger=log,
+        )
+        for package in progress.iter(alpine_packages_query.iterator(chunk_size=CHUNK_SIZE)):
+            package.type = "alpine"
+            package.namespace = ""
+
+            package.package_url = update_alpine_purl(package.package_url, "alpine", "")
+            package.plain_package_url = update_alpine_purl(package.plain_package_url, "alpine", "")
+
+            batch.append(package)
+            if len(batch) >= BATCH_SIZE:
+                bulk_update_package(Package, batch)
+                batch.clear()
+
+        bulk_update_package(Package, batch)
+
+    dependencies = [
+        ("vulnerabilities", "0087_update_alpine_advisory_created_by"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            code=fix_alpine_purl_type,
+            reverse_code=reverse_fix_alpine_purl_type,
+        ),
+    ]
+
+
+def bulk_update_package(package, batch):
+    if batch:
+        package.objects.bulk_update(
+            objs=batch,
+            fields=[
+                "type",
+                "namespace",
+                "package_url",
+                "plain_package_url",
+            ],
+        )
+
+
+def update_alpine_purl(purl, purl_type, purl_namespace):
+    package_url = PackageURL.from_string(purl).to_dict()
+    package_url["type"] = purl_type
+    package_url["namespace"] = purl_namespace
+    return str(PackageURL(**package_url))
+
+
+def log(message):
+    now_local = datetime.now(timezone.utc).astimezone()
+    timestamp = now_local.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+    message = f"{timestamp} {message}"
+    print(message)

vulnerabilities/models.py

@@ -7,16 +7,23 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
+import csv
 import hashlib
 import json
 import logging
-import typing
+import xml.etree.ElementTree as ET
 from contextlib import suppress
 from functools import cached_property
-from typing import Optional
+from itertools import groupby
+from operator import attrgetter
 from typing import Union
 
+from cvss.exceptions import CVSS2MalformedError
+from cvss.exceptions import CVSS3MalformedError
+from cvss.exceptions import CVSS4MalformedError
 from cwe2.database import Database
+from cwe2.mappings import xml_database_path
+from cwe2.weakness import Weakness as DBWeakness
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import UserManager
 from django.core import exceptions
@@ -43,8 +50,8 @@
 from univers.version_range import AlpineLinuxVersionRange
 from univers.versions import Version
 
-from aboutcode import hashid
 from vulnerabilities import utils
+from vulnerabilities.severity_systems import EPSS
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
 from vulnerabilities.utils import normalize_purl
 from vulnerabilities.utils import purl_to_dict
@@ -56,7 +63,7 @@
 models.CharField.register_lookup(Trim)
 
 # patch univers for missing entry
-RANGE_CLASS_BY_SCHEMES["alpine"] = AlpineLinuxVersionRange
+RANGE_CLASS_BY_SCHEMES["apk"] = AlpineLinuxVersionRange
 
 
 class BaseQuerySet(models.QuerySet):
@@ -373,6 +380,127 @@ def get_related_purls(self):
         """
         return [p.package_url for p in self.packages.distinct().all()]
 
+    def aggregate_fixed_and_affected_packages(self):
+        from vulnerabilities.utils import get_purl_version_class
+
+        sorted_fixed_by_packages = self.fixed_by_packages.filter(is_ghost=False).order_by(
+            "type", "namespace", "name", "qualifiers", "subpath"
+        )
+
+        if sorted_fixed_by_packages:
+            sorted_fixed_by_packages.first().calculate_version_rank
+
+        sorted_affected_packages = self.affected_packages.all()
+
+        if sorted_affected_packages:
+            sorted_affected_packages.first().calculate_version_rank
+
+        grouped_fixed_by_packages = {
+            key: list(group)
+            for key, group in groupby(
+                sorted_fixed_by_packages,
+                key=attrgetter("type", "namespace", "name", "qualifiers", "subpath"),
+            )
+        }
+
+        all_affected_fixed_by_matches = []
+
+        for sorted_affected_package in sorted_affected_packages:
+            affected_fixed_by_matches = {
+                "affected_package": sorted_affected_package,
+                "matched_fixed_by_packages": [],
+            }
+
+            # Build the key to find matching group
+            key = (
+                sorted_affected_package.type,
+                sorted_affected_package.namespace,
+                sorted_affected_package.name,
+                sorted_affected_package.qualifiers,
+                sorted_affected_package.subpath,
+            )
+
+            # Get matching group from pre-grouped fixed_by_packages
+            matching_fixed_packages = grouped_fixed_by_packages.get(key, [])
+
+            # Get version classes for comparison
+            affected_version_class = get_purl_version_class(sorted_affected_package)
+            affected_version = affected_version_class(sorted_affected_package.version)
+
+            # Compare versions and filter valid matches
+            matched_fixed_by_packages = [
+                fixed_by_package.purl
+                for fixed_by_package in matching_fixed_packages
+                if get_purl_version_class(fixed_by_package)(fixed_by_package.version)
+                > affected_version
+            ]
+
+            affected_fixed_by_matches["matched_fixed_by_packages"] = matched_fixed_by_packages
+            all_affected_fixed_by_matches.append(affected_fixed_by_matches)
+        return sorted_fixed_by_packages, sorted_affected_packages, all_affected_fixed_by_matches
+
+    def get_severity_vectors_and_values(self):
+        """
+        Collect severity vectors and values, excluding EPSS scoring systems and handling errors gracefully.
+        """
+        severity_vectors = []
+        severity_values = set()
+
+        # Exclude EPSS scoring system
+        base_severities = self.severities.exclude(scoring_system=EPSS.identifier)
+
+        # QuerySet for severities with valid scoring_elements and scoring_system in SCORING_SYSTEMS
+        valid_scoring_severities = base_severities.filter(
+            scoring_elements__isnull=False, scoring_system__in=SCORING_SYSTEMS.keys()
+        )
+
+        for severity in valid_scoring_severities:
+            try:
+                vector_values = SCORING_SYSTEMS[severity.scoring_system].get(
+                    severity.scoring_elements
+                )
+                if vector_values:
+                    severity_vectors.append(vector_values)
+            except (
+                CVSS2MalformedError,
+                CVSS3MalformedError,
+                CVSS4MalformedError,
+                NotImplementedError,
+            ) as e:
+                logging.error(f"CVSSMalformedError for {severity.scoring_elements}: {e}")
+
+        valid_value_severities = base_severities.filter(value__isnull=False).exclude(value="")
+
+        severity_values.update(valid_value_severities.values_list("value", flat=True))
+
+        return severity_vectors, severity_values
+
+
+def get_cwes(self):
+    """Yield CWE Weakness objects"""
+    for cwe_category in self.cwe_files:
+        cwe_category.seek(0)
+        reader = csv.DictReader(cwe_category)
+        for row in reader:
+            yield DBWeakness(*list(row.values())[0:-1])
+    tree = ET.parse(xml_database_path)
+    root = tree.getroot()
+    for tag_num in [1, 2]:  # Categories , Views
+        tag = root[tag_num]
+        for child in tag:
+            yield DBWeakness(
+                *[
+                    child.attrib["ID"],
+                    child.attrib.get("Name"),
+                    None,
+                    child.attrib.get("Status"),
+                    child[0].text,
+                ]
+            )
+
+
+Database.get_cwes = get_cwes
+
 
 class Weakness(models.Model):
     """
@@ -381,7 +509,15 @@ class Weakness(models.Model):
 
     cwe_id = models.IntegerField(help_text="CWE id")
     vulnerabilities = models.ManyToManyField(Vulnerability, related_name="weaknesses")
-    db = Database()
+
+    cwe_by_id = {}
+
+    def get_cwe(self, cwe_id):
+        if not self.cwe_by_id:
+            db = Database()
+            for weakness in db.get_cwes():
+                self.cwe_by_id[str(weakness.cwe_id)] = weakness
+        return self.cwe_by_id[cwe_id]
 
     @property
     def cwe(self):
@@ -393,7 +529,7 @@ def weakness(self):
         Return a queryset of Weakness for this vulnerability.
         """
         try:
-            weakness = self.db.get(self.cwe_id)
+            weakness = self.get_cwe(str(self.cwe_id))
             return weakness
         except Exception as e:
             logger.warning(f"Could not find CWE {self.cwe_id}: {e}")

vulnerabilities/pipelines/alpine_linux_importer.py

@@ -254,7 +254,8 @@ def load_advisories(
                     affected_packages.append(
                         AffectedPackage(
                             package=PackageURL(
-                                type="alpine",
+                                type="apk",
+                                namespace="alpine",
                                 name=pkg_infos["name"],
                                 qualifiers=qualifiers,
                             ),
@@ -266,7 +267,8 @@ def load_advisories(
                 affected_packages.append(
                     AffectedPackage(
                         package=PackageURL(
-                            type="alpine",
+                            type="apk",
+                            namespace="alpine",
                             name=pkg_infos["name"],
                             qualifiers=qualifiers,
                         ),

vulnerabilities/templates/index.html

@@ -6,8 +6,33 @@
 {% endblock %}
 
 {% block content %}
-    <section class="section pt-0">
-       {% include "package_search_box.html" %}
-       {% include "vulnerability_search_box.html" %}
-    </section>
-{% endblock %}
+<section class="section pt-2">
+    <div class="container">
+        <div class="columns is-centered mb-5 mt-2">
+            <div class="column is-full-tablet is-full-desktop">
+                {% include "vulnerability_search_box.html" %}
+            </div>
+        </div>
+        <div class="columns is-centered mb-5">
+            <div class="column is-full-tablet is-full-desktop">
+                {% include "package_search_box.html" %}
+            </div>
+        </div>
+        <div class="notification is-info is-light has-text-centered">
+            <p class="is-size-6 has-text-grey-dark has-text-centered mb-3">
+                <strong>VulnerableCode</strong> aggregates software
+                vulnerabilities from multiple public advisory sources
+                and presents their details along with their affected
+                packages and fixed-by packages identified by
+                Package URLs (PURLs).
+            </p>
+            <p class="is-size-5">
+                <strong>What's new in this Release:</strong>
+                <a href="{{ release_url }}" target="_blank" class="has-text-link is-underlined">
+                    Check out latest updates here!
+                </a>
+            </p>
+        </div>
+    </div>
+</section>
+{% endblock %}