From 6186c246a4e313237815584ed3acffac73e59da7 Mon Sep 17 00:00:00 2001 From: kitos Date: Fri, 3 Jul 2026 12:34:18 +0200 Subject: [PATCH] feat(campaigns): modification-request router endpoints --- backend/app/routers/campaigns.py | 158 ++++++++++++++++++ .../tests/test_campaign_approval_router.py | 152 +++++++++++++++++ 2 files changed, 310 insertions(+) diff --git a/backend/app/routers/campaigns.py b/backend/app/routers/campaigns.py index fc5c213..a29223c 100644 --- a/backend/app/routers/campaigns.py +++ b/backend/app/routers/campaigns.py @@ -100,6 +100,21 @@ from app.services.campaign_crud_service import ( from app.services.campaign_crud_service import ( reject_campaign as crud_reject, ) +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, +) # Import log_action from app.services.audit_service from app.services.audit_service import log_action @@ -194,6 +209,22 @@ class RejectPayload(BaseModel): reason: str +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 + + # --------------------------------------------------------------------------- # GET /campaigns — List campaigns with filters # --------------------------------------------------------------------------- @@ -501,6 +532,133 @@ def reject_campaign_endpoint( return serialize_campaign(db, campaign) +# --------------------------------------------------------------------------- +# 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), + # admin passes automatically via require_any_role's built-in bypass — do not add "admin" here + 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), + # admin passes automatically via require_any_role's built-in bypass — do not add "admin" here + 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) if request.test_id else None, + }, + ) + 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), + # admin passes automatically via require_any_role's built-in bypass — do not add "admin" here + 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) + + # --------------------------------------------------------------------------- # DELETE /campaigns/{id} — Delete campaign # --------------------------------------------------------------------------- diff --git a/backend/tests/test_campaign_approval_router.py b/backend/tests/test_campaign_approval_router.py index 844a1a0..e407e32 100644 --- a/backend/tests/test_campaign_approval_router.py +++ b/backend/tests/test_campaign_approval_router.py @@ -102,3 +102,155 @@ def test_manager_reject_without_reason_rejected_by_validation(api, db, red_lead_ json={"reason": ""}, ) assert resp.status_code == 400 + + +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(api, 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 = api( + "post", + f"/api/v1/campaigns/{campaign.id}/modification-requests", + red_lead_headers, + json={"action": "add_test", "test_id": str(new_test.id), "justification": "Coverage gap"}, + ) + assert resp.status_code == 201 + assert resp.json()["status"] == "pending" + + +def test_modification_request_without_justification_rejected(api, 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 = api( + "post", + f"/api/v1/campaigns/{campaign.id}/modification-requests", + red_lead_headers, + json={"action": "remove_test", "test_id": str(ct.test_id), "justification": ""}, + ) + assert resp.status_code == 400 + + +def test_manager_can_list_pending_modification_requests(api, 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() + + api( + "post", + f"/api/v1/campaigns/{campaign.id}/modification-requests", + red_lead_headers, + json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"}, + ) + + resp = api("get", "/api/v1/campaigns/modification-requests/pending", manager_headers) + assert resp.status_code == 200 + assert len(resp.json()) == 1 + + +def test_manager_can_approve_modification_request(api, 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 = api( + "post", + f"/api/v1/campaigns/{campaign.id}/modification-requests", + red_lead_headers, + json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"}, + ) + request_id = create_resp.json()["id"] + + resp = api( + "post", + f"/api/v1/campaigns/modification-requests/{request_id}/approve", + 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(api, 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 = api( + "post", + f"/api/v1/campaigns/{campaign.id}/modification-requests", + red_lead_headers, + json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"}, + ) + request_id = create_resp.json()["id"] + + resp = api( + "post", + f"/api/v1/campaigns/modification-requests/{request_id}/reject", + manager_headers, + json={"review_notes": "Still needed"}, + ) + 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(api, 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 = api( + "post", + f"/api/v1/campaigns/{campaign.id}/modification-requests", + red_lead_headers, + json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"}, + ) + request_id = create_resp.json()["id"] + + resp = api( + "post", + f"/api/v1/campaigns/modification-requests/{request_id}/approve", + red_lead_headers, + ) + assert resp.status_code == 403 + + +def test_direct_add_test_blocked_on_active_campaign_via_router(api, 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 = api( + "post", + f"/api/v1/campaigns/{campaign.id}/tests", + red_lead_headers, + json={"test_id": str(new_test.id)}, + ) + assert resp.status_code == 400