mirror of
https://github.com/projectdiscovery/nuclei-templates.git
synced 2026-01-31 15:53:33 +08:00
Fix KEV workflow to use tags field only (remove vKEV)
- Removed all vKEV functionality as requested - Fixed KEV script to only add 'kev' to tags field (not metadata) - Simplified workflow to focus on CISA KEV catalog only - Script now correctly adds/removes kev tags in tags field - Removed VulnCheck API integration placeholder 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
248
.github/scripts/update-kev.py
vendored
248
.github/scripts/update-kev.py
vendored
@@ -5,8 +5,7 @@ KEV Tagging Updater for Nuclei Templates
|
|||||||
This script updates KEV (Known Exploited Vulnerabilities) tags for all CVE templates by:
|
This script updates KEV (Known Exploited Vulnerabilities) tags for all CVE templates by:
|
||||||
1. Fetching the latest CISA KEV catalog
|
1. Fetching the latest CISA KEV catalog
|
||||||
2. Checking each CVE template against the KEV list
|
2. Checking each CVE template against the KEV list
|
||||||
3. Adding 'kev' tag and metadata for KEV CVEs
|
3. Adding 'kev' tag to the tags field for KEV CVEs
|
||||||
4. Adding 'vkev' metadata for VulnCheck KEV CVEs (requires API access)
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import os
|
import os
|
||||||
@@ -14,29 +13,23 @@ import re
|
|||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
import json
|
import json
|
||||||
import yaml
|
|
||||||
import requests
|
import requests
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Dict, List, Set, Optional
|
from typing import Set, Optional
|
||||||
|
|
||||||
# Configuration
|
# Configuration
|
||||||
CISA_KEV_JSON_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
|
CISA_KEV_JSON_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
|
||||||
TIMEOUT = 30
|
TIMEOUT = 30
|
||||||
MAX_RETRIES = 3
|
MAX_RETRIES = 3
|
||||||
|
|
||||||
# VulnCheck KEV configuration (placeholder - requires API key)
|
|
||||||
VULNCHECK_KEV_API = "https://api.vulncheck.com/v3/kev" # Placeholder
|
|
||||||
VULNCHECK_API_KEY = os.environ.get('VULNCHECK_API_KEY') # Optional API key
|
|
||||||
|
|
||||||
class KEVUpdater:
|
class KEVUpdater:
|
||||||
def __init__(self, root_dir: str):
|
def __init__(self, root_dir: str):
|
||||||
self.root_dir = Path(root_dir)
|
self.root_dir = Path(root_dir)
|
||||||
self.updated_count = 0
|
self.updated_count = 0
|
||||||
self.error_count = 0
|
self.error_count = 0
|
||||||
self.cisa_kev_cves = set()
|
self.cisa_kev_cves = set()
|
||||||
self.vulncheck_kev_cves = set()
|
|
||||||
|
|
||||||
def find_cve_templates(self) -> List[Path]:
|
def find_cve_templates(self) -> list[Path]:
|
||||||
"""Find all CVE template files."""
|
"""Find all CVE template files."""
|
||||||
cve_files = []
|
cve_files = []
|
||||||
|
|
||||||
@@ -106,44 +99,66 @@ class KEVUpdater:
|
|||||||
|
|
||||||
return kev_cves
|
return kev_cves
|
||||||
|
|
||||||
def fetch_vulncheck_kev_data(self) -> Set[str]:
|
|
||||||
"""Fetch VulnCheck KEV data (requires API key)."""
|
|
||||||
if not VULNCHECK_API_KEY:
|
|
||||||
print("VulnCheck API key not provided, skipping vKEV updates")
|
|
||||||
return set()
|
|
||||||
|
|
||||||
# Placeholder implementation - would need actual VulnCheck API integration
|
|
||||||
print("VulnCheck KEV integration not implemented yet")
|
|
||||||
print("To enable vKEV updates, implement VulnCheck API integration")
|
|
||||||
|
|
||||||
# TODO: Implement VulnCheck API integration
|
|
||||||
# This would require:
|
|
||||||
# 1. Authentication with VulnCheck API
|
|
||||||
# 2. Fetching their KEV dataset
|
|
||||||
# 3. Extracting CVE IDs from their format
|
|
||||||
|
|
||||||
return set()
|
|
||||||
|
|
||||||
def has_kev_tag(self, content: str) -> bool:
|
def has_kev_tag(self, content: str) -> bool:
|
||||||
"""Check if template already has KEV tag."""
|
"""Check if template already has KEV tag in tags field."""
|
||||||
# Check in tags field
|
# Look for tags field and check if kev is present
|
||||||
tags_match = re.search(r'tags:\s*([^\n]+)', content)
|
tags_match = re.search(r'tags:\s*([^\n]+)', content)
|
||||||
if tags_match:
|
if tags_match:
|
||||||
tags_str = tags_match.group(1)
|
tags_str = tags_match.group(1)
|
||||||
|
# Check for kev as a standalone tag (not part of another word)
|
||||||
if re.search(r'\bkev\b', tags_str, re.IGNORECASE):
|
if re.search(r'\bkev\b', tags_str, re.IGNORECASE):
|
||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
|
|
||||||
def has_kev_metadata(self, content: str) -> bool:
|
def add_kev_tag(self, content: str) -> tuple[str, bool]:
|
||||||
"""Check if template already has KEV metadata."""
|
"""Add kev tag to the tags field. Returns (updated_content, was_updated)."""
|
||||||
return bool(re.search(r'kev:\s*true', content, re.IGNORECASE))
|
if self.has_kev_tag(content):
|
||||||
|
return content, False
|
||||||
|
|
||||||
|
# Find the tags field
|
||||||
|
tags_match = re.search(r'(\s*tags:\s*)([^\n]+)', content)
|
||||||
|
if tags_match:
|
||||||
|
indent = tags_match.group(1)
|
||||||
|
existing_tags = tags_match.group(2).rstrip()
|
||||||
|
|
||||||
|
# Add kev tag at the end
|
||||||
|
new_tags = existing_tags + ',kev'
|
||||||
|
new_line = f"{indent}{new_tags}"
|
||||||
|
|
||||||
|
updated_content = content.replace(tags_match.group(0), new_line)
|
||||||
|
return updated_content, True
|
||||||
|
else:
|
||||||
|
print("Warning: No tags field found to add kev tag")
|
||||||
|
return content, False
|
||||||
|
|
||||||
def has_vkev_metadata(self, content: str) -> bool:
|
def remove_kev_tag(self, content: str) -> tuple[str, bool]:
|
||||||
"""Check if template already has vKEV metadata."""
|
"""Remove kev tag from the tags field. Returns (updated_content, was_updated)."""
|
||||||
return bool(re.search(r'vkev:\s*true', content, re.IGNORECASE))
|
if not self.has_kev_tag(content):
|
||||||
|
return content, False
|
||||||
|
|
||||||
|
# Find and update tags field
|
||||||
|
tags_match = re.search(r'(\s*tags:\s*)([^\n]+)', content)
|
||||||
|
if tags_match:
|
||||||
|
indent = tags_match.group(1)
|
||||||
|
existing_tags = tags_match.group(2)
|
||||||
|
|
||||||
|
# Remove kev tag in various positions
|
||||||
|
new_tags = existing_tags
|
||||||
|
# Remove ,kev or kev, patterns
|
||||||
|
new_tags = re.sub(r',\s*kev\b', '', new_tags)
|
||||||
|
new_tags = re.sub(r'\bkev\s*,', '', new_tags)
|
||||||
|
new_tags = re.sub(r'^\s*kev\s*$', '', new_tags) # kev is only tag
|
||||||
|
new_tags = re.sub(r'^\s*kev\s*,\s*', '', new_tags) # kev is first tag
|
||||||
|
|
||||||
|
if new_tags != existing_tags:
|
||||||
|
new_line = f"{indent}{new_tags}"
|
||||||
|
updated_content = content.replace(tags_match.group(0), new_line)
|
||||||
|
return updated_content, True
|
||||||
|
|
||||||
|
return content, False
|
||||||
|
|
||||||
def update_template_with_kev(self, file_path: Path, cve_id: str, is_cisa_kev: bool, is_vulncheck_kev: bool) -> bool:
|
def update_template_with_kev(self, file_path: Path, cve_id: str, is_kev: bool) -> bool:
|
||||||
"""Update a template file with KEV tags and metadata."""
|
"""Update a template file with KEV tags."""
|
||||||
try:
|
try:
|
||||||
with open(file_path, 'r', encoding='utf-8') as f:
|
with open(file_path, 'r', encoding='utf-8') as f:
|
||||||
content = f.read()
|
content = f.read()
|
||||||
@@ -151,81 +166,23 @@ class KEVUpdater:
|
|||||||
original_content = content
|
original_content = content
|
||||||
updated = False
|
updated = False
|
||||||
|
|
||||||
# Check current state
|
if is_kev:
|
||||||
has_kev_tag = self.has_kev_tag(content)
|
|
||||||
has_kev_meta = self.has_kev_metadata(content)
|
|
||||||
has_vkev_meta = self.has_vkev_metadata(content)
|
|
||||||
|
|
||||||
# Update CISA KEV
|
|
||||||
if is_cisa_kev:
|
|
||||||
# Add kev tag if missing
|
# Add kev tag if missing
|
||||||
if not has_kev_tag:
|
content, tag_added = self.add_kev_tag(content)
|
||||||
tags_match = re.search(r'(\s+tags:\s*)([^\n]+)', content)
|
if tag_added:
|
||||||
if tags_match:
|
print(f"Added kev tag to {file_path.name}")
|
||||||
indent = tags_match.group(1)
|
|
||||||
existing_tags = tags_match.group(2)
|
|
||||||
# Add kev tag
|
|
||||||
new_tags = existing_tags.rstrip() + ',kev'
|
|
||||||
content = content.replace(tags_match.group(0), f"{indent}{new_tags}")
|
|
||||||
updated = True
|
|
||||||
print(f"Added kev tag to {file_path.name}")
|
|
||||||
|
|
||||||
# Add kev metadata if missing
|
|
||||||
if not has_kev_meta:
|
|
||||||
# Find metadata section
|
|
||||||
metadata_match = re.search(r'(\s+metadata:\s*\n)', content)
|
|
||||||
if metadata_match:
|
|
||||||
# Add kev: true after metadata section start
|
|
||||||
insertion_point = metadata_match.end()
|
|
||||||
indent = " " # Standard YAML indent
|
|
||||||
kev_meta = f"{indent}kev: true\n"
|
|
||||||
content = content[:insertion_point] + kev_meta + content[insertion_point:]
|
|
||||||
updated = True
|
|
||||||
print(f"Added kev metadata to {file_path.name}")
|
|
||||||
else:
|
|
||||||
# Create metadata section if it doesn't exist
|
|
||||||
# Find a good place to insert it (after classification or before tags)
|
|
||||||
classification_match = re.search(r'(\s+classification:.*?\n)(\s+(?:metadata|tags))', content, re.DOTALL)
|
|
||||||
if classification_match:
|
|
||||||
insertion_point = classification_match.start(2)
|
|
||||||
metadata_section = f" metadata:\n kev: true\n"
|
|
||||||
content = content[:insertion_point] + metadata_section + content[insertion_point:]
|
|
||||||
updated = True
|
|
||||||
print(f"Created metadata section with kev for {file_path.name}")
|
|
||||||
|
|
||||||
# Update VulnCheck KEV
|
|
||||||
if is_vulncheck_kev and not has_vkev_meta:
|
|
||||||
# Add vkev metadata
|
|
||||||
metadata_match = re.search(r'(\s+metadata:\s*\n)', content)
|
|
||||||
if metadata_match:
|
|
||||||
# Find a good spot within metadata section
|
|
||||||
insertion_point = metadata_match.end()
|
|
||||||
# Look for existing metadata to maintain order
|
|
||||||
next_field_match = re.search(r'\n(\s+\w+:)', content[insertion_point:])
|
|
||||||
if next_field_match:
|
|
||||||
# Insert before next field
|
|
||||||
actual_insertion = insertion_point + next_field_match.start()
|
|
||||||
indent = " "
|
|
||||||
vkev_meta = f"{indent}vkev: true\n"
|
|
||||||
content = content[:actual_insertion] + vkev_meta + content[actual_insertion:]
|
|
||||||
else:
|
|
||||||
# Insert at end of metadata section
|
|
||||||
indent = " "
|
|
||||||
vkev_meta = f"{indent}vkev: true\n"
|
|
||||||
content = content[:insertion_point] + vkev_meta + content[insertion_point:]
|
|
||||||
updated = True
|
updated = True
|
||||||
print(f"Added vkev metadata to {file_path.name}")
|
else:
|
||||||
elif is_cisa_kev: # Only create metadata if we also added kev
|
# Remove kev tag if present but CVE is not in KEV catalog
|
||||||
# Metadata section was already created above
|
if self.has_kev_tag(content):
|
||||||
# Add vkev to existing metadata
|
# Only remove if we're confident about our KEV data
|
||||||
kev_meta_match = re.search(r'(\s+kev: true\n)', content)
|
if len(self.cisa_kev_cves) > 1000: # Sanity check
|
||||||
if kev_meta_match:
|
content, tag_removed = self.remove_kev_tag(content)
|
||||||
insertion_point = kev_meta_match.end()
|
if tag_removed:
|
||||||
indent = " "
|
print(f"Removed kev tag from {file_path.name} (no longer in KEV catalog)")
|
||||||
vkev_meta = f"{indent}vkev: true\n"
|
updated = True
|
||||||
content = content[:insertion_point] + vkev_meta + content[insertion_point:]
|
else:
|
||||||
updated = True
|
print(f"Warning: {cve_id} has kev tag but not in current KEV catalog (keeping tag)")
|
||||||
print(f"Added vkev metadata after kev for {file_path.name}")
|
|
||||||
|
|
||||||
# Write updated content if changes were made
|
# Write updated content if changes were made
|
||||||
if updated:
|
if updated:
|
||||||
@@ -240,50 +197,12 @@ class KEVUpdater:
|
|||||||
self.error_count += 1
|
self.error_count += 1
|
||||||
return False
|
return False
|
||||||
|
|
||||||
def remove_kev_tags(self, file_path: Path, cve_id: str) -> bool:
|
|
||||||
"""Remove KEV tags from templates that are no longer in KEV lists."""
|
|
||||||
try:
|
|
||||||
with open(file_path, 'r', encoding='utf-8') as f:
|
|
||||||
content = f.read()
|
|
||||||
|
|
||||||
original_content = content
|
|
||||||
updated = False
|
|
||||||
|
|
||||||
# Remove kev tag if present
|
|
||||||
if self.has_kev_tag(content):
|
|
||||||
# Remove from tags
|
|
||||||
content = re.sub(r',\s*kev\b', '', content) # Remove ,kev
|
|
||||||
content = re.sub(r'\bkev\s*,', '', content) # Remove kev,
|
|
||||||
content = re.sub(r'\s+kev\b(?![\w-])', '', content) # Remove standalone kev
|
|
||||||
updated = True
|
|
||||||
print(f"Removed kev tag from {file_path.name}")
|
|
||||||
|
|
||||||
# Remove kev metadata if present
|
|
||||||
if self.has_kev_metadata(content):
|
|
||||||
content = re.sub(r'\s+kev:\s*true\s*\n', '\n', content)
|
|
||||||
updated = True
|
|
||||||
print(f"Removed kev metadata from {file_path.name}")
|
|
||||||
|
|
||||||
# Note: We don't remove vkev metadata automatically as VulnCheck data might be incomplete
|
|
||||||
|
|
||||||
if updated:
|
|
||||||
with open(file_path, 'w', encoding='utf-8') as f:
|
|
||||||
f.write(content)
|
|
||||||
return True
|
|
||||||
|
|
||||||
return False
|
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Error cleaning KEV tags from {file_path}: {e}")
|
|
||||||
return False
|
|
||||||
|
|
||||||
def run(self):
|
def run(self):
|
||||||
"""Main execution function."""
|
"""Main execution function."""
|
||||||
print("Starting KEV tagging update...")
|
print("Starting KEV tagging update...")
|
||||||
|
|
||||||
# Fetch KEV data
|
# Fetch KEV data
|
||||||
self.cisa_kev_cves = self.fetch_cisa_kev_data()
|
self.cisa_kev_cves = self.fetch_cisa_kev_data()
|
||||||
self.vulncheck_kev_cves = self.fetch_vulncheck_kev_data()
|
|
||||||
|
|
||||||
if not self.cisa_kev_cves:
|
if not self.cisa_kev_cves:
|
||||||
print("No CISA KEV data available, exiting")
|
print("No CISA KEV data available, exiting")
|
||||||
@@ -298,30 +217,31 @@ class KEVUpdater:
|
|||||||
return
|
return
|
||||||
|
|
||||||
# Process templates
|
# Process templates
|
||||||
|
kev_found = 0
|
||||||
|
non_kev_with_tag = 0
|
||||||
|
|
||||||
for file_path in template_files:
|
for file_path in template_files:
|
||||||
cve_id = self.extract_cve_id(file_path)
|
cve_id = self.extract_cve_id(file_path)
|
||||||
if cve_id:
|
if cve_id:
|
||||||
is_cisa_kev = cve_id in self.cisa_kev_cves
|
is_kev = cve_id in self.cisa_kev_cves
|
||||||
is_vulncheck_kev = cve_id in self.vulncheck_kev_cves
|
|
||||||
|
|
||||||
if is_cisa_kev or is_vulncheck_kev:
|
if is_kev:
|
||||||
# Add KEV tags/metadata
|
kev_found += 1
|
||||||
if self.update_template_with_kev(file_path, cve_id, is_cisa_kev, is_vulncheck_kev):
|
|
||||||
self.updated_count += 1
|
if self.update_template_with_kev(file_path, cve_id, is_kev):
|
||||||
elif self.has_kev_tag(open(file_path, 'r').read()) or self.has_kev_metadata(open(file_path, 'r').read()):
|
self.updated_count += 1
|
||||||
# Template has KEV tags but CVE is not in current KEV list
|
|
||||||
# Only remove if we're confident about our data
|
# Track non-KEV CVEs that have kev tags for reporting
|
||||||
if len(self.cisa_kev_cves) > 1000: # Sanity check
|
if not is_kev and self.has_kev_tag(open(file_path, 'r').read()):
|
||||||
print(f"Warning: {cve_id} has KEV tags but not in current KEV catalog")
|
non_kev_with_tag += 1
|
||||||
# Uncomment the line below to automatically remove outdated KEV tags
|
|
||||||
# self.remove_kev_tags(file_path, cve_id)
|
|
||||||
else:
|
else:
|
||||||
print(f"Could not extract CVE ID from {file_path}")
|
print(f"Could not extract CVE ID from {file_path}")
|
||||||
|
|
||||||
print(f"\nKEV update complete!")
|
print(f"\nKEV update complete!")
|
||||||
print(f"Templates updated: {self.updated_count}")
|
print(f"Templates updated: {self.updated_count}")
|
||||||
print(f"CISA KEV CVEs processed: {len(self.cisa_kev_cves)}")
|
print(f"KEV CVEs found in templates: {kev_found}")
|
||||||
print(f"VulnCheck KEV CVEs processed: {len(self.vulncheck_kev_cves)}")
|
print(f"CISA KEV catalog size: {len(self.cisa_kev_cves)}")
|
||||||
|
print(f"Non-KEV templates with kev tags: {non_kev_with_tag}")
|
||||||
print(f"Errors: {self.error_count}")
|
print(f"Errors: {self.error_count}")
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
|
|||||||
Reference in New Issue
Block a user