a8b4518485
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
Backend:
- TechniqueOwnership model: per-technique owner, backup owner, team
- RevalidationQueueItem model: prioritised analyst work queue
(critical/high/medium/low, reasons: validation_expired/infra_change/
osint_alert/mitre_update/rule_modified/low_confidence/manual)
- Migration b035ownerq: creates technique_ownerships and
revalidation_queue_items tables with full indexes
Services:
- ownership_service: set/get technique ownership, bulk assign by tactic
or platform, orphan reports for techniques and assets
- revalidation_queue_service: smart queue generation (scans expired
validations, low-confidence techniques, recent infra changes),
list/create/update queue items, analyst dashboard
Router /api/v1/ownership:
GET/PUT /ownership/techniques/{id} — technique ownership
PATCH /ownership/assets/{id} — asset ownership
GET /ownership/orphans/techniques — orphan report
GET /ownership/orphans/assets — orphan report
POST /ownership/bulk-assign — bulk by tactic/platform
GET/POST /ownership/queue — revalidation queue CRUD
PATCH /ownership/queue/{id} — update item status/assignee
POST /ownership/queue/generate — scan & generate items
GET /ownership/analyst-dashboard — personalised daily view
Scheduler: queue_generation job daily at 02:30 (after decay engine)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
"""Phase 9: Ownership & Revalidation Queue models."""
|
|
|
|
import enum
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import Column, DateTime, Enum, ForeignKey, Index, String, Text
|
|
from sqlalchemy.dialects.postgresql import UUID, JSONB
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from app.database import Base
|
|
|
|
|
|
class QueuePriority(str, enum.Enum):
|
|
critical = "critical"
|
|
high = "high"
|
|
medium = "medium"
|
|
low = "low"
|
|
|
|
|
|
class QueueStatus(str, enum.Enum):
|
|
pending = "pending"
|
|
in_progress = "in_progress"
|
|
completed = "completed"
|
|
dismissed = "dismissed"
|
|
|
|
|
|
class QueueReason(str, enum.Enum):
|
|
validation_expired = "validation_expired"
|
|
infra_change = "infra_change"
|
|
osint_alert = "osint_alert"
|
|
mitre_update = "mitre_update"
|
|
rule_modified = "rule_modified"
|
|
low_confidence = "low_confidence"
|
|
manual = "manual"
|
|
|
|
|
|
class TechniqueOwnership(Base):
|
|
"""Ownership assignment for a MITRE technique."""
|
|
|
|
__tablename__ = "technique_ownerships"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
technique_id = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("techniques.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
unique=True,
|
|
)
|
|
owner_id = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
backup_owner_id = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
team = Column(String(200), nullable=True)
|
|
notes = Column(Text, nullable=True)
|
|
assigned_at = Column(DateTime, nullable=True)
|
|
assigned_by = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
technique = relationship("Technique", foreign_keys=[technique_id])
|
|
owner = relationship("User", foreign_keys=[owner_id])
|
|
backup_owner = relationship("User", foreign_keys=[backup_owner_id])
|
|
|
|
|
|
class RevalidationQueueItem(Base):
|
|
"""A prioritised work item for the analyst's daily queue."""
|
|
|
|
__tablename__ = "revalidation_queue_items"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
technique_id = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("techniques.id", ondelete="CASCADE"),
|
|
nullable=True,
|
|
)
|
|
detection_asset_id = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("detection_assets.id", ondelete="CASCADE"),
|
|
nullable=True,
|
|
)
|
|
|
|
priority = Column(
|
|
Enum(QueuePriority, name="queue_priority"),
|
|
nullable=False,
|
|
default=QueuePriority.medium,
|
|
)
|
|
reason = Column(
|
|
Enum(QueueReason, name="queue_reason"),
|
|
nullable=False,
|
|
)
|
|
reason_detail = Column(Text, nullable=True)
|
|
status = Column(
|
|
Enum(QueueStatus, name="queue_status"),
|
|
nullable=False,
|
|
default=QueueStatus.pending,
|
|
)
|
|
|
|
assigned_to = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
due_date = Column(DateTime, nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
completed_at = Column(DateTime, nullable=True)
|
|
dismissed_at = Column(DateTime, nullable=True)
|
|
completed_by = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
extra = Column(JSONB, nullable=True) # arbitrary metadata
|
|
|
|
technique = relationship("Technique", foreign_keys=[technique_id])
|
|
detection_asset = relationship("DetectionAsset", foreign_keys=[detection_asset_id])
|
|
assignee = relationship("User", foreign_keys=[assigned_to])
|
|
|
|
|
|
# Indexes
|
|
Index("ix_rqueue_status", RevalidationQueueItem.status)
|
|
Index("ix_rqueue_priority", RevalidationQueueItem.priority)
|
|
Index("ix_rqueue_assigned_to", RevalidationQueueItem.assigned_to)
|
|
Index("ix_rqueue_technique_id", RevalidationQueueItem.technique_id)
|
|
Index("ix_rqueue_asset_id", RevalidationQueueItem.detection_asset_id)
|
|
Index("ix_techown_owner_id", TechniqueOwnership.owner_id)
|