Merge latest main

Signed-off-by: Philippe Ombredanne <[email protected]>

Author: pombredanne

vulnerabilities/migrations/0022_alter_vulnerability_vulnerability_id.py

@@ -0,0 +1,19 @@
+# Generated by Django 4.0.4 on 2022-09-06 11:22
+
+from django.db import migrations, models
+import vulnerabilities.models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('vulnerabilities', '0021_alter_vulnerabilityreference_url'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='vulnerability',
+            name='vulnerability_id',
+            field=models.CharField(blank=True, default=vulnerabilities.models.build_vcid, help_text='Unique identifier for a vulnerability in the external representation. It is prefixed with VCID-', max_length=20, unique=True),
+        ),
+    ]

vulnerabilities/migrations/0023_vcid_migration.py

@@ -0,0 +1,29 @@
+from django.db import migrations
+from django.db.models import Q
+
+from vulnerabilities.utils import build_vcid
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('vulnerabilities', '0022_alter_vulnerability_vulnerability_id'),
+    ]
+
+    def save_vulnerability_id(apps, schema_editor):
+        Vulnerabilities = apps.get_model("vulnerabilities", "Vulnerability")
+        updatables = []
+        for vulnerability in Vulnerabilities.objects.filter(~Q(vulnerability_id__startswith="VCID-")):
+            vulnerability.vulnerability_id = build_vcid()
+            updatables.append(vulnerability)
+        # update in bulk, 500 at a time
+        updated = Vulnerabilities.objects.bulk_update(
+            objs=updatables, 
+            fields=["vulnerability_id"], 
+            batch_size=500,
+        )
+        print(f"Migrated {updated} vulnerabilities to new VCID")
+
+    operations = [
+        migrations.RunPython(save_vulnerability_id, migrations.RunPython.noop)
+    ]

vulnerabilities/models.py

@@ -18,7 +18,6 @@
 from django.db import models
 from django.dispatch import receiver
 from django.urls import reverse
-from django.utils.http import int_to_base36
 from packageurl import PackageURL
 from packageurl.contrib.django.models import PackageURLMixin
 from rest_framework.authtoken.models import Token
@@ -28,6 +27,7 @@
 from vulnerabilities.importer import Reference
 from vulnerabilities.improver import MAX_CONFIDENCE
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
+from vulnerabilities.utils import build_vcid
 
 logger = logging.getLogger(__name__)
 
@@ -41,8 +41,9 @@ class Vulnerability(models.Model):
         unique=True,
         blank=True,
         max_length=20,
+        default=build_vcid,
         help_text="Unique identifier for a vulnerability in the external representation. "
