| 1 | """ |
| 2 | Service layer for case tasks (issue #792, Phase 3). |
| 3 | |
| 4 | Responsibilities: |
| 5 | - Selecting a CaseTemplate for a newly-created Case based on |
| 6 | ``(customer_code, source)`` with a documented priority order. |
| 7 | - Snapshot-copying ``CaseTemplateTask`` rows into ``CaseTask`` rows on |
| 8 | case creation (or post-creation manual apply). |
| 9 | - CRUD on CaseTask rows (analyst-driven custom adds, status updates with |
| 10 | evidence comments, deletes). |
| 11 | - The "soft-warning on close" check that's wired into the existing |
| 12 | ``/case/status`` endpoint by the route layer. |
| 13 | |
| 14 | Authorization is enforced at the route layer; this module is auth-agnostic. |
| 15 | """ |
| 16 | |
| 17 | from datetime import datetime |
| 18 | from typing import Any |
| 19 | from typing import Dict |
| 20 | from typing import List |
| 21 | from typing import Optional |
| 22 | from typing import Tuple |
| 23 | |
| 24 | from loguru import logger |
| 25 | from sqlalchemy import select |
| 26 | from sqlalchemy import update |
| 27 | from sqlalchemy.ext.asyncio import AsyncSession |
| 28 | from sqlalchemy.orm import selectinload |
| 29 | |
| 30 | from app.incidents.models import Alert |
| 31 | from app.incidents.models import Asset |
| 32 | from app.incidents.models import Case |
| 33 | from app.incidents.models import CaseAlertLink |
| 34 | from app.incidents.models import CaseTask |
| 35 | from app.incidents.models import CaseTemplate |
| 36 | from app.incidents.schema.case_templates import CaseCloseWarningResponse |
| 37 | from app.incidents.schema.case_templates import CaseEventType |
| 38 | from app.incidents.schema.case_templates import CaseTaskCreate |
| 39 | from app.incidents.schema.case_templates import CaseTaskListResponse |
| 40 | from app.incidents.schema.case_templates import CaseTaskOperationResponse |
| 41 | from app.incidents.schema.case_templates import CaseTaskResponse |
| 42 | from app.incidents.schema.case_templates import CaseTaskStatus |
| 43 | from app.incidents.schema.case_templates import CaseTaskUpdate |
| 44 | |
| 45 | # --------------------------------------------------------------------------- |
| 46 | # Conversions |
| 47 | # --------------------------------------------------------------------------- |
| 48 | |
| 49 | |
| 50 | def _case_task_to_response(task: CaseTask) -> CaseTaskResponse: |
| 51 | return CaseTaskResponse( |
| 52 | id=task.id, |
| 53 | case_id=task.case_id, |
| 54 | alert_id=task.alert_id, |
| 55 | template_task_id=task.template_task_id, |
| 56 | title=task.title, |
| 57 | description=task.description, |
| 58 | guidelines=task.guidelines, |
| 59 | mandatory=task.mandatory, |
| 60 | order_index=task.order_index, |
| 61 | status=CaseTaskStatus(task.status), |
| 62 | evidence_comment=task.evidence_comment, |
| 63 | completed_by=task.completed_by, |
| 64 | completed_at=task.completed_at, |
| 65 | created_by=task.created_by, |
| 66 | created_at=task.created_at, |
| 67 | updated_at=task.updated_at, |
| 68 | ) |
| 69 | |
| 70 | |
| 71 | # --------------------------------------------------------------------------- |
| 72 | # Template selection (for case creation) |
| 73 | # --------------------------------------------------------------------------- |
| 74 | |
| 75 | |
| 76 | async def pick_template_for_case( |
| 77 | customer_code: Optional[str], |
| 78 | source: Optional[str], |
| 79 | session: AsyncSession, |
| 80 | ) -> Optional[CaseTemplate]: |
| 81 | """ |
| 82 | Pick the most-specific applicable template for a newly created case. |
| 83 | |
| 84 | Priority order (each step short-circuits the next): |
| 85 | |
| 86 | 1. ``customer_code`` + ``source`` exact match, prefer is_default |
| 87 | 2. ``customer_code`` only (source IS NULL), prefer is_default |
| 88 | 3. ``source`` only (customer_code IS NULL), prefer is_default |
| 89 | 4. Global default (both NULL, is_default=True) |
| 90 | |
| 91 | **Conditional templates are excluded from every tier.** A template with a |
| 92 | ``match_field`` set is saying "fire only when this condition matches" — it |
| 93 | participates in selection exclusively via ``pick_templates_for_alert``. If |
| 94 | its condition is rejected, it must not silently re-enter via the fallback |
| 95 | tier; that would defeat the whole point of having a condition. |
| 96 | |
| 97 | Returns the template with tasks eagerly loaded, or None if no match. |
| 98 | """ |
| 99 | |
| 100 | async def _query(customer_filter, source_filter) -> Optional[CaseTemplate]: |
| 101 | stmt = ( |
| 102 | select(CaseTemplate) |
| 103 | .options(selectinload(CaseTemplate.tasks)) |
| 104 | .where(CaseTemplate.match_field.is_(None)) |
| 105 | .where(customer_filter) |
| 106 | .where(source_filter) |
| 107 | .order_by(CaseTemplate.is_default.desc(), CaseTemplate.created_at.desc()) |
| 108 | ) |
| 109 | result = await session.execute(stmt) |
| 110 | return result.scalars().first() |
| 111 | |
| 112 | # Step 1 — both match |
| 113 | if customer_code is not None and source is not None: |
| 114 | match = await _query( |
| 115 | CaseTemplate.customer_code == customer_code, |
| 116 | CaseTemplate.source == source, |
| 117 | ) |
| 118 | if match is not None: |
| 119 | return match |
| 120 | |
| 121 | # Step 2 — customer only |
| 122 | if customer_code is not None: |
| 123 | match = await _query( |
| 124 | CaseTemplate.customer_code == customer_code, |
| 125 | CaseTemplate.source.is_(None), |
| 126 | ) |
| 127 | if match is not None: |
| 128 | return match |
| 129 | |
| 130 | # Step 3 — source only |
| 131 | if source is not None: |
| 132 | match = await _query( |
| 133 | CaseTemplate.customer_code.is_(None), |
| 134 | CaseTemplate.source == source, |
| 135 | ) |
| 136 | if match is not None: |
| 137 | return match |
| 138 | |
| 139 | # Step 4 — global default |
| 140 | match = await _query( |
| 141 | CaseTemplate.customer_code.is_(None), |
| 142 | CaseTemplate.source.is_(None), |
| 143 | ) |
| 144 | if match is not None and match.is_default: |
| 145 | return match |
| 146 | |
| 147 | return None |
| 148 | |
| 149 | |
| 150 | # --------------------------------------------------------------------------- |
| 151 | # Conditional template selection (field-match against the raw Wazuh document) |
| 152 | # --------------------------------------------------------------------------- |
| 153 | |
| 154 | |
| 155 | async def _fetch_raw_event_for_alert( |
| 156 | alert: Alert, |
| 157 | session: AsyncSession, |
| 158 | ) -> Optional[Dict[str, Any]]: |
| 159 | """ |
| 160 | Return the raw Wazuh document for this alert's first asset, or None if |
| 161 | there is no asset or the OpenSearch fetch raises. |
| 162 | |
| 163 | Picks the lowest-id asset deterministically; the original-event coordinates |
| 164 | are typically the same across an alert's assets, but if they diverge we |
| 165 | prefer the asset that landed first. Talks to the Wazuh indexer directly |
| 166 | rather than going through ``fetch_alert_details`` because we don't want |
| 167 | its syslog_type validation gate — for field-match we only need a key/value |
| 168 | bag. |
| 169 | """ |
| 170 | asset_stmt = select(Asset).where(Asset.alert_linked == alert.id).order_by(Asset.id.asc()).limit(1) |
| 171 | asset = (await session.execute(asset_stmt)).scalars().first() |
| 172 | if asset is None: |
| 173 | logger.info( |
| 174 | f"pick_templates_for_alert: alert id={alert.id} has no asset row; " "field-match templates are skipped for this alert.", |
| 175 | ) |
| 176 | return None |
| 177 | |
| 178 | try: |
| 179 | from app.connectors.wazuh_indexer.utils.universal import ( |
| 180 | create_wazuh_indexer_client_async, |
| 181 | ) |
| 182 | |
| 183 | es_client = await create_wazuh_indexer_client_async("Wazuh-Indexer") |
| 184 | doc = await es_client.get(index=asset.index_name, id=asset.index_id) |
| 185 | return doc.get("_source") or {} |
| 186 | except Exception as e: |
| 187 | logger.warning( |
| 188 | f"pick_templates_for_alert: raw-event fetch failed for alert id={alert.id} " |
| 189 | f"(index={asset.index_name}, id={asset.index_id}): {e}. " |
| 190 | "Falling back to non-field-match templates.", |
| 191 | ) |
| 192 | return None |
| 193 | |
| 194 | |
| 195 | async def pick_templates_for_alert( |
| 196 | case: Case, |
| 197 | alert: Alert, |
| 198 | session: AsyncSession, |
| 199 | ) -> List[CaseTemplate]: |
| 200 | """ |
| 201 | Return every CaseTemplate that should auto-apply for this (case, alert). |
| 202 | |
| 203 | Two-stage selection: |
| 204 | |
| 205 | 1. **Field-match templates** (both ``match_field`` and ``match_value`` |
| 206 | set) scoped to this case's customer + alert's source are evaluated |
| 207 | against the raw Wazuh document fetched via the alert's first asset. |
| 208 | Every template whose ``document[match_field] == match_value`` is |
| 209 | included — they layer additively. |
| 210 | 2. **Fallback.** If zero field-match templates fired (no candidates, no |
| 211 | matches, or the raw-event fetch raised), use the existing single- |
| 212 | template tier picker (``customer+source > customer > source > global |
| 213 | default``) and return its result as a one-element list. |
| 214 | |
| 215 | This split is deliberate: a generic "wazuh global" template shouldn't |
| 216 | drown a specific "sysmon event 1" one. Returns an empty list if nothing |
| 217 | applies. |
| 218 | """ |
| 219 | field_match_stmt = ( |
| 220 | select(CaseTemplate) |
| 221 | .options(selectinload(CaseTemplate.tasks)) |
| 222 | .where(CaseTemplate.match_field.is_not(None)) |
| 223 | .where(CaseTemplate.match_value.is_not(None)) |
| 224 | .where( |
| 225 | (CaseTemplate.customer_code == case.customer_code) | (CaseTemplate.customer_code.is_(None)), |
| 226 | ) |
| 227 | .where( |
| 228 | (CaseTemplate.source == alert.source) | (CaseTemplate.source.is_(None)), |
| 229 | ) |
| 230 | ) |
| 231 | candidates = list((await session.execute(field_match_stmt)).scalars().all()) |
| 232 | |
| 233 | matched: List[CaseTemplate] = [] |
| 234 | if candidates: |
| 235 | raw_event = await _fetch_raw_event_for_alert(alert, session) |
| 236 | if raw_event is not None: |
| 237 | for tmpl in candidates: |
| 238 | doc_value = raw_event.get(tmpl.match_field) |
| 239 | if doc_value is None: |
| 240 | continue |
| 241 | # Wazuh top-level values arrive as strings already (numeric |
| 242 | # fields like data_win_system_eventID are quoted: "1"). Coerce |
| 243 | # defensively so a numeric value in the document still matches |
| 244 | # the string in match_value. |
| 245 | if str(doc_value) == tmpl.match_value: |
| 246 | matched.append(tmpl) |
| 247 | |
| 248 | if matched: |
| 249 | logger.info( |
| 250 | f"pick_templates_for_alert: {len(matched)} field-match template(s) fired " |
| 251 | f"for case id={case.id} alert id={alert.id}: {[t.id for t in matched]}", |
| 252 | ) |
| 253 | return matched |
| 254 | |
| 255 | fallback = await pick_template_for_case( |
| 256 | customer_code=case.customer_code, |
| 257 | source=alert.source, |
| 258 | session=session, |
| 259 | ) |
| 260 | return [fallback] if fallback is not None else [] |
| 261 | |
| 262 | |
| 263 | # --------------------------------------------------------------------------- |
| 264 | # Template application (snapshot copy) |
| 265 | # --------------------------------------------------------------------------- |
| 266 | |
| 267 | |
| 268 | async def apply_template_to_case( |
| 269 | case_id: int, |
| 270 | template_id: int, |
| 271 | actor: str, |
| 272 | session: AsyncSession, |
| 273 | *, |
| 274 | alert_id: Optional[int] = None, |
| 275 | commit: bool = True, |
| 276 | ) -> List[CaseTask]: |
| 277 | """ |
| 278 | Snapshot-copy every CaseTemplateTask on the named template into a new |
| 279 | CaseTask row attached to the case. |
| 280 | |
| 281 | Tasks are *snapshots* — editing the source template later does not |
| 282 | mutate the CaseTask rows. ``template_task_id`` is preserved as a soft |
| 283 | link for analytics / future "this came from template X" UI hints. |
| 284 | |
| 285 | When ``alert_id`` is provided, every materialized CaseTask row is |
| 286 | stamped with that alert id (so the Tasks UI can group tasks under |
| 287 | their originating alert). The caller is responsible for ensuring the |
| 288 | alert is linked to the case; this function trusts the caller. ``None`` |
| 289 | produces case-wide / general tasks. |
| 290 | |
| 291 | Set ``commit=False`` when calling from inside another transaction |
| 292 | (e.g., immediately after a Case is created in the same flow); the |
| 293 | caller is then responsible for the commit. |
| 294 | """ |
| 295 | template_result = await session.execute( |
| 296 | select(CaseTemplate).where(CaseTemplate.id == template_id).options(selectinload(CaseTemplate.tasks)), |
| 297 | ) |
| 298 | template = template_result.scalar_one_or_none() |
| 299 | if template is None: |
| 300 | logger.warning(f"apply_template_to_case: template id={template_id} not found") |
| 301 | return [] |
| 302 | |
| 303 | case_result = await session.execute(select(Case).where(Case.id == case_id)) |
| 304 | case = case_result.scalar_one_or_none() |
| 305 | if case is None: |
| 306 | logger.warning(f"apply_template_to_case: case id={case_id} not found") |
| 307 | return [] |
| 308 | |
| 309 | new_tasks: List[CaseTask] = [] |
| 310 | for tmpl_task in sorted(template.tasks, key=lambda t: (t.order_index, t.id)): |
| 311 | case_task = CaseTask( |
| 312 | case_id=case_id, |
| 313 | alert_id=alert_id, |
| 314 | template_task_id=tmpl_task.id, |
| 315 | title=tmpl_task.title, |
| 316 | description=tmpl_task.description, |
| 317 | guidelines=tmpl_task.guidelines, |
| 318 | mandatory=tmpl_task.mandatory, |
| 319 | order_index=tmpl_task.order_index, |
| 320 | status=CaseTaskStatus.TODO.value, |
| 321 | evidence_comment=None, |
| 322 | completed_by=None, |
| 323 | completed_at=None, |
| 324 | created_by=actor, |
| 325 | ) |
| 326 | session.add(case_task) |
| 327 | new_tasks.append(case_task) |
| 328 | |
| 329 | # Audit emits (Phase 4): one template_applied event for the operation, |
| 330 | # plus one task_added event per snapshotted task. Imported lazily to |
| 331 | # avoid a circular import — case_events doesn't depend on case_tasks |
| 332 | # but isort/lint sometimes resolves these eagerly. |
| 333 | from app.incidents.services.case_events import emit_case_event |
| 334 | from app.incidents.services.case_events import payload_task |
| 335 | from app.incidents.services.case_events import payload_template_applied |
| 336 | |
| 337 | await session.flush() # ensure case_task.id is populated for the audit payloads |
| 338 | |
| 339 | template_applied_payload = payload_template_applied( |
| 340 | template_id=template.id, |
| 341 | template_name=template.name, |
| 342 | tasks_added=len(new_tasks), |
| 343 | ) |
| 344 | if alert_id is not None: |
| 345 | template_applied_payload["alert_id"] = alert_id |
| 346 | |
| 347 | await emit_case_event( |
| 348 | session=session, |
| 349 | case_id=case_id, |
| 350 | event_type=CaseEventType.TEMPLATE_APPLIED, |
| 351 | actor=actor, |
| 352 | payload=template_applied_payload, |
| 353 | commit=False, |
| 354 | ) |
| 355 | for ct in new_tasks: |
| 356 | await emit_case_event( |
| 357 | session=session, |
| 358 | case_id=case_id, |
| 359 | event_type=CaseEventType.TASK_ADDED, |
| 360 | actor=actor, |
| 361 | payload=payload_task( |
| 362 | task_id=ct.id, |
| 363 | title=ct.title, |
| 364 | mandatory=ct.mandatory, |
| 365 | source="template", |
| 366 | template_id=template.id, |
| 367 | alert_id=alert_id, |
| 368 | ), |
| 369 | commit=False, |
| 370 | ) |
| 371 | |
| 372 | if commit: |
| 373 | await session.commit() |
| 374 | for t in new_tasks: |
| 375 | await session.refresh(t) |
| 376 | logger.info( |
| 377 | f"Applied template id={template_id} ('{template.name}') to case id={case_id}: " f"{len(new_tasks)} task(s) created by {actor}", |
| 378 | ) |
| 379 | else: |
| 380 | logger.info( |
| 381 | f"Staged template id={template_id} ('{template.name}') for case id={case_id}: " f"{len(new_tasks)} task(s) (uncommitted)", |
| 382 | ) |
| 383 | |
| 384 | return new_tasks |
| 385 | |
| 386 | |
| 387 | async def auto_apply_template_for_new_case( |
| 388 | case: Case, |
| 389 | alert: Alert, |
| 390 | actor: str, |
| 391 | session: AsyncSession, |
| 392 | ) -> List[Tuple[CaseTemplate, List[CaseTask]]]: |
| 393 | """ |
| 394 | Run the per-alert auto-apply selection for a case and apply every |
| 395 | template the new ``pick_templates_for_alert`` returns. |
| 396 | |
| 397 | Field-match templates (when present and the raw event matches) layer |
| 398 | additively; otherwise the legacy tier picker contributes one fallback |
| 399 | template. All materialized CaseTask rows are stamped with ``alert.id`` |
| 400 | so they group under the originating alert in the UI. |
| 401 | |
| 402 | Returns the list of (template, tasks) pairs applied. ``commit=False`` on |
| 403 | each apply call — the caller's transaction stays in control. |
| 404 | """ |
| 405 | templates = await pick_templates_for_alert(case=case, alert=alert, session=session) |
| 406 | applied: List[Tuple[CaseTemplate, List[CaseTask]]] = [] |
| 407 | for template in templates: |
| 408 | tasks = await apply_template_to_case( |
| 409 | case_id=case.id, |
| 410 | template_id=template.id, |
| 411 | actor=actor, |
| 412 | session=session, |
| 413 | alert_id=alert.id, |
| 414 | commit=False, |
| 415 | ) |
| 416 | applied.append((template, tasks)) |
| 417 | return applied |
| 418 | |
| 419 | |
| 420 | # --------------------------------------------------------------------------- |
| 421 | # Case task CRUD |
| 422 | # --------------------------------------------------------------------------- |
| 423 | |
| 424 | |
| 425 | async def list_case_tasks(case_id: int, session: AsyncSession) -> CaseTaskListResponse: |
| 426 | try: |
| 427 | stmt = select(CaseTask).where(CaseTask.case_id == case_id).order_by(CaseTask.order_index, CaseTask.id) |
| 428 | result = await session.execute(stmt) |
| 429 | tasks = result.scalars().all() |
| 430 | return CaseTaskListResponse( |
| 431 | tasks=[_case_task_to_response(t) for t in tasks], |
| 432 | success=True, |
| 433 | message=f"Retrieved {len(tasks)} task(s) for case id={case_id}", |
| 434 | ) |
| 435 | except Exception as e: |
| 436 | logger.error(f"Failed to list tasks for case id={case_id}: {e}") |
| 437 | return CaseTaskListResponse( |
| 438 | tasks=[], |
| 439 | success=False, |
| 440 | message=f"Failed to list case tasks: {e}", |
| 441 | ) |
| 442 | |
| 443 | |
| 444 | async def is_alert_linked_to_case( |
| 445 | case_id: int, |
| 446 | alert_id: int, |
| 447 | session: AsyncSession, |
| 448 | ) -> bool: |
| 449 | """True iff a CaseAlertLink row exists for this (case, alert) pair.""" |
| 450 | stmt = select(CaseAlertLink.alert_id).where(CaseAlertLink.case_id == case_id).where(CaseAlertLink.alert_id == alert_id).limit(1) |
| 451 | result = await session.execute(stmt) |
| 452 | return result.scalar_one_or_none() is not None |
| 453 | |
| 454 | |
| 455 | async def add_case_task( |
| 456 | case_id: int, |
| 457 | request: CaseTaskCreate, |
| 458 | actor: str, |
| 459 | session: AsyncSession, |
| 460 | ) -> CaseTaskOperationResponse: |
| 461 | """Add a custom task to a case mid-investigation. template_task_id is NULL. |
| 462 | |
| 463 | When ``request.alert_id`` is provided, validates the alert is currently |
| 464 | linked to the case — analysts shouldn't be able to attach tasks to alerts |
| 465 | that aren't part of the case. Otherwise the task is created as case-wide. |
| 466 | """ |
| 467 | try: |
| 468 | case_result = await session.execute(select(Case).where(Case.id == case_id)) |
| 469 | if case_result.scalar_one_or_none() is None: |
| 470 | return CaseTaskOperationResponse( |
| 471 | task=None, |
| 472 | success=False, |
| 473 | message=f"Case id={case_id} not found", |
| 474 | ) |
| 475 | |
| 476 | if request.alert_id is not None: |
| 477 | if not await is_alert_linked_to_case(case_id, request.alert_id, session): |
| 478 | return CaseTaskOperationResponse( |
| 479 | task=None, |
| 480 | success=False, |
| 481 | message=( |
| 482 | f"Alert id={request.alert_id} is not linked to case id={case_id}; " |
| 483 | "link the alert first or omit alert_id for a case-wide task." |
| 484 | ), |
| 485 | ) |
| 486 | |
| 487 | task = CaseTask( |
| 488 | case_id=case_id, |
| 489 | alert_id=request.alert_id, |
| 490 | template_task_id=None, |
| 491 | title=request.title, |
| 492 | description=request.description, |
| 493 | guidelines=request.guidelines, |
| 494 | mandatory=request.mandatory, |
| 495 | order_index=request.order_index, |
| 496 | status=CaseTaskStatus.TODO.value, |
| 497 | created_by=actor, |
| 498 | ) |
| 499 | session.add(task) |
| 500 | await session.flush() |
| 501 | |
| 502 | from app.incidents.services.case_events import emit_case_event |
| 503 | from app.incidents.services.case_events import payload_task |
| 504 | |
| 505 | await emit_case_event( |
| 506 | session=session, |
| 507 | case_id=case_id, |
| 508 | event_type=CaseEventType.TASK_ADDED, |
| 509 | actor=actor, |
| 510 | payload=payload_task( |
| 511 | task_id=task.id, |
| 512 | title=task.title, |
| 513 | mandatory=task.mandatory, |
| 514 | source="custom", |
| 515 | alert_id=task.alert_id, |
| 516 | ), |
| 517 | commit=False, |
| 518 | ) |
| 519 | |
| 520 | await session.commit() |
| 521 | await session.refresh(task) |
| 522 | |
| 523 | return CaseTaskOperationResponse( |
| 524 | task=_case_task_to_response(task), |
| 525 | success=True, |
| 526 | message=f"Added task id={task.id} to case id={case_id}", |
| 527 | ) |
| 528 | except Exception as e: |
| 529 | logger.error(f"Failed to add task to case id={case_id}: {e}") |
| 530 | await session.rollback() |
| 531 | return CaseTaskOperationResponse( |
| 532 | task=None, |
| 533 | success=False, |
| 534 | message=f"Failed to add case task: {e}", |
| 535 | ) |
| 536 | |
| 537 | |
| 538 | async def update_case_task( |
| 539 | task_id: int, |
| 540 | request: CaseTaskUpdate, |
| 541 | actor: str, |
| 542 | session: AsyncSession, |
| 543 | ) -> CaseTaskOperationResponse: |
| 544 | """ |
| 545 | Update task status and/or evidence comment. Validates that |
| 546 | NOT_NECESSARY isn't applied to a mandatory task. Sets/unsets |
| 547 | ``completed_by`` and ``completed_at`` based on the resulting status. |
| 548 | """ |
| 549 | try: |
| 550 | result = await session.execute(select(CaseTask).where(CaseTask.id == task_id)) |
| 551 | task = result.scalar_one_or_none() |
| 552 | if task is None: |
| 553 | return CaseTaskOperationResponse( |
| 554 | task=None, |
| 555 | success=False, |
| 556 | message=f"Case task id={task_id} not found", |
| 557 | ) |
| 558 | |
| 559 | fields_set = request.__fields_set__ |
| 560 | previous_status = task.status |
| 561 | status_changed = False |
| 562 | |
| 563 | if "status" in fields_set and request.status is not None: |
| 564 | new_status = request.status |
| 565 | if new_status == CaseTaskStatus.NOT_NECESSARY and task.mandatory: |
| 566 | return CaseTaskOperationResponse( |
| 567 | task=_case_task_to_response(task), |
| 568 | success=False, |
| 569 | message="Mandatory tasks cannot be marked NOT_NECESSARY.", |
| 570 | ) |
| 571 | if new_status.value != task.status: |
| 572 | task.status = new_status.value |
| 573 | status_changed = True |
| 574 | |
| 575 | # Maintain completed_by / completed_at to reflect the resulting state. |
| 576 | if new_status in (CaseTaskStatus.DONE, CaseTaskStatus.NOT_NECESSARY): |
| 577 | task.completed_by = actor |
| 578 | task.completed_at = datetime.utcnow() |
| 579 | else: |
| 580 | # Returning to TODO clears the completion record. |
| 581 | task.completed_by = None |
| 582 | task.completed_at = None |
| 583 | |
| 584 | comment_set_this_call = "evidence_comment" in fields_set |
| 585 | if comment_set_this_call: |
| 586 | task.evidence_comment = request.evidence_comment |
| 587 | |
| 588 | task.updated_at = datetime.utcnow() |
| 589 | session.add(task) |
| 590 | |
| 591 | # Audit emits (Phase 4): two distinct events when both fire — the UI |
| 592 | # can render them as a single block but the data model keeps them |
| 593 | # separate so a comment-only update still appears in the timeline. |
| 594 | from app.incidents.services.case_events import emit_case_event |
| 595 | from app.incidents.services.case_events import payload_task |
| 596 | |
| 597 | if status_changed: |
| 598 | await emit_case_event( |
| 599 | session=session, |
| 600 | case_id=task.case_id, |
| 601 | event_type=CaseEventType.TASK_STATUS_CHANGED, |
| 602 | actor=actor, |
| 603 | payload=payload_task( |
| 604 | task_id=task.id, |
| 605 | title=task.title, |
| 606 | mandatory=task.mandatory, |
| 607 | from_status=previous_status, |
| 608 | to_status=task.status, |
| 609 | ), |
| 610 | commit=False, |
| 611 | ) |
| 612 | |
| 613 | if comment_set_this_call and request.evidence_comment: |
| 614 | await emit_case_event( |
| 615 | session=session, |
| 616 | case_id=task.case_id, |
| 617 | event_type=CaseEventType.TASK_COMMENTED, |
| 618 | actor=actor, |
| 619 | payload=payload_task( |
| 620 | task_id=task.id, |
| 621 | title=task.title, |
| 622 | mandatory=task.mandatory, |
| 623 | snippet=request.evidence_comment[:140], |
| 624 | ), |
| 625 | commit=False, |
| 626 | ) |
| 627 | |
| 628 | await session.commit() |
| 629 | await session.refresh(task) |
| 630 | |
| 631 | return CaseTaskOperationResponse( |
| 632 | task=_case_task_to_response(task), |
| 633 | success=True, |
| 634 | message=f"Updated case task id={task_id}", |
| 635 | ) |
| 636 | except Exception as e: |
| 637 | logger.error(f"Failed to update case task id={task_id}: {e}") |
| 638 | await session.rollback() |
| 639 | return CaseTaskOperationResponse( |
| 640 | task=None, |
| 641 | success=False, |
| 642 | message=f"Failed to update case task: {e}", |
| 643 | ) |
| 644 | |
| 645 | |
| 646 | async def delete_case_task( |
| 647 | task_id: int, |
| 648 | session: AsyncSession, |
| 649 | ) -> CaseTaskOperationResponse: |
| 650 | """ |
| 651 | Delete a case task. Allowed against template-derived tasks too — |
| 652 | if the analyst genuinely doesn't want the task tracked, deletion |
| 653 | is more honest than NOT_NECESSARY (which is reserved for |
| 654 | "intentionally skipped during this investigation"). |
| 655 | """ |
| 656 | try: |
| 657 | result = await session.execute(select(CaseTask).where(CaseTask.id == task_id)) |
| 658 | task = result.scalar_one_or_none() |
| 659 | if task is None: |
| 660 | return CaseTaskOperationResponse( |
| 661 | task=None, |
| 662 | success=False, |
| 663 | message=f"Case task id={task_id} not found", |
| 664 | ) |
| 665 | snapshot = _case_task_to_response(task) |
| 666 | await session.delete(task) |
| 667 | await session.commit() |
| 668 | return CaseTaskOperationResponse( |
| 669 | task=snapshot, |
| 670 | success=True, |
| 671 | message=f"Deleted case task id={task_id}", |
| 672 | ) |
| 673 | except Exception as e: |
| 674 | logger.error(f"Failed to delete case task id={task_id}: {e}") |
| 675 | await session.rollback() |
| 676 | return CaseTaskOperationResponse( |
| 677 | task=None, |
| 678 | success=False, |
| 679 | message=f"Failed to delete case task: {e}", |
| 680 | ) |
| 681 | |
| 682 | |
| 683 | # --------------------------------------------------------------------------- |
| 684 | # Soft-warning support |
| 685 | # --------------------------------------------------------------------------- |
| 686 | |
| 687 | |
| 688 | async def get_incomplete_mandatory_tasks( |
| 689 | case_id: int, |
| 690 | session: AsyncSession, |
| 691 | ) -> List[CaseTask]: |
| 692 | """ |
| 693 | Return mandatory tasks on the named case whose status is not DONE. |
| 694 | Used by the close-case route to drive the soft-warning response. |
| 695 | |
| 696 | Note: NOT_NECESSARY is never a valid status for a mandatory task |
| 697 | (enforced in update_case_task), so the only "completed" terminal |
| 698 | state for a mandatory task is DONE. |
| 699 | """ |
| 700 | stmt = ( |
| 701 | select(CaseTask) |
| 702 | .where(CaseTask.case_id == case_id) |
| 703 | .where(CaseTask.mandatory == True) # noqa: E712 |
| 704 | .where(CaseTask.status != CaseTaskStatus.DONE.value) |
| 705 | .order_by(CaseTask.order_index, CaseTask.id) |
| 706 | ) |
| 707 | result = await session.execute(stmt) |
| 708 | return list(result.scalars().all()) |
| 709 | |
| 710 | |
| 711 | def build_close_warning_response(incomplete: List[CaseTask]) -> CaseCloseWarningResponse: |
| 712 | return CaseCloseWarningResponse( |
| 713 | success=False, |
| 714 | requires_confirmation=True, |
| 715 | message=(f"{len(incomplete)} mandatory task(s) are not marked DONE. " "Re-submit with force=true to close anyway."), |
| 716 | incomplete_mandatory_tasks=[_case_task_to_response(t) for t in incomplete], |
| 717 | ) |
| 718 | |
| 719 | |
| 720 | # --------------------------------------------------------------------------- |
| 721 | # Convenience: derive first-linked-alert source for create_case path |
| 722 | # --------------------------------------------------------------------------- |
| 723 | |
| 724 | |
| 725 | async def orphan_tasks_for_alert( |
| 726 | case_id: int, |
| 727 | alert_id: int, |
| 728 | session: AsyncSession, |
| 729 | *, |
| 730 | commit: bool = True, |
| 731 | ) -> int: |
| 732 | """ |
| 733 | Set ``alert_id = NULL`` on every CaseTask in this case that was attached |
| 734 | to the unlinked alert. Tasks survive (snapshot-preserving semantics) and |
| 735 | become case-wide / general tasks. |
| 736 | |
| 737 | Returns the number of rows affected so the caller can include it in the |
| 738 | timeline payload for the unlink event. |
| 739 | """ |
| 740 | stmt = ( |
| 741 | update(CaseTask) |
| 742 | .where(CaseTask.case_id == case_id) |
| 743 | .where(CaseTask.alert_id == alert_id) |
| 744 | .values(alert_id=None, updated_at=datetime.utcnow()) |
| 745 | ) |
| 746 | result = await session.execute(stmt) |
| 747 | affected = result.rowcount or 0 |
| 748 | if commit: |
| 749 | await session.commit() |
| 750 | if affected: |
| 751 | logger.info( |
| 752 | f"Orphaned {affected} task(s) on case id={case_id} when alert id={alert_id} was unlinked", |
| 753 | ) |
| 754 | return affected |
| 755 | |
| 756 | |
| 757 | async def get_first_alert_source_for_case( |
| 758 | case_id: int, |
| 759 | session: AsyncSession, |
| 760 | ) -> Optional[str]: |
| 761 | """ |
| 762 | Return the alert.source of the lowest-id alert linked to the case, |
| 763 | or None if no alerts are linked. Used by the manual create_case path |
| 764 | (which doesn't naturally know the source) AFTER the analyst links |
| 765 | an alert and wants to apply a template. |
| 766 | """ |
| 767 | stmt = ( |
| 768 | select(Alert.source) |
| 769 | .join(CaseAlertLink, CaseAlertLink.alert_id == Alert.id) |
| 770 | .where(CaseAlertLink.case_id == case_id) |
| 771 | .order_by(Alert.id.asc()) |
| 772 | .limit(1) |
| 773 | ) |
| 774 | result = await session.execute(stmt) |
| 775 | return result.scalar_one_or_none() |