feat(templates,campaigns): validate MITRE technique IDs, surface template suggestions first, manager can re-approve rejected campaigns
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
- TestTemplate/TemplateSuggestion creation and updates now reject any mitre_technique_id that doesn't match a real, already-synced MITRE ATT&CK technique. Frontend swaps the free-text ID input for a technique picker. - Test Catalog now shows pending template suggestions before the catalog grid, with full field detail (platform, severity, tool, atomic ID, source URL, remediation) instead of just name/procedure. - A manager can edit and directly re-approve a campaign they previously rejected, instead of needing the original lead to resubmit it.
This commit is contained in:
@@ -431,15 +431,19 @@ def update_campaign(
|
||||
# Entry: db
|
||||
db: Session = Depends(get_db),
|
||||
# Entry: current_user
|
||||
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
|
||||
current_user: User = Depends(require_any_role("red_lead", "blue_lead", "manager")),
|
||||
) -> dict:
|
||||
"""Update a campaign. Only allowed in draft or active state.
|
||||
|
||||
A manager may only edit a campaign that's sitting in draft with a
|
||||
rejection_reason set (i.e. one they previously rejected) — see
|
||||
``update_campaign()``'s ownership check.
|
||||
|
||||
Args:
|
||||
campaign_id (str): UUID string of the campaign to update.
|
||||
payload (CampaignUpdate): Partial update payload; only set fields are applied.
|
||||
db (Session): SQLAlchemy database session.
|
||||
current_user (User): Authenticated red_lead or blue_lead performing the update.
|
||||
current_user (User): Authenticated red_lead, blue_lead, or manager performing the update.
|
||||
|
||||
Returns:
|
||||
dict: Serialised representation of the updated campaign.
|
||||
|
||||
@@ -367,6 +367,11 @@ def approve_campaign(
|
||||
) -> Campaign:
|
||||
"""Approve a pending campaign: fix its start date and activate it.
|
||||
|
||||
Also allows approving a previously-rejected campaign directly (status
|
||||
``draft`` with a ``rejection_reason`` set) — a manager can edit it via
|
||||
``update_campaign()`` and then approve it themselves, without needing
|
||||
the original lead to resubmit it through the normal queue.
|
||||
|
||||
Raises EntityNotFoundError, BusinessRuleViolation.
|
||||
Does not commit; caller commits.
|
||||
"""
|
||||
@@ -374,8 +379,9 @@ def approve_campaign(
|
||||
if not campaign:
|
||||
raise EntityNotFoundError("Campaign", campaign_id)
|
||||
|
||||
if campaign.status != "pending_approval":
|
||||
raise BusinessRuleViolation("Only campaigns pending approval can be approved")
|
||||
is_previously_rejected_draft = campaign.status == "draft" and campaign.rejection_reason
|
||||
if campaign.status != "pending_approval" and not is_previously_rejected_draft:
|
||||
raise BusinessRuleViolation("Only campaigns pending approval (or previously rejected) can be approved")
|
||||
|
||||
if not start_date:
|
||||
raise BusinessRuleViolation("start_date is required to approve a campaign")
|
||||
@@ -454,7 +460,7 @@ def update_campaign(
|
||||
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()
|
||||
campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first()
|
||||
# Check: not campaign
|
||||
if not campaign:
|
||||
# Raise EntityNotFoundError
|
||||
@@ -465,10 +471,16 @@ def update_campaign(
|
||||
# Raise BusinessRuleViolation
|
||||
raise BusinessRuleViolation("Can only update draft or active campaigns")
|
||||
|
||||
# Check: str(campaign.created_by) != str(updater_id) and updater_role != "ad...
|
||||
if str(campaign.created_by) != str(updater_id) and updater_role != "admin":
|
||||
# A manager may edit a campaign they (or another manager) previously
|
||||
# rejected — this is what lets them fix it up and self-approve it,
|
||||
# instead of bouncing it back to the original lead to resubmit.
|
||||
is_manager_editing_own_rejection = (
|
||||
updater_role == "manager" and campaign.status == "draft" and campaign.rejection_reason
|
||||
)
|
||||
is_owner_or_admin = str(campaign.created_by) == str(updater_id) or updater_role == "admin"
|
||||
if not is_owner_or_admin and not is_manager_editing_own_rejection:
|
||||
# Raise PermissionViolation
|
||||
raise PermissionViolation("Only the creator or admin can update this campaign")
|
||||
raise PermissionViolation("Only the creator, admin, or a manager reviewing a rejected campaign can update this campaign")
|
||||
|
||||
# Check: "scheduled_at" in fields and fields["scheduled_at"]
|
||||
if "scheduled_at" in fields and fields["scheduled_at"]:
|
||||
|
||||
@@ -19,7 +19,7 @@ from app.models.template_suggestion import TemplateSuggestion
|
||||
from app.models.test_template import TestTemplate
|
||||
from app.models.user import User
|
||||
from app.services.notification_service import notify_role
|
||||
from app.services.test_template_service import create_template
|
||||
from app.services.test_template_service import create_template, validate_mitre_technique_id
|
||||
|
||||
_TEAM_BY_ROLE = {"red_tech": "red", "blue_tech": "blue"}
|
||||
_LEAD_ROLE = {"red": "red_lead", "blue": "blue_lead"}
|
||||
@@ -45,6 +45,7 @@ def create_template_suggestion(db: Session, submitter: User, **fields: object) -
|
||||
Does not commit; caller uses UnitOfWork.
|
||||
"""
|
||||
team = team_for_submitter(submitter)
|
||||
validate_mitre_technique_id(db, fields["mitre_technique_id"])
|
||||
suggestion = TemplateSuggestion(team=team, submitted_by=submitter.id, **fields)
|
||||
db.add(suggestion)
|
||||
db.flush()
|
||||
|
||||
@@ -12,7 +12,7 @@ from sqlalchemy import func, or_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# Import EntityNotFoundError from app.domain.errors
|
||||
from app.domain.errors import EntityNotFoundError
|
||||
from app.domain.errors import BusinessRuleViolation, EntityNotFoundError
|
||||
|
||||
# Import TestTemplate from app.models.test_template
|
||||
from app.models.test_template import TestTemplate
|
||||
@@ -235,9 +235,22 @@ def get_template_or_raise(db: Session, template_id: uuid.UUID) -> TestTemplate:
|
||||
return template
|
||||
|
||||
|
||||
def validate_mitre_technique_id(db: Session, mitre_technique_id: str) -> None:
|
||||
"""Raise BusinessRuleViolation unless *mitre_technique_id* matches a real,
|
||||
already-synced MITRE ATT&CK technique — free text like "made up" or a
|
||||
typo'd ID would otherwise silently create an orphaned template that can
|
||||
never be found via technique lookups or coverage reporting."""
|
||||
exists = db.query(Technique).filter(Technique.mitre_id == mitre_technique_id).first()
|
||||
if not exists:
|
||||
raise BusinessRuleViolation(
|
||||
f"'{mitre_technique_id}' is not a known MITRE ATT&CK technique ID"
|
||||
)
|
||||
|
||||
|
||||
# Define function create_template
|
||||
def create_template(db: Session, **fields: object) -> TestTemplate:
|
||||
"""Create a test template from keyword args (e.g. payload.model_dump()). Does NOT commit."""
|
||||
validate_mitre_technique_id(db, fields["mitre_technique_id"])
|
||||
# Assign template = TestTemplate(**fields)
|
||||
template = TestTemplate(**fields)
|
||||
# Stage new record(s) for database insertion
|
||||
@@ -249,6 +262,8 @@ def create_template(db: Session, **fields: object) -> TestTemplate:
|
||||
# Define function update_template
|
||||
def update_template(db: Session, template_id: uuid.UUID, **fields: object) -> TestTemplate:
|
||||
"""Update an existing template. Raises EntityNotFoundError if not found. Does NOT commit."""
|
||||
if fields.get("mitre_technique_id"):
|
||||
validate_mitre_technique_id(db, fields["mitre_technique_id"])
|
||||
# Assign template = get_template_or_raise(db, template_id)
|
||||
template = get_template_or_raise(db, template_id)
|
||||
# Iterate over fields.items()
|
||||
|
||||
@@ -417,3 +417,56 @@ def test_lead_created_campaign_is_still_draft_not_auto_approved(api, db, red_lea
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
assert resp.json()["status"] == "draft"
|
||||
|
||||
|
||||
def test_manager_can_edit_a_campaign_they_rejected(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)
|
||||
reject = api(
|
||||
"post", f"/api/v1/campaigns/{campaign.id}/reject", manager_headers,
|
||||
json={"reason": "Needs a clearer scope"},
|
||||
)
|
||||
assert reject.status_code == 200, reject.text
|
||||
assert reject.json()["status"] == "draft"
|
||||
|
||||
resp = api(
|
||||
"patch", f"/api/v1/campaigns/{campaign.id}", manager_headers,
|
||||
json={"name": "Edited by manager after rejection"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["name"] == "Edited by manager after rejection"
|
||||
|
||||
|
||||
def test_manager_cannot_edit_a_draft_campaign_that_was_never_rejected(
|
||||
api, db, red_lead_user, red_lead_headers, manager_headers,
|
||||
):
|
||||
campaign = _make_draft_campaign(db, red_lead_user.id)
|
||||
resp = api(
|
||||
"patch", f"/api/v1/campaigns/{campaign.id}", manager_headers,
|
||||
json={"name": "Manager should not be able to do this"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_manager_can_approve_a_campaign_they_previously_rejected(
|
||||
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)
|
||||
api(
|
||||
"post", f"/api/v1/campaigns/{campaign.id}/reject", manager_headers,
|
||||
json={"reason": "Needs a clearer scope"},
|
||||
)
|
||||
api(
|
||||
"patch", f"/api/v1/campaigns/{campaign.id}", manager_headers,
|
||||
json={"description": "Scope clarified after rejection"},
|
||||
)
|
||||
|
||||
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, resp.text
|
||||
body = resp.json()
|
||||
assert body["status"] == "active"
|
||||
assert body["rejection_reason"] is None
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
"""A TestTemplate/TemplateSuggestion's mitre_technique_id must correspond
|
||||
to a real, already-synced MITRE ATT&CK technique — free text or a typo'd
|
||||
ID would otherwise create an orphaned template invisible to technique
|
||||
lookups and coverage reporting.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def technique(api, auth_headers):
|
||||
resp = api(
|
||||
"post", "/api/v1/techniques", auth_headers,
|
||||
json={"mitre_id": "T1059.500", "name": "Command Line MITRE Validation"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()["id"]
|
||||
|
||||
|
||||
class TestTemplateCreateValidation:
|
||||
def test_create_template_rejects_unknown_mitre_id(self, api, red_lead_headers, technique):
|
||||
resp = api(
|
||||
"post", "/api/v1/test-templates", red_lead_headers,
|
||||
json={"mitre_technique_id": "T9999.999", "name": "Bogus template"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "T9999.999" in resp.text
|
||||
|
||||
def test_create_template_accepts_known_mitre_id(self, api, red_lead_headers, technique):
|
||||
resp = api(
|
||||
"post", "/api/v1/test-templates", red_lead_headers,
|
||||
json={"mitre_technique_id": "T1059.500", "name": "Valid template"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
|
||||
def test_update_template_rejects_unknown_mitre_id(self, api, red_lead_headers, technique):
|
||||
created = api(
|
||||
"post", "/api/v1/test-templates", red_lead_headers,
|
||||
json={"mitre_technique_id": "T1059.500", "name": "Will be edited"},
|
||||
)
|
||||
template_id = created.json()["id"]
|
||||
|
||||
resp = api(
|
||||
"patch", f"/api/v1/test-templates/{template_id}", red_lead_headers,
|
||||
json={"mitre_technique_id": "T0000.000", "name": "Will be edited"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
class TestTemplateSuggestionValidation:
|
||||
def test_propose_template_rejects_unknown_mitre_id(self, api, red_tech_headers, technique):
|
||||
resp = api(
|
||||
"post", "/api/v1/template-suggestions", red_tech_headers,
|
||||
json={"mitre_technique_id": "T8888.888", "name": "Bogus proposal"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_propose_template_accepts_known_mitre_id(self, api, red_tech_headers, technique):
|
||||
resp = api(
|
||||
"post", "/api/v1/template-suggestions", red_tech_headers,
|
||||
json={"mitre_technique_id": "T1059.500", "name": "Valid proposal"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
|
||||
def test_approve_suggestion_with_edited_unknown_mitre_id_rejected(
|
||||
self, api, red_tech_headers, red_lead_headers, technique,
|
||||
):
|
||||
created = api(
|
||||
"post", "/api/v1/template-suggestions", red_tech_headers,
|
||||
json={"mitre_technique_id": "T1059.500", "name": "Valid proposal"},
|
||||
)
|
||||
suggestion_id = created.json()["id"]
|
||||
|
||||
resp = api(
|
||||
"post", f"/api/v1/template-suggestions/{suggestion_id}/approve", red_lead_headers,
|
||||
json={"mitre_technique_id": "T7777.777"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
Reference in New Issue
Block a user