Spaces:
Sleeping
Sleeping
File size: 4,684 Bytes
fc40cb4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | import requests
import pandas as pd
import time
import os
import json
from datetime import datetime, timedelta, timezone
API_KEY = os.environ["NVD_API_KEY"]
HEADERS = {"apiKey": API_KEY}
CSV_PATH = "data/cves_raw.csv"
TRACKER = "data/last_updated.json"
def score_to_label(score):
if score >= 9.0: return "Critical"
elif score >= 7.0: return "High"
elif score >= 4.0: return "Medium"
else: return "Low"
def fetch_chunk(start, end):
url = "https://services.nvd.nist.gov/rest/json/cves/2.0"
all_items = []
idx = 0
while True:
full_url = f"{url}?pubStartDate={start}&pubEndDate={end}&startIndex={idx}&resultsPerPage=2000"
try:
r = requests.get(full_url, headers=HEADERS, timeout=60)
r.raise_for_status()
data = r.json()
total = data.get("totalResults", 0)
items = data.get("vulnerabilities", [])
all_items.extend(items)
if len(all_items) >= total:
break
idx += 2000
time.sleep(0.7)
except Exception as e:
print(f" Error: {e}")
break
return all_items
def parse_items(items, existing_ids):
rows = []
for item in items:
try:
cve = item["cve"]
desc = ""
for d in cve.get("descriptions", []):
if d["lang"] == "en":
desc = d["value"]
break
if not desc or "** REJECT **" in desc or len(desc.split()) < 10:
continue
if cve["id"] in existing_ids:
continue
metrics = cve.get("metrics", {})
cvss_data = None
if "cvssMetricV31" in metrics:
cvss_data = metrics["cvssMetricV31"][0]["cvssData"]
elif "cvssMetricV30" in metrics:
cvss_data = metrics["cvssMetricV30"][0]["cvssData"]
else:
continue
score = cvss_data["baseScore"]
rows.append({
"cve_id": cve["id"],
"description": desc,
"cvss_score": score,
"cvss_label": score_to_label(score),
"attack_vector": cvss_data.get("attackVector", ""),
"attack_complexity": cvss_data.get("attackComplexity", ""),
"privileges_required": cvss_data.get("privilegesRequired", ""),
"user_interaction": cvss_data.get("userInteraction", ""),
"scope": cvss_data.get("scope", "")
})
except:
continue
return rows
# ── MAIN ──────────────────────────────────────────────────────────────
df = pd.read_csv(CSV_PATH)
existing_ids = set(df["cve_id"].tolist())
today = datetime.now(timezone.utc).replace(tzinfo=None)
# Read last collected date from tracker
if os.path.exists(TRACKER):
with open(TRACKER) as f:
data = json.load(f)
last = datetime.strptime(data["last_collected"], "%Y-%m-%d")
else:
last = datetime(2023, 12, 31)
gap = (today - last).days
print(f"Last collected: {last.strftime('%Y-%m-%d')}")
print(f"Today: {today.strftime('%Y-%m-%d')}")
print(f"Gap: {gap} days")
print(f"Existing CVEs: {len(df)}")
if gap < 1:
print("Already up to date.")
else:
chunks = []
current = last + timedelta(days=1)
while current < today:
chunk_end = min(current + timedelta(days=99), today)
chunks.append((
current.strftime("%Y-%m-%dT00:00:00.000"),
chunk_end.strftime("%Y-%m-%dT23:59:59.999")
))
current = chunk_end + timedelta(days=1)
print(f"Fetching {len(chunks)} chunks...")
all_new = []
for i, (start, end) in enumerate(chunks):
print(f"Chunk {i+1}/{len(chunks)}: {start[:10]} → {end[:10]}")
items = fetch_chunk(start, end)
new_rows = parse_items(items, existing_ids)
all_new.extend(new_rows)
existing_ids.update([r["cve_id"] for r in new_rows])
print(f" Added {len(new_rows)} | Total new: {len(all_new)}")
time.sleep(2)
if all_new:
combined = pd.concat([df, pd.DataFrame(all_new)], ignore_index=True)
combined.to_csv(CSV_PATH, index=False)
print(f"Saved. Total CVEs: {len(combined)}")
with open(TRACKER, "w") as f:
json.dump({"last_collected": today.strftime("%Y-%m-%d")}, f)
print("Tracker updated.") |