diff --git a/backend/app/routers/tests.py b/backend/app/routers/tests.py index 3310bd4..7d2b930 100644 --- a/backend/app/routers/tests.py +++ b/backend/app/routers/tests.py @@ -10,13 +10,19 @@ PATCH /tests/{id} — general update (draft/rejected only) PATCH /tests/{id}/red — Red Team updates (draft, red_executing) PATCH /tests/{id}/blue — Blue Team updates (blue_evaluating) POST /tests/{id}/start-execution — draft → red_executing -POST /tests/{id}/submit-red — red_executing → blue_evaluating +POST /tests/{id}/submit-red — red_executing → red_review +POST /tests/{id}/review-red — assigned Red Lead approves/reopens POST /tests/{id}/start-blue-work — blue tech picks up (sets Tempo timer) -POST /tests/{id}/submit-blue — blue_evaluating → in_review +POST /tests/{id}/submit-blue — blue_evaluating → blue_review +POST /tests/{id}/review-blue — assigned Blue Lead approves/reopens/flags gap POST /tests/{id}/validate-red — Red Lead validates POST /tests/{id}/validate-blue — Blue Lead validates POST /tests/{id}/reopen — rejected → draft GET /tests/{id}/timeline — audit-log history for this test + +GET /tests/{id} hides the other team's fields while a test is blind +(draft through blue_review) for red_tech/red_lead/blue_tech/blue_lead +viewers — admin and viewer always see everything. """ import base64 @@ -52,12 +58,14 @@ from app.models.user import User # Import from app.schemas.test from app.schemas.test import ( TestAssign, + TestBlueReview, TestBlueUpdate, TestBlueValidate, TestClassificationUpdate, TestCreate, TestHold, TestOut, + TestRedReview, TestRedUpdate, TestRedValidate, TestRemediationUpdate, @@ -126,7 +134,12 @@ from app.services.test_crud_service import ( from app.services.test_workflow_service import ( start_execution as wf_start_execution, submit_red_evidence as wf_submit_red, + approve_red_review as wf_approve_red_review, + reopen_red_review as wf_reopen_red_review, submit_blue_evidence as wf_submit_blue, + approve_blue_review as wf_approve_blue_review, + reopen_blue_review as wf_reopen_blue_review, + flag_blue_review_gap as wf_flag_blue_review_gap, start_blue_work as wf_start_blue_work, validate_as_red_lead as wf_validate_red, validate_as_blue_lead as wf_validate_blue, @@ -141,6 +154,46 @@ from app.services.test_workflow_service import ( router = APIRouter(prefix="/tests", tags=["tests"]) +# --------------------------------------------------------------------------- +# Blind visibility — hide the other team's fields until both reviews pass +# --------------------------------------------------------------------------- + +_RED_ONLY_FIELDS = [ + "procedure_text", "tool_used", "attack_success", + "execution_start_time", "execution_end_time", "red_summary", + "red_validation_status", "red_validated_by", "red_validated_at", "red_validation_notes", +] +_BLUE_ONLY_FIELDS = [ + "detection_result", "containment_result", "detection_time", "containment_time", + "blue_summary", "blue_validation_status", "blue_validated_by", "blue_validated_at", + "blue_validation_notes", "system_gaps", +] +_BLIND_STATES = {"draft", "red_executing", "red_review", "blue_evaluating", "blue_review"} + + +def _mask_for_team_blindness(test_out: TestOut, *, viewer_role: str) -> TestOut: + """Null out the other team's fields while the test is still blind. + + admin and viewer are never blinded. Once the test reaches in_review or + beyond, both sides see everything (existing cross-validation behavior). + """ + if viewer_role in ("admin", "viewer"): + return test_out + + test_state = test_out.state.value if hasattr(test_out.state, "value") else str(test_out.state) + if test_state not in _BLIND_STATES: + return test_out + + if viewer_role in ("blue_tech", "blue_lead"): + hide_fields = _RED_ONLY_FIELDS + elif viewer_role in ("red_tech", "red_lead"): + hide_fields = _BLUE_ONLY_FIELDS + else: + return test_out + + return test_out.model_copy(update={f: None for f in hide_fields}) + + # --------------------------------------------------------------------------- # GET /tests — list with filters # --------------------------------------------------------------------------- @@ -406,9 +459,12 @@ def get_test( Returns: TestOut: Full test detail including split red/blue evidence lists. + Fields belonging to the other team are nulled out while the + test is blind (see :func:`_mask_for_team_blindness`). """ - # Return crud_get_test_detail(db, test_id) - return crud_get_test_detail(db, test_id) + test = crud_get_test_detail(db, test_id) + test_out = TestOut.model_validate(test) + return _mask_for_team_blindness(test_out, viewer_role=current_user.role) # --------------------------------------------------------------------------- @@ -848,6 +904,68 @@ def start_blue_work( return test +# --------------------------------------------------------------------------- +# POST /tests/{id}/review-red — Red Lead reviews the operator's submission +# --------------------------------------------------------------------------- + + +@router.post("/{test_id}/review-red", response_model=TestOut) +def review_red( + test_id: uuid.UUID, + payload: TestRedReview, + db: Session = Depends(get_db), + current_user: User = Depends(require_any_role("red_lead", "admin")), +) -> TestOut: + """Assigned Red Lead approves or reopens a test sitting in red_review.""" + test = crud_get_test_or_raise(db, test_id) + + if current_user.role != "admin" and test.red_reviewer_assignee != current_user.id: + raise HTTPException(status_code=403, detail="You are not the assigned reviewer for this test") + + with UnitOfWork(db) as uow: + if payload.decision == "approve": + test = wf_approve_red_review(db, test, current_user, notes=payload.notes) + elif payload.decision == "reopen": + test = wf_reopen_red_review(db, test, current_user, notes=payload.notes or "") + else: + raise HTTPException(status_code=400, detail="decision must be 'approve' or 'reopen'") + uow.commit() + db.refresh(test) + return test + + +# --------------------------------------------------------------------------- +# POST /tests/{id}/review-blue — Blue Lead reviews the operator's submission +# --------------------------------------------------------------------------- + + +@router.post("/{test_id}/review-blue", response_model=TestOut) +def review_blue( + test_id: uuid.UUID, + payload: TestBlueReview, + db: Session = Depends(get_db), + current_user: User = Depends(require_any_role("blue_lead", "admin")), +) -> TestOut: + """Assigned Blue Lead approves, reopens, or flags a capability gap on a test in blue_review.""" + test = crud_get_test_or_raise(db, test_id) + + if current_user.role != "admin" and test.blue_reviewer_assignee != current_user.id: + raise HTTPException(status_code=403, detail="You are not the assigned reviewer for this test") + + with UnitOfWork(db) as uow: + if payload.decision == "approve": + test = wf_approve_blue_review(db, test, current_user, notes=payload.notes) + elif payload.decision == "reopen": + test = wf_reopen_blue_review(db, test, current_user, notes=payload.notes or "") + elif payload.decision == "gap": + test = wf_flag_blue_review_gap(db, test, current_user, system_gaps=payload.system_gaps or "", notes=payload.notes) + else: + raise HTTPException(status_code=400, detail="decision must be 'approve', 'reopen', or 'gap'") + uow.commit() + db.refresh(test) + return test + + # --------------------------------------------------------------------------- # POST /tests/{id}/pause-timer — pause the active phase timer # --------------------------------------------------------------------------- diff --git a/backend/app/schemas/test.py b/backend/app/schemas/test.py index dd8ef7d..a7d3369 100644 --- a/backend/app/schemas/test.py +++ b/backend/app/schemas/test.py @@ -125,6 +125,27 @@ class TestBlueValidate(BaseModel): blue_validation_notes: str | None = None +# ── Red Lead review gate (pre-Blue-Team) ──────────────────────────── + + +class TestRedReview(BaseModel): + """Payload sent by the assigned Red Lead reviewer.""" + + decision: str # "approve" | "reopen" + notes: str | None = None + + +# ── Blue Lead review gate (pre-cross-validation) ──────────────────── + + +class TestBlueReview(BaseModel): + """Payload sent by the assigned Blue Lead reviewer.""" + + decision: str # "approve" | "reopen" | "gap" + notes: str | None = None + system_gaps: str | None = None + + # ── Remediation update ──────────────────────────────────────────── @@ -248,6 +269,17 @@ class TestOut(BaseModel): red_tech_assignee: uuid.UUID | None = None blue_tech_assignee: uuid.UUID | None = None + # Review assignment fields + red_reviewer_assignee: uuid.UUID | None = None + red_review_by: uuid.UUID | None = None + red_review_at: datetime | None = None + red_review_notes: str | None = None + blue_reviewer_assignee: uuid.UUID | None = None + blue_review_by: uuid.UUID | None = None + blue_review_at: datetime | None = None + blue_review_notes: str | None = None + system_gaps: str | None = None + # On-hold fields is_on_hold: bool = False hold_reason: str | None = None diff --git a/backend/tests/test_review_gates_router.py b/backend/tests/test_review_gates_router.py new file mode 100644 index 0000000..89db740 --- /dev/null +++ b/backend/tests/test_review_gates_router.py @@ -0,0 +1,242 @@ +"""HTTP-level tests for the red_review / blue_review lead-review gates. + +Covers POST /tests/{id}/review-red and POST /tests/{id}/review-blue: the +assigned-reviewer-only guard, approve/reopen/gap decisions, and the state +transitions they drive. Uses the ``api`` fixture (see conftest.py) for any +step that switches role, since the TestClient cookie jar otherwise bleeds +one role's session into the next request. +""" + +import uuid + +import pytest + +from app.models.evidence import Evidence +from app.models.enums import TeamSide + + +def _create_technique(api, auth_headers, mitre_id="T1059.099"): + resp = api( + "post", "/api/v1/techniques", auth_headers, + json={"mitre_id": mitre_id, "name": "Command Line"}, + ) + assert resp.status_code == 201, resp.text + return resp.json()["id"] + + +def _create_test(api, auth_headers, technique_id, name="Review gate test"): + resp = api( + "post", "/api/v1/tests", auth_headers, + json={"technique_id": technique_id, "name": name}, + ) + assert resp.status_code == 201, resp.text + return resp.json()["id"] + + +def _add_evidence(db, test_id, team: TeamSide): + ev = Evidence( + test_id=uuid.UUID(test_id), + file_name="proof.txt", + file_path="s3://bucket/proof.txt", + sha256_hash="a" * 64, + team=team, + ) + db.add(ev) + db.commit() + + +@pytest.fixture +def technique(api, auth_headers): + return _create_technique(api, auth_headers) + + +def _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique_id, name="Review gate test"): + """Drive a fresh test from draft to red_review via real HTTP calls.""" + test_id = _create_test(api, auth_headers, technique_id, name) + + api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers) + _add_evidence(db, test_id, TeamSide.red) + resp = api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers) + assert resp.status_code == 200, resp.text + assert resp.json()["state"] == "red_review" + return test_id + + +def _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique_id, name="Review gate test"): + test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique_id, name) + + approve = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "approve"}) + assert approve.status_code == 200, approve.text + assert approve.json()["state"] == "blue_evaluating" + + api("post", f"/api/v1/tests/{test_id}/start-blue-work", blue_tech_headers) + _add_evidence(db, test_id, TeamSide.blue) + resp = api("post", f"/api/v1/tests/{test_id}/submit-blue", blue_tech_headers) + assert resp.status_code == 200, resp.text + assert resp.json()["state"] == "blue_review" + return test_id + + +# --------------------------------------------------------------------------- +# review-red +# --------------------------------------------------------------------------- + + +def test_review_red_forbidden_for_non_assigned_lead( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique +): + """A red_lead who isn't the assigned reviewer gets 403.""" + test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique) + # red_lead_user IS the only red_lead fixture, so it WILL be auto-assigned + # as reviewer (single-candidate load balancing). To exercise the 403 + # path we need a second, non-assigned red_lead. + from app.auth import hash_password + from app.models.user import User + + other_lead = User( + username="otherredlead", email="otherredlead@test.com", + hashed_password=hash_password("x"), role="red_lead", is_active=True, + must_change_password=False, + ) + db.add(other_lead) + db.commit() + + login = client.post("/api/v1/auth/login", data={"username": "otherredlead", "password": "x"}) + assert login.status_code == 200 + other_headers = {"Authorization": f"Bearer {login.json()['access_token']}"} + + resp = api("post", f"/api/v1/tests/{test_id}/review-red", other_headers, json={"decision": "approve"}) + assert resp.status_code == 403 + + +def test_review_red_approve_moves_to_blue_evaluating( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique +): + test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique) + + resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "approve", "notes": "LGTM"}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["state"] == "blue_evaluating" + assert body["red_review_notes"] == "LGTM" + assert body["blue_started_at"] is not None + + +def test_review_red_reopen_requires_notes( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique +): + test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique) + + resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "reopen"}) + assert resp.status_code == 400 + + +def test_review_red_reopen_moves_to_red_executing( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique +): + test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique) + + resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "reopen", "notes": "add more detail"}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["state"] == "red_executing" + assert body["red_review_notes"] == "add more detail" + + +def test_review_red_admin_can_always_act( + client, db, api, auth_headers, red_tech_headers, red_lead_user, technique +): + test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique) + + resp = api("post", f"/api/v1/tests/{test_id}/review-red", auth_headers, json={"decision": "approve"}) + assert resp.status_code == 200, resp.text + assert resp.json()["state"] == "blue_evaluating" + + +# --------------------------------------------------------------------------- +# review-blue +# --------------------------------------------------------------------------- + + +def test_review_blue_forbidden_for_non_assigned_lead( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers, + red_lead_user, blue_lead_user, technique, +): + test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique) + + from app.auth import hash_password + from app.models.user import User + + other_lead = User( + username="otherbluelead", email="otherbluelead@test.com", + hashed_password=hash_password("x"), role="blue_lead", is_active=True, + must_change_password=False, + ) + db.add(other_lead) + db.commit() + + login = client.post("/api/v1/auth/login", data={"username": "otherbluelead", "password": "x"}) + other_headers = {"Authorization": f"Bearer {login.json()['access_token']}"} + + resp = api("post", f"/api/v1/tests/{test_id}/review-blue", other_headers, json={"decision": "approve"}) + assert resp.status_code == 403 + + +def test_review_blue_approve_moves_to_in_review( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers, + red_lead_user, blue_lead_user, technique, +): + test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique) + + resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "approve"}) + assert resp.status_code == 200, resp.text + assert resp.json()["state"] == "in_review" + + +def test_review_blue_reopen_requires_notes( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers, + red_lead_user, blue_lead_user, technique, +): + test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique) + + resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "reopen"}) + assert resp.status_code == 400 + + +def test_review_blue_reopen_moves_to_blue_evaluating( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers, + red_lead_user, blue_lead_user, technique, +): + test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique) + + resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "reopen", "notes": "redo detection"}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["state"] == "blue_evaluating" + assert body["blue_work_started_at"] is None + + +def test_review_blue_gap_requires_system_gaps( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers, + red_lead_user, blue_lead_user, technique, +): + test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique) + + resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "gap"}) + assert resp.status_code == 400 + + +def test_review_blue_gap_moves_to_in_review_with_system_gaps( + client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers, + red_lead_user, blue_lead_user, technique, +): + test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique) + + resp = api( + "post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, + json={"decision": "gap", "system_gaps": "Missing EDR agent on host X"}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["state"] == "in_review" + assert body["system_gaps"] == "Missing EDR agent on host X"