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
+ )