Add migration to drop malformed advisories

Add a migration test

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

Author: ziadhany

vulnerabilities/migrations/0122_advisoryv2_remove_malformed_aliases_and_dvisory_id.py

@@ -0,0 +1,31 @@
+from django.db import migrations
+from django.db.models import Q
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("vulnerabilities", "0121_advisoryv2_is_latest_alter_advisoryv2_advisory_id_and_more"),
+    ]
+
+    def drop_malformed_advisory_v2(apps, _):
+        AdvisoryV2 = apps.get_model("vulnerabilities", "AdvisoryV2")
+        AdvisoryAlias = apps.get_model("vulnerabilities", "AdvisoryAlias")
+
+        valid_alias_prefix = [
+            "cve-", "osv-", "xsa-", "vsv", "zbx-", "zf2", "vu#", "gms-", "usn-",
+            "sw-", "ss-", "ts-", "osvdb-", "ysa-", "se-core-", "pysec-", "alpine-",
+            "dw2", "go-", "mal-", "zdi-can", "asa-", "ezsa-", "ghsl-", "ghsa-",
+            "talos-", "srcclr-sid-", "bit-", "gnutls-", "rustsec-", "snyk-",
+            "temp-", "TYPO3-", "wnpa-sec-", "sa-core-", "skcsirt-", "flow-", "gsd-"
+        ]
+        query = Q()
+        for alias_prefix in valid_alias_prefix:
+            query |= Q(alias__istartswith=alias_prefix)
+
+        malformed_aliases = AdvisoryAlias.objects.exclude(query)
+        AdvisoryV2.objects.filter(aliases__in=malformed_aliases).delete()
+        malformed_aliases.delete()
+
+    operations = [
+        migrations.RunPython(drop_malformed_advisory_v2, reverse_code=migrations.RunPython.noop),
+    ]

vulnerabilities/pipelines/v2_importers/alpine_linux_importer.py

