Implement impact packages and specific version with univers

Samk1710 · Samk1710 · commit 7d47d4666a75 · 2026-01-27T21:46:17.000+05:30
Signed-off-by: Sampurna Pyne &lt;sampurnapyne1710@gmail.com&gt;
diff --git a/vulnerabilities/pipelines/v2_importers/tuxcare_importer.py b/vulnerabilities/pipelines/v2_importers/tuxcare_importer.py
@@ -14,7 +14,10 @@
 from dateutil.parser import parse
 from packageurl import PackageURL
 from pytz import UTC
+from univers.version_range import AlpineLinuxVersionRange
+from univers.version_range import DebianVersionRange
 from univers.version_range import GenericVersionRange
+from univers.version_range import RpmVersionRange
 
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import AffectedPackageV2
@@ -25,6 +28,18 @@
 
 logger = logging.getLogger(__name__)
 
+# See https://docs.tuxcare.com/els-for-os/#cve-status-definition
+NON_AFFECTED_STATUSES = ["Not Vulnerable"]
+AFFECTED_STATUSES = ["Ignored", "Needs Triage", "In Testing", "In Progress", "In Rollout"]
+FIXED_STATUSES = ["Released", "Already Fixed"]
+
+VERSION_RANGE_BY_PURL_TYPE = {
+    "rpm": RpmVersionRange,
+    "deb": DebianVersionRange,
+    "apk": AlpineLinuxVersionRange,
+    "generic": GenericVersionRange,
+}
+
 
 class TuxCareImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     pipeline_id = "tuxcare_importer_v2"
@@ -43,9 +58,54 @@ def fetch(self) -> None:
         self.log(f"Fetching `{url}`")
         response = fetch_response(url)
         self.response = response.json() if response else []
+        self._grouped = self._group_records_by_cve()
+
+    def _group_records_by_cve(self) -> dict:
+        grouped = {}
+        skipped_invalid = 0
+        skipped_non_affected = 0
+
+        for record in self.response:
+            cve_id = record.get("cve", "").strip()
+            if not cve_id or not cve_id.startswith("CVE-"):
+                logger.warning(f"Skipping invalid CVE ID: {cve_id}")
+                skipped_invalid += 1
+                continue
+
+            os_name = record.get("os_name", "").strip()
+            project_name = record.get("project_name", "").strip()
+            version = record.get("version", "").strip()
+            status = record.get("status", "").strip()
+
+            if not all([os_name, project_name, version, status]):
+                logger.warning(f"Skipping {cve_id}: missing required fields")
+                skipped_invalid += 1
+                continue
+
+            # Skip records with non-affected statuses
+            if status in NON_AFFECTED_STATUSES:
+                skipped_non_affected += 1
+                continue
+
+            if status not in AFFECTED_STATUSES and status not in FIXED_STATUSES:
+                logger.warning(f"Skipping {cve_id}: unrecognized status '{status}'")
+                skipped_invalid += 1
+                continue
+
+            if cve_id not in grouped:
+                grouped[cve_id] = []
+            grouped[cve_id].append(record)
+
+        total_skipped = skipped_invalid + skipped_non_affected
+        self.log(
+            f"Grouped {len(self.response):,d} records into {len(grouped):,d} unique CVEs "
+            f"(skipped {total_skipped:,d}: {skipped_invalid:,d} invalid, "
+            f"{skipped_non_affected:,d} non-affected)"
+        )
+        return grouped
 
     def advisories_count(self) -> int:
-        return len(self.response)
+        return len(self._grouped)
 
     def _create_purl(self, project_name: str, os_name: str) -> PackageURL:
         normalized_os = os_name.lower().replace(" ", "-")
@@ -64,9 +124,6 @@ def _create_purl(self, project_name: str, os_name: str) -> PackageURL:
             "tuxcare": ("generic", "tuxcare"),
         }
 
-        pkg_type = "generic"
-        namespace = "tuxcare"
-
         for keyword, (ptype, pns) in os_mapping.items():
             if keyword in os_lower:
                 pkg_type = ptype
@@ -75,105 +132,90 @@ def _create_purl(self, project_name: str, os_name: str) -> PackageURL:
         else:
             return None
 
-        qualifiers = {}
-        if normalized_os:
-            qualifiers["distro"] = normalized_os
+        qualifiers = {"distro": normalized_os}
 
         return PackageURL(
             type=pkg_type, namespace=namespace, name=project_name, qualifiers=qualifiers
         )
 
     def collect_advisories(self) -> Iterable[AdvisoryData]:
