Add V2 Importer for Tuxcare advisories

requirements.txt

@@ -118,7 +118,7 @@ toml==0.10.2
 tomli==2.0.1
 traitlets==5.1.1
 typing_extensions==4.1.1
-univers==30.12.0
+git+https://github.com/aboutcode-org/univers.git@5e28f2cd0fbee81a2b1268e88916a52a208bb672#egg=univers
 urllib3==1.26.19
 wcwidth==0.2.5
 websocket-client==0.59.0

vulnerabilities/importers/__init__.py

@@ -79,6 +79,7 @@
 from vulnerabilities.pipelines.v2_importers import retiredotnet_importer as retiredotnet_importer_v2
 from vulnerabilities.pipelines.v2_importers import ruby_importer as ruby_importer_v2
 from vulnerabilities.pipelines.v2_importers import suse_score_importer as suse_score_importer_v2
+from vulnerabilities.pipelines.v2_importers import tuxcare_importer as tuxcare_importer_v2
 from vulnerabilities.pipelines.v2_importers import ubuntu_osv_importer as ubuntu_osv_importer_v2
 from vulnerabilities.pipelines.v2_importers import vulnrichment_importer as vulnrichment_importer_v2
 from vulnerabilities.pipelines.v2_importers import xen_importer as xen_importer_v2
@@ -113,6 +114,7 @@
         nginx_importer_v2.NginxImporterPipeline,
         debian_importer_v2.DebianImporterPipeline,
         mattermost_importer_v2.MattermostImporterPipeline,
+        tuxcare_importer_v2.TuxCareImporterPipeline,
         apache_tomcat_v2.ApacheTomcatImporterPipeline,
         suse_score_importer_v2.SUSESeverityScoreImporterPipeline,
         retiredotnet_importer_v2.RetireDotnetImporterPipeline,

vulnerabilities/pipelines/v2_importers/tuxcare_importer.py

