006c4c2c5f
- Simulation model with full field set (redteam + SOC sides) and cascade delete - Alembic migration 0002 for simulations table - simulation_workflow service: PATCH RBAC field-level + auto-transition pending→in_progress + state machine - mitre service: STIX bundle loader (boot-safe) + ranked search (exact-id > prefix-id > name) - 7 new API endpoints: list/create/get/patch/delete simulations, transition, MITRE autocomplete - serialize_simulation added to serializers.py - Makefile update-mitre target with real curl + optional docker restart - Dockerfile updated to copy backend/data/ into image - MITRE enterprise-attack.json bundle committed (~45 MB) - 67 new tests (total 130 passing), ruff clean, mypy introduces no new errors Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
273 lines
8.9 KiB
Python
273 lines
8.9 KiB
Python
"""Simulation PATCH tests: auto-transition, RBAC field-level, SOC restrictions."""
|
|
from __future__ import annotations
|
|
|
|
from flask.testing import FlaskClient
|
|
|
|
from backend.tests.conftest import auth_headers as _h
|
|
|
|
|
|
def _make_engagement(client: FlaskClient, token: str) -> dict:
|
|
resp = client.post(
|
|
"/api/engagements",
|
|
headers=_h(token),
|
|
json={"name": "Op Beta", "start_date": "2026-06-01"},
|
|
)
|
|
assert resp.status_code == 201
|
|
return resp.get_json()
|
|
|
|
|
|
def _make_sim(client: FlaskClient, token: str, eid: int) -> dict:
|
|
resp = client.post(
|
|
f"/api/engagements/{eid}/simulations",
|
|
headers=_h(token),
|
|
json={"name": "Test Sim"},
|
|
)
|
|
assert resp.status_code == 201
|
|
return resp.get_json()
|
|
|
|
|
|
def _patch(client: FlaskClient, token: str, sid: int, payload: dict) -> dict:
|
|
resp = client.patch(
|
|
f"/api/simulations/{sid}", headers=_h(token), json=payload
|
|
)
|
|
return resp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auto-transition pending → in_progress (AC-8.2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_patch_triggers_auto_transition(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
assert sim["status"] == "pending"
|
|
|
|
resp = _patch(client, redteam_token, sim["id"], {"description": "some desc"})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "in_progress"
|
|
|
|
|
|
def test_patch_name_triggers_auto_transition(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
|
|
resp = _patch(client, redteam_token, sim["id"], {"name": "Updated name"})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "in_progress"
|
|
|
|
|
|
def test_patch_commands_triggers_auto_transition(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
|
|
resp = _patch(client, redteam_token, sim["id"], {"commands": "cmd1\ncmd2"})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "in_progress"
|
|
|
|
|
|
def test_patch_null_value_does_not_trigger_auto_transition(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
|
|
resp = _patch(client, redteam_token, sim["id"], {"description": None})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "pending"
|
|
|
|
|
|
def test_patch_empty_string_does_not_trigger_auto_transition(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
|
|
resp = _patch(client, redteam_token, sim["id"], {"description": ""})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "pending"
|
|
|
|
|
|
def test_patch_admin_triggers_auto_transition(
|
|
client: FlaskClient, admin_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, admin_token)
|
|
sim = _make_sim(client, admin_token, eng["id"])
|
|
|
|
resp = _patch(client, admin_token, sim["id"], {"execution_result": "success"})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "in_progress"
|
|
|
|
|
|
def test_patch_soc_does_not_trigger_auto_transition(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
"""SOC patch on review_required must not trigger auto-transition."""
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
|
|
# Manually advance to review_required.
|
|
client.post(
|
|
f"/api/simulations/{sim['id']}/transition",
|
|
headers=_h(redteam_token),
|
|
json={"to": "review_required"},
|
|
)
|
|
|
|
resp = _patch(client, soc_token, sim["id"], {"soc_comment": "looks good"})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "review_required"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Field updates
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_patch_updates_commands_as_text(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
commands = "whoami\nnet user\nipconfig"
|
|
|
|
resp = _patch(client, redteam_token, sim["id"], {"commands": commands})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["commands"] == commands
|
|
|
|
|
|
def test_patch_updates_executed_at(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
|
|
resp = _patch(
|
|
client, redteam_token, sim["id"], {"executed_at": "2026-06-01T12:00:00"}
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "2026-06-01" in resp.get_json()["executed_at"]
|
|
|
|
|
|
def test_patch_invalid_executed_at(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
|
|
resp = _patch(
|
|
client, redteam_token, sim["id"], {"executed_at": "not-a-date"}
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()["error"] == "invalid executed_at"
|
|
|
|
|
|
def test_patch_clear_executed_at(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
_patch(client, redteam_token, sim["id"], {"executed_at": "2026-06-01T12:00:00"})
|
|
|
|
resp = _patch(client, redteam_token, sim["id"], {"executed_at": None})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["executed_at"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SOC RBAC field-level (AC-9.1, AC-9.2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_soc_cannot_patch_before_review_required(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
assert sim["status"] == "pending"
|
|
|
|
resp = _patch(client, soc_token, sim["id"], {"soc_comment": "hello"})
|
|
assert resp.status_code == 403
|
|
assert "not ready" in resp.get_json()["error"]
|
|
|
|
|
|
def test_soc_cannot_patch_in_progress(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
_patch(client, redteam_token, sim["id"], {"description": "in progress now"})
|
|
|
|
resp = _patch(client, soc_token, sim["id"], {"soc_comment": "hello"})
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_soc_can_patch_when_review_required(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
client.post(
|
|
f"/api/simulations/{sim['id']}/transition",
|
|
headers=_h(redteam_token),
|
|
json={"to": "review_required"},
|
|
)
|
|
|
|
resp = _patch(
|
|
client,
|
|
soc_token,
|
|
sim["id"],
|
|
{"soc_comment": "Detected", "log_source": "SIEM", "incident_number": "INC-001"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.get_json()
|
|
assert body["soc_comment"] == "Detected"
|
|
assert body["log_source"] == "SIEM"
|
|
assert body["incident_number"] == "INC-001"
|
|
|
|
|
|
def test_soc_can_patch_when_done(
|
|
client: FlaskClient, redteam_token: str, soc_token: str, admin_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
client.post(
|
|
f"/api/simulations/{sim['id']}/transition",
|
|
headers=_h(redteam_token),
|
|
json={"to": "review_required"},
|
|
)
|
|
client.post(
|
|
f"/api/simulations/{sim['id']}/transition",
|
|
headers=_h(soc_token),
|
|
json={"to": "done"},
|
|
)
|
|
|
|
resp = _patch(client, soc_token, sim["id"], {"soc_comment": "Final note"})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_soc_cannot_edit_redteam_fields(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
client.post(
|
|
f"/api/simulations/{sim['id']}/transition",
|
|
headers=_h(redteam_token),
|
|
json={"to": "review_required"},
|
|
)
|
|
|
|
resp = _patch(client, soc_token, sim["id"], {"description": "redteam field"})
|
|
assert resp.status_code == 403
|
|
assert resp.get_json()["error"] == "soc cannot edit redteam fields"
|
|
|
|
|
|
def test_patch_simulation_404(client: FlaskClient, redteam_token: str) -> None:
|
|
resp = _patch(client, redteam_token, 9999, {"name": "x"})
|
|
assert resp.status_code == 404
|