feat(tests): review-red/review-blue router endpoints
This commit is contained in:
kitos
@@ -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"
|
||||
Reference in New Issue
Block a user