diff --git a/backend/app/routers/tests.py b/backend/app/routers/tests.py index ffdb075..b1f5af3 100644 --- a/backend/app/routers/tests.py +++ b/backend/app/routers/tests.py @@ -218,6 +218,10 @@ def list_tests( pending_validation_side: Optional[str] = Query( None, description="Filter in_review tests pending validation on 'red' or 'blue' side" ), + reviewer_id: Optional[uuid.UUID] = Query( + None, + description="\"My reviews\" queue — filter red_review/blue_review tests assigned to this reviewer", + ), not_in_any_campaign: bool = Query( False, description="Only return tests not linked to any campaign" ), @@ -259,6 +263,7 @@ def list_tests( created_by=created_by, # Keyword argument: pending_validation_side pending_validation_side=pending_validation_side, + reviewer_id=reviewer_id, not_in_any_campaign=not_in_any_campaign, offset=offset, # Keyword argument: limit diff --git a/backend/app/schemas/test_template.py b/backend/app/schemas/test_template.py index 7d6aed6..4ff2830 100644 --- a/backend/app/schemas/test_template.py +++ b/backend/app/schemas/test_template.py @@ -102,6 +102,9 @@ class TestTemplateSummary(BaseModel): severity: str | None = None # Assign is_active = True is_active: bool = True + # Number of existing Test rows for this template's technique — lets the + # catalog UI warn before creating a likely-duplicate test. + existing_test_count: int = 0 # Assign model_config = ConfigDict(from_attributes=True) model_config = ConfigDict(from_attributes=True) diff --git a/backend/app/services/metrics_query_service.py b/backend/app/services/metrics_query_service.py index debf38a..8085218 100644 --- a/backend/app/services/metrics_query_service.py +++ b/backend/app/services/metrics_query_service.py @@ -38,6 +38,26 @@ from app.schemas.metrics import ( ValidationRate, ) +# Canonical MITRE ATT&CK kill-chain order — used to sort "coverage by +# tactic" consistently everywhere it's rendered (dashboard table, executive +# bar chart), instead of each consumer inventing its own order/subset. +MITRE_TACTIC_ORDER: list[str] = [ + "reconnaissance", + "resource-development", + "initial-access", + "execution", + "persistence", + "privilege-escalation", + "defense-evasion", + "credential-access", + "discovery", + "lateral-movement", + "collection", + "command-and-control", + "exfiltration", + "impact", +] + # Define function get_coverage_summary def get_coverage_summary(db: Session) -> CoverageSummary: @@ -112,6 +132,12 @@ def get_coverage_by_tactic(db: Session) -> list[TacticCoverage]: lambda: {s.value: 0 for s in TechniqueStatus} ) + # Materialize every canonical tactic up front so the response always + # includes all 14, zero-filled, rather than only whichever tactics + # happen to have techniques seeded. + for tactic in MITRE_TACTIC_ORDER: + tactic_data[tactic] # noqa: B018 — touch to create the defaultdict entry + # Iterate over techniques for tactic_str, status in techniques: # Check: not tactic_str @@ -128,10 +154,17 @@ def get_coverage_by_tactic(db: Session) -> list[TacticCoverage]: # Assign tactic_data[tactic][status.value] = 1 tactic_data[tactic][status.value] += 1 + def _tactic_sort_key(tactic: str) -> tuple[int, str]: + try: + return (MITRE_TACTIC_ORDER.index(tactic), tactic) + except ValueError: + # Anything outside the canonical 14 (e.g. "unknown") sorts last. + return (len(MITRE_TACTIC_ORDER), tactic) + # Assign result = [] result = [] # Iterate over sorted(tactic_data) - for tactic in sorted(tactic_data): + for tactic in sorted(tactic_data, key=_tactic_sort_key): # Assign counts = tactic_data[tactic] counts = tactic_data[tactic] # Assign total = sum(counts.values()) diff --git a/backend/app/services/test_crud_service.py b/backend/app/services/test_crud_service.py index 27550e3..db239c7 100644 --- a/backend/app/services/test_crud_service.py +++ b/backend/app/services/test_crud_service.py @@ -10,6 +10,7 @@ from datetime import datetime from typing import Any # Import Session, joinedload from sqlalchemy.orm +from sqlalchemy import or_ from sqlalchemy.orm import Session, joinedload # Import from app.domain.errors @@ -63,6 +64,8 @@ def list_tests( created_by: uuid.UUID | None = None, # Entry: pending_validation_side pending_validation_side: str | None = None, + # Entry: reviewer_id — "my reviews" queue for the red_review/blue_review gate stage + reviewer_id: uuid.UUID | None = None, not_in_any_campaign: bool = False, offset: int = 0, # Entry: limit @@ -101,6 +104,23 @@ def list_tests( Test.state == TestState.in_review, Test.blue_validation_status.in_(["pending", None]), ) + # "My reviews" — tests currently assigned to *reviewer_id* at the + # red_review/blue_review lead-review gate (distinct from + # pending_validation_side, which covers the later in_review stage). + if reviewer_id: + if state == TestState.red_review.value: + query = query.filter(Test.red_reviewer_assignee == reviewer_id) + elif state == TestState.blue_review.value: + query = query.filter(Test.blue_reviewer_assignee == reviewer_id) + else: + query = query.filter( + Test.state.in_([TestState.red_review, TestState.blue_review]), + or_( + Test.red_reviewer_assignee == reviewer_id, + Test.blue_reviewer_assignee == reviewer_id, + ), + ) + if not_in_any_campaign: linked = db.query(CampaignTest.test_id).distinct().subquery() query = query.filter(~Test.id.in_(linked)) diff --git a/backend/app/services/test_template_service.py b/backend/app/services/test_template_service.py index 0340328..ceff86c 100644 --- a/backend/app/services/test_template_service.py +++ b/backend/app/services/test_template_service.py @@ -16,6 +16,8 @@ from app.domain.errors import EntityNotFoundError # Import TestTemplate from app.models.test_template from app.models.test_template import TestTemplate +from app.models.technique import Technique +from app.models.test import Test # Import escape_like from app.utils from app.utils import escape_like @@ -91,6 +93,21 @@ def list_templates( # Chain .all() call .all() ) + + # Attach existing_test_count per template — lets the catalog warn before + # creating a likely-duplicate test for a technique that already has one. + if templates: + mitre_ids = {t.mitre_technique_id for t in templates} + counts = dict( + db.query(Technique.mitre_id, func.count(Test.id)) + .join(Test, Test.technique_id == Technique.id) + .filter(Technique.mitre_id.in_(mitre_ids)) + .group_by(Technique.mitre_id) + .all() + ) + for t in templates: + t.existing_test_count = counts.get(t.mitre_technique_id, 0) + # Return templates return templates diff --git a/backend/app/services/test_workflow_service.py b/backend/app/services/test_workflow_service.py index ff8a72e..defc13e 100644 --- a/backend/app/services/test_workflow_service.py +++ b/backend/app/services/test_workflow_service.py @@ -28,6 +28,7 @@ from sqlalchemy.orm import Session from app.config import settings from app.domain.exceptions import BusinessRuleViolation, InvalidOperationError from app.domain.test_entity import TestEntity +from app.models.campaign import Campaign, CampaignTest from app.models.enums import TestState, TeamSide from app.models.evidence import Evidence from app.models.test import Test @@ -174,8 +175,26 @@ def transition_state( # --------------------------------------------------------------------------- +def _get_campaign_start_date(db: Session, test_id) -> datetime | None: + """Return the scheduled start_date of the campaign *test_id* belongs to, if any.""" + campaign = ( + db.query(Campaign) + .join(CampaignTest, CampaignTest.campaign_id == Campaign.id) + .filter(CampaignTest.test_id == test_id) + .first() + ) + return campaign.start_date if campaign else None + + def start_execution(db: Session, test: Test, user: User) -> Test: """Move from ``draft`` → ``red_executing``.""" + campaign_start_date = _get_campaign_start_date(db, test.id) + if campaign_start_date and campaign_start_date > datetime.utcnow(): + raise BusinessRuleViolation( + "This test's campaign is scheduled to start on " + f"{campaign_start_date.date().isoformat()} — it cannot be started before then." + ) + entity = TestEntity.from_orm(test) # Call entity.start_execution() entity.start_execution() diff --git a/backend/tests/test_campaign_schedule_gate.py b/backend/tests/test_campaign_schedule_gate.py new file mode 100644 index 0000000..e1666f9 --- /dev/null +++ b/backend/tests/test_campaign_schedule_gate.py @@ -0,0 +1,76 @@ +"""Tests for the campaign start_date scheduling gate on test execution. + +Block 3 fix: `Campaign.start_date` was persisted and displayed but never +actually enforced — a test belonging to a campaign scheduled for the future +could still be started immediately. `start_execution` now blocks this. +""" + +from datetime import datetime, timedelta + +import pytest + +from app.domain.errors import BusinessRuleViolation +from app.models.campaign import Campaign, CampaignTest +from app.models.enums import TestState +from app.models.technique import Technique +from app.models.test import Test +from app.services.test_workflow_service import start_execution + + +def _seed_test_in_campaign(db, owner_id, start_date): + tech = Technique(mitre_id="T1059.003", name="Windows Command Shell", tactic="execution", platforms=["windows"]) + db.add(tech) + db.flush() + + campaign = Campaign(name="Scheduled Campaign", type="custom", status="active", created_by=owner_id, start_date=start_date) + db.add(campaign) + db.flush() + + test = Test(technique_id=tech.id, name="Scheduled 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(test) + return test + + +def test_start_execution_blocked_before_campaign_start_date(db, red_tech_user): + future = datetime.utcnow() + timedelta(days=7) + test = _seed_test_in_campaign(db, red_tech_user.id, future) + + with pytest.raises(BusinessRuleViolation): + start_execution(db, test, red_tech_user) + + +def test_start_execution_allowed_after_campaign_start_date(db, red_tech_user): + past = datetime.utcnow() - timedelta(days=1) + test = _seed_test_in_campaign(db, red_tech_user.id, past) + + result = start_execution(db, test, red_tech_user) + + assert result.state == TestState.red_executing + + +def test_start_execution_allowed_for_standalone_test(db, red_tech_user): + """A test with no campaign at all is never gated by this check.""" + tech = Technique(mitre_id="T1059.004", name="Unix Shell", tactic="execution", platforms=["linux"]) + db.add(tech) + db.flush() + test = Test(technique_id=tech.id, name="Standalone test", state=TestState.draft, created_by=red_tech_user.id) + db.add(test) + db.commit() + db.refresh(test) + + result = start_execution(db, test, red_tech_user) + + assert result.state == TestState.red_executing + + +def test_start_execution_allowed_when_campaign_has_no_start_date(db, red_tech_user): + test = _seed_test_in_campaign(db, red_tech_user.id, None) + + result = start_execution(db, test, red_tech_user) + + assert result.state == TestState.red_executing diff --git a/backend/tests/test_coverage_by_tactic_order.py b/backend/tests/test_coverage_by_tactic_order.py new file mode 100644 index 0000000..2bb8cfb --- /dev/null +++ b/backend/tests/test_coverage_by_tactic_order.py @@ -0,0 +1,48 @@ +"""Tests for canonical MITRE kill-chain ordering in coverage-by-tactic. + +Block 3 fix: TacticCoverageChart.tsx (dashboard) rendered whatever order +the backend happened to return (alphabetical, and only tactics with +seeded techniques), while ExecutiveDashboardPage.tsx re-sorted client-side +into the correct kill-chain order. get_coverage_by_tactic now returns the +canonical order — and all 14 tactics, zero-filled — so both consumers get +consistent results without needing their own reordering logic. +""" + +from app.models.technique import Technique +from app.services.metrics_query_service import MITRE_TACTIC_ORDER, get_coverage_by_tactic + + +def test_coverage_by_tactic_returns_canonical_kill_chain_order(db): + # Seed a few tactics out of order to prove the response isn't alphabetical. + db.add(Technique(mitre_id="T3001", name="A", tactic="impact")) + db.add(Technique(mitre_id="T3002", name="B", tactic="execution")) + db.add(Technique(mitre_id="T3003", name="C", tactic="reconnaissance")) + db.commit() + + rows = get_coverage_by_tactic(db) + tactics = [r.tactic for r in rows] + + canonical_present = [t for t in tactics if t in MITRE_TACTIC_ORDER] + assert canonical_present == MITRE_TACTIC_ORDER + + +def test_coverage_by_tactic_zero_fills_tactics_with_no_techniques(db): + db.add(Technique(mitre_id="T3004", name="D", tactic="execution")) + db.commit() + + rows = get_coverage_by_tactic(db) + by_tactic = {r.tactic: r for r in rows} + + assert "persistence" in by_tactic + assert by_tactic["persistence"].total == 0 + assert by_tactic["execution"].total == 1 + + +def test_coverage_by_tactic_unknown_tactic_sorts_last(db): + db.add(Technique(mitre_id="T3005", name="E", tactic=None)) + db.commit() + + rows = get_coverage_by_tactic(db) + tactics = [r.tactic for r in rows] + + assert tactics[-1] == "unknown" diff --git a/backend/tests/test_reviewer_queue_filter.py b/backend/tests/test_reviewer_queue_filter.py new file mode 100644 index 0000000..209361d --- /dev/null +++ b/backend/tests/test_reviewer_queue_filter.py @@ -0,0 +1,60 @@ +"""Tests for the `reviewer_id` "my reviews" queue filter on GET /tests. + +Block 3 fix: red_lead/blue_lead had no way to see tests specifically +assigned to THEM at the red_review/blue_review lead-review gate — the +existing `pending_validation_side` filter only covers the later, unrelated +in_review manager-validation stage and ignores assignment entirely. +""" + +import uuid + +from app.models.enums import TestState +from app.models.technique import Technique +from app.models.test import Test +from app.services.test_crud_service import list_tests + + +def _make_test(db, owner_id, state, name, red_reviewer=None, blue_reviewer=None): + tech = Technique(mitre_id=f"T9{uuid.uuid4().hex[:6]}", name="X", tactic="execution", platforms=["windows"]) + db.add(tech) + db.flush() + test = Test( + technique_id=tech.id, + name=name, + state=state, + created_by=owner_id, + red_reviewer_assignee=red_reviewer, + blue_reviewer_assignee=blue_reviewer, + ) + db.add(test) + db.commit() + db.refresh(test) + return test + + +def test_reviewer_id_with_explicit_state_filters_by_matching_assignee(db, red_lead_user, blue_lead_user): + mine = _make_test(db, red_lead_user.id, TestState.red_review, "Mine", red_reviewer=red_lead_user.id) + _make_test(db, red_lead_user.id, TestState.red_review, "Someone else's", red_reviewer=blue_lead_user.id) + + results = list_tests(db, state="red_review", reviewer_id=red_lead_user.id) + + assert [t.id for t in results] == [mine.id] + + +def test_reviewer_id_without_state_covers_both_red_and_blue_review(db, red_lead_user, blue_lead_user): + red = _make_test(db, red_lead_user.id, TestState.red_review, "Red review mine", red_reviewer=red_lead_user.id) + blue = _make_test(db, red_lead_user.id, TestState.blue_review, "Blue review mine", blue_reviewer=red_lead_user.id) + _make_test(db, red_lead_user.id, TestState.red_review, "Not mine", red_reviewer=blue_lead_user.id) + _make_test(db, red_lead_user.id, TestState.draft, "Unrelated state") + + results = list_tests(db, reviewer_id=red_lead_user.id) + + assert {t.id for t in results} == {red.id, blue.id} + + +def test_reviewer_id_excludes_unassigned_tests_in_that_state(db, red_lead_user): + _make_test(db, red_lead_user.id, TestState.red_review, "Unassigned", red_reviewer=None) + + results = list_tests(db, state="red_review", reviewer_id=red_lead_user.id) + + assert results == [] diff --git a/backend/tests/test_t106_workflow_service.py b/backend/tests/test_t106_workflow_service.py index 862dbf0..cb2fa22 100644 --- a/backend/tests/test_t106_workflow_service.py +++ b/backend/tests/test_t106_workflow_service.py @@ -60,7 +60,11 @@ def _make_user(role: str = "red_tech") -> MagicMock: def _make_db() -> MagicMock: - return MagicMock() + db = MagicMock() + # start_execution's campaign-schedule lookup (.query(Campaign).join(...).filter(...).first()) + # defaults to "no campaign found" so unrelated tests aren't gated by a MagicMock date. + db.query.return_value.join.return_value.filter.return_value.first.return_value = None + return db # --------------------------------------------------------------------------- diff --git a/backend/tests/test_template_existing_test_count.py b/backend/tests/test_template_existing_test_count.py new file mode 100644 index 0000000..a87a1e6 --- /dev/null +++ b/backend/tests/test_template_existing_test_count.py @@ -0,0 +1,41 @@ +"""Tests for the `existing_test_count` field on TestTemplateSummary. + +Block 3 fix: the Test Catalog page had no way to know a technique already +had one or more tests, so operators could unknowingly instantiate a +duplicate from a template. list_templates now attaches a per-technique +test count so the frontend can show a warning badge. +""" + +from app.models.enums import TestState +from app.models.technique import Technique +from app.models.test import Test +from app.models.test_template import TestTemplate +from app.services.test_template_service import list_templates + + +def test_existing_test_count_reflects_tests_for_that_technique(db, red_lead_user): + tech = Technique(mitre_id="T1055", name="Process Injection", tactic="defense-evasion", platforms=["windows"]) + db.add(tech) + db.flush() + db.add(TestTemplate(mitre_technique_id="T1055", name="Injection template", source="custom")) + db.add(Test(technique_id=tech.id, name="Existing test 1", state=TestState.draft, created_by=red_lead_user.id)) + db.add(Test(technique_id=tech.id, name="Existing test 2", state=TestState.validated, created_by=red_lead_user.id)) + db.commit() + + templates = list_templates(db, mitre_technique_id="T1055") + + assert len(templates) == 1 + assert templates[0].existing_test_count == 2 + + +def test_existing_test_count_zero_when_no_tests(db): + tech = Technique(mitre_id="T1056", name="Input Capture", tactic="collection", platforms=["windows"]) + db.add(tech) + db.flush() + db.add(TestTemplate(mitre_technique_id="T1056", name="Capture template", source="custom")) + db.commit() + + templates = list_templates(db, mitre_technique_id="T1056") + + assert len(templates) == 1 + assert templates[0].existing_test_count == 0 diff --git a/backend/tests/test_workflow.py b/backend/tests/test_workflow.py index ac46b65..50d6b69 100644 --- a/backend/tests/test_workflow.py +++ b/backend/tests/test_workflow.py @@ -153,7 +153,11 @@ def _make_user(role: str = "red_tech") -> MagicMock: def _make_db() -> MagicMock: - return MagicMock() + db = MagicMock() + # start_execution's campaign-schedule lookup (.query(Campaign).join(...).filter(...).first()) + # defaults to "no campaign found" so unrelated tests aren't gated by a MagicMock date. + db.query.return_value.join.return_value.filter.return_value.first.return_value = None + return db # ===========================================================================