# Campaign Manager Approval Workflow Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Add a `manager` role that approves campaigns leads build, fixes their launch dates, and reviews justified requests to add/remove tests from an already-active campaign — with every action logged to a queryable timeline. **Architecture:** Campaign gains a `pending_approval` status between `draft` and `active`. Leads submit drafts into that queue; a manager (or admin) approves (setting `start_date`) or rejects (with a mandatory reason, returning it to `draft`). Once `active`, leads can no longer edit the test list directly — they file a `CampaignModificationRequest` (with mandatory justification) that a manager must approve before the underlying test is actually added/removed. All of it reuses the existing generic `AuditLog` (new action names) and exposes a `GET /campaigns/{id}/timeline` endpoint, mirroring the one that already exists for tests. **Tech Stack:** FastAPI + SQLAlchemy + Alembic (backend), React 19 + TypeScript + TanStack Query (frontend), pytest + in-memory SQLite (tests). --- ## Context for the implementing engineer - Read [backend/app/services/campaign_crud_service.py](backend/app/services/campaign_crud_service.py) fully before starting — nearly every backend task edits this file. - Read [backend/app/routers/campaigns.py](backend/app/routers/campaigns.py) fully before starting Task 9 onward. - `require_any_role(*roles)` (in `app/dependencies/auth.py`) already lets `admin` through unconditionally — so `require_any_role("manager")` alone is enough to satisfy "admin can also approve." You never need to write `require_any_role("manager", "admin")`. - Domain errors (`EntityNotFoundError`, `BusinessRuleViolation`, `PermissionViolation`) are raised in the service layer and converted to HTTP 404/400/403 automatically by `app/middleware/error_handler.py` — routers never raise `HTTPException` for these cases, they just let the exception propagate. - Service functions never call `db.commit()` — only `db.flush()`. The router wraps calls in `with UnitOfWork(db) as uow: ... uow.commit()`. Follow this exactly. - Backend tests use the fixtures in `backend/tests/conftest.py` (`client`, `db`, `admin_user`/`admin_token`, `red_lead_user`/`red_lead_headers`, etc.) against a real (in-memory SQLite) FastAPI app — no mocking. Run pytest from the `backend/` directory. - Frontend has no test runner configured. Frontend tasks are verified with `npm run build` (TypeScript check) inside `frontend/` and manual verification in the browser via the dev server. - This plan deliberately leaves `app/services/campaign_scheduler_service.py` (recurring campaign auto-clone, which sets `status="active"` directly) untouched — a recurring child campaign is a re-run of an already-approved template, not a new campaign a lead is proposing, so it doesn't need to re-enter the approval queue. --- ## Task 1: Domain entity — add `pending_approval` state and transition methods **Files:** - Modify: `backend/app/domain/entities/campaign.py` - Test: `backend/tests/test_campaign_entity.py` - [ ] **Step 1: Write the failing tests** Append to `backend/tests/test_campaign_entity.py`: ```python # ── 9. Submit for approval ──────────────────────────────────────────── def test_submit_for_approval_from_draft_with_tests_success(): e = _entity("draft", test_count=1) e.submit_for_approval() assert e.status == CampaignStatus.pending_approval def test_submit_for_approval_from_draft_with_zero_tests_raises(): e = _entity("draft", test_count=0) with pytest.raises(BusinessRuleViolation, match="at least one test"): e.submit_for_approval() assert e.status == CampaignStatus.draft def test_submit_for_approval_from_active_raises(): e = _entity("active", test_count=1) with pytest.raises(InvalidStateTransition): e.submit_for_approval() # ── 10. Approve ──────────────────────────────────────────────────────── def test_approve_from_pending_approval_success(): e = _entity("pending_approval", test_count=1) e.approve() assert e.status == CampaignStatus.active def test_approve_from_draft_raises(): e = _entity("draft", test_count=1) with pytest.raises(InvalidStateTransition): e.approve() # ── 11. Reject ───────────────────────────────────────────────────────── def test_reject_from_pending_approval_returns_to_draft(): e = _entity("pending_approval", test_count=1) e.reject() assert e.status == CampaignStatus.draft def test_reject_from_draft_raises(): e = _entity("draft", test_count=1) with pytest.raises(InvalidStateTransition): e.reject() ``` - [ ] **Step 2: Run tests to verify they fail** Run (from `backend/`): `pytest tests/test_campaign_entity.py -v -k "submit_for_approval or test_approve or test_reject"` Expected: FAIL — `AttributeError: 'CampaignEntity' object has no attribute 'submit_for_approval'` (and `CampaignStatus.pending_approval` doesn't exist yet). - [ ] **Step 3: Implement the state and methods** In `backend/app/domain/entities/campaign.py`, update `CampaignStatus`: ```python class CampaignStatus(str, enum.Enum): """Lifecycle states for a campaign.""" draft = "draft" pending_approval = "pending_approval" active = "active" completed = "completed" archived = "archived" ``` Update `VALID_TRANSITIONS`: ```python VALID_TRANSITIONS: dict[CampaignStatus, list[CampaignStatus]] = { CampaignStatus.draft: [CampaignStatus.pending_approval], CampaignStatus.pending_approval: [CampaignStatus.active, CampaignStatus.draft], CampaignStatus.active: [CampaignStatus.completed], CampaignStatus.completed: [CampaignStatus.archived], CampaignStatus.archived: [], } ``` Replace the `activate` method and add `submit_for_approval`, `approve`, `reject` right after it: ```python def submit_for_approval(self) -> None: """Transition the campaign from ``draft`` to ``pending_approval``. Returns: None """ if not self.can_transition_to(CampaignStatus.pending_approval): raise InvalidStateTransition( self.status.value, CampaignStatus.pending_approval.value, [s.value for s in VALID_TRANSITIONS[self.status]], ) if self.test_count == 0: raise BusinessRuleViolation( "Campaign must have at least one test to submit for approval" ) self.status = CampaignStatus.pending_approval def approve(self) -> None: """Transition the campaign from ``pending_approval`` to ``active``. Returns: None """ if not self.can_transition_to(CampaignStatus.active): raise InvalidStateTransition( self.status.value, CampaignStatus.active.value, [s.value for s in VALID_TRANSITIONS[self.status]], ) self.status = CampaignStatus.active def reject(self) -> None: """Transition the campaign from ``pending_approval`` back to ``draft``. Returns: None """ if not self.can_transition_to(CampaignStatus.draft): raise InvalidStateTransition( self.status.value, CampaignStatus.draft.value, [s.value for s in VALID_TRANSITIONS[self.status]], ) self.status = CampaignStatus.draft ``` Leave the existing `activate` method as-is (it's superseded by `submit_for_approval`/`approve` but removing it would break `test_activate_from_draft_with_tests_success` etc. — out of scope to touch those). Update `ensure_modifiable` so campaigns can still be read while pending approval isn't treated as an error state for callers that only check draft/active — leave this method unchanged; it already raises for anything other than draft/active, which is correct (pending_approval campaigns are frozen). - [ ] **Step 4: Run tests to verify they pass** Run: `pytest tests/test_campaign_entity.py -v` Expected: PASS — all tests including the pre-existing ones (the old `activate`/`ensure_modifiable` tests must still pass unchanged). - [ ] **Step 5: Commit** ```bash git add backend/app/domain/entities/campaign.py backend/tests/test_campaign_entity.py git commit -m "feat(campaigns): add pending_approval state to campaign domain entity" ``` --- ## Task 2: ORM models — Campaign approval fields + CampaignModificationRequest **Files:** - Modify: `backend/app/models/campaign.py` - Modify: `backend/app/models/__init__.py` - [ ] **Step 1: Add approval fields to `Campaign`** In `backend/app/models/campaign.py`, add these columns right after `start_date`/`scheduled_at` (after line 77): ```python approved_by = Column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True, ) approved_at = Column(DateTime, nullable=True) rejection_reason = Column(Text, nullable=True) ``` - [ ] **Step 2: Add the `CampaignModificationRequest` model** Append to the bottom of `backend/app/models/campaign.py`, after the `CampaignTest` class: ```python class CampaignModificationRequest(Base): """A lead's request to add or remove a test from an already-active campaign. The underlying ``CampaignTest`` row is only created/deleted once a manager approves the request — see ``approve_modification_request`` in ``campaign_crud_service.py``. """ __tablename__ = "campaign_modification_requests" 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, ) requested_by = Column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True, ) action = Column(String, nullable=False) # add_test, remove_test test_id = Column( UUID(as_uuid=True), ForeignKey("tests.id", ondelete="CASCADE"), nullable=False, ) order_index = Column(Integer, nullable=True) phase = Column(String, nullable=True) justification = Column(Text, nullable=False) status = Column(String, nullable=False, default="pending") # pending, approved, rejected reviewed_by = Column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True, ) reviewed_at = Column(DateTime, nullable=True) review_notes = Column(Text, nullable=True) created_at = Column(DateTime(timezone=True), server_default=func.now()) campaign = relationship("Campaign") test = relationship("Test") requester = relationship("User", foreign_keys=[requested_by]) reviewer = relationship("User", foreign_keys=[reviewed_by]) __table_args__ = ( Index("ix_campaign_mod_requests_campaign", "campaign_id"), Index("ix_campaign_mod_requests_status", "status"), ) ``` - [ ] **Step 3: Register the new model** In `backend/app/models/__init__.py`, change line 11: ```python from app.models.campaign import Campaign, CampaignTest, CampaignModificationRequest ``` And in `__all__` (line 62), change: ```python "Campaign", "CampaignTest", "CampaignModificationRequest", ``` - [ ] **Step 4: Verify the app still imports cleanly** Run (from `backend/`): `python -c "import app.models; print('ok')"` Expected: prints `ok` with no traceback. - [ ] **Step 5: Commit** ```bash git add backend/app/models/campaign.py backend/app/models/__init__.py git commit -m "feat(campaigns): add approval fields and CampaignModificationRequest model" ``` --- ## Task 3: Migration — campaign approval columns **Files:** - Create: `backend/alembic/versions/b054_add_campaign_approval_fields.py` - [ ] **Step 1: Write the migration** ```python """Add approved_by, approved_at, rejection_reason to campaigns. Revision ID: b054 Revises: b053 Create Date: 2026-07-02 """ from alembic import op import sqlalchemy as sa from sqlalchemy.dialects import postgresql revision = "b054" down_revision = "b053" branch_labels = None depends_on = None def upgrade() -> None: op.add_column( "campaigns", sa.Column("approved_by", postgresql.UUID(as_uuid=True), nullable=True), ) op.add_column("campaigns", sa.Column("approved_at", sa.DateTime(), nullable=True)) op.add_column("campaigns", sa.Column("rejection_reason", sa.Text(), nullable=True)) op.create_foreign_key( "fk_campaigns_approved_by_users", "campaigns", "users", ["approved_by"], ["id"], ondelete="SET NULL", ) def downgrade() -> None: op.drop_constraint("fk_campaigns_approved_by_users", "campaigns", type_="foreignkey") op.drop_column("campaigns", "rejection_reason") op.drop_column("campaigns", "approved_at") op.drop_column("campaigns", "approved_by") ``` - [ ] **Step 2: Verify migration chain is valid** Run (from `backend/`): `alembic heads` Expected: prints `b054 (head)` — confirming no branch conflicts. - [ ] **Step 3: Commit** ```bash git add backend/alembic/versions/b054_add_campaign_approval_fields.py git commit -m "feat(campaigns): migration for campaign approval fields" ``` --- ## Task 4: Migration — campaign_modification_requests table **Files:** - Create: `backend/alembic/versions/b055_add_campaign_modification_requests.py` - [ ] **Step 1: Write the migration** ```python """Add campaign_modification_requests table. Revision ID: b055 Revises: b054 Create Date: 2026-07-02 """ from alembic import op import sqlalchemy as sa from sqlalchemy.dialects import postgresql revision = "b055" down_revision = "b054" branch_labels = None depends_on = None def upgrade() -> None: op.create_table( "campaign_modification_requests", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column( "campaign_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("campaigns.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "requested_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, ), sa.Column("action", sa.String(), nullable=False), sa.Column( "test_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("tests.id", ondelete="CASCADE"), nullable=False, ), sa.Column("order_index", sa.Integer(), nullable=True), sa.Column("phase", sa.String(), nullable=True), sa.Column("justification", sa.Text(), nullable=False), sa.Column("status", sa.String(), nullable=False, server_default="pending"), sa.Column( "reviewed_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, ), sa.Column("reviewed_at", sa.DateTime(), nullable=True), sa.Column("review_notes", sa.Text(), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), ) op.create_index( "ix_campaign_mod_requests_campaign", "campaign_modification_requests", ["campaign_id"], ) op.create_index( "ix_campaign_mod_requests_status", "campaign_modification_requests", ["status"], ) def downgrade() -> None: op.drop_index("ix_campaign_mod_requests_status", table_name="campaign_modification_requests") op.drop_index("ix_campaign_mod_requests_campaign", table_name="campaign_modification_requests") op.drop_table("campaign_modification_requests") ``` - [ ] **Step 2: Verify migration chain is valid** Run (from `backend/`): `alembic heads` Expected: prints `b055 (head)`. - [ ] **Step 3: Commit** ```bash git add backend/alembic/versions/b055_add_campaign_modification_requests.py git commit -m "feat(campaigns): migration for campaign_modification_requests table" ``` --- ## Task 5: Test fixtures — manager role **Files:** - Modify: `backend/tests/conftest.py` - [ ] **Step 1: Add the `manager_user` fixture** Insert right after the `blue_lead_user` fixture (after line 217) in `backend/tests/conftest.py`: ```python @pytest.fixture(scope="function") def manager_user(db): """Create a manager user for testing.""" user = User( username="manager", email="manager@test.com", hashed_password=hash_password("manager123"), role="manager", is_active=True, must_change_password=False, ) db.add(user) db.commit() db.refresh(user) return user ``` - [ ] **Step 2: Add the `manager_token` and `manager_headers` fixtures** Insert right after the `blue_lead_headers` fixture (after line 297): ```python @pytest.fixture(scope="function") def manager_token(client, manager_user): """Get an auth token for the manager user.""" response = client.post( "/api/v1/auth/login", data={"username": "manager", "password": "manager123"}, ) return response.json()["access_token"] @pytest.fixture(scope="function") def manager_headers(manager_token): """Return authorization headers for manager user.""" return {"Authorization": f"Bearer {manager_token}"} ``` - [ ] **Step 3: Verify the fixtures load without error** Run (from `backend/`): `pytest tests/test_user_api_validation.py -v` Expected: PASS (unrelated test file — this just confirms conftest.py still parses and collects cleanly after the edit). - [ ] **Step 4: Commit** ```bash git add backend/tests/conftest.py git commit -m "test(campaigns): add manager role fixtures" ``` --- ## Task 6: Service layer — submit / approve / reject campaign **Files:** - Modify: `backend/app/services/campaign_crud_service.py` - Test: `backend/tests/test_campaign_approval.py` (new file) - [ ] **Step 1: Write the failing tests** Create `backend/tests/test_campaign_approval.py`: ```python """Tests for the campaign manager-approval workflow.""" import pytest from app.models.campaign import Campaign, CampaignTest from app.models.technique import Technique from app.models.test import Test from app.models.enums import TestState from app.domain.errors import BusinessRuleViolation, EntityNotFoundError, PermissionViolation from app.services.campaign_crud_service import ( create_campaign, submit_campaign_for_approval, approve_campaign, reject_campaign, ) @pytest.fixture def draft_campaign_with_test(db, admin_user): """A draft campaign with exactly one test, created by admin_user.""" tech = Technique(mitre_id="T1059", name="Command Line", tactic="execution", platforms=["windows"]) db.add(tech) db.flush() campaign = Campaign(name="Approval Test Campaign", type="custom", status="draft", created_by=admin_user.id) db.add(campaign) db.flush() test = Test(technique_id=tech.id, name="T1059 test", state=TestState.draft, created_by=admin_user.id) db.add(test) db.flush() ct = CampaignTest(campaign_id=campaign.id, test_id=test.id, order_index=0) db.add(ct) db.commit() db.refresh(campaign) return campaign class TestSubmitCampaignForApproval: def test_submit_moves_draft_to_pending_approval(self, db, draft_campaign_with_test, admin_user): campaign = submit_campaign_for_approval( db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin", ) assert campaign.status == "pending_approval" def test_submit_without_tests_raises(self, db, admin_user): campaign = Campaign(name="Empty", type="custom", status="draft", created_by=admin_user.id) db.add(campaign) db.commit() with pytest.raises(BusinessRuleViolation, match="at least one test"): submit_campaign_for_approval( db, str(campaign.id), submitter_id=admin_user.id, submitter_role="admin", ) def test_submit_by_non_creator_non_admin_raises(self, db, draft_campaign_with_test, red_lead_user): with pytest.raises(PermissionViolation): submit_campaign_for_approval( db, str(draft_campaign_with_test.id), submitter_id=red_lead_user.id, submitter_role="red_lead", ) def test_submit_non_draft_raises(self, db, draft_campaign_with_test, admin_user): draft_campaign_with_test.status = "active" db.commit() with pytest.raises(BusinessRuleViolation, match="Only draft campaigns"): submit_campaign_for_approval( db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin", ) class TestApproveCampaign: def test_approve_sets_start_date_and_activates(self, db, draft_campaign_with_test, admin_user, manager_user=None): submit_campaign_for_approval( db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin", ) db.commit() campaign = approve_campaign( db, str(draft_campaign_with_test.id), approver_id=admin_user.id, start_date="2026-08-01T00:00:00", ) assert campaign.status == "active" assert campaign.start_date is not None assert campaign.approved_by == admin_user.id assert campaign.approved_at is not None def test_approve_without_start_date_raises(self, db, draft_campaign_with_test, admin_user): submit_campaign_for_approval( db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin", ) db.commit() with pytest.raises(BusinessRuleViolation, match="start_date is required"): approve_campaign(db, str(draft_campaign_with_test.id), approver_id=admin_user.id, start_date="") def test_approve_non_pending_raises(self, db, draft_campaign_with_test, admin_user): with pytest.raises(BusinessRuleViolation, match="pending approval"): approve_campaign( db, str(draft_campaign_with_test.id), approver_id=admin_user.id, start_date="2026-08-01T00:00:00", ) class TestRejectCampaign: def test_reject_returns_to_draft_with_reason(self, db, draft_campaign_with_test, admin_user): submit_campaign_for_approval( db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin", ) db.commit() campaign = reject_campaign( db, str(draft_campaign_with_test.id), rejecter_id=admin_user.id, reason="Missing scope details", ) assert campaign.status == "draft" assert campaign.rejection_reason == "Missing scope details" def test_reject_without_reason_raises(self, db, draft_campaign_with_test, admin_user): submit_campaign_for_approval( db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin", ) db.commit() with pytest.raises(BusinessRuleViolation, match="reason is required"): reject_campaign(db, str(draft_campaign_with_test.id), rejecter_id=admin_user.id, reason="") def test_reject_non_pending_raises(self, db, draft_campaign_with_test, admin_user): with pytest.raises(BusinessRuleViolation, match="pending approval"): reject_campaign(db, str(draft_campaign_with_test.id), rejecter_id=admin_user.id, reason="No") ``` - [ ] **Step 2: Run tests to verify they fail** Run (from `backend/`): `pytest tests/test_campaign_approval.py -v` Expected: FAIL — `ImportError: cannot import name 'submit_campaign_for_approval' from 'app.services.campaign_crud_service'`. - [ ] **Step 3: Implement the service functions** In `backend/app/services/campaign_crud_service.py`, add right after `create_campaign` (after line 306, before `get_campaign_detail`): ```python def submit_campaign_for_approval( db: Session, campaign_id: str, *, submitter_id: uuid.UUID, submitter_role: str, ) -> Campaign: """Submit a draft campaign into the manager's approval queue. Raises EntityNotFoundError, PermissionViolation, BusinessRuleViolation. Does not commit; caller commits. """ campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() if not campaign: raise EntityNotFoundError("Campaign", campaign_id) if str(campaign.created_by) != str(submitter_id) and submitter_role != "admin": raise PermissionViolation("Only the creator or admin can submit this campaign for approval") if campaign.status != "draft": raise BusinessRuleViolation("Only draft campaigns can be submitted for approval") test_count = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign_id).count() if test_count == 0: raise BusinessRuleViolation("Campaign must have at least one test to submit for approval") campaign.status = "pending_approval" db.flush() return campaign def approve_campaign( db: Session, campaign_id: str, *, approver_id: uuid.UUID, start_date: str, ) -> Campaign: """Approve a pending campaign: fix its start date and activate it. Raises EntityNotFoundError, BusinessRuleViolation. Does not commit; caller commits. """ campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() if not campaign: raise EntityNotFoundError("Campaign", campaign_id) if campaign.status != "pending_approval": raise BusinessRuleViolation("Only campaigns pending approval can be approved") if not start_date: raise BusinessRuleViolation("start_date is required to approve a campaign") campaign.start_date = datetime.fromisoformat(start_date) campaign.status = "active" campaign.approved_by = approver_id campaign.approved_at = datetime.utcnow() campaign.rejection_reason = None db.flush() return campaign def reject_campaign( db: Session, campaign_id: str, *, rejecter_id: uuid.UUID, reason: str, ) -> Campaign: """Reject a pending campaign, returning it to draft with a mandatory reason. Raises EntityNotFoundError, BusinessRuleViolation. Does not commit; caller commits. """ campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() if not campaign: raise EntityNotFoundError("Campaign", campaign_id) if campaign.status != "pending_approval": raise BusinessRuleViolation("Only campaigns pending approval can be rejected") if not reason or not reason.strip(): raise BusinessRuleViolation("A rejection reason is required") campaign.status = "draft" campaign.rejection_reason = reason.strip() campaign.approved_by = None campaign.approved_at = None db.flush() return campaign ``` Note: `rejecter_id` is accepted for symmetry with `approve_campaign`'s `approver_id` but isn't stored (there's no `rejected_by` column — `rejection_reason` plus the audit log entry the router writes are the record of who rejected). Leave the parameter in the signature since the router will always pass it and a future column could use it. - [ ] **Step 4: Run tests to verify they pass** Run: `pytest tests/test_campaign_approval.py -v` Expected: PASS — all 10 tests green. - [ ] **Step 5: Commit** ```bash git add backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval.py git commit -m "feat(campaigns): submit/approve/reject service functions" ``` --- ## Task 7: Service layer — gate test add/remove to draft, add modification requests **Files:** - Modify: `backend/app/services/campaign_crud_service.py` - Test: `backend/tests/test_campaign_approval.py` - [ ] **Step 1: Write the failing tests** Append to `backend/tests/test_campaign_approval.py`: ```python from app.services.campaign_crud_service import ( add_test_to_campaign, remove_test_from_campaign, create_modification_request, approve_modification_request, reject_modification_request, list_modification_requests, ) @pytest.fixture def active_campaign_with_test(db, draft_campaign_with_test, admin_user): """An active campaign (bypasses the approval flow directly for test setup).""" draft_campaign_with_test.status = "active" db.commit() db.refresh(draft_campaign_with_test) return draft_campaign_with_test class TestDirectTestEditingGate: def test_add_test_blocked_on_active_campaign(self, db, active_campaign_with_test, admin_user): tech = Technique(mitre_id="T1078", name="Valid Accounts", tactic="persistence", platforms=["windows"]) db.add(tech) db.flush() new_test = Test(technique_id=tech.id, name="New test", state=TestState.draft, created_by=admin_user.id) db.add(new_test) db.commit() with pytest.raises(BusinessRuleViolation, match="draft"): add_test_to_campaign(db, str(active_campaign_with_test.id), test_id=str(new_test.id)) def test_remove_test_blocked_on_active_campaign(self, db, active_campaign_with_test): ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first() with pytest.raises(BusinessRuleViolation, match="draft"): remove_test_from_campaign(db, str(active_campaign_with_test.id), str(ct.id)) class TestModificationRequests: def test_create_add_test_request(self, db, active_campaign_with_test, admin_user): tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"]) db.add(tech) db.flush() new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=admin_user.id) db.add(new_test) db.commit() request = create_modification_request( db, str(active_campaign_with_test.id), requester_id=admin_user.id, action="add_test", test_id=str(new_test.id), justification="Coverage gap found in review", ) assert request.status == "pending" # Underlying campaign is untouched until approval cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all() assert len(cts) == 1 def test_create_request_without_justification_raises(self, db, active_campaign_with_test, admin_user): ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first() with pytest.raises(BusinessRuleViolation, match="justification is required"): create_modification_request( db, str(active_campaign_with_test.id), requester_id=admin_user.id, action="remove_test", test_id=str(ct.test_id), justification="", ) def test_create_request_on_non_active_campaign_raises(self, db, draft_campaign_with_test, admin_user): ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == draft_campaign_with_test.id).first() with pytest.raises(BusinessRuleViolation, match="active campaigns"): create_modification_request( db, str(draft_campaign_with_test.id), requester_id=admin_user.id, action="remove_test", test_id=str(ct.test_id), justification="Not needed", ) def test_approve_add_test_request_applies_change(self, db, active_campaign_with_test, admin_user): tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"]) db.add(tech) db.flush() new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=admin_user.id) db.add(new_test) db.commit() request = create_modification_request( db, str(active_campaign_with_test.id), requester_id=admin_user.id, action="add_test", test_id=str(new_test.id), justification="Coverage gap found in review", ) db.commit() approved = approve_modification_request(db, str(request.id), reviewer_id=admin_user.id) assert approved.status == "approved" cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all() assert len(cts) == 2 assert any(ct.test_id == new_test.id for ct in cts) def test_approve_remove_test_request_applies_change(self, db, active_campaign_with_test, admin_user): ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first() request = create_modification_request( db, str(active_campaign_with_test.id), requester_id=admin_user.id, action="remove_test", test_id=str(ct.test_id), justification="Test superseded", ) db.commit() approve_modification_request(db, str(request.id), reviewer_id=admin_user.id) cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all() assert len(cts) == 0 def test_reject_request_leaves_campaign_untouched(self, db, active_campaign_with_test, admin_user): ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first() request = create_modification_request( db, str(active_campaign_with_test.id), requester_id=admin_user.id, action="remove_test", test_id=str(ct.test_id), justification="Test superseded", ) db.commit() rejected = reject_modification_request( db, str(request.id), reviewer_id=admin_user.id, review_notes="Still needed for coverage", ) assert rejected.status == "rejected" assert rejected.review_notes == "Still needed for coverage" cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all() assert len(cts) == 1 def test_reject_without_notes_raises(self, db, active_campaign_with_test, admin_user): ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first() request = create_modification_request( db, str(active_campaign_with_test.id), requester_id=admin_user.id, action="remove_test", test_id=str(ct.test_id), justification="Test superseded", ) db.commit() with pytest.raises(BusinessRuleViolation, match="Review notes are required"): reject_modification_request(db, str(request.id), reviewer_id=admin_user.id, review_notes="") def test_list_modification_requests_filters_by_status(self, db, active_campaign_with_test, admin_user): ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first() create_modification_request( db, str(active_campaign_with_test.id), requester_id=admin_user.id, action="remove_test", test_id=str(ct.test_id), justification="Test superseded", ) db.commit() pending = list_modification_requests(db, status="pending") assert len(pending) == 1 assert pending[0]["action"] == "remove_test" ``` - [ ] **Step 2: Run tests to verify they fail** Run: `pytest tests/test_campaign_approval.py -v -k Modification or DirectTestEditingGate` Expected: FAIL — `ImportError` for the new functions, and the existing `add_test_to_campaign`/`remove_test_from_campaign` gate tests fail because those functions currently accept `active` campaigns too. - [ ] **Step 3: Tighten the status gate on `add_test_to_campaign` and `remove_test_from_campaign`** In `backend/app/services/campaign_crud_service.py`, change the `add_test_to_campaign` signature (around line 379) to accept an `allowed_statuses` keyword, defaulting to draft-only: ```python def add_test_to_campaign( db: Session, campaign_id: str, *, test_id: str, order_index: Optional[int] = None, depends_on: Optional[str] = None, phase: Optional[str] = None, allowed_statuses: tuple[str, ...] = ("draft",), ) -> dict: ``` And change its status check (currently `if campaign.status not in ("draft", "active"):`) to: ```python if campaign.status not in allowed_statuses: raise BusinessRuleViolation( f"Can only add tests to a campaign in one of: {', '.join(allowed_statuses)}" ) ``` Do the same for `remove_test_from_campaign` (around line 495): add `allowed_statuses: tuple[str, ...] = ("draft",)` to its signature, and change its check to: ```python if campaign.status not in allowed_statuses: raise BusinessRuleViolation( f"Can only modify tests on a campaign in one of: {', '.join(allowed_statuses)}" ) ``` This is the only change to either function's body — everything else (order_index calculation, circular-dependency validation, phase inference, dependent-cleanup) stays exactly as-is. - [ ] **Step 4: Add the modification-request functions** Add the import for the new model at the top of `backend/app/services/campaign_crud_service.py` (change the existing campaign model import line): ```python from app.models.campaign import Campaign, CampaignModificationRequest, CampaignTest ``` Append these functions at the end of the file: ```python # ── Campaign Modification Requests ────────────────────────────────────── def serialize_modification_request(db: Session, request: CampaignModificationRequest) -> dict: """Serialize a modification request, including the test's display name.""" test = request.test return { "id": str(request.id), "campaign_id": str(request.campaign_id), "campaign_name": request.campaign.name if request.campaign else None, "requested_by": str(request.requested_by) if request.requested_by else None, "action": request.action, "test_id": str(request.test_id), "test_name": test.name if test else None, "order_index": request.order_index, "phase": request.phase, "justification": request.justification, "status": request.status, "reviewed_by": str(request.reviewed_by) if request.reviewed_by else None, "reviewed_at": request.reviewed_at.isoformat() if request.reviewed_at else None, "review_notes": request.review_notes, "created_at": request.created_at.isoformat() if request.created_at else None, } def create_modification_request( db: Session, campaign_id: str, *, requester_id: uuid.UUID, action: str, test_id: str, justification: str, order_index: Optional[int] = None, phase: Optional[str] = None, ) -> CampaignModificationRequest: """File a request to add/remove a test on an active campaign. Raises EntityNotFoundError, BusinessRuleViolation. Does not commit; caller commits. """ campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() if not campaign: raise EntityNotFoundError("Campaign", campaign_id) if campaign.status != "active": raise BusinessRuleViolation("Modification requests can only be created for active campaigns") if action not in ("add_test", "remove_test"): raise BusinessRuleViolation("action must be 'add_test' or 'remove_test'") if not justification or not justification.strip(): raise BusinessRuleViolation("A justification is required") test = db.query(Test).filter(Test.id == test_id).first() if not test: raise EntityNotFoundError("Test", test_id) if action == "remove_test": ct = ( db.query(CampaignTest) .filter(CampaignTest.campaign_id == campaign_id, CampaignTest.test_id == test_id) .first() ) if not ct: raise EntityNotFoundError("CampaignTest", test_id) request = CampaignModificationRequest( campaign_id=campaign_id, requested_by=requester_id, action=action, test_id=test_id, order_index=order_index, phase=phase, justification=justification.strip(), status="pending", ) db.add(request) db.flush() return request def approve_modification_request( db: Session, request_id: str, *, reviewer_id: uuid.UUID, ) -> CampaignModificationRequest: """Approve a pending modification request and apply the underlying test change. Raises EntityNotFoundError, BusinessRuleViolation. Does not commit; caller commits. """ request = ( db.query(CampaignModificationRequest) .filter(CampaignModificationRequest.id == request_id) .first() ) if not request: raise EntityNotFoundError("CampaignModificationRequest", request_id) if request.status != "pending": raise BusinessRuleViolation("Only pending modification requests can be approved") if request.action == "add_test": add_test_to_campaign( db, str(request.campaign_id), test_id=str(request.test_id), order_index=request.order_index, phase=request.phase, allowed_statuses=("active",), ) else: ct = ( db.query(CampaignTest) .filter( CampaignTest.campaign_id == request.campaign_id, CampaignTest.test_id == request.test_id, ) .first() ) if not ct: raise EntityNotFoundError("CampaignTest", str(request.test_id)) remove_test_from_campaign( db, str(request.campaign_id), str(ct.id), allowed_statuses=("active",), ) request.status = "approved" request.reviewed_by = reviewer_id request.reviewed_at = datetime.utcnow() db.flush() return request def reject_modification_request( db: Session, request_id: str, *, reviewer_id: uuid.UUID, review_notes: str, ) -> CampaignModificationRequest: """Reject a pending modification request. No change is applied to the campaign. Raises EntityNotFoundError, BusinessRuleViolation. Does not commit; caller commits. """ request = ( db.query(CampaignModificationRequest) .filter(CampaignModificationRequest.id == request_id) .first() ) if not request: raise EntityNotFoundError("CampaignModificationRequest", request_id) if request.status != "pending": raise BusinessRuleViolation("Only pending modification requests can be rejected") if not review_notes or not review_notes.strip(): raise BusinessRuleViolation("Review notes are required to reject a modification request") request.status = "rejected" request.reviewed_by = reviewer_id request.reviewed_at = datetime.utcnow() request.review_notes = review_notes.strip() db.flush() return request def list_modification_requests( db: Session, *, campaign_id: Optional[str] = None, status: Optional[str] = None, ) -> list[dict]: """List modification requests, optionally filtered by campaign and/or status.""" query = db.query(CampaignModificationRequest) if campaign_id: query = query.filter(CampaignModificationRequest.campaign_id == campaign_id) if status: query = query.filter(CampaignModificationRequest.status == status) requests = query.order_by(CampaignModificationRequest.created_at.desc()).all() return [serialize_modification_request(db, r) for r in requests] ``` - [ ] **Step 5: Run tests to verify they pass** Run: `pytest tests/test_campaign_approval.py -v` Expected: PASS — all tests from Task 6 and Task 7 green. - [ ] **Step 6: Run the full campaign test suite to check for regressions** Run: `pytest tests/test_campaigns_and_snapshots.py -v` Expected: PASS — the existing `test_circular_dependency_prevention` etc. tests still pass (they operate on draft campaigns, unaffected by the tightened gate). - [ ] **Step 7: Commit** ```bash git add backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval.py git commit -m "feat(campaigns): gate test edits to draft, add modification request workflow" ``` --- ## Task 8: Service layer — campaign timeline **Files:** - Modify: `backend/app/services/campaign_crud_service.py` - Test: `backend/tests/test_campaign_approval.py` - [ ] **Step 1: Write the failing test** Append to `backend/tests/test_campaign_approval.py`: ```python from app.services.audit_service import log_action from app.services.campaign_crud_service import get_campaign_timeline class TestCampaignTimeline: def test_timeline_returns_entries_in_chronological_order(self, db, draft_campaign_with_test, admin_user): log_action( db, user_id=admin_user.id, action="create_campaign", entity_type="campaign", entity_id=str(draft_campaign_with_test.id), details={}, ) log_action( db, user_id=admin_user.id, action="submit_campaign_for_approval", entity_type="campaign", entity_id=str(draft_campaign_with_test.id), details={}, ) db.commit() timeline = get_campaign_timeline(db, str(draft_campaign_with_test.id)) assert len(timeline) == 2 assert timeline[0]["action"] == "create_campaign" assert timeline[1]["action"] == "submit_campaign_for_approval" def test_timeline_missing_campaign_raises(self, db): import uuid as uuid_mod with pytest.raises(EntityNotFoundError): get_campaign_timeline(db, str(uuid_mod.uuid4())) ``` - [ ] **Step 2: Run test to verify it fails** Run: `pytest tests/test_campaign_approval.py -v -k CampaignTimeline` Expected: FAIL — `ImportError: cannot import name 'get_campaign_timeline'`. - [ ] **Step 3: Implement `get_campaign_timeline`** Add the import at the top of `backend/app/services/campaign_crud_service.py`: ```python from app.models.audit import AuditLog ``` Append the function at the end of the file: ```python def get_campaign_timeline(db: Session, campaign_id: str) -> list[dict]: """Return the chronological audit-log history for a campaign. Raises EntityNotFoundError if the campaign does not exist. """ campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() if not campaign: raise EntityNotFoundError("Campaign", campaign_id) logs = ( db.query(AuditLog) .filter(AuditLog.entity_type == "campaign", AuditLog.entity_id == str(campaign_id)) .order_by(AuditLog.timestamp.asc()) .all() ) return [ { "id": str(log.id), "action": log.action, "user_id": str(log.user_id) if log.user_id else None, "timestamp": log.timestamp.isoformat() if log.timestamp else None, "details": log.details or {}, } for log in logs ] ``` - [ ] **Step 4: Run test to verify it passes** Run: `pytest tests/test_campaign_approval.py -v` Expected: PASS — full file green. - [ ] **Step 5: Commit** ```bash git add backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval.py git commit -m "feat(campaigns): add get_campaign_timeline service function" ``` --- ## Task 9: Router — submit / approve / reject endpoints **Files:** - Modify: `backend/app/routers/campaigns.py` - Test: `backend/tests/test_campaign_approval_router.py` (new file) - [ ] **Step 1: Write the failing tests** Create `backend/tests/test_campaign_approval_router.py`: ```python """Router-level tests for the campaign manager-approval workflow.""" from app.models.campaign import Campaign, CampaignTest from app.models.technique import Technique from app.models.test import Test from app.models.enums import TestState def _make_draft_campaign(db, owner_id): tech = Technique(mitre_id="T1059", name="Command Line", tactic="execution", platforms=["windows"]) db.add(tech) db.flush() campaign = Campaign(name="Router Test Campaign", type="custom", status="draft", created_by=owner_id) db.add(campaign) db.flush() test = Test(technique_id=tech.id, name="T1059 test", state=TestState.draft, created_by=owner_id) db.add(test) db.flush() db.add(CampaignTest(campaign_id=campaign.id, test_id=test.id, order_index=0)) db.commit() db.refresh(campaign) return campaign def test_lead_can_submit_own_campaign(client, db, red_lead_user, red_lead_headers): campaign = _make_draft_campaign(db, red_lead_user.id) resp = client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers) assert resp.status_code == 200 assert resp.json()["status"] == "pending_approval" def test_red_tech_cannot_submit(client, db, red_lead_user, red_tech_headers): campaign = _make_draft_campaign(db, red_lead_user.id) resp = client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_tech_headers) assert resp.status_code == 403 def test_manager_can_approve_and_sets_start_date(client, db, red_lead_user, red_lead_headers, manager_headers): campaign = _make_draft_campaign(db, red_lead_user.id) client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers) resp = client.post( f"/api/v1/campaigns/{campaign.id}/approve", json={"start_date": "2026-09-01T00:00:00"}, headers=manager_headers, ) assert resp.status_code == 200 body = resp.json() assert body["status"] == "active" assert body["start_date"] is not None def test_lead_cannot_approve(client, db, red_lead_user, red_lead_headers): campaign = _make_draft_campaign(db, red_lead_user.id) client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers) resp = client.post( f"/api/v1/campaigns/{campaign.id}/approve", json={"start_date": "2026-09-01T00:00:00"}, headers=red_lead_headers, ) assert resp.status_code == 403 def test_admin_can_approve_as_manager_backup(client, db, red_lead_user, red_lead_headers, auth_headers): campaign = _make_draft_campaign(db, red_lead_user.id) client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers) resp = client.post( f"/api/v1/campaigns/{campaign.id}/approve", json={"start_date": "2026-09-01T00:00:00"}, headers=auth_headers, ) assert resp.status_code == 200 def test_manager_can_reject_with_reason(client, db, red_lead_user, red_lead_headers, manager_headers): campaign = _make_draft_campaign(db, red_lead_user.id) client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers) resp = client.post( f"/api/v1/campaigns/{campaign.id}/reject", json={"reason": "Needs more detail"}, headers=manager_headers, ) assert resp.status_code == 200 assert resp.json()["status"] == "draft" def test_manager_reject_without_reason_rejected_by_validation(client, db, red_lead_user, red_lead_headers, manager_headers): campaign = _make_draft_campaign(db, red_lead_user.id) client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers) resp = client.post( f"/api/v1/campaigns/{campaign.id}/reject", json={"reason": ""}, headers=manager_headers, ) assert resp.status_code == 400 ``` - [ ] **Step 2: Run tests to verify they fail** Run: `pytest tests/test_campaign_approval_router.py -v` Expected: FAIL — `404 Not Found` for `/submit`, `/approve`, `/reject` (routes don't exist yet). - [ ] **Step 3: Add the Pydantic payloads and endpoints** In `backend/app/routers/campaigns.py`, add the import for the new service functions (extend the existing multi-import block for `campaign_crud_service` — add right after the `activate_campaign as crud_activate` import around line 92): ```python from app.services.campaign_crud_service import ( submit_campaign_for_approval as crud_submit, ) from app.services.campaign_crud_service import ( approve_campaign as crud_approve, ) from app.services.campaign_crud_service import ( reject_campaign as crud_reject, ) ``` Add two new Pydantic payloads near the other payload classes (after `SchedulePayload`, around line 173): ```python class ApprovePayload(BaseModel): """Payload for a manager approving a pending campaign.""" start_date: str # ISO date/datetime — required, campaign won't activate without it class RejectPayload(BaseModel): """Payload for a manager rejecting a pending campaign.""" reason: str ``` Add the three endpoints right after `update_campaign` (after line 386, before the `DELETE /campaigns/{id}` section): ```python # --------------------------------------------------------------------------- # POST /campaigns/{id}/submit — Submit draft for manager approval # --------------------------------------------------------------------------- @router.post("/{campaign_id}/submit") def submit_campaign( campaign_id: str, db: Session = Depends(get_db), current_user: User = Depends(require_any_role("red_lead", "blue_lead")), ) -> dict: """Submit a draft campaign into the manager's approval queue.""" with UnitOfWork(db) as uow: campaign = crud_submit( db, campaign_id, submitter_id=current_user.id, submitter_role=current_user.role, ) log_action( db, user_id=current_user.id, action="submit_campaign_for_approval", entity_type="campaign", entity_id=campaign.id, details={"name": campaign.name}, ) uow.commit() db.refresh(campaign) return serialize_campaign(db, campaign) # --------------------------------------------------------------------------- # POST /campaigns/{id}/approve — Manager approves a pending campaign # --------------------------------------------------------------------------- @router.post("/{campaign_id}/approve") def approve_campaign_endpoint( campaign_id: str, payload: ApprovePayload, db: Session = Depends(get_db), current_user: User = Depends(require_any_role("manager")), ) -> dict: """Manager approves a pending campaign, fixing its start date and activating it.""" with UnitOfWork(db) as uow: campaign = crud_approve( db, campaign_id, approver_id=current_user.id, start_date=payload.start_date, ) log_action( db, user_id=current_user.id, action="approve_campaign", entity_type="campaign", entity_id=campaign.id, details={"start_date": payload.start_date}, ) uow.commit() db.refresh(campaign) return serialize_campaign(db, campaign) # --------------------------------------------------------------------------- # POST /campaigns/{id}/reject — Manager rejects a pending campaign # --------------------------------------------------------------------------- @router.post("/{campaign_id}/reject") def reject_campaign_endpoint( campaign_id: str, payload: RejectPayload, db: Session = Depends(get_db), current_user: User = Depends(require_any_role("manager")), ) -> dict: """Manager rejects a pending campaign, returning it to draft with a reason.""" with UnitOfWork(db) as uow: campaign = crud_reject( db, campaign_id, rejecter_id=current_user.id, reason=payload.reason, ) log_action( db, user_id=current_user.id, action="reject_campaign", entity_type="campaign", entity_id=campaign.id, details={"reason": payload.reason}, ) uow.commit() db.refresh(campaign) return serialize_campaign(db, campaign) ``` - [ ] **Step 4: Add `approved_by`, `approved_at`, `rejection_reason` to the serialized campaign** In `serialize_campaign` in `backend/app/services/campaign_crud_service.py`, add these three keys right after `"completed_at"` (around line 127): ```python "approved_by": str(campaign.approved_by) if campaign.approved_by else None, "approved_at": campaign.approved_at.isoformat() if campaign.approved_at else None, "rejection_reason": campaign.rejection_reason, ``` - [ ] **Step 5: Run tests to verify they pass** Run: `pytest tests/test_campaign_approval_router.py -v` Expected: PASS — all 7 tests green. - [ ] **Step 6: Run the full backend test suite to check for regressions** Run (from `backend/`): `pytest tests/ -v` Expected: PASS — no pre-existing test broken. Pay special attention to `test_campaigns_and_snapshots.py`. - [ ] **Step 7: Commit** ```bash git add backend/app/routers/campaigns.py backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval_router.py git commit -m "feat(campaigns): submit/approve/reject router endpoints" ``` --- ## Task 10: Router — modification request endpoints **Files:** - Modify: `backend/app/routers/campaigns.py` - Test: `backend/tests/test_campaign_approval_router.py` - [ ] **Step 1: Write the failing tests** Append to `backend/tests/test_campaign_approval_router.py`: ```python def _activate_campaign_directly(db, campaign): """Test helper — bypass the approval flow to get straight to 'active' for setup.""" campaign.status = "active" db.commit() db.refresh(campaign) return campaign def test_lead_can_create_modification_request(client, db, red_lead_user, red_lead_headers): campaign = _make_draft_campaign(db, red_lead_user.id) _activate_campaign_directly(db, campaign) tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"]) db.add(tech) db.flush() new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=red_lead_user.id) db.add(new_test) db.commit() resp = client.post( f"/api/v1/campaigns/{campaign.id}/modification-requests", json={"action": "add_test", "test_id": str(new_test.id), "justification": "Coverage gap"}, headers=red_lead_headers, ) assert resp.status_code == 201 assert resp.json()["status"] == "pending" def test_modification_request_without_justification_rejected(client, db, red_lead_user, red_lead_headers): campaign = _make_draft_campaign(db, red_lead_user.id) _activate_campaign_directly(db, campaign) ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first() resp = client.post( f"/api/v1/campaigns/{campaign.id}/modification-requests", json={"action": "remove_test", "test_id": str(ct.test_id), "justification": ""}, headers=red_lead_headers, ) assert resp.status_code == 400 def test_manager_can_list_pending_modification_requests(client, db, red_lead_user, red_lead_headers, manager_headers): campaign = _make_draft_campaign(db, red_lead_user.id) _activate_campaign_directly(db, campaign) ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first() client.post( f"/api/v1/campaigns/{campaign.id}/modification-requests", json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"}, headers=red_lead_headers, ) resp = client.get("/api/v1/campaigns/modification-requests/pending", headers=manager_headers) assert resp.status_code == 200 assert len(resp.json()) == 1 def test_manager_can_approve_modification_request(client, db, red_lead_user, red_lead_headers, manager_headers): campaign = _make_draft_campaign(db, red_lead_user.id) _activate_campaign_directly(db, campaign) ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first() create_resp = client.post( f"/api/v1/campaigns/{campaign.id}/modification-requests", json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"}, headers=red_lead_headers, ) request_id = create_resp.json()["id"] resp = client.post( f"/api/v1/campaigns/modification-requests/{request_id}/approve", headers=manager_headers, ) assert resp.status_code == 200 assert resp.json()["status"] == "approved" remaining = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).all() assert len(remaining) == 0 def test_manager_can_reject_modification_request(client, db, red_lead_user, red_lead_headers, manager_headers): campaign = _make_draft_campaign(db, red_lead_user.id) _activate_campaign_directly(db, campaign) ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first() create_resp = client.post( f"/api/v1/campaigns/{campaign.id}/modification-requests", json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"}, headers=red_lead_headers, ) request_id = create_resp.json()["id"] resp = client.post( f"/api/v1/campaigns/modification-requests/{request_id}/reject", json={"review_notes": "Still needed"}, headers=manager_headers, ) assert resp.status_code == 200 assert resp.json()["status"] == "rejected" remaining = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).all() assert len(remaining) == 1 def test_lead_cannot_approve_modification_request(client, db, red_lead_user, red_lead_headers): campaign = _make_draft_campaign(db, red_lead_user.id) _activate_campaign_directly(db, campaign) ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first() create_resp = client.post( f"/api/v1/campaigns/{campaign.id}/modification-requests", json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"}, headers=red_lead_headers, ) request_id = create_resp.json()["id"] resp = client.post( f"/api/v1/campaigns/modification-requests/{request_id}/approve", headers=red_lead_headers, ) assert resp.status_code == 403 def test_direct_add_test_blocked_on_active_campaign_via_router(client, db, red_lead_user, red_lead_headers): campaign = _make_draft_campaign(db, red_lead_user.id) _activate_campaign_directly(db, campaign) tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"]) db.add(tech) db.flush() new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=red_lead_user.id) db.add(new_test) db.commit() resp = client.post( f"/api/v1/campaigns/{campaign.id}/tests", json={"test_id": str(new_test.id)}, headers=red_lead_headers, ) assert resp.status_code == 400 ``` - [ ] **Step 2: Run tests to verify they fail** Run: `pytest tests/test_campaign_approval_router.py -v -k modification` Expected: FAIL — `404 Not Found` for the new routes (`test_direct_add_test_blocked...` should already pass since Task 7 tightened the service layer, but the route-level test confirms it end-to-end). - [ ] **Step 3: Add the payloads and endpoints** In `backend/app/routers/campaigns.py`, add the service imports (extend the multi-import block, same place as Task 9's imports): ```python from app.services.campaign_crud_service import ( create_modification_request as crud_create_mod_request, ) from app.services.campaign_crud_service import ( approve_modification_request as crud_approve_mod_request, ) from app.services.campaign_crud_service import ( reject_modification_request as crud_reject_mod_request, ) from app.services.campaign_crud_service import ( list_modification_requests as crud_list_mod_requests, ) from app.services.campaign_crud_service import ( serialize_modification_request as crud_serialize_mod_request, ) ``` Add the payloads near `ApprovePayload`/`RejectPayload`: ```python class ModificationRequestPayload(BaseModel): """Payload for a lead requesting to add/remove a test on an active campaign.""" action: str # add_test | remove_test test_id: str justification: str order_index: Optional[int] = None phase: Optional[str] = None class RejectModificationPayload(BaseModel): """Payload for a manager rejecting a modification request.""" review_notes: str ``` Add the endpoints right after the `reject_campaign_endpoint` function from Task 9: ```python # --------------------------------------------------------------------------- # POST /campaigns/{id}/modification-requests — Request a test add/remove # --------------------------------------------------------------------------- @router.post("/{campaign_id}/modification-requests", status_code=201) def create_modification_request_endpoint( campaign_id: str, payload: ModificationRequestPayload, db: Session = Depends(get_db), current_user: User = Depends(require_any_role("red_lead", "blue_lead")), ) -> dict: """File a request to add/remove a test on an active campaign — needs manager approval.""" with UnitOfWork(db) as uow: request = crud_create_mod_request( db, campaign_id, requester_id=current_user.id, action=payload.action, test_id=payload.test_id, justification=payload.justification, order_index=payload.order_index, phase=payload.phase, ) log_action( db, user_id=current_user.id, action="request_campaign_modification", entity_type="campaign", entity_id=campaign_id, details={ "action": payload.action, "test_id": payload.test_id, "justification": payload.justification, }, ) uow.commit() return crud_serialize_mod_request(db, request) # --------------------------------------------------------------------------- # GET /campaigns/{id}/modification-requests — List requests for one campaign # --------------------------------------------------------------------------- @router.get("/{campaign_id}/modification-requests") def list_campaign_modification_requests_endpoint( campaign_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ) -> list: """List modification requests filed against a specific campaign.""" return crud_list_mod_requests(db, campaign_id=campaign_id) # --------------------------------------------------------------------------- # GET /campaigns/modification-requests/pending — Manager's global queue # --------------------------------------------------------------------------- @router.get("/modification-requests/pending") def list_pending_modification_requests_endpoint( db: Session = Depends(get_db), current_user: User = Depends(require_any_role("manager")), ) -> list: """List all modification requests awaiting manager review, across all campaigns.""" return crud_list_mod_requests(db, status="pending") # --------------------------------------------------------------------------- # POST /campaigns/modification-requests/{id}/approve — Apply the requested change # --------------------------------------------------------------------------- @router.post("/modification-requests/{request_id}/approve") def approve_modification_request_endpoint( request_id: str, db: Session = Depends(get_db), current_user: User = Depends(require_any_role("manager")), ) -> dict: """Manager approves a modification request — the test change is applied now.""" with UnitOfWork(db) as uow: request = crud_approve_mod_request(db, request_id, reviewer_id=current_user.id) log_action( db, user_id=current_user.id, action="approve_campaign_modification", entity_type="campaign", entity_id=str(request.campaign_id), details={ "request_id": request_id, "action": request.action, "test_id": str(request.test_id), }, ) uow.commit() return crud_serialize_mod_request(db, request) # --------------------------------------------------------------------------- # POST /campaigns/modification-requests/{id}/reject — Deny the requested change # --------------------------------------------------------------------------- @router.post("/modification-requests/{request_id}/reject") def reject_modification_request_endpoint( request_id: str, payload: RejectModificationPayload, db: Session = Depends(get_db), current_user: User = Depends(require_any_role("manager")), ) -> dict: """Manager rejects a modification request. No change is applied.""" with UnitOfWork(db) as uow: request = crud_reject_mod_request( db, request_id, reviewer_id=current_user.id, review_notes=payload.review_notes, ) log_action( db, user_id=current_user.id, action="reject_campaign_modification", entity_type="campaign", entity_id=str(request.campaign_id), details={"request_id": request_id, "review_notes": payload.review_notes}, ) uow.commit() return crud_serialize_mod_request(db, request) ``` Note: `GET /campaigns/modification-requests/pending` must be declared before `GET /campaigns/{campaign_id}` would otherwise be — but since that route has a different path segment count (2 segments vs. 1), there's no actual routing conflict regardless of declaration order in this file. Add these new routes in the order shown above for readability. - [ ] **Step 4: Run tests to verify they pass** Run: `pytest tests/test_campaign_approval_router.py -v` Expected: PASS — all tests green, including `test_direct_add_test_blocked_on_active_campaign_via_router`. - [ ] **Step 5: Commit** ```bash git add backend/app/routers/campaigns.py backend/tests/test_campaign_approval_router.py git commit -m "feat(campaigns): modification-request router endpoints" ``` --- ## Task 11: Router — timeline endpoint, admin-only manual activation, drop lead-set dates **Files:** - Modify: `backend/app/routers/campaigns.py` - Modify: `backend/app/services/campaign_service.py` - Test: `backend/tests/test_campaign_approval_router.py` - [ ] **Step 1: Write the failing tests** Append to `backend/tests/test_campaign_approval_router.py`: ```python def test_get_campaign_timeline(client, db, red_lead_user, red_lead_headers): campaign = _make_draft_campaign(db, red_lead_user.id) client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers) resp = client.get(f"/api/v1/campaigns/{campaign.id}/timeline", headers=red_lead_headers) assert resp.status_code == 200 actions = [e["action"] for e in resp.json()] assert "submit_campaign_for_approval" in actions def test_lead_cannot_directly_activate_draft_campaign(client, db, red_lead_user, red_lead_headers): campaign = _make_draft_campaign(db, red_lead_user.id) resp = client.post(f"/api/v1/campaigns/{campaign.id}/activate", headers=red_lead_headers) assert resp.status_code == 403 def test_admin_can_still_directly_activate_as_override(client, db, red_lead_user, red_lead_headers, auth_headers): campaign = _make_draft_campaign(db, red_lead_user.id) resp = client.post(f"/api/v1/campaigns/{campaign.id}/activate", headers=auth_headers) assert resp.status_code == 200 assert resp.json()["status"] == "active" def test_create_campaign_payload_has_no_start_date_field(client, db, red_lead_headers): resp = client.post( "/api/v1/campaigns", json={"name": "No date campaign", "start_date": "2026-01-01"}, headers=red_lead_headers, ) assert resp.status_code == 201 # start_date silently ignored — only the manager can ever set it, via /approve assert resp.json()["start_date"] is None ``` - [ ] **Step 2: Run tests to verify they fail** Run: `pytest tests/test_campaign_approval_router.py -v -k "timeline or activate or start_date"` Expected: FAIL — `404` for `/timeline`, `403`-expecting tests get `200` (lead can still activate directly today), and the create-campaign test gets a non-null `start_date`. - [ ] **Step 3: Add the timeline endpoint** Import the new service function in `backend/app/routers/campaigns.py` (extend the multi-import block): ```python from app.services.campaign_crud_service import ( get_campaign_timeline as crud_get_timeline, ) ``` Add the endpoint right after `get_campaign_history` (after line 856): ```python # --------------------------------------------------------------------------- # GET /campaigns/{id}/timeline — Audit-log history for this campaign # --------------------------------------------------------------------------- @router.get("/{campaign_id}/timeline") def get_campaign_timeline_endpoint( campaign_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ) -> list: """Return the chronological audit-log history for a campaign.""" return crud_get_timeline(db, campaign_id) ``` - [ ] **Step 4: Restrict direct `/activate` to admin only** In `backend/app/routers/campaigns.py`, change the `activate_campaign` endpoint's dependency (around line 516) from: ```python current_user: User = Depends(require_any_role("red_lead", "blue_lead")), ``` to: ```python current_user: User = Depends(require_any_role("admin")), ``` This closes the loophole where a lead could bypass the approval queue entirely by calling the old direct-activation endpoint. `require_any_role("admin")` only ever lets `admin` through (per its own always-pass rule), so this becomes an admin-only emergency override, matching the "admin also approves" pattern used for `/approve` and `/reject`. - [ ] **Step 5: Remove `start_date` from lead-facing create/update/generate payloads** In `backend/app/routers/campaigns.py`: 1. Remove the `start_date` field from `CampaignCreate` (delete the line `start_date: Optional[str] = None # ISO date...` around line 127). 2. Remove the `start_date` field from `CampaignUpdate` (delete the equivalent line around line 146). 3. Remove `start_date: Optional[str] = None` from `GenerateFromActorPayload` (around line 696) — replace the class body with `pass`: ```python class GenerateFromActorPayload(BaseModel): pass ``` 4. In the `generate_campaign_from_actor` endpoint, remove the two lines that parse and forward `payload.start_date`: ```python campaign = generate_campaign_from_threat_actor( db, uuid.UUID(actor_id), current_user, ) ``` (delete the `start_date_parsed = ...` line entirely and the `start_date=start_date_parsed` kwarg). In `backend/app/services/campaign_service.py`, remove the `start_date` parameter from `generate_campaign_from_threat_actor`'s signature and drop `start_date=start_date` from the `Campaign(...)` construction inside it (around lines 183 and 241). In `backend/app/services/campaign_crud_service.py`, remove the `start_date` parameter from `create_campaign`'s signature and drop `start_date=datetime.fromisoformat(start_date) if start_date else None,` from its `Campaign(...)` construction (around lines 278 and 299). Also remove the `start_date` handling block from `update_campaign` (the `if "start_date" in fields and fields["start_date"]:` block around line 364) — with `start_date` no longer in the request schema this is dead code, but removing it makes clear the field is intentionally unreachable via this function now. - [ ] **Step 6: Run tests to verify they pass** Run: `pytest tests/test_campaign_approval_router.py -v` Expected: PASS — all tests in the file green. - [ ] **Step 7: Run the full backend test suite** Run (from `backend/`): `pytest tests/ -v` Expected: PASS. If `test_campaigns_and_snapshots.py` or any other file passes `start_date` to `create_campaign`/`generate_campaign_from_threat_actor` directly, update that call site to drop the argument — grep first: Run: `grep -rn "start_date=" backend/tests/ backend/app/services/campaign_scheduler_service.py` Fix any remaining call sites the grep turns up (the scheduler's `_clone_campaign` does not set `start_date` today, so it should need no change). - [ ] **Step 8: Commit** ```bash git add backend/app/routers/campaigns.py backend/app/services/campaign_service.py backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval_router.py git commit -m "feat(campaigns): timeline endpoint, admin-only manual activation, remove lead-set dates" ``` --- ## Task 12: Backend — full-suite regression check **Files:** none (verification only) - [ ] **Step 1: Run the complete backend test suite** Run (from `backend/`): `pytest tests/ -v` Expected: PASS — every test green, including all new files from Tasks 1–11. - [ ] **Step 2: Run Ruff lint** Run (from `backend/`): `ruff check app/ tests/` Expected: no errors. Fix any import-order or unused-import issues the edits introduced (e.g. if `Optional` or `datetime` end up unused anywhere after the `start_date` removal in Task 11). - [ ] **Step 3: Commit any lint fixes** ```bash git add -A git commit -m "chore(campaigns): fix lint issues from approval workflow" ``` (Skip this commit if step 2 found nothing to fix.) --- ## Task 13: Frontend — API client for the approval workflow **Files:** - Modify: `frontend/src/api/campaigns.ts` - [ ] **Step 1: Add new types and drop `start_date` from create/update payloads** In `frontend/src/api/campaigns.ts`: 1. Add `approved_by`, `approved_at`, `rejection_reason` to the `Campaign` interface, right after `completed_at`: ```typescript approved_by: string | null; approved_at: string | null; rejection_reason: string | null; ``` 2. Remove `start_date` from `CampaignCreatePayload` (it's now ignored by the backend and only ever settable via `approveCampaign`): ```typescript export interface CampaignCreatePayload { name: string; description?: string; type?: string; threat_actor_id?: string; target_platform?: string; tags?: string[]; scheduled_at?: string; } ``` 3. Add the modification-request type and campaign timeline entry type at the end of the "Types" section: ```typescript export type ModificationRequestAction = "add_test" | "remove_test"; export type ModificationRequestStatus = "pending" | "approved" | "rejected"; export interface CampaignModificationRequest { id: string; campaign_id: string; campaign_name: string | null; requested_by: string | null; action: ModificationRequestAction; test_id: string; test_name: string | null; order_index: number | null; phase: string | null; justification: string; status: ModificationRequestStatus; reviewed_by: string | null; reviewed_at: string | null; review_notes: string | null; created_at: string | null; } export interface CampaignTimelineEntry { id: string; action: string; user_id: string | null; timestamp: string; details: Record; } ``` - [ ] **Step 2: Add the API functions** Append to `frontend/src/api/campaigns.ts`: ```typescript // ── Approval workflow ────────────────────────────────────────────── /** Submit a draft campaign into the manager's approval queue. */ export async function submitCampaignForApproval(campaignId: string): Promise { const { data } = await client.post(`/campaigns/${campaignId}/submit`); return data; } /** Manager approves a pending campaign, fixing its start date. */ export async function approveCampaign(campaignId: string, startDate: string): Promise { const { data } = await client.post(`/campaigns/${campaignId}/approve`, { start_date: startDate, }); return data; } /** Manager rejects a pending campaign with a mandatory reason. */ export async function rejectCampaign(campaignId: string, reason: string): Promise { const { data } = await client.post(`/campaigns/${campaignId}/reject`, { reason }); return data; } // ── Modification requests ─────────────────────────────────────────── /** File a request to add/remove a test on an active campaign. */ export async function createModificationRequest( campaignId: string, payload: { action: ModificationRequestAction; test_id: string; justification: string; order_index?: number; phase?: string; }, ): Promise { const { data } = await client.post( `/campaigns/${campaignId}/modification-requests`, payload, ); return data; } /** List modification requests filed against a specific campaign. */ export async function listCampaignModificationRequests( campaignId: string, ): Promise { const { data } = await client.get( `/campaigns/${campaignId}/modification-requests`, ); return data; } /** Manager's global queue of modification requests awaiting review. */ export async function listPendingModificationRequests(): Promise { const { data } = await client.get( "/campaigns/modification-requests/pending", ); return data; } /** Manager approves a modification request — the test change is applied now. */ export async function approveModificationRequest( requestId: string, ): Promise { const { data } = await client.post( `/campaigns/modification-requests/${requestId}/approve`, ); return data; } /** Manager rejects a modification request with mandatory review notes. */ export async function rejectModificationRequest( requestId: string, reviewNotes: string, ): Promise { const { data } = await client.post( `/campaigns/modification-requests/${requestId}/reject`, { review_notes: reviewNotes }, ); return data; } // ── Timeline ───────────────────────────────────────────────────────── /** Get the audit-log timeline for a campaign. */ export async function getCampaignTimeline(campaignId: string): Promise { const { data } = await client.get(`/campaigns/${campaignId}/timeline`); return data; } ``` - [ ] **Step 3: Verify TypeScript compiles** Run (from `frontend/`): `npm run build` Expected: build succeeds. It's expected/fine that `CampaignsPage.tsx` and `CampaignDetailPage.tsx` still reference `newCampaign.start_date` — those get fixed in Tasks 14–15. If `npm run build` fails only on those two files' `start_date` usage, that confirms the type change took effect correctly; proceed to Task 14. - [ ] **Step 4: Commit** ```bash git add frontend/src/api/campaigns.ts git commit -m "feat(campaigns): API client for approval workflow" ``` --- ## Task 14: Frontend — CampaignsPage: drop lead-set start dates, add pending-approval filter **Files:** - Modify: `frontend/src/pages/CampaignsPage.tsx` - [ ] **Step 1: Remove the "Start date" field from the Create Campaign modal** In `frontend/src/pages/CampaignsPage.tsx`: 1. Remove `start_date: ""` from the `newCampaign` state initializer (line 58) and from the `setNewCampaign` reset call in `createMutation.onSuccess` (line 78): ```typescript const [newCampaign, setNewCampaign] = useState({ name: "", description: "", type: "custom", target_platform: "", }); ``` ```typescript setNewCampaign({ name: "", description: "", type: "custom", target_platform: "" }); ``` 2. Delete the entire "Start date" `
...
` block (lines 232–253) from the Create Campaign modal. - [ ] **Step 2: Remove the "Start date" field from the Generate-from-Threat-Actor modal** Remove the `actorStartDate` state (line 52) and its setter usages: ```typescript const [actorSearch, setActorSearch] = useState(""); ``` (delete the `actorStartDate` line entirely) Update `generateMutation` (lines 90–99) to stop passing a start date: ```typescript const generateMutation = useMutation({ mutationFn: (actorId: string) => generateCampaignFromThreatActor(actorId), onSuccess: (campaign) => { queryClient.invalidateQueries({ queryKey: ["campaigns"] }); setShowActorSelector(false); navigate(`/campaigns/${campaign.id}`); }, }); ``` Delete the entire "Start date" `
...
` block from the actor selector modal (the one right after the search input, roughly lines 295–316). The `setActorStartDate("")` call that used to live in `generateMutation`'s `onSuccess` is already gone since Step 2 rewrote that whole mutation block above — no other call sites reference `actorStartDate` after this change. - [ ] **Step 3: Add a "Pending Approval" option to the status filter** In the status ` setApproveStartDate(e.target.value)} className="w-full rounded-lg border border-gray-700 bg-gray-800 px-3 py-2 text-sm text-gray-200 focus:border-cyan-500 focus:outline-none [color-scheme:dark]" />
)} {/* Reject modal */} {rejectModalOpen && (

Reject Campaign