Merge pull request #15 from krystianbajno/feature/cisa-kev

Add CISA-KEV, more enrichment, general refactor, more features, filte…

Author: krystianbajno

.gitignore

@@ -1,5 +1,6 @@
 # Byte-compiled / optimized / DLL files
 .DS_Store
+cache/
 cveseeker_*_report.csv
 cveseeker_*_report.json
 cveseeker_*_report.html

README.md

@@ -18,19 +18,21 @@ python3 cveseeker.py windows remote code execution
 python3 cveseeker.py cve-2024
 python3 cveseeker.py cve-2024 --max-per-provider 2000 # max results per provider, default 100
 python3 cveseeker.py cve-2024 --report # generate CSV, JSON and HTML report
+python3 cveseeker.py cve-2024 --critical --high --medium --low # include critical, high, medium, and low severities
+
 ```
 
 # Sources
 - [www.exploit-db.com](https://www.exploit-db.com) (IMPLEMENTED)
 - [services.nvd.nist.gov](https://services.nvd.nist.gov/rest/json/cves/2.0?noRejected) (IMPLEMENTED)
 - [www.opencve.io](https://www.opencve.io) (IMPLEMENTED)
 - [www.packetstormsecurity.com](https://packetstormsecurity.com) (IMPLEMENTED)
-- [github.com advisories](https://github.com/advisories) (IMPLEMENTED)
 - [vulners.com](https://vulners.com/search) (IMPLEMENTED)
-- [www.cisa.gov](https://www.cisa.gov/known-exploited-vulnerabilities-catalog) (WIP)
+- [www.cisa.gov - KEV](https://www.cisa.gov/known-exploited-vulnerabilities-catalog) (IMPLEMENTED)
 - [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 advisories](https://github.com/advisories) (IMPLEMENTED)
 - [github.com/trickest/cve](https://github.com/search?q=repo%3Atrickest%2Fcve%20cve-2024&type=code) (IMPLEMENTED)
 
 # Reporting

config.yaml

@@ -5,4 +5,10 @@ providers:
   ExploitDBAPI: true
   GitHubAdvisoryAPI: true
   VulnersAPI: false
-enrichment: true
+  CISAKEVAPI: true
+
+enrichment:
+  sources:
+    vulners: true
+    github: true
+    cisa_kev: true

cveseeker.py

@@ -27,6 +27,11 @@ def main():
         help="Generate CSV report"
     )
 
+    parser.add_argument('--low', action='store_true', help='Include low severity vulnerabilities')
+    parser.add_argument('--medium', action='store_true', help='Include medium severity vulnerabilities')
+    parser.add_argument('--high', action='store_true', help='Include high severity vulnerabilities')
+    parser.add_argument('--critical', action='store_true', help='Include critical severity vulnerabilities')
+    
     parser.add_argument(
         '--playwright',
         action="store_true",
@@ -40,10 +45,14 @@ def main():
 
     keywords = args.keywords
 
-    search_provider = SearchProvider(playwright_enabled=args.playwright)
+    search_provider = SearchProvider(playwright_enabled=args.playwright, config_file='config.yaml')
     search_service = search_provider.make_service_api()
 
-    results = search_service.search(keywords, args.max_per_provider)
+    desired_severities = [
+        severity for severity in ['low', 'medium', 'high', 'critical'] if getattr(args, severity)
+    ]
+
+    results = search_service.search(keywords, args.max_per_provider, desired_severities=desired_severities)
 
     VulnerabilityIntelligencePrinter.print(results)
 

models/__init__.py

@@ -1,5 +1,5 @@
 import yaml
-from services.search_manager import SearchManager
+from services.api.sources.cisa_kev import CISAKEVAPI
 
 from services.api.sources.exploitdb import ExploitDBAPI
 from services.api.sources.github_advisories import GitHubAdvisoryAPI
@@ -8,7 +8,12 @@
 from services.api.sources.packetstormsecurity import PacketStormSecurityAPI
 from services.api.sources.vulners import VulnersAPI
 
-class SearchProvider():
+from typing import Dict
+
+from services.search.search_manager import SearchManager
+from services.search.engine.progress import ProgressManager
+
+class SearchProvider:
     def __init__(self, playwright_enabled=False, config_file='config.yaml'):
         self.search_service: SearchManager = None
         self.playwright_enabled = playwright_enabled
@@ -21,17 +26,18 @@ def __init__(self, playwright_enabled=False, config_file='config.yaml'):
             'ExploitDBAPI': ExploitDBAPI,
             'GitHubAdvisoryAPI': GitHubAdvisoryAPI,
             'VulnersAPI': VulnersAPI,
+            "CISAKEVAPI": CISAKEVAPI
         }
 
     def make_service_api(self) -> SearchManager:
         if self.search_service is None:
             self.boot()
         return self.search_service
-        
+            
     def boot(self):
         config = self.load_config()
         providers_config = config.get('providers', {})
-        enrichment_config = config.get("enrichment", False)
+        enrichment_config = config.get("enrichment", {})
 
         providers = []
 
@@ -48,10 +54,11 @@ def boot(self):
         if self.playwright_enabled:
             playwright_providers = []
             providers.extend(playwright_providers)
+            
+        progress_manager = ProgressManager()
+        self.search_service = SearchManager(providers, enrichment_config, progress_manager=progress_manager)
 
-        self.search_service = SearchManager(providers, enrichment_enabled=enrichment_config)
-        
-    def load_config(self):
+    def load_config(self) -> Dict:
         try:
             with open(self.config_file, 'r') as f:
                 config = yaml.safe_load(f)

providers/__init__.py

@@ -0,0 +1,115 @@
+import os
+import time
+import json
+from typing import List, Dict
+import httpx
+from dateutil import parser as dateutil_parser
+
+from models.vulnerability import Vulnerability
+from services.api.source import Source
+from services.vulnerabilities.factories.vulnerability_factory import VulnerabilityFactory
+
+class CISAKEVAPI(Source):
+    CACHE_DIR = "cache"
+    CACHE_FILE = os.path.join(CACHE_DIR, "cisa_kev_cache.json")
+    CACHE_DURATION = 600
+
+    def __init__(self):
+        self.url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
+        self.ensure_cache_dir()
+
+    def ensure_cache_dir(self):
+        if not os.path.exists(self.CACHE_DIR):
+            try:
+                os.makedirs(self.CACHE_DIR)
+                print(f"[*] Created cache directory at '{self.CACHE_DIR}'.")
+            except Exception as e:
+                print(f"[!] Failed to create cache directory '{self.CACHE_DIR}': {e}")
+
+    def is_cache_valid(self) -> bool:
+        if os.path.exists(self.CACHE_FILE):
+            cache_mtime = os.path.getmtime(self.CACHE_FILE)
+            current_time = time.time()
+            if (current_time - cache_mtime) < self.CACHE_DURATION:
+                return True
+        return False
+
+    def load_cache(self) -> Dict:
+        try:
+            with open(self.CACHE_FILE, 'r', encoding='utf-8') as f:
+                data = json.load(f)
+                print("[*] Loaded CISA KEV catalog from cache.")
+                return data
+        except Exception as e:
+            print(f"[!] Error reading CISA KEV cache: {e}")
+            return {}
+
+    def update_cache(self, data: Dict):
+        try:
+            with open(self.CACHE_FILE, 'w', encoding='utf-8') as f:
+                json.dump(data, f, ensure_ascii=False, indent=4)
+                print("[+] CISA KEV catalog downloaded and cached.")
+        except Exception as e:
+            print(f"[!] Error updating CISA KEV cache: {e}")
+
+    def fetch_data(self) -> Dict:
+        try:
+            print("[*] Downloading CISA KEV catalog...")
+            response = httpx.get(self.url, timeout=15)
+            if response.status_code == 200:
+                data = response.json()
+                self.update_cache(data)
+                return data
+            else:
+                print(f"[!] Failed to fetch CISA KEV catalog. Status code: {response.status_code}")
+        except Exception as e:
+            print(f"[!] Error fetching CISA KEV data: {e}")
+        return {}
+
+    def get_data(self) -> Dict:
+        if self.is_cache_valid():
+            return self.load_cache()
+        else:
+            return self.fetch_data()
+
+    def search(self, keywords: List[str], max_results: int) -> List[Vulnerability]:
+
+        vulnerabilities = []
+        try:
+            data = self.get_data()
+            kev_vulnerabilities = data.get("vulnerabilities", [])
+
+            for item in kev_vulnerabilities:
+                cve_id = item.get("cveID")
+                
+                if not cve_id:
+                    continue
+
+                date_added = item.get("dateAdded")
+                try:
+                    parsed_date = dateutil_parser.parse(date_added)
+                    date = parsed_date.strftime('%Y-%m-%d')
+                except Exception:
+                    date = date_added or "N/A"
+
+                notes = item.get("notes", "")
+                reference_urls = [url.strip() for url in notes.split(" ; ") if url.strip()]
+                weaknesses = item.get("cwes", [])
+
+                vulnerability = VulnerabilityFactory.make(
+                    id=cve_id,
+                    source=self.__class__.__name__,
+                    url="https://www.cisa.gov/known-exploited-vulnerabilities-catalog",
+                    date=date,
+                    reference_urls=reference_urls,
+                    description=item.get("shortDescription", "N/A"),
+                    vulnerable_components=[item.get("product", "N/A")],
+                    tags=[item.get("vendorProject", "N/A")],
+                    weaknesses=weaknesses
+                )
+                vulnerabilities.append(vulnerability)
+
+        except Exception as e:
+            print(f"[!] Error processing CISA KEV data: {e}")
+
+        return vulnerabilities