fix(campaigns,tests): admin cannot create campaigns, manager can delete unstarted tests, intensify red/blue team colors
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
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
- Campaign creation and generate-from-threat-actor now use the strict
role check — admin no longer gets a free pass into campaign content,
same principle already applied to test-template creation.
- New manager-only DELETE /tests/{id}: removes a standalone test still
in draft (not started, not linked to any campaign).
- Replaced the orange/indigo stand-ins used across team badges, tabs,
action buttons, and timers with true red/blue so Red Team and Blue
Team read unambiguously at a glance.
This commit is contained in:
@@ -25,7 +25,7 @@ from sqlalchemy.orm import Session
|
||||
from app.database import get_db
|
||||
|
||||
# Import get_current_user, require_any_role from app.dependencies.auth
|
||||
from app.dependencies.auth import get_current_user, require_any_role
|
||||
from app.dependencies.auth import get_current_user, require_any_role, require_any_role_strict
|
||||
|
||||
# Import UnitOfWork from app.domain.unit_of_work
|
||||
from app.domain.unit_of_work import UnitOfWork
|
||||
@@ -319,7 +319,10 @@ def create_campaign(
|
||||
# Entry: db
|
||||
db: Session = Depends(get_db),
|
||||
# Entry: current_user
|
||||
current_user: User = Depends(require_any_role("red_lead", "blue_lead", "manager")),
|
||||
# Strict variant — admin must NOT get a free pass here. Admin
|
||||
# administers the site, not campaign content; the only admin path to
|
||||
# an active campaign is the emergency /activate override below.
|
||||
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead", "manager")),
|
||||
) -> dict:
|
||||
"""Create a new campaign.
|
||||
|
||||
@@ -987,7 +990,9 @@ def generate_campaign_from_actor(
|
||||
payload: GenerateFromActorPayload = GenerateFromActorPayload(),
|
||||
db: Session = Depends(get_db),
|
||||
# Entry: current_user
|
||||
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
|
||||
# Strict variant — admin must NOT get a free pass here, same as the
|
||||
# plain create_campaign endpoint above.
|
||||
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
|
||||
) -> dict:
|
||||
"""Auto-generate a campaign from a threat actor's uncovered techniques.
|
||||
|
||||
|
||||
@@ -94,6 +94,11 @@ from app.services.test_crud_service import (
|
||||
create_test_from_template as crud_create_from_template,
|
||||
)
|
||||
|
||||
# Import from app.services.test_crud_service
|
||||
from app.services.test_crud_service import (
|
||||
delete_test as crud_delete_test,
|
||||
)
|
||||
|
||||
# Import from app.services.test_crud_service
|
||||
from app.services.test_crud_service import (
|
||||
get_test_detail as crud_get_test_detail,
|
||||
@@ -543,6 +548,38 @@ def update_test(
|
||||
return test
|
||||
|
||||
|
||||
@router.delete("/{test_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_test(
|
||||
test_id: uuid.UUID,
|
||||
db: Session = Depends(get_db),
|
||||
# manager-only, and admin does NOT get a free pass — this is a
|
||||
# queue-cleanup action for the manager role specifically, not a site
|
||||
# administration task.
|
||||
current_user: User = Depends(require_any_role_strict("manager")),
|
||||
) -> None:
|
||||
"""Delete a standalone test that hasn't started yet.
|
||||
|
||||
Only tests still in ``draft`` state and not linked to any campaign can
|
||||
be deleted this way.
|
||||
|
||||
Args:
|
||||
test_id (uuid.UUID): Primary key of the test to delete.
|
||||
db (Session): SQLAlchemy database session.
|
||||
current_user (User): Authenticated manager performing the deletion.
|
||||
"""
|
||||
with UnitOfWork(db) as uow:
|
||||
crud_delete_test(db, test_id)
|
||||
log_action(
|
||||
db,
|
||||
user_id=current_user.id,
|
||||
action="delete_test",
|
||||
entity_type="test",
|
||||
entity_id=test_id,
|
||||
details={},
|
||||
)
|
||||
uow.commit()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PATCH /tests/{id}/classification — admin data classification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -438,6 +438,28 @@ def get_test_or_raise(db: Session, test_id: uuid.UUID) -> Test:
|
||||
return test
|
||||
|
||||
|
||||
def delete_test(db: Session, test_id: uuid.UUID) -> None:
|
||||
"""Delete a standalone, not-yet-started test from the queue.
|
||||
|
||||
Only a test still in ``draft`` (execution hasn't started) and not
|
||||
linked to any campaign may be deleted this way — a test already being
|
||||
worked on, or one that's part of a campaign's plan, must be handled
|
||||
through the normal workflow/campaign-modification paths instead.
|
||||
Raises EntityNotFoundError, BusinessRuleViolation. Does not commit;
|
||||
caller commits.
|
||||
"""
|
||||
test = get_test_or_raise(db, test_id)
|
||||
|
||||
if test.state != TestState.draft:
|
||||
raise BusinessRuleViolation("Only tests that haven't started yet can be deleted")
|
||||
|
||||
in_campaign = db.query(CampaignTest).filter(CampaignTest.test_id == test.id).first()
|
||||
if in_campaign:
|
||||
raise BusinessRuleViolation("Cannot delete a test that belongs to a campaign")
|
||||
|
||||
db.delete(test)
|
||||
|
||||
|
||||
# Define function get_test_with_technique
|
||||
def get_test_with_technique(db: Session, test_id: uuid.UUID) -> Test:
|
||||
"""Fetch a test with technique joined. Raises EntityNotFoundError if not found.
|
||||
|
||||
@@ -39,6 +39,8 @@ DUMMY_ID = str(uuid.uuid4())
|
||||
("post", "/api/v1/tests", {"technique_id": DUMMY_ID, "name": "x"}),
|
||||
("post", "/api/v1/tests/from-template", {"template_id": DUMMY_ID, "technique_id": DUMMY_ID}),
|
||||
("post", "/api/v1/tests/import-rt", {"name": "x", "techniques": []}),
|
||||
("post", "/api/v1/campaigns", {"name": "x"}),
|
||||
("post", f"/api/v1/campaigns/from-threat-actor/{DUMMY_ID}", None),
|
||||
],
|
||||
)
|
||||
def test_admin_forbidden(client, auth_headers, method, path, payload):
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
"""A manager (and only a manager) can delete a standalone test from the
|
||||
queue, but only while it hasn't started yet (draft) and isn't linked to
|
||||
any campaign — a test already in progress, or one that's part of a
|
||||
campaign's plan, must go through the normal workflow instead.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def technique(api, auth_headers):
|
||||
resp = api(
|
||||
"post", "/api/v1/techniques", auth_headers,
|
||||
json={"mitre_id": "T1059.600", "name": "Command Line Delete Test"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()["id"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def draft_test(api, red_lead_headers, technique):
|
||||
resp = api(
|
||||
"post", "/api/v1/tests", red_lead_headers,
|
||||
json={"technique_id": technique, "name": "Standalone draft test"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()["id"]
|
||||
|
||||
|
||||
def test_manager_can_delete_draft_standalone_test(api, manager_headers, draft_test):
|
||||
resp = api("delete", f"/api/v1/tests/{draft_test}", manager_headers)
|
||||
assert resp.status_code == 204, resp.text
|
||||
|
||||
reloaded = api("get", f"/api/v1/tests/{draft_test}", manager_headers)
|
||||
assert reloaded.status_code == 404
|
||||
|
||||
|
||||
def test_red_lead_cannot_delete_test(api, red_lead_headers, draft_test):
|
||||
resp = api("delete", f"/api/v1/tests/{draft_test}", red_lead_headers)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_admin_cannot_delete_test(api, auth_headers, draft_test):
|
||||
resp = api("delete", f"/api/v1/tests/{draft_test}", auth_headers)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_manager_cannot_delete_started_test(api, manager_headers, red_tech_headers, draft_test):
|
||||
started = api("post", f"/api/v1/tests/{draft_test}/start-execution", red_tech_headers)
|
||||
assert started.status_code == 200, started.text
|
||||
|
||||
resp = api("delete", f"/api/v1/tests/{draft_test}", manager_headers)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_manager_cannot_delete_test_linked_to_a_campaign(
|
||||
api, db, manager_headers, red_lead_headers, draft_test,
|
||||
):
|
||||
campaign = api(
|
||||
"post", "/api/v1/campaigns", red_lead_headers,
|
||||
json={"name": "Holds the test"},
|
||||
)
|
||||
assert campaign.status_code == 201, campaign.text
|
||||
campaign_id = campaign.json()["id"]
|
||||
|
||||
add = api(
|
||||
"post", f"/api/v1/campaigns/{campaign_id}/tests", red_lead_headers,
|
||||
json={"test_id": draft_test},
|
||||
)
|
||||
assert add.status_code in (200, 201), add.text
|
||||
|
||||
resp = api("delete", f"/api/v1/tests/{draft_test}", manager_headers)
|
||||
assert resp.status_code == 400
|
||||
Reference in New Issue
Block a user