Merge branch 'ref-type-exploitdb' into vul-exploit

Author: ziadhany

vulnerabilities/improvers/__init__.py

@@ -7,6 +7,7 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
+from vulnerabilities.improvers import exploitdb
 from vulnerabilities.improvers import valid_versions
 from vulnerabilities.improvers import vulnerability_kev
 from vulnerabilities.improvers import vulnerability_status
@@ -29,6 +30,7 @@
     valid_versions.GithubOSVImprover,
     vulnerability_status.VulnerabilityStatusImprover,
     vulnerability_kev.VulnerabilityKevImprover,
+    exploitdb.ExploitDBImprover,
 ]
 
 IMPROVERS_REGISTRY = {x.qualified_name: x for x in IMPROVERS_REGISTRY}

vulnerabilities/improvers/exploitdb.py

@@ -0,0 +1,92 @@
+import csv
+import io
+import logging
+from typing import Iterable
+
+from django.db import IntegrityError
+from django.db.models import QuerySet
+from sphinx.util import requests
+
+from vulnerabilities.improver import Improver
+from vulnerabilities.improver import Inference
+from vulnerabilities.models import Advisory
+from vulnerabilities.models import Alias
+from vulnerabilities.models import VulnerabilityReference
+from vulnerabilities.models import VulnerabilityRelatedReference
+
+logger = logging.getLogger(__name__)
+
+
+class ExploitDBImprover(Improver):
+    """
+    ExploitDB Improver
+    """
+
+    license_expression = "GPL-2.0"
+
+    @property
+    def interesting_advisories(self) -> QuerySet:
+        # TODO Modify ExploitDB Improver to iterate over the vulnerabilities alias, not the advisory
+        return [Advisory.objects.first()]
+
+    def get_inferences(self, advisory_data) -> Iterable[Inference]:
+        """
+        Fetch ExploitDB data, iterate over it to find the vulnerability with the specified alias, and create or update
+        the ref and ref-type accordingly.
+        """
+
+        exploit_db_url = (
+            "https://gitlab.com/exploit-database/exploitdb/-/raw/main/files_exploits.csv"
+        )
+        response = requests.get(exploit_db_url)
+
+        if response.status_code != 200:
+            logger.error(f"Failed to fetch ExploitDB URL: {exploit_db_url}")
+            return []
+
+        raw_data = io.StringIO(response.text)
+
+        csvreader = csv.reader(raw_data)
+
+        # Ignore the csv header
+        next(csvreader)  # header
+        for row in csvreader:
+            try:
+                aliases = row[11].split(";")
+                for raw_alias in aliases:
+                    alias = Alias.objects.get(alias=raw_alias)
+                    if not alias:
+                        continue
+
+                    vul = alias.vulnerability
+                    if not vul:
+                        continue
+
+                    if raw_alias:
+                        url_map = {
+                            "file_url": f"https://gitlab.com/exploit-database/exploitdb/-/blob/main/{row[1]}"
+                            if row[1]
+                            else None,
+                            "direct_url": row[16] if row[16] else None,
+                        }
+
+                        for key, url in url_map.items():
+                            if url:
+                                ref, created = VulnerabilityReference.objects.update_or_create(
+                                    reference_id=row[11],
+                                    reference_type=VulnerabilityReference.EXPLOIT,
+                                    defaults={"url": url},
+                                )
+
+                                if created:
+                                    VulnerabilityRelatedReference.objects.create(
+                                        vulnerability=vul,
+                                        reference=ref,
+                                    )
+
+            except Alias.DoesNotExist as e:
+                logger.error(f"No Alias found for exploit id {row[0]}: {e}")
+            except Exception as e:
+                logger.error(e)
+
+        return []