feat(phase-26): add Campaign models, endpoints, service with kill chain timeline UI (T-217 to T-220)

2026-02-09 16:52:52 +01:00
parent f4de12d8ab
commit 57b47c296d
12 changed files with 2032 additions and 0 deletions
--- a/backend/app/models/campaign.py
+++ b/backend/app/models/campaign.py
@@ -0,0 +1,132 @@
+"""Campaign and CampaignTest models.
+
+Campaigns group multiple tests into a kill chain sequence,
+enabling simulation of complete attack chains and APT emulations.
+"""
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import (
+    Column, String, Text, Integer, DateTime,
+    ForeignKey, Index,
+)
+from sqlalchemy.dialects.postgresql import UUID, JSONB
+from sqlalchemy.orm import relationship
+
+from app.database import Base
+
+
+class Campaign(Base):
+    """
+    A campaign groups multiple tests into a sequenced attack chain.
+
+    Types:
+    - custom: manually created campaign
+    - apt_emulation: generated from a threat actor profile
+    - kill_chain: structured around kill chain phases
+    - compliance: targeting specific compliance requirements
+
+    Status:
+    - draft: being configured
+    - active: tests are being executed
+    - completed: all tests done
+    - archived: historical record
+    """
+    __tablename__ = "campaigns"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    name = Column(String, nullable=False)
+    description = Column(Text, nullable=True)
+    type = Column(String, nullable=False, default="custom")       # custom, apt_emulation, kill_chain, compliance
+    threat_actor_id = Column(
+        UUID(as_uuid=True),
+        ForeignKey("threat_actors.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    status = Column(String, nullable=False, default="draft")      # draft, active, completed, archived
+    created_by = Column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    scheduled_at = Column(DateTime, nullable=True)
+    completed_at = Column(DateTime, nullable=True)
+    target_platform = Column(String, nullable=True)
+    tags = Column(JSONB, nullable=True, default=[])
+    created_at = Column(DateTime, default=datetime.utcnow)
+
+    # Relationships
+    threat_actor = relationship("ThreatActor")
+    creator = relationship("User", foreign_keys=[created_by])
+    campaign_tests = relationship(
+        "CampaignTest",
+        back_populates="campaign",
+        cascade="all, delete-orphan",
+        order_by="CampaignTest.order_index",
+    )
+
+    __table_args__ = (
+        Index('ix_campaigns_status', 'status'),
+        Index('ix_campaigns_type', 'type'),
+        Index('ix_campaigns_threat_actor', 'threat_actor_id'),
+        Index('ix_campaigns_created_by', 'created_by'),
+    )
+
+
+# Kill chain phases in order (for sorting and validation)
+KILL_CHAIN_PHASES = [
+    "reconnaissance",
+    "resource_development",
+    "initial_access",
+    "execution",
+    "persistence",
+    "privilege_escalation",
+    "defense_evasion",
+    "credential_access",
+    "discovery",
+    "lateral_movement",
+    "collection",
+    "command_and_control",
+    "exfiltration",
+    "impact",
+]
+
+
+class CampaignTest(Base):
+    """
+    A test within a campaign, with ordering and dependency information.
+
+    ``depends_on`` creates a self-referential chain (A -> B -> C).
+    Circular dependencies are validated at the service layer.
+    """
+    __tablename__ = "campaign_tests"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    campaign_id = Column(
+        UUID(as_uuid=True),
+        ForeignKey("campaigns.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    test_id = Column(
+        UUID(as_uuid=True),
+        ForeignKey("tests.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    order_index = Column(Integer, nullable=False, default=0)
+    depends_on = Column(
+        UUID(as_uuid=True),
+        ForeignKey("campaign_tests.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    phase = Column(String, nullable=True)   # kill chain phase
+
+    # Relationships
+    campaign = relationship("Campaign", back_populates="campaign_tests")
+    test = relationship("Test")
+    dependency = relationship("CampaignTest", remote_side="CampaignTest.id")
+
+    __table_args__ = (
+        Index('ix_campaign_tests_campaign', 'campaign_id'),
+        Index('ix_campaign_tests_test', 'test_id'),
+    )