Fix : Added Pipeline to detect Reserved/Non-Existent CVEs

.github/workflows/check-mitre-api.yml

@@ -0,0 +1,27 @@
+name: Detect Non-Existent CVEs
+
+on:
+  workflow_dispatch: 
+  schedule:
+    - cron: '0 2 * * 0'  # Run weekly on Sunday at 2:00 AM UTC
+
+jobs:
+  detect-nonexistent-cves:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.10'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      - name: Run pipeline
+        run: |
+          python manage.py improve detect_nonexistent_cves

vulnerabilities/improvers/__init__.py

@@ -11,6 +11,7 @@
 from vulnerabilities.improvers import vulnerability_status
 from vulnerabilities.pipelines import VulnerableCodePipeline
 from vulnerabilities.pipelines import add_cvss31_to_CVEs
+from vulnerabilities.pipelines import check_mitre_api
 from vulnerabilities.pipelines import collect_commits
 from vulnerabilities.pipelines import compute_package_risk
 from vulnerabilities.pipelines import compute_package_version_rank
@@ -47,6 +48,7 @@
     collect_commits.CollectFixCommitsPipeline,
     add_cvss31_to_CVEs.CVEAdvisoryMappingPipeline,
     remove_duplicate_advisories.RemoveDuplicateAdvisoriesPipeline,
+    check_mitre_api.DetectNonExistentCvesPipeline,
 ]
 
 IMPROVERS_REGISTRY = {

vulnerabilities/pipelines/check_mitre_api.py

@@ -0,0 +1,161 @@
+#
+# 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 re
+import time
+from datetime import datetime
+from datetime import timedelta
+
+import requests
+from aboutcode.pipeline import LoopProgress
+
+from vulnerabilities.improvers.vulnerability_status import MITRE_API_URL
+from vulnerabilities.models import Alias
+from vulnerabilities.models import VulnerabilityChangeLog
+from vulnerabilities.models import VulnerabilityStatusType
+from vulnerabilities.pipelines import VulnerableCodePipeline
+from vulnerabilities.utils import fetch_response
+from vulnerabilities.utils import get_item
+
+
+class DetectNonExistentCvesPipeline(VulnerableCodePipeline):
+    """
+    Pipeline to detect and properly mark reserved or non-existent CVEs.
+    """
+
+    pipeline_id = "detect_nonexistent_cves"
+
+    @classmethod
+    def steps(cls):
+        return (cls.process_cves,)
+
+    def process_cves(self):
+        """
+        Process all CVE IDs in the database to detect reserved or non-existent CVEs.
+        """
+        # Get all CVE aliases
+        cve_aliases = Alias.objects.filter(alias__regex=r"^CVE-\d{4}-\d{4,}$").select_related(
+            "vulnerability"
+        )
+
+        self.log(f"Processing {cve_aliases.count():,d} CVE IDs for reserved/non-existent status")
+
+        progress = LoopProgress(
+            total_iterations=cve_aliases.count(),
+            logger=self.log,
+            progress_step=5,
+        )
+
+        reserved_count = 0
+        invalid_count = 0
+        error_count = 0
+        rate_limited = False
+
+        batch_size = 100
+        for cve_alias in progress.iter(
+            cve_aliases.order_by("alias").paginated(per_page=batch_size)
+        ):
+            if rate_limited:
+                # Add a delay if we hit rate limits
+                time.sleep(1)
+
+            cve_id = cve_alias.alias
+            vulnerability = cve_alias.vulnerability
+
+            # Skip if vulnerability doesn't exist
+            if not vulnerability:
+                continue
+
+            # Skip if vulnerability already has a non-PUBLISHED status
+            if vulnerability.status != VulnerabilityStatusType.PUBLISHED:
+                continue
+
+            try:
+                status = self.check_cve_status(cve_id)
+
+                if status == VulnerabilityStatusType.RESERVED:
+                    self.update_vulnerability_status(vulnerability, status)
+                    reserved_count += 1
+                elif status == VulnerabilityStatusType.INVALID:
+                    self.update_vulnerability_status(vulnerability, status)
+                    invalid_count += 1
+
+            except requests.exceptions.HTTPError as http_error:
+                if http_error.response.status_code == 429:  # Rate limited
+                    rate_limited = True
+                    self.log(f"Rate limited by MITRE API. Adding delay.")
+                    continue
+                else:
+                    self.log(f"HTTP error for {cve_id}: {http_error}")
+                    error_count += 1
+            except Exception as e:
+                self.log(f"Error processing {cve_id}: {e}")
+                error_count += 1
+
+        self.log(
+            f"Completed. Found {reserved_count} reserved CVEs, {invalid_count} invalid CVEs. Encountered {error_count} errors."
+        )
+
+    def check_cve_status(self, cve_id):
+        """
+        Check the status of a CVE ID using the MITRE API.
+        Returns the appropriate VulnerabilityStatusType.
+        """
+        url = f"{MITRE_API_URL}{cve_id}"
+
+        try:
+            response = fetch_response(url=url)
+            response_json = response.json()
+
+            cve_state = get_item(response_json, "cveMetadata", "state") or None
+
+            try:
+                tags = get_item(response_json, "containers", "cna", "tags") or []
+            except (TypeError, AttributeError, KeyError) as e:
+                tags = []
+                self.log(f"Missing attribute tags in {response_json}")
+
+            if "disputed" in tags:
+                return VulnerabilityStatusType.DISPUTED
+
+            if cve_state:
+                if cve_state == "REJECTED":
+                    return VulnerabilityStatusType.INVALID
+                elif cve_state == "RESERVED":
+                    return VulnerabilityStatusType.RESERVED
+                else:
+                    return VulnerabilityStatusType.PUBLISHED
+
+            return VulnerabilityStatusType.PUBLISHED
+
+        except requests.exceptions.HTTPError as http_error:
+            if http_error.response.status_code == 404:
+                # CVE not found in MITRE database
+                return VulnerabilityStatusType.INVALID
+            raise
+
+    def update_vulnerability_status(self, vulnerability, status):
+        """
+        Update the status of a vulnerability and create a change log entry.
+        """
+
+        old_status = vulnerability.status
+        vulnerability.status = status
+        vulnerability.save()
+
+        # Create change log entry
+        VulnerabilityChangeLog.objects.create(
+            vulnerability=vulnerability,
+            field="status",
+            old_value=old_status,
+            new_value=status,
+            automated=True,
+            message=f"Updated CVE status via MITRE API check",
+            data_source=self.pipeline_id,
+        )