From dc5f206bf57620739174a044865e682f6a6e15c6 Mon Sep 17 00:00:00 2001 From: kitos Date: Fri, 3 Jul 2026 10:18:32 +0200 Subject: [PATCH] feat(campaigns): gate test edits to draft, add modification request workflow --- backend/app/services/campaign_crud_service.py | 243 ++++++++++++++++-- backend/tests/test_campaign_approval.py | 154 +++++++++++ 2 files changed, 377 insertions(+), 20 deletions(-) diff --git a/backend/app/services/campaign_crud_service.py b/backend/app/services/campaign_crud_service.py index 384d12a..ae8f5f4 100644 --- a/backend/app/services/campaign_crud_service.py +++ b/backend/app/services/campaign_crud_service.py @@ -23,8 +23,8 @@ from app.domain.errors import ( PermissionViolation, ) -# Import Campaign, CampaignTest from app.models.campaign -from app.models.campaign import Campaign, CampaignTest +# Import Campaign, CampaignModificationRequest, CampaignTest from app.models.campaign +from app.models.campaign import Campaign, CampaignModificationRequest, CampaignTest # Import Technique from app.models.technique from app.models.technique import Technique @@ -482,6 +482,8 @@ def add_test_to_campaign( depends_on: Optional[str] = None, # Entry: phase phase: Optional[str] = None, + # Entry: allowed_statuses + allowed_statuses: tuple[str, ...] = ("draft",), ) -> dict: """Add a test to a campaign with optional ordering and dependency. @@ -489,20 +491,22 @@ def add_test_to_campaign( Raises BusinessRuleViolation for invalid state or circular dependency. Does not commit; caller commits. """ - # Assign campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() - campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() + # Assign campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first() + campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first() # Check: not campaign if not campaign: # Raise EntityNotFoundError raise EntityNotFoundError("Campaign", campaign_id) - # Check: campaign.status not in ("draft", "active") - if campaign.status not in ("draft", "active"): + # Check: campaign.status not in allowed_statuses + if campaign.status not in allowed_statuses: # Raise BusinessRuleViolation - raise BusinessRuleViolation("Can only add tests to draft or active campaigns") + raise BusinessRuleViolation( + f"Can only add tests to a campaign in one of: {', '.join(allowed_statuses)}" + ) - # Assign test = db.query(Test).filter(Test.id == test_id).first() - test = db.query(Test).filter(Test.id == test_id).first() + # Assign test = db.query(Test).filter(Test.id == uuid.UUID(test_id)).first() + test = db.query(Test).filter(Test.id == uuid.UUID(test_id)).first() # Check: not test if not test: # Raise EntityNotFoundError @@ -518,7 +522,7 @@ def add_test_to_campaign( max_order = ( db.query(CampaignTest.order_index) # Chain .filter() call - .filter(CampaignTest.campaign_id == campaign_id) + .filter(CampaignTest.campaign_id == uuid.UUID(campaign_id)) # Chain .order_by() call .order_by(CampaignTest.order_index.desc()) # Chain .first() call @@ -551,9 +555,9 @@ def add_test_to_campaign( # Keyword argument: id id=ct_id, # Keyword argument: campaign_id - campaign_id=campaign_id, + campaign_id=uuid.UUID(campaign_id), # Keyword argument: test_id - test_id=test_id, + test_id=uuid.UUID(test_id), # Keyword argument: order_index order_index=final_order_index, # Keyword argument: depends_on @@ -584,32 +588,41 @@ def add_test_to_campaign( # Define function remove_test_from_campaign -def remove_test_from_campaign(db: Session, campaign_id: str, campaign_test_id: str) -> None: +def remove_test_from_campaign( + db: Session, + campaign_id: str, + campaign_test_id: str, + *, + # Entry: allowed_statuses + allowed_statuses: tuple[str, ...] = ("draft",), +) -> None: """Remove a test from a campaign. Raises EntityNotFoundError for missing campaign or campaign test. Raises BusinessRuleViolation for invalid state. Does not commit; caller commits. """ - # Assign campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() - campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first() + # Assign campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first() + campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first() # Check: not campaign if not campaign: # Raise EntityNotFoundError raise EntityNotFoundError("Campaign", campaign_id) - # Check: campaign.status not in ("draft", "active") - if campaign.status not in ("draft", "active"): + # Check: campaign.status not in allowed_statuses + if campaign.status not in allowed_statuses: # Raise BusinessRuleViolation - raise BusinessRuleViolation("Can only modify draft or active campaigns") + raise BusinessRuleViolation( + f"Can only modify tests on a campaign in one of: {', '.join(allowed_statuses)}" + ) # Assign ct = ( ct = ( db.query(CampaignTest) # Chain .filter() call .filter( - CampaignTest.id == campaign_test_id, - CampaignTest.campaign_id == campaign_id, + CampaignTest.id == uuid.UUID(campaign_test_id), + CampaignTest.campaign_id == uuid.UUID(campaign_id), ) # Chain .first() call .first() @@ -930,3 +943,193 @@ def get_campaign_history(db: Session, campaign_id: str) -> dict: for child in children ], } + + +# ── 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 == uuid.UUID(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 == uuid.UUID(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 == uuid.UUID(request_id)) + .first() + ) + if not request: + raise EntityNotFoundError("CampaignModificationRequest", request_id) + + if request.status != "pending": + raise BusinessRuleViolation("Only pending modification requests can be approved") + + # Mark the request approved *before* applying the underlying change: a + # "remove_test" approval deletes the Test row, which cascades (ON DELETE + # CASCADE on test_id) and would otherwise wipe out this very request row + # before we get a chance to persist its reviewed/approved fields. + request.status = "approved" + request.reviewed_by = reviewer_id + request.reviewed_at = datetime.utcnow() + db.flush() + + 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",), + ) + + 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 == uuid.UUID(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 == uuid.UUID(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] diff --git a/backend/tests/test_campaign_approval.py b/backend/tests/test_campaign_approval.py index 536a755..c128382 100644 --- a/backend/tests/test_campaign_approval.py +++ b/backend/tests/test_campaign_approval.py @@ -168,3 +168,157 @@ class TestRejectCampaign: def test_reject_nonexistent_campaign_raises(self, db, admin_user): with pytest.raises(EntityNotFoundError): reject_campaign(db, str(uuid.uuid4()), rejecter_id=admin_user.id, reason="No") + + +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"