d4b147da7c
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
AlertRule + AlertInstance models (b041alerts migration), 8 pre-seeded system rules (high_risk x2, stale_technique, coverage_regression, low_coverage, expiry_wave, new_technique, orphan_spike), evaluation engine with per-rule cooldown, full alert lifecycle (acknowledge/resolve/dismiss), custom rule CRUD, and summary endpoint. Rules seeded at app startup. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
145 lines
5.4 KiB
Python
145 lines
5.4 KiB
Python
"""Phase 13: Operational Alerts — AlertRule and AlertInstance models."""
|
|
|
|
import enum
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import (
|
|
Boolean, Column, DateTime, ForeignKey,
|
|
Index, Integer, String, Text,
|
|
)
|
|
from sqlalchemy.dialects.postgresql import JSONB, UUID
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from app.database import Base
|
|
|
|
|
|
# ── Enumerations ──────────────────────────────────────────────────────────────
|
|
|
|
class AlertSeverity(str, enum.Enum):
|
|
critical = "critical"
|
|
high = "high"
|
|
medium = "medium"
|
|
low = "low"
|
|
info = "info"
|
|
|
|
|
|
class AlertStatus(str, enum.Enum):
|
|
open = "open"
|
|
acknowledged = "acknowledged"
|
|
resolved = "resolved"
|
|
dismissed = "dismissed"
|
|
|
|
|
|
class AlertRuleType(str, enum.Enum):
|
|
high_risk = "high_risk" # risk_score >= threshold
|
|
stale_technique = "stale_technique" # not validated in N days
|
|
coverage_regression = "coverage_regression" # coverage_pct dropped
|
|
low_coverage = "low_coverage" # coverage below min
|
|
expiry_wave = "expiry_wave" # many pending queue items
|
|
new_technique = "new_technique" # new MITRE techniques added
|
|
orphan_spike = "orphan_spike" # many unowned techniques
|
|
custom = "custom" # future extension placeholder
|
|
|
|
|
|
# ── AlertRule ─────────────────────────────────────────────────────────────────
|
|
|
|
class AlertRule(Base):
|
|
"""
|
|
Defines a condition that, when satisfied, fires an AlertInstance.
|
|
|
|
System rules (is_system=True) are seeded at startup and cannot be deleted.
|
|
Custom rules (is_system=False) can be created by admins.
|
|
"""
|
|
|
|
__tablename__ = "alert_rules"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
name = Column(String(300), nullable=False)
|
|
description = Column(Text, nullable=True)
|
|
rule_type = Column(String(50), nullable=False)
|
|
severity = Column(String(20), nullable=False, default=AlertSeverity.medium.value)
|
|
is_enabled = Column(Boolean, nullable=False, default=True)
|
|
is_system = Column(Boolean, nullable=False, default=False) # seeded, not deletable
|
|
|
|
# Rule-specific thresholds/config (varies by rule_type)
|
|
config = Column(JSONB, nullable=False, default={})
|
|
|
|
# Delivery
|
|
notify_in_app = Column(Boolean, nullable=False, default=True)
|
|
notify_webhook = Column(Boolean, nullable=False, default=False)
|
|
webhook_id = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("webhook_configs.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
|
|
# Cooldown — don't re-fire within N hours of last firing
|
|
cooldown_hours = Column(Integer, nullable=False, default=24)
|
|
|
|
# Meta
|
|
created_by = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
last_fired_at = Column(DateTime, nullable=True)
|
|
|
|
creator = relationship("User", foreign_keys=[created_by])
|
|
instances = relationship("AlertInstance", back_populates="rule",
|
|
cascade="all, delete-orphan")
|
|
|
|
__table_args__ = (
|
|
Index("ix_alert_rules_type", "rule_type"),
|
|
Index("ix_alert_rules_enabled", "is_enabled"),
|
|
)
|
|
|
|
|
|
# ── AlertInstance ─────────────────────────────────────────────────────────────
|
|
|
|
class AlertInstance(Base):
|
|
"""
|
|
A single firing of an AlertRule.
|
|
|
|
Transitions: open → acknowledged → resolved
|
|
open → dismissed
|
|
"""
|
|
|
|
__tablename__ = "alert_instances"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
rule_id = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("alert_rules.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
# Denormalised fields kept for history even after rule deletion
|
|
rule_name = Column(String(300), nullable=False)
|
|
rule_type = Column(String(50), nullable=False)
|
|
severity = Column(String(20), nullable=False)
|
|
|
|
title = Column(String(500), nullable=False)
|
|
message = Column(Text, nullable=False)
|
|
details = Column(JSONB, nullable=True) # structured context
|
|
|
|
status = Column(String(20), nullable=False, default=AlertStatus.open.value)
|
|
acknowledged_by = Column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
)
|
|
acknowledged_at = Column(DateTime, nullable=True)
|
|
resolved_at = Column(DateTime, nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
rule = relationship("AlertRule", back_populates="instances")
|
|
acknowledger = relationship("User", foreign_keys=[acknowledged_by])
|
|
|
|
__table_args__ = (
|
|
Index("ix_alert_instances_rule_id", "rule_id"),
|
|
Index("ix_alert_instances_status", "status"),
|
|
Index("ix_alert_instances_severity", "severity"),
|
|
Index("ix_alert_instances_created", "created_at"),
|
|
)
|