Merge pull request #16 from krystianbajno/feature/github-poc

Added Proof of Concept exploits from GitHub

README.md

@@ -32,6 +32,7 @@ python3 cveseeker.py cve-2024 --critical --high --medium --low # include critica
 - [www.rapid7.com](https://www.rapid7.com) (WIP)
 - [cve.mitre.org](https://cve.mitre.org/cve/search_cve_list.html) (WIP)
 - [github.com](https://github.com)  (WIP)
+- [github.com PoC](https://github.com/nomi-sec/PoC-in-GitHub)  (IMPLEMENTED)
 - [github.com advisories](https://github.com/advisories) (IMPLEMENTED)
 - [github.com/trickest/cve](https://github.com/search?q=repo%3Atrickest%2Fcve%20cve-2024&type=code) (IMPLEMENTED)
 

config.yaml

@@ -11,4 +11,5 @@ enrichment:
   sources:
     vulners: true
     github: true
-    cisa_kev: true
+    cisa_kev: true
+    github_poc: true

services/search/engine/collection.py

@@ -2,9 +2,7 @@
 import time
 from typing import List
 from models.vulnerability import Vulnerability
-from models.vulnerability_intelligence import VulnerabilityIntelligence
 from services.api.source import Source
-from services.vulnerability_intelligence.enrichment.vulnerability_intelligence_enrichment_manager import VulnerabilityIntelligenceEnrichmentManager
 
 def collect_from_source_with_retries(manager, source: Source, keywords: List[str], max_results: int) -> List[Vulnerability]:
     attempts = 0
@@ -24,13 +22,6 @@ def collect_from_source_with_retries(manager, source: Source, keywords: List[str
             time.sleep(retry_delay)
             retry_delay *= 2
 
-def is_enrichment_enabled(config: dict) -> bool:
-    return any(config.get('sources', {}).values())
-
-def perform_enrichment(vulnerabilities: List[VulnerabilityIntelligence], config: dict) -> List[VulnerabilityIntelligence]:
-    enrichment_manager = VulnerabilityIntelligenceEnrichmentManager(vulnerabilities, config)
-    return enrichment_manager.enrich()
-
 def collect_results(manager, keywords: List[str], max_results: int) -> List[Vulnerability]:
     collected_results = []
     with ThreadPoolExecutor(max_workers=256) as executor:

services/search/engine/enrichment.py

@@ -0,0 +1,10 @@
+from typing import List
+from models.vulnerability_intelligence import VulnerabilityIntelligence
+from services.vulnerability_intelligence.enrichment.vulnerability_intelligence_enrichment_manager import VulnerabilityIntelligenceEnrichmentManager
+
+def is_enrichment_enabled(config: dict) -> bool:
+    return any(config.get('sources', {}).values())
+
+def perform_enrichment(vulnerabilities: List[VulnerabilityIntelligence], config: dict) -> List[VulnerabilityIntelligence]:
+    enrichment_manager = VulnerabilityIntelligenceEnrichmentManager(vulnerabilities, config)
+    return enrichment_manager.enrich()

services/search/engine/intelligence.py

@@ -0,0 +1,6 @@
+from typing import List
+from models.vulnerability import Vulnerability
+from services.vulnerability_intelligence.processors.vulnerability_intelligence_processor import VulnerabilityIntelligenceProcessor
+
+def prepare_intelligence_from_vulnerabilities(vulnerabilities: List[Vulnerability], keywords):
+    return VulnerabilityIntelligenceProcessor.process(vulnerabilities, keywords)

services/search/search_manager.py

@@ -1,11 +1,12 @@
 from typing import List, Dict
 from models.vulnerability_intelligence import VulnerabilityIntelligence
 from services.api.source import Source
-from services.search.engine.collection import collect_results, is_enrichment_enabled, perform_enrichment
+from services.search.engine.collection import collect_results
+from services.search.engine.enrichment import is_enrichment_enabled, perform_enrichment
 from services.search.engine.filtering import filter_by_severity
+from services.search.engine.intelligence import prepare_intelligence_from_vulnerabilities
 from services.search.engine.modifiers import prepare_descriptions
 from services.search.engine.progress import ProgressManager
-from services.vulnerability_intelligence.processors.vulnerability_intelligence_processor import VulnerabilityIntelligenceProcessor
 
 class SearchManager:
     def __init__(
@@ -25,11 +26,11 @@ def __init__(
     def search(self, keywords: List[str], max_results: int, desired_severities=[]) -> List[VulnerabilityIntelligence]:
         print(f"[*] Initiating search for: \"{' '.join(keywords)}\" with a maximum of {max_results} results per source.\n")
 
-        collected_results = collect_results(self, keywords, max_results)
+        results = collect_results(self, keywords, max_results)
 
         print("[+] Collection process complete.")
-
-        results = VulnerabilityIntelligenceProcessor.process(collected_results, keywords)
+        
+        results = prepare_intelligence_from_vulnerabilities(results, keywords)
 
         if is_enrichment_enabled(self.enrichment_config):
             print("\n[*] Initiating enrichment process.")

services/vulnerabilities/validators/vulnerability_validator.py

@@ -1,7 +1,6 @@
 import re
 from typing import List, Tuple
 from models.vulnerability import Vulnerability
-from services.api.sources.cisa_kev import CISAKEVAPI
 
 class VulnerabilityValidator:
     version_pattern = r'\b\d+(\.\d+)*\b'

services/vulnerability_intelligence/enrichment/enrichment/github_poc.py

@@ -0,0 +1,38 @@
+import httpx
+import logging
+from typing import Dict
+
+def fetch_github_poc_data(cve: str) -> Dict:
+    year = cve.split('-')[1]
+    url = f"https://raw.githubusercontent.com/nomi-sec/PoC-in-GitHub/refs/heads/master/{year}/{cve}.json"
+
+    pocs = []
+
+    try:
+        response = httpx.get(url, timeout=15)
+
+        if response.status_code == 200:
+            json_data = response.json()
+
+
+            for entry in json_data:
+                data = {
+                    'github_url': 'N/A',
+                    'github_description': None,
+                    'github_date': 'N/A',
+                    'github_tags': [],
+                    'github_stars': 0
+                }
+
+                data['github_url'] = entry["html_url"]
+                data['github_description'] = entry["description"]
+                data['github_date'] = entry["updated_at"]
+                data['github_tags'].extend(entry["topics"])
+                data['github_stars'] = entry["stargazers_count"]
+
+                pocs.append(data)
+
+    except Exception as e:
+        logging.error(f"Error fetching JSON PoC data for CVE {cve}: {e}")
+
+    return pocs

services/vulnerability_intelligence/enrichment/vulnerability_intelligence_enrichment_manager.py

@@ -4,7 +4,9 @@
 
 from services.vulnerability_intelligence.enrichment.enrichment.cisa_kev import cisa_kev_enrich_vulnerability
 from services.vulnerability_intelligence.enrichment.enrichment.github import github_fetch_cve_data
+from services.vulnerability_intelligence.enrichment.enrichment.github_poc import fetch_github_poc_data
 from services.vulnerability_intelligence.enrichment.enrichment.vulners import vulners_find_related_cve_data
+from services.vulnerability_intelligence.handlers.github_poc_handler import GitHubPoCHandler
 from services.vulnerability_intelligence.handlers.vulners_handler import VulnersHandler
 from services.vulnerability_intelligence.handlers.github_handler import GitHubHandler
 from services.vulnerability_intelligence.handlers.cisa_kev_handler import CisaKevHandler
@@ -17,12 +19,14 @@ def __init__(self, vulnerability_intelligence_list: List[VulnerabilityIntelligen
         self.enrichment_functions = {
             "Vulners": vulners_find_related_cve_data,
             "GitHub": github_fetch_cve_data,
+            "GitHubPoc": fetch_github_poc_data,
             "CISA KEV": cisa_kev_enrich_vulnerability 
         }
 
         self.handlers = {
             "Vulners": VulnersHandler,
             "GitHub": GitHubHandler,
+            "GitHubPoc": GitHubPoCHandler,
             "CISA KEV": CisaKevHandler
         }
 

services/vulnerability_intelligence/handlers/github_poc_handler.py

@@ -0,0 +1,34 @@
+from services.vulnerability_intelligence.handlers.base_handler import BaseHandler
+from models.vulnerability_intelligence import VulnerabilityIntelligence
+import logging
+
+class GitHubPoCHandler(BaseHandler):
+    def apply(self, vuln_intelligence: VulnerabilityIntelligence):
+        try:   
+            sorted_data = sorted(self.data, key=lambda entry: entry.get('github_stars', 0), reverse=True)
+
+            for entry in sorted_data:
+                github_url = entry.get('github_url', "N/A")
+                github_date = entry.get('github_date', "N/A")
+                github_description = entry.get('github_description')
+                github_tags = entry.get('github_tags', [])
+                github_stars = entry.get('github_stars')
+
+                if github_description:
+                    vuln_intelligence.descriptions.append({
+                        "source": self.enrich_source_name("GitHub PoC"),
+                        "text": github_description,
+                        "date": github_date
+                    })
+
+                vuln_intelligence.urls.append({
+                    "source": self.enrich_source_name(f"GitHub - PoC Exploit [{github_stars} ⭐]"),
+                    "url": github_url,
+                    "date": github_date
+                })
+
+                vuln_intelligence.reference_urls.update([github_url]) 
+                vuln_intelligence.tags.update(github_tags)
+
+        except Exception as e:
+            logging.error(f"Error applying GitHub JSON PoC enrichment: {e}")