fix: validate aliases for alpine linux, mattermost, fireeye, istio advisories

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

Author: ziadhany

vulnerabilities/pipelines/v2_importers/alpine_linux_importer.py

@@ -9,6 +9,7 @@
 
 import json
 import logging
+import re
 from pathlib import Path
 from typing import Any
 from typing import Iterable
@@ -28,6 +29,7 @@
 from vulnerabilities.references import XsaReferenceV2
 from vulnerabilities.references import ZbxReferenceV2
 from vulnerabilities.utils import get_advisory_url
+from vulnerabilities.utils import is_cve
 from vulnerabilities.utils import load_json
 
 
@@ -163,8 +165,10 @@ 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:
-            aliases = vuln_ids.strip().split(" ")
-            vuln_id = aliases[0]
+            vuln_id, aliases = parse_vuln_ids(vuln_ids)
+
+            if not vuln_id:
+                continue
 
             references = []
             if vuln_id.startswith("XSA"):
@@ -248,3 +252,34 @@ def load_advisories(
                 url=url,
                 original_advisory_text=json.dumps(pkg_infos, indent=2, ensure_ascii=False),
             )
+
+
+PARENTHESES_RE = re.compile(r"\(.*?\)")
+
+
+def parse_vuln_ids(vuln_ids_string):
+    """
+    Parses a raw vulnerability ids, removes parentheses and returns the advisory_id and a list of all valid aliases.
+    """
+    vuln_ids = PARENTHESES_RE.sub("", vuln_ids_string)
+    if not vuln_ids_string:
+        return None, []
+
+    cleaned_vuln_ids = []
+    for alias in vuln_ids.split():
+        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))
+    ]
+
+    if not aliases:
+        return None, []
+
+    vuln_id = aliases[0]
+    return vuln_id, aliases

vulnerabilities/pipelines/v2_importers/fireeye_importer_v2.py

@@ -104,15 +104,16 @@ def parse_advisory_data(raw_data, file_path, base_path) -> AdvisoryDataV2:
     cve_refs = md_dict.get("## CVE Reference") or []
     cve_ids = md_dict.get("## CVE ID") or []
     cleaned_cve_ids = []
-    for line in cve_ids:
+    for line in cve_ids + cve_refs:
         found_cves = find_all_cve(line)
         cleaned_cve_ids.extend(found_cves)
 
     references = md_dict.get("## References") or []
     cwe_data = md_dict.get("## Common Weakness Enumeration") or []
 
     advisory_id = file_path.stem
-    aliases = dedupe([cve.strip() for cve in cleaned_cve_ids + cve_refs])
+    aliases = dedupe([cve.strip() for cve in cleaned_cve_ids])
+
     aliases = [aliase for aliase in aliases if aliase != advisory_id]
     advisory_url = get_advisory_url(
         file=file_path,

vulnerabilities/pipelines/v2_importers/istio_importer.py

@@ -28,6 +28,7 @@
 from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
 from vulnerabilities.utils import get_advisory_url
+from vulnerabilities.utils import is_cve
 from vulnerabilities.utils import split_markdown_front_matter
 
 is_release = re.compile(r"^[\d.]+$", re.IGNORECASE).match
@@ -88,6 +89,7 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
                 data.get("releases", []), GolangVersion
             )
             cves = data.get("cves", [])
+            aliases = [cve for cve in cves if is_cve(cve)]
 
             affected_packages = []
             if semver_constraints:
@@ -116,10 +118,10 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
                         url=f"https://istio.io/latest/news/security/{title}/",
                     )
                 )