@@ -165,7 +165,7 @@ def load_advisories(
         # 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:
-            vuln_id, aliases = parse_vuln_ids(vuln_ids)
+            vuln_id, aliases = parse_vuln_ids(vuln_ids, logger=logger)
 
             if not vuln_id:
                 continue
@@ -257,7 +257,7 @@ def load_advisories(
 PARENTHESES_RE = re.compile(r"\(.*?\)")
 
 
-def parse_vuln_ids(vuln_ids_string):
+def parse_vuln_ids(vuln_ids_string, logger=print):
     """
     Parses a raw vulnerability ids, removes parentheses and returns the advisory_id and a list of all valid aliases.
     """
@@ -270,13 +270,30 @@ def parse_vuln_ids(vuln_ids_string):
         clean_alias = alias.replace("_", "-").replace(".patch", "")
         cleaned_vuln_ids.append(clean_alias)
 
-    aliases = [
-        alias
-        for alias in cleaned_vuln_ids
-        if alias
-        and alias not in ["N/A", "CVE"]
-        and not (alias.startswith("CVE") and not is_cve(alias))
-    ]
+    aliases = []
+    valid_prefixes = (
+        "XSA-",
+        "GHSL-",
+        "TALOS-",
+        "RUSTSEC-",
+        "GHSA-",
+        "GNUTLS-",
+        "VSV",
+        "ZDI-CAN-",
+        "DW",
+        "YSA-",
+        "ZBX-",
+        "ALPINE-",
+        "TS-",
+        "wnpa-sec-",
+    )
+    for alias in cleaned_vuln_ids:
+        if alias and (
+            (alias.startswith("CVE-") and is_cve(alias)) or alias.startswith(valid_prefixes)
+        ):
+            aliases.append(alias)
+        else:
+            logger(f"Malformed aliases found: {alias}")
 
     if not aliases:
         return None, []

vulnerabilities/pipelines/v2_importers/istio_importer.py

@@ -109,6 +109,9 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
                 )
 
             title = data.get("title") or ""
+            if not title.startswith("ISTIO-"):
+                self.log(f"Invalid advisory_id: {title}")
+
             summary = data.get("description") or ""
             references = []
             if title:

vulnerabilities/tests/test_data_migrations.py

@@ -6,6 +6,7 @@
 # See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
+
 from datetime import datetime
 
 from django.apps import apps
@@ -1088,3 +1089,98 @@ def test_latest_is_actually_recent(self):
 
         latest = AdvisoryV2.objects.get(avid="test_pipeline/test_adv", is_latest=True)
         self.assertEqual("New advisory", latest.summary)
+
+
+class TestMalformedAliasesAVIDMigration(TestMigrations):
+    app_name = "vulnerabilities"
+    migrate_from = "0121_advisoryv2_is_latest_alter_advisoryv2_advisory_id_and_more"
+    migrate_to = "0122_advisoryv2_remove_malformed_aliases_and_dvisory_id"
+    raw_alias_inputs = [
+        ("CVE-2023-1111", True),
+        ("GHSA-abcd-1234", True),
+        ("", False),
+        ("(not", False),
+        ("applicable)", False),
+        ("(BABEL)", False),
+        ("(was", False),
+        ("--with-systemd)", False),
+        ("fixed", False),
+        ("printing", False),
+        ("(AFS/RX)", False),
+        ("unreliably", False),
+        ("(ICMP)", False),
+        ("CVE", False),
+        ("(Not", False),
+        ("(RSVP)", False),
+        ("libpcap)", False),
+        ("(SMB", False),
+        ("fix)", False),
+        ("(DCCP)", False),
+        ("(HNCP)", False),
+        ("(+", False),
+        ("(IKEv1)", False),
+        ("(FrameRelay)", False),
+        ("XPTI", False),
+        ("CVE_2019-2426", False),
+        ("(BGP)", False),
+        ("disabled)", False),
+        ("(RPL)", False),
+        ("regression", False),
+        ("actually", False),
+        ("(VRRP)", False),
+        ("-V)", False),
+        ("2025-48379", False),
+        ("fixed,", False),
+        ("(802.11)", False),
+        ("affected,", False),
+        ("SMB", False),
+        ("(OSPF6)", False),
+        ("too", False),
+        ("partially", False),
+        ("in", False),
+        ("(SMB)", False),
+        ("but", False),
+        ("-", False),
+        ("(LDP)", False),
+        ("reproduced,", False),
+        ("N/A", False),
+        ("(tcpdump", False),
+        ("requires", False),
+        ("(AoE)", False),
+        ("(LMP)", False),
+        (" CVE-2025-55070", False),
+        ("n/a", False),
+        ("No CVE assigned", False),
+        ("- CVE-2026-26365", False),
+    ]
+
+    def setUpBeforeMigration(self, apps):
+        AdvisoryV2 = apps.get_model("vulnerabilities", "AdvisoryV2")
+        AdvisoryAlias = apps.get_model("vulnerabilities", "AdvisoryAlias")
+
+        for i, (raw_input, _) in enumerate(self.raw_alias_inputs):
+            adv = AdvisoryV2.objects.create(
+                unique_content_id=f"content_{i}",
+                url="https://example.com",
+                summary=f"Advisory for {raw_input}",
+                advisory_id=raw_input,
+                avid=f"test_pipeline/{raw_input}",
+                datasource_id="test_pipeline",
+            )
+            alias = AdvisoryAlias.objects.create(alias=raw_input)
+            adv.aliases.add(alias)
+
+    def test_migration_processes_malformed_aliases(self):
+        AdvisoryV2 = self.apps.get_model("vulnerabilities", "AdvisoryV2")
+        AdvisoryAlias = self.apps.get_model("vulnerabilities", "AdvisoryAlias")
+
+        for i, (raw_input, expected_to_survive) in enumerate(self.raw_alias_inputs):
+            adv_exists = AdvisoryV2.objects.filter(unique_content_id=f"content_{i}").exists()
+            alias_exists = AdvisoryAlias.objects.filter(alias=raw_input).exists()
+
+            if expected_to_survive:
+                assert adv_exists == True
+                assert alias_exists == True
+            else:
+                assert adv_exists == False
+                assert alias_exists == False