-        for record in self.response:
-            cve_id = record.get("cve", "").strip()
-            if not cve_id or not cve_id.startswith("CVE-"):
-                logger.warning(f"Skipping record with invalid CVE ID: {cve_id}")
-                continue
-
-            os_name = record.get("os_name", "").strip()
-            project_name = record.get("project_name", "").strip()
-            version = record.get("version", "").strip()
-            score = record.get("score", "").strip()
-            severity = record.get("severity", "").strip()
-            status = record.get("status", "").strip()
-            last_updated = record.get("last_updated", "").strip()
+        grouped_by_cve = self._grouped
 
-            if not all([os_name, project_name, version, status]):
-                logger.warning(f"Skipping {cve_id} - missing required fields")
-                continue
+        for cve_id, records in grouped_by_cve.items():
+            affected_packages = []
+            severities = []
+            date_published = None
+            all_records = []
+            severity_added = False
+
+            for record in records:
+                os_name = record.get("os_name", "").strip()
+                project_name = record.get("project_name", "").strip()
+                version = record.get("version", "").strip()
+                score = record.get("score", "").strip()
+                severity = record.get("severity", "").strip()
+                status = record.get("status", "").strip()
+                last_updated = record.get("last_updated", "").strip()
+
+                purl = self._create_purl(project_name, os_name)
+                if not purl:
+                    logger.warning(
+                        f"Skipping package {project_name} on {os_name} for {cve_id} - unexpected OS type"
+                    )
+                    continue
 
-            # See https://docs.tuxcare.com/els-for-os/#cve-status-definition
-            non_affected_statuses = ["Not Vulnerable"]
-            affected_statuses = [
-                "Ignored",
-                "Needs Triage",
-                "In Testing",
-                "In Progress",
-                "In Rollout",
-            ]
-            fixed_statuses = ["Released", "Already Fixed"]
-
-            # Skip CVEs that are not vulnerable
-            if status in non_affected_statuses:
-                continue
+                version_range_class = VERSION_RANGE_BY_PURL_TYPE.get(purl.type, GenericVersionRange)
+                try:
+                    version_range = version_range_class.from_versions([version])
+                except ValueError as e:
+                    logger.warning(f"Failed to parse version {version} for {cve_id}: {e}")
+                    continue
+
+                affected_version_range = None
+                fixed_version_range = None
+
+                if status in AFFECTED_STATUSES:
+                    affected_version_range = version_range
+                elif status in FIXED_STATUSES:
+                    fixed_version_range = version_range
+
+                affected_packages.append(
+                    AffectedPackageV2(
+                        package=purl,
+                        affected_version_range=affected_version_range,
+                        fixed_version_range=fixed_version_range,
+                    )
+                )
 
-            if status not in affected_statuses and status not in fixed_statuses:
-                logger.warning(f"Skipping {cve_id} - unknown status: {status}")
-                continue
+                if severity and score and not severity_added:
+                    severities.append(
+                        VulnerabilitySeverity(
+                            system=GENERIC,
+                            value=score,
+                            scoring_elements=severity,
+                        )
+                    )
+                    severity_added = True
 
-            normalized_os = os_name.lower().replace(" ", "-")
-            advisory_id = f"{cve_id}-{normalized_os}-{project_name.lower()}-{version}"
+                if last_updated:
+                    try:
+                        current_date = parse(last_updated).replace(tzinfo=UTC)
+                        if date_published is None or current_date > date_published:
+                            date_published = current_date
+                    except ValueError as e:
+                        logger.warning(f"Failed to parse date {last_updated} for {cve_id}: {e}")
 
-            purl = self._create_purl(project_name, os_name)
-            if not purl:
-                logger.warning(f"Skipping {cve_id} - unexpected OS type: '{os_name}'")
-                continue
+                all_records.append(record)
 
-            try:
-                version_range = GenericVersionRange.from_versions([version])
-            except ValueError as e:
-                logger.warning(f"Failed to parse version {version} for {cve_id}: {e}")
+            if not affected_packages:
+                logger.warning(f"Skipping {cve_id} - no valid affected packages")
                 continue
 