@@ -0,0 +1,208 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import json
+from typing import Iterable
+
+from dateutil.parser import parse
+from packageurl import PackageURL
+from pytz import UTC
+from univers.version_range import RANGE_CLASS_BY_SCHEMES
+
+from vulnerabilities.importer import AdvisoryDataV2
+from vulnerabilities.importer import AffectedPackageV2
+from vulnerabilities.importer import VulnerabilitySeverity
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+from vulnerabilities.severity_systems import GENERIC
+from vulnerabilities.utils import fetch_response
+
+# 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"]
+
+
+class TuxCareImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
+    pipeline_id = "tuxcare_importer_v2"
+    spdx_license_expression = "Apache-2.0"
+    license_url = "https://tuxcare.com/legal"
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.fetch,
+            cls.group_records_by_cve,
+            cls.collect_and_store_advisories,
+        )
+
+    def fetch(self) -> None:
+        url = "https://cve.tuxcare.com/els/download-json?orderBy=updated-desc"
+        self.log(f"Fetching `{url}`")
+        response = fetch_response(url)
+        self.response = response.json() if response else []
+
+    def group_records_by_cve(self):
+        """
+        A single CVE can appear in multiple records across different operating systems, distributions, or package versions. This method groups all records with the same CVE together and skips entries that are invalid or marked as not affected. The result is a dictionary keyed by CVE ID, with each value containing the related records.
+        """
+        self.cve_to_records = {}
+        skipped_invalid = 0
+        skipped_non_affected = 0
+
+        for record in self.response:
+            cve_id = record.get("cve", "").strip()
+            if not cve_id:
+                self.log(f"Skipping record with empty 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]):
+                self.log(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:
+                self.log(f"Skipping {cve_id}: unrecognized status '{status}'")
+                skipped_invalid += 1
+                continue
+
+            if cve_id not in self.cve_to_records:
+                self.cve_to_records[cve_id] = []
+            self.cve_to_records[cve_id].append(record)
+
+        total_skipped = skipped_invalid + skipped_non_affected
+        self.log(
+            f"Grouped {len(self.response):,d} records into {len(self.cve_to_records):,d} unique CVEs "
+            f"(skipped {total_skipped:,d}: {skipped_invalid:,d} invalid, "
+            f"{skipped_non_affected:,d} non-affected)"
+        )
+
+    def advisories_count(self) -> int:
+        return len(self.cve_to_records)
+
+    def _create_purl(self, project_name: str, os_name: str) -> PackageURL:
+        normalized_os = os_name.lower().replace(" ", "-")
+        os_lower = os_name.lower()
+
+        os_mapping = {
+            "ubuntu": ("deb", "ubuntu"),
+            "debian": ("deb", "debian"),
+            "centos": ("rpm", "centos"),
+            "almalinux": ("rpm", "almalinux"),
+            "rhel": ("rpm", "rhel"),
+            "oracle": ("rpm", "oracle"),
+            "cloudlinux": ("rpm", "cloudlinux"),
+            "alpine": ("apk", "alpine"),
+            "unknown": ("generic", "tuxcare"),
+            "tuxcare": ("generic", "tuxcare"),
+        }
+
+        for keyword, (ptype, pns) in os_mapping.items():
+            if keyword in os_lower:
+                pkg_type = ptype
+                namespace = pns
+                break
+        else:
+            return None
+
+        qualifiers = {"distro": normalized_os}
+
+        return PackageURL(
+            type=pkg_type, namespace=namespace, name=project_name, qualifiers=qualifiers
+        )
+
+    def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
+        for cve_id, records in self.cve_to_records.items():
+            affected_packages = []
+            severities = []
+            date_published = None
+            all_records = []
+
+            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:
+                    self.log(
+                        f"Skipping package {project_name} on {os_name} for {cve_id} - unexpected OS type"
+                    )
+                    continue
+
+                version_range_class = RANGE_CLASS_BY_SCHEMES.get(purl.type)
+
+                try:
+                    version_range = version_range_class.from_versions([version])
+                except ValueError as e:
+                    self.log(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,
+                    )
+                )
+
+                # Severity is per-CVE hence we add it only once
+                if severity and score and not severities:
+                    severities.append(
+                        VulnerabilitySeverity(
+                            system=GENERIC,
+                            value=score,
+                            scoring_elements=severity,
+                        )
+                    )
+
+                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:
+                        self.log(f"Failed to parse date {last_updated} for {cve_id}: {e}")
+
+                all_records.append(record)
+
+            if not affected_packages:
+                self.log(f"Skipping {cve_id} - no valid affected packages")
+                continue
+
+            yield AdvisoryDataV2(
+                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(all_records, indent=2, ensure_ascii=False),
+            )

vulnerabilities/tests/pipelines/v2_importers/test_tuxcare_importer_v2.py

@@ -0,0 +1,67 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import json
+from pathlib import Path
+from unittest import TestCase
+from unittest.mock import Mock
+from unittest.mock import patch
+
+from vulnerabilities.pipelines.v2_importers.tuxcare_importer import TuxCareImporterPipeline
+from vulnerabilities.tests import util_tests
+
+TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "tuxcare"
+
+
+class TestTuxCareImporterPipeline(TestCase):
+    @patch("vulnerabilities.pipelines.v2_importers.tuxcare_importer.fetch_response")
+    def test_collect_advisories(self, mock_fetch):
+        sample_path = TEST_DATA / "data.json"
+        sample_data = json.loads(sample_path.read_text(encoding="utf-8"))
+
+        mock_fetch.return_value = Mock(json=lambda: sample_data)
+
+        pipeline = TuxCareImporterPipeline()
+        pipeline.fetch()
+        pipeline.group_records_by_cve()
+
+        advisories = [data.to_dict() for data in list(pipeline.collect_advisories())]
+
+        expected_file = TEST_DATA / "expected.json"
+        util_tests.check_results_against_json(advisories, expected_file)
+
+        assert len(advisories) == 14
+
+    def test_create_purl(self):
+        pipeline = TuxCareImporterPipeline()
+
+        cases = [
+            ("squid", "CloudLinux 7 ELS", "rpm", "cloudlinux", "cloudlinux-7-els"),
+            ("squid", "Oracle Linux 7 ELS", "rpm", "oracle", "oracle-linux-7-els"),
+            ("kernel", "CentOS 8.5 ELS", "rpm", "centos", "centos-8.5-els"),
+            ("squid", "CentOS Stream 8 ELS", "rpm", "centos", "centos-stream-8-els"),
+            ("libpng", "CentOS 7 ELS", "rpm", "centos", "centos-7-els"),
+            ("java-11-openjdk", "RHEL 7 ELS", "rpm", "rhel", "rhel-7-els"),
+            ("mysql", "AlmaLinux 9.2 ESU", "rpm", "almalinux", "almalinux-9.2-esu"),
+            ("linux", "Ubuntu 16.04 ELS", "deb", "ubuntu", "ubuntu-16.04-els"),
+            ("samba", "Debian 10 ELS", "deb", "debian", "debian-10-els"),
+            ("dpkg", "Alpine Linux 3.18 ELS", "apk", "alpine", "alpine-linux-3.18-els"),
+            ("kernel", "Unknown OS", "generic", "tuxcare", "unknown-os"),
+            ("webkit2gtk3", "TuxCare 9.6 ESU", "generic", "tuxcare", "tuxcare-9.6-esu"),
+        ]
+
+        for name, os_name, expected_type, expected_ns, expected_distro in cases:
+            purl = pipeline._create_purl(name, os_name)
+            assert purl is not None, f"Expected purl for os_name={os_name!r}"
+            assert purl.type == expected_type
+            assert purl.namespace == expected_ns
+            assert purl.qualifiers == {"distro": expected_distro}
+
+        # Invalid PURL
+        assert pipeline._create_purl("foo", "Foo 123") is None