feat(phase-31): add campaign scheduling and recurring automation (T-233 to T-234)

backend/app/services/campaign_scheduler_service.py

@@ -0,0 +1,193 @@
+"""Campaign scheduler service — recurring campaign execution.
+
+Handles checking which recurring campaigns are due, cloning them with
+fresh tests, and computing the next run date.
+"""
+
+import logging
+import uuid
+from datetime import datetime, timedelta
+
+from sqlalchemy.orm import Session
+
+from app.models.campaign import Campaign, CampaignTest
+from app.models.test import Test
+from app.models.enums import TestState
+from app.services.notification_service import create_notification
+from app.services.audit_service import log_action
+from app.models.user import User
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Next-run calculation
+# ---------------------------------------------------------------------------
+
+
+def calculate_next_run(current_date: datetime, pattern: str) -> datetime:
+    """Compute the next run date from *current_date* and a recurrence pattern.
+
+    Supported patterns:
+    - ``weekly``  : +7 days
+    - ``monthly`` : +30 days
+    - ``quarterly``: +90 days
+    """
+    offsets = {
+        "weekly": timedelta(days=7),
+        "monthly": timedelta(days=30),
+        "quarterly": timedelta(days=90),
+    }
+    return current_date + offsets.get(pattern, timedelta(days=30))
+
+
+# ---------------------------------------------------------------------------
+# Clone a campaign
+# ---------------------------------------------------------------------------
+
+
+def _clone_campaign(db: Session, original: Campaign) -> Campaign:
+    """Create a new child campaign from a recurring template.
+
+    1. Clone the campaign with a date-stamped name.
+    2. For each ``CampaignTest`` in the original, create a new ``Test``
+       with the same base data (in ``draft`` state) and link it.
+    3. Activate the new campaign.
+    """
+    now = datetime.utcnow()
+    run_label = now.strftime("%Y-%m-%d")
+
+    child = Campaign(
+        name=f"{original.name} (Run {run_label})",
+        description=original.description,
+        type=original.type,
+        threat_actor_id=original.threat_actor_id,
+        status="active",
+        created_by=original.created_by,
+        target_platform=original.target_platform,
+        tags=original.tags or [],
+        parent_campaign_id=original.id,
+    )
+    db.add(child)
+    db.flush()  # get child.id
+
+    # Clone each campaign_test with a fresh Test
+    original_cts = (
+        db.query(CampaignTest)
+        .filter(CampaignTest.campaign_id == original.id)
+        .order_by(CampaignTest.order_index)
+        .all()
+    )
+
+    for ct in original_cts:
+        src_test = ct.test
+        if not src_test:
+            continue
+
+        new_test = Test(
+            technique_id=src_test.technique_id,
+            name=src_test.name,
+            description=src_test.description,
+            platform=src_test.platform,
+            procedure_text=src_test.procedure_text,
+            tool_used=src_test.tool_used,
+            created_by=original.created_by,
+            state=TestState.draft,
+        )
+        db.add(new_test)
+        db.flush()  # get new_test.id
+
+        new_ct = CampaignTest(
+            campaign_id=child.id,
+            test_id=new_test.id,
+            order_index=ct.order_index,
+            phase=ct.phase,
+            # depends_on is not copied — would need ID remapping
+        )
+        db.add(new_ct)
+
+    db.flush()
+    return child
+
+
+# ---------------------------------------------------------------------------
+# Check and run recurring campaigns (daily job)
+# ---------------------------------------------------------------------------
+
+
+def check_and_run_recurring_campaigns(db: Session) -> int:
+    """Check all recurring campaigns and clone any that are due.
+
+    Returns the number of campaigns spawned.
+    """
+    now = datetime.utcnow()
+
+    due_campaigns = (
+        db.query(Campaign)
+        .filter(
+            Campaign.is_recurring == True,  # noqa: E712
+            Campaign.next_run_at <= now,
+        )
+        .all()
+    )
+
+    spawned = 0
+
+    for campaign in due_campaigns:
+        try:
+            child = _clone_campaign(db, campaign)
+
+            # Update the original's scheduling fields
+            campaign.last_run_at = now
+            campaign.next_run_at = calculate_next_run(now, campaign.recurrence_pattern or "monthly")
+
+            db.commit()
+            db.refresh(child)
+
+            # Audit
+            log_action(
+                db,
+                user_id=campaign.created_by,
+                action="recurring_campaign_run",
+                entity_type="campaign",
+                entity_id=child.id,
+                details={
+                    "parent_campaign_id": str(campaign.id),
+                    "child_campaign_name": child.name,
+                    "pattern": campaign.recurrence_pattern,
+                },
+            )
+
+            # Notify
+            if campaign.created_by:
+                create_notification(
+                    db,
+                    user_id=campaign.created_by,
+                    type="recurring_campaign_run",
+                    title="Recurring campaign executed",
+                    message=f'Campaign "{child.name}" was automatically created from recurring template "{campaign.name}".',
+                    entity_type="campaign",
+                    entity_id=child.id,
+                )
+
+            # Notify red_tech users
+            red_techs = db.query(User).filter(User.role == "red_tech", User.is_active == True).all()  # noqa: E712
+            for user in red_techs:
+                create_notification(
+                    db,
+                    user_id=user.id,
+                    type="campaign_activated",
+                    title="New recurring campaign active",
+                    message=f'Campaign "{child.name}" is now active and ready for execution.',
+                    entity_type="campaign",
+                    entity_id=child.id,
+                )
+
+            spawned += 1
+            logger.info("Spawned child campaign '%s' from parent '%s'", child.name, campaign.name)
+
+        except Exception:
+            db.rollback()
+            logger.exception("Failed to run recurring campaign '%s'", campaign.name)
+
+    return spawned