-
+            print(title, aliases)
             yield AdvisoryDataV2(
                 advisory_id=title,
-                aliases=cves,
+                aliases=aliases,
                 summary=summary,
                 affected_packages=affected_packages,
                 references=references,

vulnerabilities/pipelines/v2_importers/mattermost_importer.py

@@ -20,6 +20,7 @@
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
 from vulnerabilities.utils import fetch_response
+from vulnerabilities.utils import is_cve
 
 MM_REPO = {
     "Mattermost Mobile Apps": "mattermost-mobile",
@@ -62,11 +63,17 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
             return
 
         for advisory in data:
-            vuln_id = advisory.get("issue_id")
-            if not vuln_id or not vuln_id.startswith("MMSA-"):
-                self.log(f"Skipping advisory with missing issue_id. {vuln_id}")
+            issue_id = advisory.get("issue_id") or ""
+            cve_id = advisory.get("cve_id") or ""
+
+            advisory_id, aliases = parse_vuln_ids(issue_id, cve_id)
+
+            if not advisory_id:
+                self.log(
+                    f"Skipping advisory with missing advisory_id. issue_id:{issue_id} cve_id:{cve_id}"
+                )
                 continue
-            cve_id = advisory.get("cve_id")
+
             details = advisory.get("details")
 
             platform = advisory.get("platform")
@@ -78,7 +85,7 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
             affected_packages = []
             severity = advisory.get("severity")
             if not package_name:
-                self.log(f"Unknown platform '{platform}' in advisory '{vuln_id}'.")
+                self.log(f"Unknown platform '{platform}' in advisory '{advisory_id}'.")
 
             else:
                 package = PackageURL(
@@ -105,7 +112,7 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
                     )
                 except Exception as e:
                     self.log(
-                        f"Error processing fixed versions '{fixed_versions}' for advisory '{vuln_id}': {e}"
+                        f"Error processing fixed versions '{fixed_versions}' for advisory '{advisory_id}': {e}"
                     )
 
             severities = []
@@ -118,12 +125,36 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
             )
 
             yield AdvisoryDataV2(
-                advisory_id=vuln_id,
-                aliases=[cve_id],
+                advisory_id=advisory_id,
+                aliases=aliases,
                 summary=details,
                 references=[reference],
                 affected_packages=affected_packages,
                 severities=severities,
                 url=self.url,
                 original_advisory_text=json.dumps(advisory, indent=2, ensure_ascii=False),
             )
+
+
+def parse_vuln_ids(issue_id, cve_id):
+    """
+    Parses a raw issue_id, cve_id, validate and returns the advisory id and a list of all valid aliases.
+    """
+    advisory_id = None
+    aliases = []
+
+    cve_id = cve_id.strip()
+    issue_id = issue_id.strip()
+
+    for vuln_id in issue_id.split(","):
+        vuln_id = vuln_id.strip()
+        if vuln_id.startswith("MMSA-") or vuln_id.startswith("CVE-"):
+            aliases.append(vuln_id)
+
+    if cve_id and is_cve(cve_id):
+        aliases.append(cve_id)
+
+    if aliases:
+        advisory_id = aliases.pop(0)
+
+    return advisory_id, aliases

vulnerabilities/tests/pipelines/v2_importers/test_alpine_linux_importer_pipeline.py

@@ -13,6 +13,7 @@
 import pytest
 
 from vulnerabilities.pipelines.v2_importers.alpine_linux_importer import load_advisories
+from vulnerabilities.pipelines.v2_importers.alpine_linux_importer import parse_vuln_ids
 from vulnerabilities.pipelines.v2_importers.alpine_linux_importer import process_record
 from vulnerabilities.tests import util_tests
 from vulnerabilities.tests.pipelines import TestLogger
@@ -95,3 +96,82 @@ def test_load_advisories_package_with_invalid_alpine_version(test_case):
     }
     result = list(load_advisories(package, "v3.11", "main", archs=[], url="", logger=logger.write))
     assert result != []
