Merge remote-tracking branch 'origin/main' into openssl-advisories

Author: kunalsz

.github/workflows/docs.yml

@@ -4,7 +4,7 @@ on: [push, pull_request]
 
 jobs:
   build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
 
     strategy:
       max-parallel: 4

.github/workflows/main.yml

@@ -9,7 +9,7 @@ env:
 
 jobs:
   build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
 
     services:
       postgres:

.github/workflows/pypi-release.yml

@@ -21,7 +21,7 @@ on:
 jobs:
   build-pypi-distribs:
     name: Build and publish library to PyPI
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
 
     steps:
       - uses: actions/checkout@master

docs/source/conf.py

@@ -36,6 +36,8 @@
     "https://www.softwaretestinghelp.com/how-to-write-good-bug-report/",  # Cloudflare protection
     "https://www.openssl.org/news/vulnerabilities.xml",  # OpenSSL legacy advisory URL, not longer available
     "https://example.org/api/non-existent-packages",
+    "https://github.com/aboutcode-org/vulnerablecode/pull/495/commits",
+    "https://nvd.nist.gov/products/cpe",
 ]
 
 # Add any Sphinx extension module names here, as strings. They can be

vulnerabilities/import_runner.py

@@ -104,24 +104,30 @@ def process_advisories(
         advisories = []
         for data in advisory_datas:
             content_id = compute_content_id(advisory_data=data)
+            advisory = {
+                "summary": data.summary,
+                "affected_packages": [pkg.to_dict() for pkg in data.affected_packages],
+                "references": [ref.to_dict() for ref in data.references],
+                "date_published": data.date_published,
+                "weaknesses": data.weaknesses,
+                "created_by": importer_name,
+                "date_collected": datetime.datetime.now(tz=datetime.timezone.utc),
+            }
             try:
                 aliases = get_or_create_aliases(aliases=data.aliases)
                 obj, created = Advisory.objects.get_or_create(
                     unique_content_id=content_id,
                     url=data.url,
-                    defaults={
-                        "summary": data.summary,
-                        "affected_packages": [pkg.to_dict() for pkg in data.affected_packages],
-                        "references": [ref.to_dict() for ref in data.references],
-                        "date_published": data.date_published,
-                        "weaknesses": data.weaknesses,
-                        "created_by": importer_name,
-                        "date_collected": datetime.datetime.now(tz=datetime.timezone.utc),
-                    },
+                    defaults=advisory,
                 )
                 obj.aliases.add(*aliases)
                 if not obj.date_imported:
                     advisories.append(obj)
+            except Advisory.MultipleObjectsReturned:
+                logger.error(
+                    f"Multiple Advisories returned: unique_content_id: {content_id}, url: {data.url}, advisory: {advisory!r}"
+                )
+                raise
             except Exception as e:
                 logger.error(
                     f"Error while processing {data!r} with aliases {data.aliases!r}: {e!r} \n {traceback_format_exc()}"

vulnerabilities/migrations/0091_alter_advisory_unique_together_and_more.py

@@ -0,0 +1,31 @@
+# Generated by Django 4.2.17 on 2025-04-04 16:08
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0090_migrate_advisory_aliases"),
+    ]
+
+    operations = [
+        migrations.AlterUniqueTogether(
+            name="advisory",
+            unique_together=set(),
+        ),
+        migrations.AlterField(
+            model_name="advisory",
+            name="unique_content_id",
+            field=models.CharField(
+                help_text="A 64 character unique identifier for the content of the advisory since we use sha256 as hex",
+                max_length=64,
+                unique=True,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="advisory",
+            name="url",
+            field=models.URLField(help_text="Link to the advisory on the upstream website"),
+        ),
+    ]

vulnerabilities/models.py

@@ -1321,6 +1321,7 @@ class Advisory(models.Model):
         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",
     )
     aliases = models.ManyToManyField(
@@ -1355,14 +1356,14 @@ class Advisory(models.Model):
         "vulnerabilities.pipeline.nginx_importer.NginxImporterPipeline",
     )
     url = models.URLField(
-        blank=True,
+        blank=False,
+        null=False,
         help_text="Link to the advisory on the upstream website",
     )
 
     objects = AdvisoryQuerySet.as_manager()
 
     class Meta:
-        unique_together = ["unique_content_id", "date_published", "url"]
         ordering = ["date_published", "unique_content_id"]
 
     def save(self, *args, **kwargs):

vulnerabilities/pipelines/alpine_linux_importer.py

@@ -195,7 +195,8 @@ def load_advisories(
                     level=logging.DEBUG,
                 )
             continue
-
+        # fixed_vulns is a list of strings and each string is a space-separated
+        # list of aliases and CVES
         for vuln_ids in fixed_vulns:
             if not isinstance(vuln_ids, str):
                 if logger:
@@ -204,15 +205,16 @@ def load_advisories(
                         level=logging.DEBUG,
                     )
                 continue
-            vuln_ids = vuln_ids.split()
-            aliases = []
-            vuln_id = vuln_ids[0]
-            # check for valid vuln ID, if there is valid vuln ID then iterate over
-            # the remaining elements of the list else iterate over the whole list
-            # and also check if the initial element is a reference or not
-            if is_cve(vuln_id):
-                aliases = [vuln_id]
-                vuln_ids = vuln_ids[1:]
+            vuln_ids = vuln_ids.strip().split()
+            if not vuln_ids:
+                if logger:
+                    logger(
+                        f"{vuln_ids!r} is empty",
+                        level=logging.DEBUG,
+                    )
+                continue
+            aliases = vuln_ids
+
             references = []
             for reference_id in vuln_ids:
 
@@ -225,6 +227,10 @@ def load_advisories(
                 elif reference_id.startswith("wnpa-sec"):
                     references.append(WireSharkReference.from_id(wnpa_sec_id=reference_id))
 
+                elif not reference_id.startswith("CVE"):
+                    if logger:
+                        logger(f"Unknown reference id {reference_id!r}", level=logging.DEBUG)
+
             qualifiers = {
                 "distroversion": distroversion,
                 "reponame": reponame,

vulnerabilities/pipes/advisory.py

@@ -43,20 +43,27 @@ def insert_advisory(advisory: AdvisoryData, pipeline_id: str, logger: Callable =
     aliases = get_or_create_aliases(aliases=advisory.aliases)
     content_id = compute_content_id(advisory_data=advisory)
     try:
+        default_data = {
+            "summary": advisory.summary,
+            "affected_packages": [pkg.to_dict() for pkg in advisory.affected_packages],
+            "references": [ref.to_dict() for ref in advisory.references],
+            "date_published": advisory.date_published,
+            "weaknesses": advisory.weaknesses,
+            "created_by": pipeline_id,
+            "date_collected": datetime.now(timezone.utc),
+        }
+
         advisory_obj, _ = Advisory.objects.get_or_create(
             unique_content_id=content_id,
             url=advisory.url,
-            defaults={
-                "summary": advisory.summary,
-                "affected_packages": [pkg.to_dict() for pkg in advisory.affected_packages],
-                "references": [ref.to_dict() for ref in advisory.references],
-                "date_published": advisory.date_published,
-                "weaknesses": advisory.weaknesses,
-                "created_by": pipeline_id,
-                "date_collected": datetime.now(timezone.utc),
-            },
+            defaults=default_data,
         )
         advisory_obj.aliases.add(*aliases)
+    except Advisory.MultipleObjectsReturned:
+        logger.error(
+            f"Multiple Advisories returned: unique_content_id: {content_id}, url: {advisory.url}, advisory: {advisory!r}"
+        )
+        raise
     except Exception as e:
         if logger:
             logger(
@@ -137,19 +144,18 @@ def import_advisory(
                     },
                 )
                 vulnerability.severities.add(vulnerability_severity)
+                if not created and logger:
+                    logger(
+                        f"Severity updated for reference {ref.url!r} to value: {severity.value!r} "
+                        f"and scoring_elements: {severity.scoring_elements!r}",
+                        level=logging.DEBUG,
+                    )
             except:
                 if logger:
                     logger(
                         f"Failed to create VulnerabilitySeverity for: {severity} with error:\n{traceback_format_exc()}",
                         level=logging.ERROR,
                     )
-            if not created:
-                if logger:
-                    logger(
-                        f"Severity updated for reference {ref.url!r} to value: {severity.value!r} "
-                        f"and scoring_elements: {severity.scoring_elements!r}",
-                        level=logging.DEBUG,
-                    )
 
     for affected_purl in affected_purls or []:
         vulnerable_package, _ = Package.objects.get_or_create_from_purl(purl=affected_purl)

vulnerabilities/tests/conftest.py

@@ -25,7 +25,6 @@ def no_rmtree(monkeypatch):
 # Step 2: Run test for importer only if it is activated (pytestmark = pytest.mark.skipif(...))
 # Step 3: Migrate all the tests
 collect_ignore = [
-    "test_models.py",
     "test_rust.py",
     "test_suse_backports.py",
     "test_suse.py",