sync
This commit is contained in:
@@ -0,0 +1,149 @@
|
||||
import nvdlib
|
||||
from datetime import datetime, timedelta
|
||||
import pandas as pd
|
||||
|
||||
def get_latest_cves(days_ago=3, api_key=None):
|
||||
"""
|
||||
Fetches the latest CVEs registered by NIST over the past N days.
|
||||
|
||||
Args:
|
||||
days_ago (int): The number of days back from today to search for CVEs.
|
||||
api_key (str, optional): Your NVD API key. Defaults to None.
|
||||
|
||||
Returns:
|
||||
list: A list of dictionaries, each representing a CVE with relevant details.
|
||||
"""
|
||||
end_date = datetime.now(tz=None)
|
||||
start_date = end_date - timedelta(days=days_ago)
|
||||
|
||||
# Format dates to ISO-8601 for the NVD API
|
||||
pub_start_date = start_date.isoformat(timespec='seconds') + 'Z'
|
||||
pub_end_date = end_date.isoformat(timespec='seconds') + 'Z'
|
||||
|
||||
print(f"Searching for CVEs published between {pub_start_date} and {pub_end_date}...")
|
||||
|
||||
cve_data = []
|
||||
try:
|
||||
# The NVD API has a limit on the number of results per page and date range.
|
||||
# We'll use pagination to retrieve all results within the date range.
|
||||
# The 'lastModStartDate' and 'lastModEndDate' parameters are for when a CVE was last modified.
|
||||
# We are interested in 'pubStartDate' and 'pubEndDate' for newly registered CVEs.
|
||||
results = nvdlib.searchCVE(
|
||||
pubStartDate=pub_start_date,
|
||||
pubEndDate=pub_end_date,
|
||||
key=api_key,
|
||||
delay=6 if not api_key else 0, # Add delay if no API key is provided
|
||||
results_per_page=2000 # Max results per page
|
||||
)
|
||||
|
||||
for cve in results:
|
||||
cve_id = cve.id
|
||||
published_date = cve.published
|
||||
last_modified_date = cve.lastModified
|
||||
|
||||
# Description
|
||||
description = "No description available."
|
||||
if cve.descriptions:
|
||||
for desc in cve.descriptions:
|
||||
if desc.lang == 'en':
|
||||
description = desc.value
|
||||
break
|
||||
|
||||
# Source (often derived from the first reference or CNA)
|
||||
source = "N/A"
|
||||
if cve.cve:
|
||||
if cve.cve.sourceIdentifier:
|
||||
source = cve.cve.sourceIdentifier
|
||||
elif cve.cve.CVE_data_meta and cve.cve.CVE_data_meta.ASSIGNER:
|
||||
source = cve.cve.CVE_data_meta.ASSIGNER
|
||||
|
||||
# Vulnerability Status (not directly available as a single field, infer from analysis status)
|
||||
# The NVD API v2.0 doesn't expose a direct "Vulnerability Status" field
|
||||
# like "Undergoing Analysis" or "Analyzed". We can infer based on the presence
|
||||
# of CVSS scores or other enrichment. For simplicity, we'll assume "Analyzed" if CVSS exists.
|
||||
vulnerability_status = "Unknown"
|
||||
if cve.vulnStatus:
|
||||
vulnerability_status = cve.vulnStatus
|
||||
elif cve.metrics and (cve.metrics.cvssMetricV31 or cve.metrics.cvssMetricV2):
|
||||
vulnerability_status = "Analyzed (CVSS available)"
|
||||
else:
|
||||
vulnerability_status = "Awaiting Analysis"
|
||||
|
||||
|
||||
# CVSS Score
|
||||
cvss_score = "N/A"
|
||||
if cve.metrics and cve.metrics.cvssMetricV31:
|
||||
cvss_score = cve.metrics.cvssMetricV31[0].cvssData.baseScore
|
||||
elif cve.metrics and cve.metrics.cvssMetricV2:
|
||||
cvss_score = cve.metrics.cvssMetricV2[0].cvssData.baseScore
|
||||
|
||||
# Products (CPEs)
|
||||
products = []
|
||||
if cve.configurations:
|
||||
for config in cve.configurations:
|
||||
if config.nodes:
|
||||
for node in config.nodes:
|
||||
if node.cpeMatch:
|
||||
for cpe_match in node.cpeMatch:
|
||||
products.append(cpe_match.criteria)
|
||||
|
||||
cve_data.append({
|
||||
"CVE-ID": cve_id,
|
||||
"Source": source,
|
||||
"Published Date": published_date,
|
||||
"Last Modified Date": last_modified_date,
|
||||
"Vulnerability Status": vulnerability_status,
|
||||
"Description": description,
|
||||
"Product": "\n".join(products) if products else "N/A",
|
||||
"CVSS Score": cvss_score
|
||||
})
|
||||
except nvdlib.NVDAPIError as e:
|
||||
print(f"Error fetching CVEs: {e}")
|
||||
print("Please check your API key and try again, or consider adding a delay if not using an API key.")
|
||||
except Exception as e:
|
||||
print(f"An unexpected error occurred: {e}")
|
||||
|
||||
return cve_data
|
||||
|
||||
def generate_report(cve_list):
|
||||
"""
|
||||
Generates a formatted report from the list of CVE data.
|
||||
|
||||
Args:
|
||||
cve_list (list): A list of dictionaries, each representing a CVE.
|
||||
"""
|
||||
if not cve_list:
|
||||
print("No new CVEs found for the specified period.")
|
||||
return
|
||||
|
||||
df = pd.DataFrame(cve_list)
|
||||
|
||||
# Reorder columns for the report
|
||||
report_columns = [
|
||||
"CVE-ID",
|
||||
"Source",
|
||||
"Published Date",
|
||||
"Last Modified Date",
|
||||
"Vulnerability Status",
|
||||
"Description",
|
||||
"Product",
|
||||
"CVSS Score"
|
||||
]
|
||||
df = df[report_columns]
|
||||
|
||||
print("\n--- Latest NIST CVE Report ---")
|
||||
print(df.to_string(index=False)) # Use to_string for full display without truncation
|
||||
|
||||
# You can also save this to a CSV or other formats
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
output_filename = f"nist_cve_report_{timestamp}.csv"
|
||||
df.to_csv(output_filename, index=False)
|
||||
print(f"\nReport saved to {output_filename}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Replace 'YOUR_NVD_API_KEY' with your actual API key if you have one
|
||||
# If you don't have one, leave it as None, but be aware of rate limits.
|
||||
NVD_API_KEY = None
|
||||
#2586b8ea-afbb-4309-9853-311f161f5568
|
||||
cves = get_latest_cves(days_ago=3, api_key=NVD_API_KEY)
|
||||
generate_report(cves)
|
||||
Reference in New Issue
Block a user