+
+
+@pytest.mark.parametrize(
+    "raw_input, expected_advisory_id, expected_aliases",
+    [
+        ("CVE-2022-42332 XSA-427", "CVE-2022-42332", ["CVE-2022-42332", "XSA-427"]),
+        (
+            "CVE-2022-42333 CVE-2022-43334 XSA-428",
+            "CVE-2022-42333",
+            ["CVE-2022-42333", "CVE-2022-43334", "XSA-428"],
+        ),
+        (
+            "CVE-2020-11501 GNUTLS-SA-2020-03-31 CVE-2020-11501",
+            "CVE-2020-11501",
+            ["CVE-2020-11501", "GNUTLS-SA-2020-03-31", "CVE-2020-11501"],
+        ),
+        ("CVE_2019-2426", "CVE-2019-2426", ["CVE-2019-2426"]),
+        (
+            "CVE-2024-22195 GHSA-h5c8-rqwp-cp95",
+            "CVE-2024-22195",
+            ["CVE-2024-22195", "GHSA-h5c8-rqwp-cp95"],
+        ),
+        ("CVE-2023-44441 ZDI-CAN-22093", "CVE-2023-44441", ["CVE-2023-44441", "ZDI-CAN-22093"]),
+        ("CVE-2022-45059 VSV00010", "CVE-2022-45059", ["CVE-2022-45059", "VSV00010"]),
+        ("CVE-2021-35940.patch", "CVE-2021-35940", ["CVE-2021-35940"]),
+        ("XSA-207", "XSA-207", ["XSA-207"]),
+        ("ALPINE-13661", "ALPINE-13661", ["ALPINE-13661"]),
+        ("GHSA-vv2x-vrpj-qqpq", "GHSA-vv2x-vrpj-qqpq", ["GHSA-vv2x-vrpj-qqpq"]),
+        ("CVE N/A ZBX-11023", "ZBX-11023", ["ZBX-11023"]),
+        ("CVE-2017-2616 (+ regression fix)", "CVE-2017-2616", ["CVE-2017-2616"]),
+        (
+            "CVE-2020-14342 (Not affected, requires --with-systemd)",
+            "CVE-2020-14342",
+            ["CVE-2020-14342"],
+        ),
+        ("CVE-2017-16808 (AoE)", "CVE-2017-16808", ["CVE-2017-16808"]),
+        ("CVE-2018-14468 (FrameRelay)", "CVE-2018-14468", ["CVE-2018-14468"]),
+        ("CVE-2018-14469 (IKEv1)", "CVE-2018-14469", ["CVE-2018-14469"]),
+        ("CVE-2018-14470 (BABEL)", "CVE-2018-14470", ["CVE-2018-14470"]),
+        ("CVE-2018-14466 (AFS/RX)", "CVE-2018-14466", ["CVE-2018-14466"]),
+        ("CVE-2018-14461 (LDP)", "CVE-2018-14461", ["CVE-2018-14461"]),
+        ("CVE-2018-14462 (ICMP)", "CVE-2018-14462", ["CVE-2018-14462"]),
+        ("CVE-2018-14465 (RSVP)", "CVE-2018-14465", ["CVE-2018-14465"]),
+        ("CVE-2018-14881 (BGP)", "CVE-2018-14881", ["CVE-2018-14881"]),
+        ("CVE-2018-14464 (LMP)", "CVE-2018-14464", ["CVE-2018-14464"]),
+        ("CVE-2018-14463 (VRRP)", "CVE-2018-14463", ["CVE-2018-14463"]),
+        ("CVE-2018-14467 (BGP)", "CVE-2018-14467", ["CVE-2018-14467"]),
+        (
+            "CVE-2018-10103 (SMB - partially fixed, but SMB printing disabled)",
+            "CVE-2018-10103",
+            ["CVE-2018-10103"],
+        ),
+        (
+            "CVE-2018-10105 (SMB - too unreliably reproduced, SMB printing disabled)",
+            "CVE-2018-10105",
+            ["CVE-2018-10105"],
+        ),
+        ("CVE-2018-14880 (OSPF6)", "CVE-2018-14880", ["CVE-2018-14880"]),
+        ("CVE-2018-16451 (SMB)", "CVE-2018-16451", ["CVE-2018-16451"]),
+        ("CVE-2018-14882 (RPL)", "CVE-2018-14882", ["CVE-2018-14882"]),
+        ("CVE-2018-16227 (802.11)", "CVE-2018-16227", ["CVE-2018-16227"]),
+        ("CVE-2018-16229 (DCCP)", "CVE-2018-16229", ["CVE-2018-16229"]),
+        ("CVE-2018-16301 (was fixed in libpcap)", "CVE-2018-16301", ["CVE-2018-16301"]),
+        ("CVE-2018-16230 (BGP)", "CVE-2018-16230", ["CVE-2018-16230"]),
+        ("CVE-2018-16452 (SMB)", "CVE-2018-16452", ["CVE-2018-16452"]),
+        ("CVE-2018-16300 (BGP)", "CVE-2018-16300", ["CVE-2018-16300"]),
+        ("CVE-2018-16228 (HNCP)", "CVE-2018-16228", ["CVE-2018-16228"]),
+        ("CVE-2019-15166 (LMP)", "CVE-2019-15166", ["CVE-2019-15166"]),
+        ("CVE-2019-15167 (VRRP)", "CVE-2019-15167", ["CVE-2019-15167"]),
+        ("CVE-????-????? TS-2024-005", "TS-2024-005", ["TS-2024-005"]),
+        ("CVE-????-????? TS-2024-005", "TS-2024-005", ["TS-2024-005"]),
+        ("CVE-2018-14879 (tcpdump -V)", "CVE-2018-14879", ["CVE-2018-14879"]),
+        ("CVE-46838", None, []),  # invalid CVE
+    ],
+)
+def test_parse_vuln_ids(raw_input, expected_advisory_id, expected_aliases):
+    advisory_id, aliases = parse_vuln_ids(raw_input)
+    assert advisory_id == expected_advisory_id
+    assert aliases == expected_aliases