-        "It is prefixed with VULCOID-",
+        "It is prefixed with VCID-",
     )
 
     summary = models.TextField(
@@ -70,12 +71,6 @@ def severities(self):
         for reference in self.references.all():
             yield from VulnerabilitySeverity.objects.filter(reference=reference.id)
 
-    def save(self, *args, **kwargs):
-        super().save(*args, **kwargs)
-        if not self.vulnerability_id:
-            self.vulnerability_id = f"VULCOID-{int_to_base36(self.id).upper()}"
-            super().save(update_fields=["vulnerability_id"])
-
     @property
     def vulnerable_to(self):
         """

vulnerabilities/tests/test_fix_api.py

@@ -59,7 +59,7 @@ def test_api_with_single_vulnerability(self):
         ).data
         assert response == {
             "url": f"http://testserver/api/vulnerabilities/{self.vulnerability.id}",
-            "vulnerability_id": f"VULCOID-{int_to_base36(self.vulnerability.id).upper()}",
+            "vulnerability_id": self.vulnerability.vulnerability_id,
             "summary": "test",
             "aliases": [],
             "fixed_packages": [
@@ -84,7 +84,7 @@ def test_api_with_single_vulnerability_with_filters(self):
         ).data
         assert response == {
             "url": f"http://testserver/api/vulnerabilities/{self.vulnerability.id}",
-            "vulnerability_id": f"VULCOID-{int_to_base36(self.vulnerability.id).upper()}",
+            "vulnerability_id": self.vulnerability.vulnerability_id,
             "summary": "test",
             "aliases": [],
             "fixed_packages": [
@@ -182,7 +182,7 @@ def test_api_with_single_vulnerability_and_fixed_package(self):
             "affected_by_vulnerabilities": [
                 {
                     "url": f"http://testserver/api/vulnerabilities/{self.vuln1.id}",
-                    "vulnerability_id": f"VULCOID-{int_to_base36(self.vuln1.id).upper()}",
+                    "vulnerability_id": self.vuln1.vulnerability_id,
                     "summary": "test-vuln1",
                     "references": [],
                     "fixed_packages": [],
@@ -191,7 +191,7 @@ def test_api_with_single_vulnerability_and_fixed_package(self):
             "fixing_vulnerabilities": [
                 {
                     "url": f"http://testserver/api/vulnerabilities/{self.vuln.id}",
-                    "vulnerability_id": f"VULCOID-{int_to_base36(self.vuln.id).upper()}",
+                    "vulnerability_id": self.vuln.vulnerability_id,
                     "summary": "test-vuln",
                     "references": [],
                     "fixed_packages": [
@@ -206,7 +206,7 @@ def test_api_with_single_vulnerability_and_fixed_package(self):
             "unresolved_vulnerabilities": [
                 {
                     "url": f"http://testserver/api/vulnerabilities/{self.vuln1.id}",
-                    "vulnerability_id": f"VULCOID-{int_to_base36(self.vuln1.id).upper()}",
+                    "vulnerability_id": self.vuln1.vulnerability_id,
                     "summary": "test-vuln1",
                     "references": [],
                     "fixed_packages": [],
@@ -228,7 +228,7 @@ def test_api_with_single_vulnerability_and_vulnerable_package(self):
             "affected_by_vulnerabilities": [
                 {
                     "url": f"http://testserver/api/vulnerabilities/{self.vuln.id}",
-                    "vulnerability_id": f"VULCOID-{int_to_base36(self.vuln.id).upper()}",
+                    "vulnerability_id": self.vuln.vulnerability_id,
                     "summary": "test-vuln",
                     "references": [],
                     "fixed_packages": [
@@ -244,7 +244,7 @@ def test_api_with_single_vulnerability_and_vulnerable_package(self):
             "unresolved_vulnerabilities": [
                 {
                     "url": f"http://testserver/api/vulnerabilities/{self.vuln.id}",
-                    "vulnerability_id": f"VULCOID-{int_to_base36(self.vuln.id).upper()}",
+                    "vulnerability_id": self.vuln.vulnerability_id,
                     "summary": "test-vuln",
                     "references": [],
                     "fixed_packages": [

vulnerabilities/utils.py

@@ -15,10 +15,12 @@
 import re
 from collections import defaultdict
 from functools import total_ordering
+from hashlib import sha256
 from typing import List
 from typing import Optional
 from typing import Tuple
 from unittest.mock import MagicMock
+from uuid import uuid4
 
 import requests
 import saneyaml
@@ -351,3 +353,60 @@ def resolve_version_range(
             )
             continue
     return affected_versions, unaffected_versions
+
+
+def build_vcid(prefix="VCID"):
+    """
+    Return a new VulnerableCode VCID unique identifier string using the ``prefix``.
+
+    For example::
+    >>> import re
+    >>> vcid = build_vcid()
+    >>> # VCID-6npv-94wz-hhuq
+    >>> assert re.match('VCID(-[a-z1-9]{4}){3}', vcid), vcid
+    """
+    # we keep only 64 bits (e.g. 8 bytes)
+    uid = sha256(uuid4().bytes).digest()[:8]
+    # we keep only 12 encoded bytes (which corresponds to 60 bits)
+    uid = base32_custom(uid)[:12].decode("utf-8").lower()
+    return f"{prefix}-{uid[:4]}-{uid[4:8]}-{uid[8:12]}"
+
+
+_base32_alphabet = b"ABCDEFGHJKMNPQRSTUVWXYZ123456789"
+_base32_table = None
+
+
+def base32_custom(btes):
+    """
+    Encode the ``btes`` bytes object using a Base32 encoding using a custom
+    alphabet and return a bytes object.
+
+    Code copied and modified from the Python Standard Library:
+    base64.b32encode function
+
+    SPDX-License-Identifier: Python-2.0
+    Copyright (c) The Python Software Foundation
+
+    For example::
+    >>> assert base32_custom(b'abcd') == b'ABTZE25E', base32_custom(b'abcd')
+    >>> assert base32_custom(b'abcde00000xxxxxPPPPP') == b'PFUGG3DFGA2DAPBTSB6HT8D2MBJFAXCT'
+    """
+    global _base32_table
+    # Delay the initialization of the table to not waste memory
+    # if the function is never called
+    if _base32_table is None:
+        b32tab = [bytes((i,)) for i in _base32_alphabet]
+        _base32_table = [a + b for a in b32tab for b in b32tab]
+
+    encoded = bytearray()
+    from_bytes = int.from_bytes
+
+    for i in range(0, len(btes), 5):
+        c = from_bytes(btes[i : i + 5], "big")
+        encoded += (
+            _base32_table[c >> 30]
+            + _base32_table[(c >> 20) & 0x3FF]  # bits 1 - 10
+            + _base32_table[(c >> 10) & 0x3FF]  # bits 11 - 20
+            + _base32_table[c & 0x3FF]  # bits 21 - 30  # bits 31 - 40
+        )
+    return bytes(encoded)