-            affected_version_range = None
-            fixed_version_range = None
-
-            if status in affected_statuses:
-                affected_version_range = version_range
-            elif status in fixed_statuses:
-                fixed_version_range = version_range
-
-            affected_packages = [
-                AffectedPackageV2(
-                    package=purl,
-                    affected_version_range=affected_version_range,
-                    fixed_version_range=fixed_version_range,
-                )
-            ]
-
-            severities = []
-            if severity and score:
-                severities.append(
-                    VulnerabilitySeverity(
-                        system=GENERIC,
-                        value=score,
-                        scoring_elements=severity,
-                    )
-                )
-
-            date_published = None
-            if last_updated:
-                try:
-                    date_published = parse(last_updated).replace(tzinfo=UTC)
-                except ValueError as e:
-                    logger.warning(f"Failed to parse date {last_updated} for {cve_id}: {e}")
-
             yield AdvisoryData(
-                advisory_id=advisory_id,
-                aliases=[cve_id],
+                advisory_id=cve_id,
                 affected_packages=affected_packages,
                 severities=severities,
                 date_published=date_published,
                 url=f"https://cve.tuxcare.com/els/cve/{cve_id}",
-                original_advisory_text=json.dumps(record, indent=2, ensure_ascii=False),
+                original_advisory_text=json.dumps(all_records, indent=2, ensure_ascii=False),
             )
diff --git a/vulnerabilities/tests/pipelines/v2_importers/test_tuxcare_importer_v2.py b/vulnerabilities/tests/pipelines/v2_importers/test_tuxcare_importer_v2.py
@@ -35,4 +35,4 @@ def test_collect_advisories(self, mock_fetch):
         expected_file = TEST_DATA / "expected.json"
         util_tests.check_results_against_json(advisories, expected_file)
 
-        assert pipeline.advisories_count() == 13
+        assert len(advisories) == 14
diff --git a/vulnerabilities/tests/test_data/tuxcare/data.json b/vulnerabilities/tests/test_data/tuxcare/data.json
@@ -19,6 +19,46 @@
     "status": "In Testing",
     "last_updated": "2025-12-23 10:08:35.944749"
   },
+  {
+    "cve": "CVE-2023-52922",
+    "os_name": "CentOS 8.5 ELS",
+    "project_name": "kernel",
+    "version": "4.18.0",
+    "score": "7.8",
+    "severity": "HIGH",
+    "status": "Released",
+    "last_updated": "2025-05-21 01:43:28.677045"
+  },
+  {
+    "cve": "CVE-2023-52922",
+    "os_name": "AlmaLinux 9.2 ESU",
+    "project_name": "squid",
+    "version": "5.5",
+    "score": "7.8",
+    "severity": "HIGH",
+    "status": "Not Vulnerable",
+    "last_updated": "2025-08-28 00:52:25.579518"
+  },
+  {
+    "cve": "CVE-2023-52922",
+    "os_name": "RHEL 7 ELS",
+    "project_name": "squid",
+    "version": "3.5.20",
+    "score": "7.8",
+    "severity": "HIGH",
+    "status": "Not Vulnerable",
+    "last_updated": "2025-11-27 10:32:13.088814"
+  },
+  {
+    "cve": "CVE-2023-52922",
+    "os_name": "CentOS Stream 8 ELS",
+    "project_name": "squid",
+    "version": "4.15",
+    "score": "7.8",
+    "severity": "HIGH",
+    "status": "Ignored",
+    "last_updated": "2025-08-28 22:56:12.357818"
+  },
   {
     "cve": "CVE-2023-48161",
     "os_name": "RHEL 7 ELS",
@@ -128,5 +168,35 @@
     "severity": "MEDIUM",
     "status": "In Rollout",
     "last_updated": "2025-12-20 04:34:51.112485"
+  },
+  {
+    "cve": "CVE-2022-50268",
+    "os_name": "Ubuntu 16.04 ELS",
+    "project_name": "linux",
+    "version": "4.4.0",
+    "score": "5.5",
+    "severity": "MEDIUM",
+    "status": "Needs Triage",
+    "last_updated": "2025-12-23 07:31:24.920518"
+  },
+  {
+    "cve": "CVE-2020-1472",
+    "os_name": "Debian 10 ELS",
+    "project_name": "samba",
+    "version": "4.9.5",
+    "score": "0.0",
+    "severity": "",
+    "status": "In Testing",
+    "last_updated": "2025-12-22 16:58:13.189255"
+  },
+  {
+    "cve": "CVE-2025-6297",
+    "os_name": "Alpine Linux 3.18 ELS",
+    "project_name": "dpkg",
+    "version": "1.21.21",
+    "score": "0.0",
+    "severity": "",
+    "status": "Released",
+    "last_updated": "2025-12-19 05:00:40.874276"
   }
 ]
diff --git a/vulnerabilities/tests/test_data/tuxcare/expected.json b/vulnerabilities/tests/test_data/tuxcare/expected.json
