kitos
4809c4a662
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
A manager organizes and validates work, so their own campaigns skip the draft -> submit -> pending_approval queue and go straight to active with the start_date they provide (they're the same role that would otherwise approve it). Manager can also now create tests from the catalog, same as red_lead/blue_lead.
420 lines
16 KiB
Python
420 lines
16 KiB
Python
"""Router-level tests for the campaign manager-approval workflow."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
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(api, db, red_lead_user, red_lead_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
resp = api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "pending_approval"
|
|
|
|
|
|
def test_red_tech_cannot_submit(api, db, red_lead_user, red_tech_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
resp = api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_tech_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_manager_can_approve_and_sets_start_date(api, db, red_lead_user, red_lead_headers, manager_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = api(
|
|
"post",
|
|
f"/api/v1/campaigns/{campaign.id}/approve",
|
|
manager_headers,
|
|
json={"start_date": "2020-01-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] == "active"
|
|
assert body["start_date"] is not None
|
|
|
|
|
|
def test_manager_approval_creates_jira_tickets_when_start_date_is_due(
|
|
api, db, red_lead_user, red_lead_headers, manager_headers
|
|
):
|
|
"""The normal manager-approval path must create Jira tickets too, not just
|
|
the admin-only emergency /activate override — this was the Block 1 gap:
|
|
campaigns approved through the standard flow never got a Jira ticket.
|
|
Uses a past start_date so ticket creation is due immediately."""
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
with patch("app.services.jira_service.get_campaign_jira_key", return_value=None) as mock_get_key, \
|
|
patch("app.services.jira_service.auto_create_campaign_issue", return_value="PT-100") as mock_create_campaign, \
|
|
patch("app.services.jira_service.get_test_jira_key", return_value=None), \
|
|
patch("app.services.jira_service.auto_create_test_issue") as mock_create_test:
|
|
resp = api(
|
|
"post",
|
|
f"/api/v1/campaigns/{campaign.id}/approve",
|
|
manager_headers,
|
|
json={"start_date": "2020-01-01T00:00:00"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
mock_get_key.assert_called_once()
|
|
mock_create_campaign.assert_called_once()
|
|
mock_create_test.assert_called_once()
|
|
|
|
|
|
def test_manager_approval_skips_jira_tickets_when_start_date_is_future(
|
|
api, db, red_lead_user, red_lead_headers, manager_headers
|
|
):
|
|
"""A campaign scheduled to start in the future must not get its Jira
|
|
tickets created at approval time — that must wait for the periodic
|
|
due-campaign sync job, once the scheduled start_date actually arrives."""
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
with patch("app.services.jira_service.get_campaign_jira_key", return_value=None) as mock_get_key, \
|
|
patch("app.services.jira_service.auto_create_campaign_issue", return_value="PT-100") as mock_create_campaign:
|
|
resp = api(
|
|
"post",
|
|
f"/api/v1/campaigns/{campaign.id}/approve",
|
|
manager_headers,
|
|
json={"start_date": "2099-01-01T00:00:00"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
mock_get_key.assert_not_called()
|
|
mock_create_campaign.assert_not_called()
|
|
|
|
|
|
def test_lead_cannot_approve(api, db, red_lead_user, red_lead_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = api(
|
|
"post",
|
|
f"/api/v1/campaigns/{campaign.id}/approve",
|
|
red_lead_headers,
|
|
json={"start_date": "2026-09-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_admin_can_approve_as_manager_backup(api, db, red_lead_user, red_lead_headers, auth_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = api(
|
|
"post",
|
|
f"/api/v1/campaigns/{campaign.id}/approve",
|
|
auth_headers,
|
|
json={"start_date": "2026-09-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_manager_can_reject_with_reason(api, db, red_lead_user, red_lead_headers, manager_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = api(
|
|
"post",
|
|
f"/api/v1/campaigns/{campaign.id}/reject",
|
|
manager_headers,
|
|
json={"reason": "Needs more detail"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "draft"
|
|
|
|
|
|
def test_manager_reject_without_reason_rejected_by_validation(api, db, red_lead_user, red_lead_headers, manager_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = api(
|
|
"post",
|
|
f"/api/v1/campaigns/{campaign.id}/reject",
|
|
manager_headers,
|
|
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
|
|
|
|
|
|
def test_get_campaign_timeline(api, db, red_lead_user, red_lead_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = api("get", f"/api/v1/campaigns/{campaign.id}/timeline", 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(api, db, red_lead_user, red_lead_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
resp = api("post", f"/api/v1/campaigns/{campaign.id}/activate", red_lead_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_admin_can_still_directly_activate_as_override(api, db, red_lead_user, red_lead_headers, auth_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
resp = api("post", f"/api/v1/campaigns/{campaign.id}/activate", auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "active"
|
|
|
|
|
|
def test_manager_cannot_directly_activate_draft_campaign(api, db, red_lead_user, manager_headers):
|
|
"""A manager approves via /approve — the direct /activate bypass is admin-only."""
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
resp = api("post", f"/api/v1/campaigns/{campaign.id}/activate", manager_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_create_campaign_payload_has_no_start_date_field(api, db, red_lead_headers):
|
|
resp = api(
|
|
"post",
|
|
"/api/v1/campaigns",
|
|
red_lead_headers,
|
|
json={"name": "No date campaign", "start_date": "2026-01-01"},
|
|
)
|
|
assert resp.status_code == 201
|
|
# start_date silently ignored for a lead — only a manager's own
|
|
# creation (or a later /approve) can ever set it.
|
|
assert resp.json()["start_date"] is None
|
|
|
|
|
|
def test_manager_can_create_campaign_auto_approved(api, db, manager_headers):
|
|
"""A manager is the same role that would otherwise approve a campaign,
|
|
so their own campaign skips the draft -> submit -> pending_approval
|
|
queue and goes straight to active with the start_date they provide."""
|
|
resp = api(
|
|
"post",
|
|
"/api/v1/campaigns",
|
|
manager_headers,
|
|
json={"name": "Manager-created campaign", "start_date": "2020-01-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
body = resp.json()
|
|
assert body["status"] == "active"
|
|
assert body["start_date"] is not None
|
|
|
|
|
|
def test_manager_campaign_creation_requires_start_date(api, db, manager_headers):
|
|
resp = api(
|
|
"post",
|
|
"/api/v1/campaigns",
|
|
manager_headers,
|
|
json={"name": "Manager-created campaign, no date"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_manager_creation_creates_jira_tickets_when_start_date_is_due(api, db, manager_headers):
|
|
with patch(
|
|
"app.services.jira_service.get_campaign_jira_key", return_value=None
|
|
) as mock_get_key, patch(
|
|
"app.services.jira_service.auto_create_campaign_issue", return_value="PT-300"
|
|
) as mock_create_campaign:
|
|
resp = api(
|
|
"post",
|
|
"/api/v1/campaigns",
|
|
manager_headers,
|
|
json={"name": "Manager-created, due now", "start_date": "2020-01-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
mock_get_key.assert_called_once()
|
|
mock_create_campaign.assert_called_once()
|
|
|
|
|
|
def test_manager_creation_skips_jira_tickets_when_start_date_is_future(api, db, manager_headers):
|
|
with patch(
|
|
"app.services.jira_service.get_campaign_jira_key", return_value=None
|
|
) as mock_get_key:
|
|
resp = api(
|
|
"post",
|
|
"/api/v1/campaigns",
|
|
manager_headers,
|
|
json={"name": "Manager-created, future", "start_date": "2099-01-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
mock_get_key.assert_not_called()
|
|
|
|
|
|
def test_lead_created_campaign_is_still_draft_not_auto_approved(api, db, red_lead_headers):
|
|
resp = api(
|
|
"post",
|
|
"/api/v1/campaigns",
|
|
red_lead_headers,
|
|
json={"name": "Lead-created campaign"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
assert resp.json()["status"] == "draft"
|