| 1 | from typing import List |
| 2 | from typing import Optional |
| 3 | |
| 4 | from fastapi import APIRouter |
| 5 | from fastapi import Query |
| 6 | from fastapi import Security |
| 7 | from loguru import logger |
| 8 | |
| 9 | from app.auth.routes.auth import AuthHandler |
| 10 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 11 | AvailableCyclesResponse, |
| 12 | ) |
| 13 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 14 | PatchTuesdayRequest, |
| 15 | ) |
| 16 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 17 | PatchTuesdayResponse, |
| 18 | ) |
| 19 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 20 | PatchTuesdaySummaryResponse, |
| 21 | ) |
| 22 | from app.integrations.microsoft_patch_tuesday.services.microsoft_patch_tuesday import ( |
| 23 | get_available_cycles, |
| 24 | ) |
| 25 | from app.integrations.microsoft_patch_tuesday.services.microsoft_patch_tuesday import ( |
| 26 | get_patch_tuesday, |
| 27 | ) |
| 28 | from app.integrations.microsoft_patch_tuesday.services.microsoft_patch_tuesday import ( |
| 29 | get_patch_tuesday_summary, |
| 30 | ) |
| 31 | from app.integrations.microsoft_patch_tuesday.services.microsoft_patch_tuesday import ( |
| 32 | search_cves_in_patch_tuesday, |
| 33 | ) |
| 34 | |
| 35 | microsoft_patch_tuesday_router = APIRouter() |
| 36 | |
| 37 | |
| 38 | @microsoft_patch_tuesday_router.get( |
| 39 | "", |
| 40 | response_model=PatchTuesdayResponse, |
| 41 | description="Get full Patch Tuesday data for a specific cycle", |
| 42 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 43 | ) |
| 44 | async def get_patch_tuesday_data( |
| 45 | cycle: Optional[str] = Query( |
| 46 | None, |
| 47 | description="Cycle in YYYY-Mmm format (e.g., 2026-Jan). Defaults to current month.", |
| 48 | pattern=r"^\d{4}-(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)$", |
| 49 | ), |
| 50 | include_epss: bool = Query(True, description="Include EPSS scores in the response"), |
| 51 | include_kev: bool = Query(True, description="Include CISA KEV data in the response"), |
| 52 | ) -> PatchTuesdayResponse: |
| 53 | """ |
| 54 | Get full Patch Tuesday vulnerability data for a specific cycle. |
| 55 | |
| 56 | This endpoint fetches data from the Microsoft Security Response Center (MSRC) CVRF API, |
| 57 | enriches it with EPSS scores and CISA KEV information, and returns prioritized results. |
| 58 | |
| 59 | **Priority Levels:** |
| 60 | - **P0 (Emergency)**: Known exploited (CISA KEV), very high EPSS (>=0.9), or Critical + high EPSS |
| 61 | - **P1 (High)**: Critical severity, or high CVSS (>=8.0) with elevated EPSS or core enterprise products |
| 62 | - **P2 (Medium)**: Important/Moderate severity, CVSS >=6.0, or EPSS >=0.1 |
| 63 | - **P3 (Low)**: All other vulnerabilities |
| 64 | |
| 65 | **Product Families:** |
| 66 | - Windows, Windows Server, Office/M365, Exchange, SharePoint, SQL Server, |
| 67 | Developer Platform, Edge, Azure, Dynamics, Other |
| 68 | """ |
| 69 | logger.info(f"Fetching Patch Tuesday data for cycle: {cycle or 'current'}") |
| 70 | |
| 71 | request = PatchTuesdayRequest( |
| 72 | cycle=cycle, |
| 73 | include_epss=include_epss, |
| 74 | include_kev=include_kev, |
| 75 | ) |
| 76 | |
| 77 | return await get_patch_tuesday(request) |
| 78 | |
| 79 | |
| 80 | @microsoft_patch_tuesday_router.get( |
| 81 | "/summary", |
| 82 | response_model=PatchTuesdaySummaryResponse, |
| 83 | description="Get Patch Tuesday summary with top prioritized items", |
| 84 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 85 | ) |
| 86 | async def get_patch_tuesday_summary_endpoint( |
| 87 | cycle: Optional[str] = Query( |
| 88 | None, |
| 89 | description="Cycle in YYYY-Mmm format (e.g., 2026-Jan). Defaults to current month.", |
| 90 | pattern=r"^\d{4}-(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)$", |
| 91 | ), |
| 92 | include_epss: bool = Query(True, description="Include EPSS scores in the response"), |
| 93 | include_kev: bool = Query(True, description="Include CISA KEV data in the response"), |
| 94 | top_n: int = Query(25, ge=1, le=100, description="Number of top items to include"), |
| 95 | ) -> PatchTuesdaySummaryResponse: |
| 96 | """ |
| 97 | Get Patch Tuesday summary with top prioritized vulnerabilities. |
| 98 | |
| 99 | This is a lighter endpoint that returns only the summary statistics and |
| 100 | top N prioritized items, suitable for dashboards and quick overviews. |
| 101 | """ |
| 102 | logger.info(f"Fetching Patch Tuesday summary for cycle: {cycle or 'current'}") |
| 103 | |
| 104 | request = PatchTuesdayRequest( |
| 105 | cycle=cycle, |
| 106 | include_epss=include_epss, |
| 107 | include_kev=include_kev, |
| 108 | ) |
| 109 | |
| 110 | return await get_patch_tuesday_summary(request, top_n=top_n) |
| 111 | |
| 112 | |
| 113 | @microsoft_patch_tuesday_router.get( |
| 114 | "/cycles", |
| 115 | response_model=AvailableCyclesResponse, |
| 116 | description="Get available Patch Tuesday cycles", |
| 117 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 118 | ) |
| 119 | async def get_cycles() -> AvailableCyclesResponse: |
| 120 | """ |
| 121 | Get available Patch Tuesday cycles. |
| 122 | |
| 123 | Returns a list of recent cycles (last 12 months), the current cycle, |
| 124 | and the date of the next Patch Tuesday. |
| 125 | """ |
| 126 | logger.info("Fetching available Patch Tuesday cycles") |
| 127 | return await get_available_cycles() |
| 128 | |
| 129 | |
| 130 | @microsoft_patch_tuesday_router.get( |
| 131 | "/search", |
| 132 | response_model=PatchTuesdayResponse, |
| 133 | description="Search for specific CVEs in Patch Tuesday data", |
| 134 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 135 | ) |
| 136 | async def search_cves( |
| 137 | cve_ids: List[str] = Query(..., description="List of CVE IDs to search for"), |
| 138 | cycle: Optional[str] = Query( |
| 139 | None, |
| 140 | description="Cycle in YYYY-Mmm format to limit search to. Defaults to current month.", |
| 141 | pattern=r"^\d{4}-(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)$", |
| 142 | ), |
| 143 | ) -> PatchTuesdayResponse: |
| 144 | """ |
| 145 | Search for specific CVEs in Patch Tuesday data. |
| 146 | |
| 147 | Returns detailed information about the specified CVEs including |
| 148 | EPSS scores, CISA KEV status, and prioritization recommendations. |
| 149 | """ |
| 150 | logger.info(f"Searching for CVEs: {cve_ids} in cycle: {cycle or 'current'}") |
| 151 | return await search_cves_in_patch_tuesday(cve_ids, cycle) |
| 152 | |
| 153 | |
| 154 | @microsoft_patch_tuesday_router.get( |
| 155 | "/priority/{priority_level}", |
| 156 | response_model=PatchTuesdayResponse, |
| 157 | description="Get vulnerabilities by priority level", |
| 158 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 159 | ) |
| 160 | async def get_by_priority( |
| 161 | priority_level: str, |
| 162 | cycle: Optional[str] = Query( |
| 163 | None, |
| 164 | description="Cycle in YYYY-Mmm format. Defaults to current month.", |
| 165 | pattern=r"^\d{4}-(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)$", |
| 166 | ), |
| 167 | include_epss: bool = Query(True, description="Include EPSS scores"), |
| 168 | include_kev: bool = Query(True, description="Include CISA KEV data"), |
| 169 | ) -> PatchTuesdayResponse: |
| 170 | """ |
| 171 | Get vulnerabilities filtered by priority level. |
| 172 | |
| 173 | **Valid priority levels:** P0, P1, P2, P3 |
| 174 | """ |
| 175 | logger.info(f"Fetching {priority_level} priority items for cycle: {cycle or 'current'}") |
| 176 | |
| 177 | priority_level = priority_level.upper() |
| 178 | if priority_level not in ["P0", "P1", "P2", "P3"]: |
| 179 | return PatchTuesdayResponse( |
| 180 | success=False, |
| 181 | message=f"Invalid priority level '{priority_level}'. Valid values: P0, P1, P2, P3", |
| 182 | summary=None, |
| 183 | items=[], |
| 184 | ) |
| 185 | |
| 186 | request = PatchTuesdayRequest( |
| 187 | cycle=cycle, |
| 188 | include_epss=include_epss, |
| 189 | include_kev=include_kev, |
| 190 | ) |
| 191 | |
| 192 | response = await get_patch_tuesday(request) |
| 193 | |
| 194 | if not response.success: |
| 195 | return response |
| 196 | |
| 197 | # Filter by priority |
| 198 | filtered_items = [item for item in response.items if item.prioritization.priority == priority_level] |
| 199 | |
| 200 | # Update counts |
| 201 | if response.summary: |
| 202 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 203 | PatchTuesdaySummary, |
| 204 | ) |
| 205 | from app.integrations.microsoft_patch_tuesday.schema.microsoft_patch_tuesday import ( |
| 206 | PriorityCounts, |
| 207 | ) |
| 208 | |
| 209 | family_counts = {} |
| 210 | for item in filtered_items: |
| 211 | family_counts[item.affected.family] = family_counts.get(item.affected.family, 0) + 1 |
| 212 | |
| 213 | # Create new priority counts with only the selected priority |
| 214 | new_prio_counts = {"P0": 0, "P1": 0, "P2": 0, "P3": 0} |
| 215 | new_prio_counts[priority_level] = len(filtered_items) |
| 216 | |
| 217 | summary = PatchTuesdaySummary( |
| 218 | cycle=response.summary.cycle, |
| 219 | patch_tuesday_date=response.summary.patch_tuesday_date, |
| 220 | generated_utc=response.summary.generated_utc, |
| 221 | unique_cves=len(set([x.cve for x in filtered_items])), |
| 222 | total_records=len(filtered_items), |
| 223 | by_priority=PriorityCounts(**new_prio_counts), |
| 224 | by_family=dict(sorted(family_counts.items(), key=lambda kv: (-kv[1], kv[0]))), |
| 225 | by_severity=response.summary.by_severity, |
| 226 | ) |
| 227 | else: |
| 228 | summary = None |
| 229 | |
| 230 | return PatchTuesdayResponse( |
| 231 | success=True, |
| 232 | message=f"Found {len(filtered_items)} {priority_level} priority items", |
| 233 | summary=summary, |
| 234 | items=filtered_items, |
| 235 | ) |
| 236 | |
| 237 | |
| 238 | @microsoft_patch_tuesday_router.get( |
| 239 | "/kev", |
| 240 | response_model=PatchTuesdayResponse, |
| 241 | description="Get only CISA KEV (Known Exploited Vulnerabilities) items", |
| 242 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 243 | ) |
| 244 | async def get_kev_items( |
| 245 | cycle: Optional[str] = Query( |
| 246 | None, |
| 247 | description="Cycle in YYYY-Mmm format. Defaults to current month.", |
| 248 | pattern=r"^\d{4}-(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)$", |
| 249 | ), |
| 250 | ) -> PatchTuesdayResponse: |
| 251 | """ |
| 252 | Get only vulnerabilities that are in the CISA Known Exploited Vulnerabilities catalog. |
| 253 | |
| 254 | These are actively exploited vulnerabilities that should be prioritized for immediate remediation. |
| 255 | """ |
| 256 | logger.info(f"Fetching KEV items for cycle: {cycle or 'current'}") |
| 257 | |
| 258 | request = PatchTuesdayRequest( |
| 259 | cycle=cycle, |
| 260 | include_epss=True, |
| 261 | include_kev=True, |
| 262 | ) |
| 263 | |
| 264 | response = await get_patch_tuesday(request) |
| 265 | |
| 266 | if not response.success: |
| 267 | return response |
| 268 | |
| 269 | # Filter to only KEV items |
| 270 | kev_items = [item for item in response.items if item.kev.in_kev] |
| 271 | |
| 272 | return PatchTuesdayResponse( |
| 273 | success=True, |
| 274 | message=f"Found {len(kev_items)} known exploited vulnerabilities", |
| 275 | summary=response.summary, |
| 276 | items=kev_items, |
| 277 | ) |