vulnerabilities/tests/pipelines/v2_importers/test_istio_importer_v2.py

@@ -19,7 +19,7 @@
 
 
 @pytest.mark.django_db
-def test_istio_advisory_parsing():
+def test_istio_advisory_parsing1():
     sample_md = dedent(
         """\
         ---
@@ -79,3 +79,55 @@ def test_istio_advisory_parsing():
             str(advisory.affected_packages[1].affected_version_range)
             == "vers:golang/>=1.0.0|<=1.0.8|>=1.1.0|<=1.1.9|>=1.2.0|<=1.2.1"
         )
+
+
+@pytest.mark.django_db
+def test_istio_advisory_parsing2():
+    sample_md = dedent(
+        """\
+---
+title: ISTIO-SECURITY-2021-004
+subtitle: Security Bulletin
+description: Potential misuse of mTLS-only fields in AuthorizationPolicy with plain text traffic. 
+cves: [N/A]
+cvss: "N/A"
+vector: ""
+releases: ["All releases 1.5 and later"]
+publishdate: 2021-04-15
+keywords: [CVE]
+skip_seealso: true
+---
+
+{{< security_bulletin >}}
+
+This is a security advisory for customers to check the authorization policy to make sure [mTLS (STRICT mode) is enabled](/docs/tasks/security/authentication/authn-policy/#globally-enabling-istio-mutual-tls-in-strict-mode)
+when using [mTLS-only fields](/docs/concepts/security/#dependency-on-mutual-tls) in the authorization policy.
+...
+        """
+    )
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        base_path = Path(tmp_dir)
+        advisory_dir = base_path / "content/en/news/security"
+        advisory_dir.mkdir(parents=True)
+        advisory_file = advisory_dir / "ISTIO-SECURITY-2021-004.md"
+        advisory_file.write_text(sample_md, encoding="utf-8")
+
+        importer = IstioImporterPipeline()
+        importer.vcs_response = type(
+            "FakeVCS", (), {"dest_dir": tmp_dir, "delete": lambda x: None}
+        )()
+
+        advisories = list(importer.collect_advisories())
+
+        assert len(advisories) == 1
+        advisory = advisories[0]
+
+        assert isinstance(advisory, AdvisoryDataV2)
+        assert advisory.advisory_id == "ISTIO-SECURITY-2021-004"
+        assert advisory.aliases == []
+        assert advisory.summary.startswith(
+            "Potential misuse of mTLS-only fields in AuthorizationPolicy with plain text traffic"
+        )
+        assert advisory.date_published.isoformat() == "2021-04-15T00:00:00+00:00"
+        assert advisory.url.endswith("ISTIO-SECURITY-2021-004.md")

vulnerabilities/tests/pipelines/v2_importers/test_mattermost_importer_v2.py

@@ -14,6 +14,7 @@
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import AdvisoryDataV2
 from vulnerabilities.pipelines.v2_importers.mattermost_importer import MattermostImporterPipeline
+from vulnerabilities.pipelines.v2_importers.mattermost_importer import parse_vuln_ids
 
 
 @pytest.fixture
@@ -212,3 +213,24 @@ def json(self):
     importer.collect_advisories()
 
     assert call_count["count"] == 1
+
+
+@pytest.mark.parametrize(
+    "issue_id, cve_id, expected_advisory_id, expected_aliases",
+    [
+        ("MMSA-2026-00624", "CVE-2026-3590", "MMSA-2026-00624", ["CVE-2026-3590"]),
+        ("MMSA-2026-00605", "", "MMSA-2026-00605", []),
+        ("MMSA-2021-0055,CVE-2021-37859", "", "MMSA-2021-0055", ["CVE-2021-37859"]),
+        ("MMSA-2021-0055, CVE-2021-37859", "", "MMSA-2021-0055", ["CVE-2021-37859"]),
+        ("MMSA-2024-00344", "n/a", "MMSA-2024-00344", []),
+        ("5.17.2.4", "", None, []),
+        ("na", "", None, []),
+        ("N/A", "", None, []),
+        ("Issue Identifier", "", None, []),
+        ("", "", None, []),
+    ],
+)
+def test_parse_vuln_ids(issue_id, cve_id, expected_advisory_id, expected_aliases):
+    advisory_id, aliases = parse_vuln_ids(issue_id, cve_id)
+    assert advisory_id == expected_advisory_id
+    assert aliases == expected_aliases