kitos
3a01facd46
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
Request Discussion only ever nudged the peer lead, and the one-time
manager notification when a dispute starts had no real follow-up
action attached to it — a manager could be told a dispute existed but
had no way to actually do anything about it. Adds:
- POST /tests/{id}/escalate-to-manager — either lead can explicitly
ask managers to step in, distinct from the passive initial notice.
- POST /tests/{id}/manager-resolve-dispute — a manager decides the
final outcome (validated/rejected) directly, bypassing both leads'
votes entirely, reusing the same dual-validation event dispatch so
Jira/notifications behave like any other resolution.
Both leads are notified of the manager's final call, win or lose.
208 lines
7.9 KiB
Python
208 lines
7.9 KiB
Python
"""HTTP-level tests for dispute resolution: POST /tests/{id}/resolve-dispute
|
|
and the manager notification fired when a test enters 'disputed'.
|
|
"""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from app.models.evidence import Evidence
|
|
from app.models.enums import TeamSide
|
|
|
|
|
|
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):
|
|
resp = api(
|
|
"post", "/api/v1/techniques", auth_headers,
|
|
json={"mitre_id": "T1059.200", "name": "Command Line"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
return resp.json()["id"]
|
|
|
|
|
|
def _reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique):
|
|
"""Drive a fresh test all the way to disputed: red approves, blue rejects."""
|
|
resp = api(
|
|
"post", "/api/v1/tests", red_lead_headers,
|
|
json={"technique_id": technique, "name": "Dispute test"},
|
|
)
|
|
test_id = resp.json()["id"]
|
|
|
|
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
|
_add_evidence(db, test_id, TeamSide.red)
|
|
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
|
api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "approve"})
|
|
|
|
api("post", f"/api/v1/tests/{test_id}/start-blue-work", blue_tech_headers)
|
|
_add_evidence(db, test_id, TeamSide.blue)
|
|
api("post", f"/api/v1/tests/{test_id}/submit-blue", blue_tech_headers)
|
|
api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "approve"})
|
|
|
|
red_vote = api(
|
|
"post", f"/api/v1/tests/{test_id}/validate-red", red_lead_headers,
|
|
json={"red_validation_status": "approved"},
|
|
)
|
|
assert red_vote.status_code == 200, red_vote.text
|
|
|
|
blue_vote = api(
|
|
"post", f"/api/v1/tests/{test_id}/validate-blue", blue_lead_headers,
|
|
json={"blue_validation_status": "rejected", "blue_validation_notes": "Detection insufficient"},
|
|
)
|
|
assert blue_vote.status_code == 200, blue_vote.text
|
|
assert blue_vote.json()["state"] == "disputed"
|
|
|
|
return test_id
|
|
|
|
|
|
def test_manager_notified_on_dispute(
|
|
client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, manager_headers, manager_user, technique,
|
|
):
|
|
_reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique)
|
|
|
|
resp = api("get", "/api/v1/notifications", manager_headers)
|
|
assert resp.status_code == 200
|
|
notifications = resp.json()
|
|
assert any(n["type"] == "validation_disputed" for n in notifications)
|
|
|
|
|
|
def test_resolve_dispute_forbidden_for_non_approver(
|
|
client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique,
|
|
):
|
|
"""Blue Lead rejected (didn't approve) — they cannot flip the vote here."""
|
|
test_id = _reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique)
|
|
|
|
resp = api(
|
|
"post", f"/api/v1/tests/{test_id}/resolve-dispute", blue_lead_headers,
|
|
json={"target_team": "blue"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_resolve_dispute_routes_to_red_queue(
|
|
client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique,
|
|
):
|
|
"""Red Lead approved; flips to reject and sends it to the Red queue."""
|
|
test_id = _reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique)
|
|
|
|
resp = api(
|
|
"post", f"/api/v1/tests/{test_id}/resolve-dispute", red_lead_headers,
|
|
json={"target_team": "red", "notes": "redo the attack with more detail"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["state"] == "red_executing"
|
|
assert body["red_validation_status"] is None
|
|
assert body["blue_validation_status"] is None
|
|
|
|
|
|
def test_escalate_to_manager_notifies_managers(
|
|
client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, manager_headers, manager_user, technique,
|
|
):
|
|
test_id = _reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique)
|
|
|
|
resp = api("post", f"/api/v1/tests/{test_id}/escalate-to-manager", red_lead_headers)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["status"] == "escalated"
|
|
|
|
notifications = api("get", "/api/v1/notifications", manager_headers).json()
|
|
assert any("escalated" in (n.get("title") or "").lower() for n in notifications)
|
|
|
|
|
|
def test_escalate_to_manager_requires_disputed_state(
|
|
client, db, api, auth_headers, red_tech_headers, red_lead_headers, technique,
|
|
):
|
|
resp = api(
|
|
"post", "/api/v1/tests", red_lead_headers,
|
|
json={"technique_id": technique, "name": "Not disputed"},
|
|
)
|
|
test_id = resp.json()["id"]
|
|
|
|
resp = api("post", f"/api/v1/tests/{test_id}/escalate-to-manager", red_lead_headers)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_manager_resolve_dispute_as_validated(
|
|
client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, manager_headers, technique,
|
|
):
|
|
test_id = _reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique)
|
|
|
|
resp = api(
|
|
"post", f"/api/v1/tests/{test_id}/manager-resolve-dispute", manager_headers,
|
|
json={"outcome": "validated", "notes": "Detection was sufficient after all"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["state"] == "validated"
|
|
|
|
red_notifs = api("get", "/api/v1/notifications", red_lead_headers).json()
|
|
assert any(n["type"] == "manager_dispute_resolution" for n in red_notifs)
|
|
blue_notifs = api("get", "/api/v1/notifications", blue_lead_headers).json()
|
|
assert any(n["type"] == "manager_dispute_resolution" for n in blue_notifs)
|
|
|
|
|
|
def test_manager_resolve_dispute_as_rejected(
|
|
client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, manager_headers, technique,
|
|
):
|
|
test_id = _reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique)
|
|
|
|
resp = api(
|
|
"post", f"/api/v1/tests/{test_id}/manager-resolve-dispute", manager_headers,
|
|
json={"outcome": "rejected"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["state"] == "rejected"
|
|
|
|
|
|
def test_manager_resolve_dispute_forbidden_for_lead(
|
|
client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique,
|
|
):
|
|
test_id = _reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers, technique)
|
|
|
|
resp = api(
|
|
"post", f"/api/v1/tests/{test_id}/manager-resolve-dispute", red_lead_headers,
|
|
json={"outcome": "validated"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_manager_resolve_dispute_requires_disputed_state(
|
|
client, db, api, red_lead_headers, manager_headers, technique,
|
|
):
|
|
resp = api(
|
|
"post", "/api/v1/tests", red_lead_headers,
|
|
json={"technique_id": technique, "name": "Not disputed either"},
|
|
)
|
|
test_id = resp.json()["id"]
|
|
|
|
resp = api(
|
|
"post", f"/api/v1/tests/{test_id}/manager-resolve-dispute", manager_headers,
|
|
json={"outcome": "validated"},
|
|
)
|
|
assert resp.status_code == 400
|