| 1 | import datetime as dt |
| 2 | import json |
| 3 | import re |
| 4 | from typing import Any |
| 5 | from typing import Dict |
| 6 | from typing import List |
| 7 | from typing import Optional |
| 8 | from typing import Tuple |
| 9 | |
| 10 | import httpx |
| 11 | from loguru import logger |
| 12 | |
| 13 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 14 | AffectedProduct, |
| 15 | ) |
| 16 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 17 | AvailableCyclesResponse, |
| 18 | ) |
| 19 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 20 | CVSSInfo, |
| 21 | ) |
| 22 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 23 | EPSSInfo, |
| 24 | ) |
| 25 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 26 | KEVInfo, |
| 27 | ) |
| 28 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 29 | PatchTuesdayItem, |
| 30 | ) |
| 31 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 32 | PatchTuesdayRequest, |
| 33 | ) |
| 34 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 35 | PatchTuesdayResponse, |
| 36 | ) |
| 37 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 38 | PatchTuesdaySummary, |
| 39 | ) |
| 40 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 41 | PatchTuesdaySummaryResponse, |
| 42 | ) |
| 43 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 44 | PrioritizationInfo, |
| 45 | ) |
| 46 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 47 | PriorityCounts, |
| 48 | ) |
| 49 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 50 | RemediationInfo, |
| 51 | ) |
| 52 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 53 | SourceInfo, |
| 54 | ) |
| 55 | |
| 56 | # Constants |
| 57 | MSRC_CVRF_BASE = "https://api.msrc.microsoft.com/cvrf/v3.0/cvrf" |
| 58 | EPSS_BASE = "https://api.first.org/data/v1/epss" |
| 59 | CISA_KEV_JSON = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json" |
| 60 | |
| 61 | MONTH_ABBR = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] |
| 62 | |
| 63 | |
| 64 | def second_tuesday(year: int, month: int) -> dt.date: |
| 65 | """Return date of the second Tuesday for a given year/month.""" |
| 66 | d = dt.date(year, month, 1) |
| 67 | days_to_tuesday = (1 - d.weekday()) % 7 |
| 68 | first_tuesday = d + dt.timedelta(days=days_to_tuesday) |
| 69 | return first_tuesday + dt.timedelta(days=7) |
| 70 | |
| 71 | |
| 72 | def get_default_cycle(today: Optional[dt.date] = None) -> str: |
| 73 | """Default cycle is current month in 'YYYY-Mmm' format.""" |
| 74 | if today is None: |
| 75 | today = dt.date.today() |
| 76 | return f"{today.year}-{MONTH_ABBR[today.month - 1]}" |
| 77 | |
| 78 | |
| 79 | def parse_cycle(cycle: str) -> Tuple[int, int, str]: |
| 80 | """ |
| 81 | Parse cycle in 'YYYY-Mmm' format. |
| 82 | Returns (year, month_int, normalized_cycle). |
| 83 | """ |
| 84 | m = re.fullmatch(r"(\d{4})-([A-Za-z]{3})", cycle.strip()) |
| 85 | if not m: |
| 86 | raise ValueError("Cycle must be in format YYYY-Mmm (e.g., 2026-Jan)") |
| 87 | year = int(m.group(1)) |
| 88 | mon = m.group(2).title() |
| 89 | if mon not in MONTH_ABBR: |
| 90 | raise ValueError(f"Invalid month abbreviation '{mon}'. Use one of: {', '.join(MONTH_ABBR)}") |
| 91 | month = MONTH_ABBR.index(mon) + 1 |
| 92 | return year, month, f"{year}-{mon}" |
| 93 | |
| 94 | |
| 95 | def safe_float(x: Any) -> Optional[float]: |
| 96 | """Safely convert a value to float.""" |
| 97 | try: |
| 98 | if x is None or x == "": |
| 99 | return None |
| 100 | return float(x) |
| 101 | except Exception: |
| 102 | return None |
| 103 | |
| 104 | |
| 105 | def normalize_text(s: Any) -> Optional[str]: |
| 106 | """Best-effort normalization to a single-line string.""" |
| 107 | if s is None: |
| 108 | return None |
| 109 | |
| 110 | if isinstance(s, dict): |
| 111 | for k in ("Value", "value", "Description", "description", "Text", "text", "Title", "title", "Name", "name"): |
| 112 | if k in s and isinstance(s[k], (str, int, float)): |
| 113 | s = s[k] |
| 114 | break |
| 115 | else: |
| 116 | try: |
| 117 | s = json.dumps(s, ensure_ascii=False) |
| 118 | except Exception: |
| 119 | s = str(s) |
| 120 | |
| 121 | if isinstance(s, list): |
| 122 | parts = [] |
| 123 | for x in s: |
| 124 | nx = normalize_text(x) |
| 125 | if nx: |
| 126 | parts.append(nx) |
| 127 | s = " | ".join(parts) |
| 128 | |
| 129 | if not isinstance(s, str): |
| 130 | s = str(s) |
| 131 | |
| 132 | s = re.sub(r"\s+", " ", s).strip() |
| 133 | return s or None |
| 134 | |
| 135 | |
| 136 | def product_family(name: Any) -> str: |
| 137 | """Classify product into a coarse family bucket.""" |
| 138 | name_s = normalize_text(name) or "" |
| 139 | n = name_s.lower() |
| 140 | |
| 141 | if "windows server" in n: |
| 142 | return "Windows Server" |
| 143 | if "windows" in n: |
| 144 | return "Windows" |
| 145 | if "office" in n or "microsoft 365" in n: |
| 146 | return "Office/M365" |
| 147 | if "exchange" in n: |
| 148 | return "Exchange" |
| 149 | if "sharepoint" in n: |
| 150 | return "SharePoint" |
| 151 | if "sql server" in n: |
| 152 | return "SQL Server" |
| 153 | if "visual studio" in n or ".net" in n: |
| 154 | return "Developer Platform" |
| 155 | if "edge" in n: |
| 156 | return "Edge" |
| 157 | if "azure" in n: |
| 158 | return "Azure" |
| 159 | if "dynamics" in n: |
| 160 | return "Dynamics" |
| 161 | return "Other" |
| 162 | |
| 163 | |
| 164 | def chunk_by_max_chars(items: List[str], max_chars: int = 1900) -> List[List[str]]: |
| 165 | """Chunk items to keep URL parameter size under limits.""" |
| 166 | chunks: List[List[str]] = [] |
| 167 | current: List[str] = [] |
| 168 | current_len = 0 |
| 169 | for it in items: |
| 170 | add_len = len(it) + (1 if current else 0) |
| 171 | if current and current_len + add_len > max_chars: |
| 172 | chunks.append(current) |
| 173 | current = [it] |
| 174 | current_len = len(it) |
| 175 | else: |
| 176 | current.append(it) |
| 177 | current_len += add_len |
| 178 | if current: |
| 179 | chunks.append(current) |
| 180 | return chunks |
| 181 | |
| 182 | |
| 183 | async def fetch_json(url: str, params: Optional[Dict[str, str]] = None, timeout: int = 60) -> Dict[str, Any]: |
| 184 | """Fetch JSON from a URL with proper error handling.""" |
| 185 | headers = { |
| 186 | "Accept": "application/json", |
| 187 | "User-Agent": "CoPilot-PatchTuesday/1.0", |
| 188 | } |
| 189 | |
| 190 | async with httpx.AsyncClient(timeout=timeout) as client: |
| 191 | response = await client.get(url, headers=headers, params=params) |
| 192 | |
| 193 | if response.status_code != 200: |
| 194 | raise RuntimeError(f"GET {url} failed: HTTP {response.status_code} - {response.text[:300]}") |
| 195 | |
| 196 | ctype = (response.headers.get("content-type") or "").lower() |
| 197 | body = response.text.lstrip() |
| 198 | if "application/json" not in ctype and body.startswith("<"): |
| 199 | raise RuntimeError( |
| 200 | f"Endpoint returned non-JSON (Content-Type: {ctype}). " "The MSRC API may have returned XML instead of JSON.", |
| 201 | ) |
| 202 | |
| 203 | try: |
| 204 | return response.json() |
| 205 | except Exception as e: |
| 206 | raise RuntimeError(f"Response was not valid JSON: {e}. First 300 chars: {response.text[:300]}") from e |
| 207 | |
| 208 | |
| 209 | async def fetch_epss_scores(cves: List[str]) -> Dict[str, Dict[str, Any]]: |
| 210 | """Fetch EPSS scores for a list of CVEs.""" |
| 211 | out: Dict[str, Dict[str, Any]] = {} |
| 212 | cves = sorted(set([c.strip().upper() for c in cves if c and str(c).strip()])) |
| 213 | |
| 214 | if not cves: |
| 215 | return out |
| 216 | |
| 217 | logger.info(f"Fetching EPSS scores for {len(cves)} CVEs") |
| 218 | |
| 219 | for chunk in chunk_by_max_chars(cves): |
| 220 | try: |
| 221 | params = {"cve": ",".join(chunk)} |
| 222 | data = await fetch_json(EPSS_BASE, params=params) |
| 223 | |
| 224 | for row in data.get("data", []) or []: |
| 225 | cve = (row.get("cve") or "").upper() |
| 226 | if not cve: |
| 227 | continue |
| 228 | out[cve] = { |
| 229 | "epss": safe_float(row.get("epss")), |
| 230 | "percentile": safe_float(row.get("percentile")), |
| 231 | "date": row.get("date"), |
| 232 | } |
| 233 | except Exception as e: |
| 234 | logger.warning(f"Error fetching EPSS chunk: {e}") |
| 235 | continue |
| 236 | |
| 237 | logger.info(f"Retrieved EPSS scores for {len(out)} CVEs") |
| 238 | return out |
| 239 | |
| 240 | |
| 241 | async def fetch_kev_data() -> Dict[str, Dict[str, Any]]: |
| 242 | """Fetch and parse CISA KEV data.""" |
| 243 | logger.info("Fetching CISA KEV data") |
| 244 | |
| 245 | try: |
| 246 | kev_doc = await fetch_json(CISA_KEV_JSON) |
| 247 | except Exception as e: |
| 248 | logger.error(f"Failed to fetch KEV data: {e}") |
| 249 | return {} |
| 250 | |
| 251 | out: Dict[str, Dict[str, Any]] = {} |
| 252 | vulns = kev_doc.get("vulnerabilities") or [] |
| 253 | if isinstance(vulns, dict): |
| 254 | vulns = [vulns] |
| 255 | |
| 256 | for v in vulns: |
| 257 | if not isinstance(v, dict): |
| 258 | continue |
| 259 | cve = (normalize_text(v.get("cveID") or v.get("cveId") or v.get("CVE")) or "").upper() |
| 260 | if not cve.startswith("CVE-"): |
| 261 | continue |
| 262 | |
| 263 | out[cve] = { |
| 264 | "in_kev": True, |
| 265 | "date_added": normalize_text(v.get("dateAdded")), |
| 266 | "due_date": normalize_text(v.get("dueDate")), |
| 267 | "required_action": normalize_text(v.get("requiredAction")), |
| 268 | "known_ransomware_campaign_use": normalize_text(v.get("knownRansomwareCampaignUse")), |
| 269 | "vendor_project": normalize_text(v.get("vendorProject")), |
| 270 | "product": normalize_text(v.get("product")), |
| 271 | "vulnerability_name": normalize_text(v.get("vulnerabilityName")), |
| 272 | "short_description": normalize_text(v.get("shortDescription")), |
| 273 | "notes": normalize_text(v.get("notes")), |
| 274 | } |
| 275 | |
| 276 | logger.info(f"Loaded {len(out)} CVEs from CISA KEV") |
| 277 | return out |
| 278 | |
| 279 | |
| 280 | def extract_cvss(vuln: Dict[str, Any]) -> Tuple[Optional[float], Optional[str]]: |
| 281 | """Extract CVSS score and vector from vulnerability data.""" |
| 282 | sets = vuln.get("CVSSScoreSets") or vuln.get("CvssScoreSets") or [] |
| 283 | if isinstance(sets, dict): |
| 284 | sets = [sets] |
| 285 | |
| 286 | best_score = None |
| 287 | best_vector = None |
| 288 | |
| 289 | for s in sets: |
| 290 | if not isinstance(s, dict): |
| 291 | continue |
| 292 | score = safe_float(s.get("BaseScore") or s.get("baseScore") or s.get("Score")) |
| 293 | vector = normalize_text(s.get("Vector") or s.get("vectorString") or s.get("VectorString")) |
| 294 | if score is None: |
| 295 | continue |
| 296 | if best_score is None or score > best_score: |
| 297 | best_score = score |
| 298 | best_vector = vector |
| 299 | |
| 300 | return best_score, best_vector |
| 301 | |
| 302 | |
| 303 | def extract_severity(vuln: Dict[str, Any]) -> Optional[str]: |
| 304 | """Extract severity from vulnerability data.""" |
| 305 | threats = vuln.get("Threats") or [] |
| 306 | if isinstance(threats, dict): |
| 307 | threats = [threats] |
| 308 | for t in threats: |
| 309 | if not isinstance(t, dict): |
| 310 | continue |
| 311 | ttype = normalize_text(t.get("Type") or t.get("type") or "") |
| 312 | desc = normalize_text(t.get("Description") or t.get("description")) |
| 313 | if ttype and "severity" in ttype.lower() and desc: |
| 314 | return desc.title() |
| 315 | |
| 316 | sev = normalize_text(vuln.get("Severity") or vuln.get("severity")) |
| 317 | return sev.title() if sev else None |
| 318 | |
| 319 | |
| 320 | def extract_kbs(vuln: Dict[str, Any]) -> List[str]: |
| 321 | """Extract KB article numbers from remediations.""" |
| 322 | kbs = set() |
| 323 | rems = vuln.get("Remediations") or [] |
| 324 | if isinstance(rems, dict): |
| 325 | rems = [rems] |
| 326 | for r in rems: |
| 327 | if not isinstance(r, dict): |
| 328 | continue |
| 329 | for field in ["Description", "URL", "Url", "description", "url"]: |
| 330 | val = r.get(field) |
| 331 | if not val: |
| 332 | continue |
| 333 | for kb in re.findall(r"\bKB\d{6,8}\b", str(val), flags=re.IGNORECASE): |
| 334 | kbs.add(kb.upper()) |
| 335 | return sorted(kbs) |
| 336 | |
| 337 | |
| 338 | def build_product_lookup(doc: Dict[str, Any]) -> Dict[str, str]: |
| 339 | """Build ProductID to product name lookup.""" |
| 340 | lookup: Dict[str, str] = {} |
| 341 | |
| 342 | pt = doc.get("ProductTree") or {} |
| 343 | fp = pt.get("FullProductName") or [] |
| 344 | if isinstance(fp, dict): |
| 345 | fp = [fp] |
| 346 | |
| 347 | for item in fp: |
| 348 | if not isinstance(item, dict): |
| 349 | continue |
| 350 | pid = item.get("ProductID") or item.get("productId") or item.get("ProductId") |
| 351 | val = normalize_text(item.get("Value") or item.get("value") or item.get("Name") or item.get("name")) |
| 352 | if pid and val: |
| 353 | lookup[str(pid)] = val |
| 354 | |
| 355 | return lookup |
| 356 | |
| 357 | |
| 358 | def extract_affected_products(vuln: Dict[str, Any], product_lookup: Dict[str, str]) -> List[str]: |
| 359 | """Extract affected products from vulnerability data.""" |
| 360 | names = set() |
| 361 | |
| 362 | statuses = vuln.get("ProductStatuses") or vuln.get("ProductStatus") or [] |
| 363 | if isinstance(statuses, dict): |
| 364 | statuses = [statuses] |
| 365 | |
| 366 | for st in statuses: |
| 367 | if not isinstance(st, dict): |
| 368 | continue |
| 369 | for _, v in st.items(): |
| 370 | if isinstance(v, list): |
| 371 | for pid in v: |
| 372 | pname = product_lookup.get(str(pid)) |
| 373 | if pname: |
| 374 | names.add(pname) |
| 375 | elif isinstance(v, (str, int)): |
| 376 | pname = product_lookup.get(str(v)) |
| 377 | if pname: |
| 378 | names.add(pname) |
| 379 | |
| 380 | prods = vuln.get("Products") |
| 381 | if isinstance(prods, list): |
| 382 | for pid in prods: |
| 383 | pname = product_lookup.get(str(pid)) |
| 384 | if pname: |
| 385 | names.add(pname) |
| 386 | |
| 387 | return sorted(names) |
| 388 | |
| 389 | |
| 390 | def calculate_priority( |
| 391 | severity: Optional[str], |
| 392 | cvss: Optional[float], |
| 393 | epss: Optional[float], |
| 394 | family: str, |
| 395 | in_kev: bool, |
| 396 | ) -> Tuple[str, List[str], str]: |
| 397 | """ |
| 398 | Calculate priority level (P0-P3) based on various factors. |
| 399 | |
| 400 | P0: CISA KEV, OR EPSS >= 0.9, OR (Critical and EPSS >= 0.7) |
| 401 | P1: Critical OR (CVSS >= 8.0 and (EPSS >= 0.3 or core enterprise families)) |
| 402 | P2: Important/Moderate OR CVSS >= 6.0 OR EPSS >= 0.1 |
| 403 | P3: Otherwise |
| 404 | """ |
| 405 | if in_kev: |
| 406 | reasons = ["CISA KEV (known exploited)"] |
| 407 | if epss is not None: |
| 408 | reasons.append(f"EPSS={epss:.3f}") |
| 409 | if cvss is not None: |
| 410 | reasons.append(f"CVSS={cvss:.1f}") |
| 411 | sla = "Emergency: known exploited. Patch/mitigate immediately; prioritize exposed and Tier-0 assets first." |
| 412 | return "P0", reasons, sla |
| 413 | |
| 414 | reasons: List[str] = [] |
| 415 | sev = (severity or "").lower() |
| 416 | e = epss if epss is not None else 0.0 |
| 417 | s = cvss if cvss is not None else 0.0 |
| 418 | |
| 419 | core_families = {"Windows", "Windows Server", "Office/M365", "Exchange", "Edge"} |
| 420 | |
| 421 | if e >= 0.9 or (sev == "critical" and e >= 0.7): |
| 422 | reasons += ["Very high EPSS" if e >= 0.9 else "Critical severity + high EPSS"] |
| 423 | sla = "Emergency: validate mitigations immediately; deploy to highest-risk assets within 24h." |
| 424 | return "P0", reasons, sla |
| 425 | |
| 426 | if sev == "critical": |
| 427 | reasons.append("Critical severity") |
| 428 | if s >= 8.0: |
| 429 | reasons.append("High CVSS (>= 8.0)") |
| 430 | if e >= 0.3: |
| 431 | reasons.append("Elevated EPSS (>= 0.3)") |
| 432 | if family in core_families: |
| 433 | reasons.append(f"High enterprise footprint ({family})") |
| 434 | |
| 435 | if (sev == "critical") or (s >= 8.0 and (e >= 0.3 or family in core_families)): |
| 436 | sla = "High: deploy to pilot in 24–48h; broad rollout in 72h." |
| 437 | return "P1", reasons, sla |
| 438 | |
| 439 | if sev in {"important", "moderate"}: |
| 440 | reasons.append(f"{severity.title() if severity else 'Unknown'} severity") |
| 441 | if s >= 6.0: |
| 442 | reasons.append("CVSS (>= 6.0)") |
| 443 | if e >= 0.1: |
| 444 | reasons.append("EPSS (>= 0.1)") |
| 445 | |
| 446 | if sev in {"important", "moderate"} or s >= 6.0 or e >= 0.1: |
| 447 | sla = "Medium: deploy in the next standard patch window (7–14 days), sooner for exposed assets." |
| 448 | return "P2", reasons, sla |
| 449 | |
| 450 | sla = "Low: patch in routine maintenance (30 days) unless asset criticality dictates otherwise." |
| 451 | return "P3", reasons, sla |
| 452 | |
| 453 | |
| 454 | async def fetch_patch_tuesday_data( |
| 455 | cycle: Optional[str] = None, |
| 456 | include_epss: bool = True, |
| 457 | include_kev: bool = True, |
| 458 | ) -> Tuple[List[PatchTuesdayItem], PatchTuesdaySummary]: |
| 459 | """ |
| 460 | Fetch and parse Patch Tuesday data for a given cycle. |
| 461 | |
| 462 | Args: |
| 463 | cycle: Cycle in YYYY-Mmm format (e.g., 2026-Jan). Defaults to current month. |
| 464 | include_epss: Whether to fetch and include EPSS scores. |
| 465 | include_kev: Whether to fetch and include CISA KEV data. |
| 466 | |
| 467 | Returns: |
| 468 | Tuple of (items, summary) |
| 469 | """ |
| 470 | # Parse cycle |
| 471 | if cycle: |
| 472 | year, month, cycle_norm = parse_cycle(cycle) |
| 473 | else: |
| 474 | cycle_norm = get_default_cycle() |
| 475 | year, month, cycle_norm = parse_cycle(cycle_norm) |
| 476 | |
| 477 | pt_date = second_tuesday(year, month) |
| 478 | url = f"{MSRC_CVRF_BASE}/{cycle_norm}" |
| 479 | |
| 480 | logger.info(f"Fetching Patch Tuesday data for cycle {cycle_norm} (date: {pt_date.isoformat()})") |
| 481 | |
| 482 | # Fetch CVRF document |
| 483 | doc = await fetch_json(url) |
| 484 | |
| 485 | # Build product lookup |
| 486 | product_lookup = build_product_lookup(doc) |
| 487 | |
| 488 | # Parse vulnerabilities |
| 489 | vulnerabilities = doc.get("Vulnerability") or doc.get("Vulnerabilities") or [] |
| 490 | if isinstance(vulnerabilities, dict): |
| 491 | vulnerabilities = [vulnerabilities] |
| 492 | |
| 493 | # Collect all CVEs for enrichment |
| 494 | cves_all: List[str] = [] |
| 495 | for v in vulnerabilities: |
| 496 | if not isinstance(v, dict): |
| 497 | continue |
| 498 | cve = (normalize_text(v.get("CVE") or v.get("Cve") or v.get("cve")) or "").upper() |
| 499 | if cve.startswith("CVE-"): |
| 500 | cves_all.append(cve) |
| 501 | |
| 502 | # Fetch enrichment data |
| 503 | epss_map: Dict[str, Dict[str, Any]] = {} |
| 504 | kev_map: Dict[str, Dict[str, Any]] = {} |
| 505 | |
| 506 | if include_epss: |
| 507 | epss_map = await fetch_epss_scores(cves_all) |
| 508 | |
| 509 | if include_kev: |
| 510 | kev_map = await fetch_kev_data() |
| 511 | |
| 512 | now_utc = dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") |
| 513 | |
| 514 | items: List[PatchTuesdayItem] = [] |
| 515 | prio_counts = {"P0": 0, "P1": 0, "P2": 0, "P3": 0} |
| 516 | family_counts: Dict[str, int] = {} |
| 517 | severity_counts: Dict[str, int] = {} |
| 518 | |
| 519 | for v in vulnerabilities: |
| 520 | if not isinstance(v, dict): |
| 521 | continue |
| 522 | |
| 523 | cve = (normalize_text(v.get("CVE") or v.get("Cve") or v.get("cve")) or "").upper() |
| 524 | if not cve.startswith("CVE-"): |
| 525 | continue |
| 526 | |
| 527 | title = normalize_text(v.get("Title") or v.get("title")) |
| 528 | severity = extract_severity(v) |
| 529 | cvss_score, cvss_vector = extract_cvss(v) |
| 530 | kbs = extract_kbs(v) |
| 531 | |
| 532 | affected_products = extract_affected_products(v, product_lookup) |
| 533 | if not affected_products: |
| 534 | affected_products = ["(Unknown product mapping)"] |
| 535 | |
| 536 | # Get enrichment data |
| 537 | epss = epss_map.get(cve, {}) |
| 538 | epss_score = epss.get("epss") |
| 539 | epss_pct = epss.get("percentile") |
| 540 | epss_date = epss.get("date") |
| 541 | |
| 542 | kev = kev_map.get(cve, {"in_kev": False}) |
| 543 | in_kev = bool(kev.get("in_kev")) |
| 544 | |
| 545 | # Track severity counts |
| 546 | if severity: |
| 547 | severity_counts[severity] = severity_counts.get(severity, 0) + 1 |
| 548 | |
| 549 | for prod in affected_products: |
| 550 | prod_norm = normalize_text(prod) or prod |
| 551 | fam = product_family(prod_norm) |
| 552 | family_counts[fam] = family_counts.get(fam, 0) + 1 |
| 553 | |
| 554 | prio, reasons, sla = calculate_priority(severity, cvss_score, epss_score, fam, in_kev) |
| 555 | prio_counts[prio] = prio_counts.get(prio, 0) + 1 |
| 556 | |
| 557 | item = PatchTuesdayItem( |
| 558 | cycle=cycle_norm, |
| 559 | release_type="patch_tuesday", |
| 560 | cve=cve, |
| 561 | title=title, |
| 562 | severity=severity, |
| 563 | cvss=CVSSInfo(base=cvss_score, vector=cvss_vector), |
| 564 | epss=EPSSInfo(score=epss_score, percentile=epss_pct, date=epss_date), |
| 565 | kev=KEVInfo( |
| 566 | in_kev=in_kev, |
| 567 | date_added=kev.get("date_added"), |
| 568 | due_date=kev.get("due_date"), |
| 569 | required_action=kev.get("required_action"), |
| 570 | known_ransomware_campaign_use=kev.get("known_ransomware_campaign_use"), |
| 571 | vendor_project=kev.get("vendor_project"), |
| 572 | product=kev.get("product"), |
| 573 | vulnerability_name=kev.get("vulnerability_name"), |
| 574 | short_description=kev.get("short_description"), |
| 575 | notes=kev.get("notes"), |
| 576 | ), |
| 577 | affected=AffectedProduct( |
| 578 | product=prod_norm, |
| 579 | family=fam, |
| 580 | component_hint=title, |
| 581 | ), |
| 582 | remediation=RemediationInfo(kbs=kbs), |
| 583 | prioritization=PrioritizationInfo( |
| 584 | priority=prio, |
| 585 | reason=reasons, |
| 586 | suggested_sla=sla, |
| 587 | ), |
| 588 | source=SourceInfo( |
| 589 | msrc_cvrf_id=cycle_norm, |
| 590 | msrc_cvrf_url=url, |
| 591 | cisa_kev_url=CISA_KEV_JSON, |
| 592 | ), |
| 593 | timestamp_utc=now_utc, |
| 594 | ) |
| 595 | items.append(item) |
| 596 | |
| 597 | # Sort by priority, then EPSS desc, then CVSS desc |
| 598 | prio_rank = {"P0": 0, "P1": 1, "P2": 2, "P3": 3} |
| 599 | items.sort( |
| 600 | key=lambda r: ( |
| 601 | prio_rank.get(r.prioritization.priority, 9), |
| 602 | -(r.epss.score or 0.0), |
| 603 | -(r.cvss.base or 0.0), |
| 604 | r.cve, |
| 605 | r.affected.product, |
| 606 | ), |
| 607 | ) |
| 608 | |
| 609 | summary = PatchTuesdaySummary( |
| 610 | cycle=cycle_norm, |
| 611 | patch_tuesday_date=pt_date.isoformat(), |
| 612 | generated_utc=now_utc, |
| 613 | unique_cves=len(set([x.cve for x in items])), |
| 614 | total_records=len(items), |
| 615 | by_priority=PriorityCounts(**prio_counts), |
| 616 | by_family=dict(sorted(family_counts.items(), key=lambda kv: (-kv[1], kv[0]))), |
| 617 | by_severity=dict(sorted(severity_counts.items(), key=lambda kv: (-kv[1], kv[0]))), |
| 618 | ) |
| 619 | |
| 620 | logger.info(f"Processed {summary.unique_cves} unique CVEs, {summary.total_records} total records for cycle {cycle_norm}") |
| 621 | |
| 622 | return items, summary |
| 623 | |
| 624 | |
| 625 | async def get_patch_tuesday(request: PatchTuesdayRequest) -> PatchTuesdayResponse: |
| 626 | """ |
| 627 | Get full Patch Tuesday data for a cycle. |
| 628 | |
| 629 | Args: |
| 630 | request: Request containing cycle and enrichment options. |
| 631 | |
| 632 | Returns: |
| 633 | PatchTuesdayResponse with all items and summary. |
| 634 | """ |
| 635 | try: |
| 636 | items, summary = await fetch_patch_tuesday_data( |
| 637 | cycle=request.cycle, |
| 638 | include_epss=request.include_epss, |
| 639 | include_kev=request.include_kev, |
| 640 | ) |
| 641 | |
| 642 | return PatchTuesdayResponse( |
| 643 | success=True, |
| 644 | message=f"Successfully retrieved {summary.unique_cves} CVEs for cycle {summary.cycle}", |
| 645 | summary=summary, |
| 646 | items=items, |
| 647 | ) |
| 648 | except ValueError as e: |
| 649 | logger.error(f"Invalid request: {e}") |
| 650 | return PatchTuesdayResponse( |
| 651 | success=False, |
| 652 | message=str(e), |
| 653 | summary=None, |
| 654 | items=[], |
| 655 | ) |
| 656 | except Exception as e: |
| 657 | logger.error(f"Error fetching Patch Tuesday data: {e}") |
| 658 | return PatchTuesdayResponse( |
| 659 | success=False, |
| 660 | message=f"Failed to fetch Patch Tuesday data: {e}", |
| 661 | summary=None, |
| 662 | items=[], |
| 663 | ) |
| 664 | |
| 665 | |
| 666 | async def get_patch_tuesday_summary(request: PatchTuesdayRequest, top_n: int = 25) -> PatchTuesdaySummaryResponse: |
| 667 | """ |
| 668 | Get Patch Tuesday summary with top prioritized items. |
| 669 | |
| 670 | Args: |
| 671 | request: Request containing cycle and enrichment options. |
| 672 | top_n: Number of top items to include (default 25). |
| 673 | |
| 674 | Returns: |
| 675 | PatchTuesdaySummaryResponse with summary and top items. |
| 676 | """ |
| 677 | try: |
| 678 | items, summary = await fetch_patch_tuesday_data( |
| 679 | cycle=request.cycle, |
| 680 | include_epss=request.include_epss, |
| 681 | include_kev=request.include_kev, |
| 682 | ) |
| 683 | |
| 684 | return PatchTuesdaySummaryResponse( |
| 685 | success=True, |
| 686 | message=f"Successfully retrieved summary for cycle {summary.cycle}", |
| 687 | summary=summary, |
| 688 | top_items=items[:top_n], |
| 689 | ) |
| 690 | except ValueError as e: |
| 691 | logger.error(f"Invalid request: {e}") |
| 692 | return PatchTuesdaySummaryResponse( |
| 693 | success=False, |
| 694 | message=str(e), |
| 695 | summary=None, |
| 696 | top_items=[], |
| 697 | ) |
| 698 | except Exception as e: |
| 699 | logger.error(f"Error fetching Patch Tuesday summary: {e}") |
| 700 | return PatchTuesdaySummaryResponse( |
| 701 | success=False, |
| 702 | message=f"Failed to fetch Patch Tuesday summary: {e}", |
| 703 | summary=None, |
| 704 | top_items=[], |
| 705 | ) |
| 706 | |
| 707 | |
| 708 | async def get_available_cycles() -> AvailableCyclesResponse: |
| 709 | """ |
| 710 | Get available Patch Tuesday cycles. |
| 711 | |
| 712 | Returns recent cycles and information about the current/next Patch Tuesday. |
| 713 | """ |
| 714 | try: |
| 715 | today = dt.date.today() |
| 716 | current_cycle = get_default_cycle(today) |
| 717 | |
| 718 | # Generate list of recent cycles (last 12 months) |
| 719 | cycles = [] |
| 720 | for i in range(12): |
| 721 | d = today - dt.timedelta(days=i * 30) |
| 722 | cycle = get_default_cycle(d) |
| 723 | if cycle not in cycles: |
| 724 | cycles.append(cycle) |
| 725 | |
| 726 | # Calculate next Patch Tuesday |
| 727 | year, month, _ = parse_cycle(current_cycle) |
| 728 | pt_date = second_tuesday(year, month) |
| 729 | |
| 730 | if today > pt_date: |
| 731 | # Move to next month |
| 732 | if month == 12: |
| 733 | next_year, next_month = year + 1, 1 |
| 734 | else: |
| 735 | next_year, next_month = year, month + 1 |
| 736 | next_pt_date = second_tuesday(next_year, next_month) |
| 737 | else: |
| 738 | next_pt_date = pt_date |
| 739 | |
| 740 | return AvailableCyclesResponse( |
| 741 | success=True, |
| 742 | message="Successfully retrieved available cycles", |
| 743 | cycles=cycles, |
| 744 | current_cycle=current_cycle, |
| 745 | next_patch_tuesday=next_pt_date.isoformat(), |
| 746 | ) |
| 747 | except Exception as e: |
| 748 | logger.error(f"Error getting available cycles: {e}") |
| 749 | return AvailableCyclesResponse( |
| 750 | success=False, |
| 751 | message=f"Failed to get available cycles: {e}", |
| 752 | cycles=[], |
| 753 | current_cycle=get_default_cycle(), |
| 754 | next_patch_tuesday="", |
| 755 | ) |
| 756 | |
| 757 | |
| 758 | async def search_cves_in_patch_tuesday(cve_ids: List[str], cycle: Optional[str] = None) -> PatchTuesdayResponse: |
| 759 | """ |
| 760 | Search for specific CVEs in Patch Tuesday data. |
| 761 | |
| 762 | Args: |
| 763 | cve_ids: List of CVE IDs to search for. |
| 764 | cycle: Optional cycle to limit search to. |
| 765 | |
| 766 | Returns: |
| 767 | PatchTuesdayResponse with matching items. |
| 768 | """ |
| 769 | try: |
| 770 | request = PatchTuesdayRequest(cycle=cycle) |
| 771 | response = await get_patch_tuesday(request) |
| 772 | |
| 773 | if not response.success: |
| 774 | return response |
| 775 | |
| 776 | # Filter items to only matching CVEs |
| 777 | cve_set = set([c.upper() for c in cve_ids]) |
| 778 | matching_items = [item for item in response.items if item.cve.upper() in cve_set] |
| 779 | |
| 780 | # Recalculate summary for matched items |
| 781 | if matching_items and response.summary: |
| 782 | prio_counts = {"P0": 0, "P1": 0, "P2": 0, "P3": 0} |
| 783 | family_counts: Dict[str, int] = {} |
| 784 | |
| 785 | for item in matching_items: |
| 786 | prio_counts[item.prioritization.priority] = prio_counts.get(item.prioritization.priority, 0) + 1 |
| 787 | family_counts[item.affected.family] = family_counts.get(item.affected.family, 0) + 1 |
| 788 | |
| 789 | summary = PatchTuesdaySummary( |
| 790 | cycle=response.summary.cycle, |
| 791 | patch_tuesday_date=response.summary.patch_tuesday_date, |
| 792 | generated_utc=response.summary.generated_utc, |
| 793 | unique_cves=len(set([x.cve for x in matching_items])), |
| 794 | total_records=len(matching_items), |
| 795 | by_priority=PriorityCounts(**prio_counts), |
| 796 | by_family=dict(sorted(family_counts.items(), key=lambda kv: (-kv[1], kv[0]))), |
| 797 | by_severity=response.summary.by_severity, |
| 798 | ) |
| 799 | else: |
| 800 | summary = None |
| 801 | |
| 802 | return PatchTuesdayResponse( |
| 803 | success=True, |
| 804 | message=f"Found {len(matching_items)} items matching {len(cve_ids)} CVE(s)", |
| 805 | summary=summary, |
| 806 | items=matching_items, |
| 807 | ) |
| 808 | except Exception as e: |
| 809 | logger.error(f"Error searching CVEs: {e}") |
| 810 | return PatchTuesdayResponse( |
| 811 | success=False, |
| 812 | message=f"Failed to search CVEs: {e}", |
| 813 | summary=None, |
| 814 | items=[], |
| 815 | ) |