"""Phase 9: Revalidation Queue service — generation, management, analyst dashboard.""" import logging from datetime import datetime, timedelta from typing import Optional from uuid import UUID from sqlalchemy.orm import Session from app.models.ownership_queue import ( RevalidationQueueItem, QueuePriority, QueueStatus, QueueReason, TechniqueOwnership, ) from app.models.detection_lifecycle import ( DetectionAsset, DetectionValidation, TechniqueConfidenceScore, InfrastructureChangeLog, DetectionConfidence, ) from app.models.technique import Technique from app.domain.exceptions import EntityNotFoundError logger = logging.getLogger(__name__) def _now() -> datetime: return datetime.utcnow() # ── Queue Generation ────────────────────────────────────────────────────────── def generate_queue_items(db: Session) -> dict: """ Scan the system and create new revalidation queue items for: - Assets with no valid detection validation (validation_expired) - Low-confidence techniques (low_confidence) - Recent infrastructure changes that affected assets (infra_change) Skips already-pending/in_progress items for the same asset+reason. Returns counts of created and skipped items. """ created = 0 skipped = 0 now = _now() def _has_active_item(technique_id=None, asset_id=None, reason=None) -> bool: q = db.query(RevalidationQueueItem).filter( RevalidationQueueItem.status.in_(["pending", "in_progress"]) ) if technique_id: q = q.filter(RevalidationQueueItem.technique_id == technique_id) if asset_id: q = q.filter(RevalidationQueueItem.detection_asset_id == asset_id) if reason: q = q.filter(RevalidationQueueItem.reason == reason) return q.first() is not None # 1) Assets with no active valid validation active_assets = db.query(DetectionAsset).filter(DetectionAsset.is_active == True).all() for asset in active_assets: has_valid = db.query(DetectionValidation).filter( DetectionValidation.detection_asset_id == asset.id, DetectionValidation.is_valid == True, DetectionValidation.expires_at > now, ).first() if has_valid: continue if _has_active_item(asset_id=asset.id, reason=QueueReason.validation_expired): skipped += 1 continue # Determine priority: assets not validated for >30 days are high last_val = db.query(DetectionValidation).filter( DetectionValidation.detection_asset_id == asset.id, ).order_by(DetectionValidation.validated_at.desc()).first() if last_val is None: priority = QueuePriority.high detail = "Asset has never been validated" else: days_stale = (now - last_val.validated_at).days if days_stale > 60: priority = QueuePriority.critical elif days_stale > 30: priority = QueuePriority.high else: priority = QueuePriority.medium detail = f"Last validation expired {days_stale} days ago" item = RevalidationQueueItem( detection_asset_id=asset.id, priority=priority, reason=QueueReason.validation_expired, reason_detail=detail, due_date=now + timedelta(days=7), ) db.add(item) created += 1 # 2) Low-confidence techniques (stale or broken) low_scores = db.query(TechniqueConfidenceScore).filter( TechniqueConfidenceScore.confidence_level.in_([ DetectionConfidence.stale, DetectionConfidence.broken, ]) ).all() for score in low_scores: if _has_active_item(technique_id=score.technique_id, reason=QueueReason.low_confidence): skipped += 1 continue priority = ( QueuePriority.critical if score.confidence_level == DetectionConfidence.broken else QueuePriority.high ) detail = ( f"Technique confidence {score.confidence_level.value} " f"(score={score.confidence_score}). " f"Risk: {', '.join(score.risk_factors or [])}" ) item = RevalidationQueueItem( technique_id=score.technique_id, priority=priority, reason=QueueReason.low_confidence, reason_detail=detail, due_date=now + timedelta(days=14), ) db.add(item) created += 1 # 3) Recent infrastructure changes (last 7 days) → one item per change recent_changes = db.query(InfrastructureChangeLog).filter( InfrastructureChangeLog.change_date >= now - timedelta(days=7), InfrastructureChangeLog.auto_invalidate == True, ).all() for change in recent_changes: if _has_active_item(reason=QueueReason.infra_change) and False: # Don't deduplicate infra changes — each change is separate pass existing = db.query(RevalidationQueueItem).filter( RevalidationQueueItem.reason == QueueReason.infra_change, RevalidationQueueItem.status.in_(["pending", "in_progress"]), RevalidationQueueItem.extra["change_id"].astext == str(change.id), ).first() if existing: skipped += 1 continue item = RevalidationQueueItem( priority=QueuePriority.high, reason=QueueReason.infra_change, reason_detail=( f"{change.change_type}: {change.description[:120]} " f"({change.invalidated_count or 0} validations invalidated)" ), due_date=now + timedelta(days=3), extra={"change_id": str(change.id), "change_type": change.change_type, "affected_platforms": change.affected_platforms or []}, ) db.add(item) created += 1 if created > 0: db.commit() logger.info("Queue generation: created=%d skipped=%d", created, skipped) return {"created": created, "skipped": skipped} # ── Queue CRUD ──────────────────────────────────────────────────────────────── def list_queue( db: Session, status: Optional[str] = None, priority: Optional[str] = None, reason: Optional[str] = None, assigned_to: Optional[UUID] = None, technique_id: Optional[UUID] = None, detection_asset_id: Optional[UUID] = None, limit: int = 100, offset: int = 0, ) -> list[RevalidationQueueItem]: q = db.query(RevalidationQueueItem) if status: q = q.filter(RevalidationQueueItem.status == status) if priority: q = q.filter(RevalidationQueueItem.priority == priority) if reason: q = q.filter(RevalidationQueueItem.reason == reason) if assigned_to: q = q.filter(RevalidationQueueItem.assigned_to == assigned_to) if technique_id: q = q.filter(RevalidationQueueItem.technique_id == technique_id) if detection_asset_id: q = q.filter(RevalidationQueueItem.detection_asset_id == detection_asset_id) # Priority order: critical > high > medium > low priority_order = { "critical": 0, "high": 1, "medium": 2, "low": 3, } from sqlalchemy import case q = q.order_by( case( {"critical": 0, "high": 1, "medium": 2, "low": 3}, value=RevalidationQueueItem.priority, ), RevalidationQueueItem.due_date.asc().nullslast(), RevalidationQueueItem.created_at.asc(), ) return q.offset(offset).limit(limit).all() def create_queue_item(db: Session, data: dict, user_id: UUID) -> RevalidationQueueItem: item = RevalidationQueueItem( technique_id=data.get("technique_id"), detection_asset_id=data.get("detection_asset_id"), priority=data.get("priority", "medium"), reason=data.get("reason", "manual"), reason_detail=data.get("reason_detail"), assigned_to=data.get("assigned_to"), due_date=data.get("due_date"), extra={"created_by": str(user_id)}, ) db.add(item) db.commit() db.refresh(item) return item def update_queue_item(db: Session, item_id: UUID, data: dict, user_id: UUID) -> RevalidationQueueItem: item = db.query(RevalidationQueueItem).filter(RevalidationQueueItem.id == item_id).first() if not item: raise EntityNotFoundError("RevalidationQueueItem", str(item_id)) now = _now() if "status" in data and data["status"] is not None: new_status = data["status"] item.status = new_status if new_status == "completed": item.completed_at = now item.completed_by = user_id elif new_status == "dismissed": item.dismissed_at = now if "assigned_to" in data: item.assigned_to = data["assigned_to"] if "priority" in data and data["priority"] is not None: item.priority = data["priority"] if "due_date" in data: item.due_date = data["due_date"] db.commit() db.refresh(item) return item # ── Analyst Dashboard ───────────────────────────────────────────────────────── def get_analyst_dashboard(db: Session, user_id: UUID) -> dict: """Return a personalised daily workday view for the logged-in analyst.""" now = _now() # 1) My pending/in_progress queue items (assigned to me) my_items = list_queue(db, status=None, assigned_to=user_id, limit=50) my_items = [i for i in my_items if i.status in ("pending", "in_progress")] # Also include unassigned items where I'm the technique/asset owner owned_tech_ids = [ row.technique_id for row in db.query(TechniqueOwnership.technique_id).filter( (TechniqueOwnership.owner_id == user_id) | (TechniqueOwnership.backup_owner_id == user_id) ).all() ] owned_asset_ids = [ row.id for row in db.query(DetectionAsset.id).filter( DetectionAsset.is_active == True, (DetectionAsset.owner_id == user_id) | (DetectionAsset.backup_owner_id == user_id), ).all() ] # Unassigned items for my techniques/assets if owned_tech_ids or owned_asset_ids: from sqlalchemy import or_ unassigned_q = db.query(RevalidationQueueItem).filter( RevalidationQueueItem.status.in_(["pending", "in_progress"]), RevalidationQueueItem.assigned_to.is_(None), ) filters = [] if owned_tech_ids: filters.append(RevalidationQueueItem.technique_id.in_(owned_tech_ids)) if owned_asset_ids: filters.append(RevalidationQueueItem.detection_asset_id.in_(owned_asset_ids)) unassigned_q = unassigned_q.filter(or_(*filters)) unassigned_items = unassigned_q.limit(20).all() # Merge, deduplicate by id seen = {i.id for i in my_items} for item in unassigned_items: if item.id not in seen: my_items.append(item) seen.add(item.id) # 2) Validations expiring in next 7 days on assets I own expiring = [] if owned_asset_ids: expiring_vals = db.query(DetectionValidation).filter( DetectionValidation.detection_asset_id.in_(owned_asset_ids), DetectionValidation.is_valid == True, DetectionValidation.expires_at <= now + timedelta(days=7), DetectionValidation.expires_at > now, ).order_by(DetectionValidation.expires_at.asc()).limit(20).all() for v in expiring_vals: asset = db.query(DetectionAsset).filter(DetectionAsset.id == v.detection_asset_id).first() expiring.append({ "validation_id": str(v.id), "asset_id": str(v.detection_asset_id), "asset_name": asset.name if asset else None, "expires_at": v.expires_at.isoformat() if v.expires_at else None, "days_until_expiry": (v.expires_at - now).days if v.expires_at else None, }) # 3) Recent infra changes (last 7 days) recent_changes = db.query(InfrastructureChangeLog).filter( InfrastructureChangeLog.change_date >= now - timedelta(days=7), ).order_by(InfrastructureChangeLog.change_date.desc()).limit(5).all() infra_list = [ { "change_id": str(c.id), "change_type": c.change_type, "description": c.description, "affected_platforms": c.affected_platforms or [], "invalidated_count": c.invalidated_count or 0, "change_date": c.change_date.isoformat() if c.change_date else None, } for c in recent_changes ] # 4) My techniques with low confidence low_confidence = [] if owned_tech_ids: low_scores = db.query(TechniqueConfidenceScore).filter( TechniqueConfidenceScore.technique_id.in_(owned_tech_ids), TechniqueConfidenceScore.confidence_score < 50, ).order_by(TechniqueConfidenceScore.confidence_score.asc()).limit(10).all() for score in low_scores: tech = db.query(Technique).filter(Technique.id == score.technique_id).first() low_confidence.append({ "technique_id": str(score.technique_id), "mitre_id": tech.mitre_id if tech else None, "name": tech.name if tech else None, "confidence_level": score.confidence_level.value if score.confidence_level else None, "confidence_score": score.confidence_score, "risk_factors": score.risk_factors or [], }) # Summary total_pending = db.query(RevalidationQueueItem).filter( RevalidationQueueItem.status == QueueStatus.pending ).count() total_critical = db.query(RevalidationQueueItem).filter( RevalidationQueueItem.status.in_(["pending", "in_progress"]), RevalidationQueueItem.priority == QueuePriority.critical, ).count() return { "my_pending_items": my_items, "expiring_validations_7d": expiring, "recent_infra_changes": infra_list, "my_low_confidence_techniques": low_confidence, "summary": { "my_assigned_items": len([i for i in my_items if i.assigned_to == user_id]), "my_owned_techniques": len(owned_tech_ids), "my_owned_assets": len(owned_asset_ids), "total_pending_system": total_pending, "critical_system": total_critical, "expiring_soon": len(expiring), }, }