kitos
c753797019
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
Leads get a Create Template button that adds directly to the catalog (existing POST /test-templates). Operators (red_tech/blue_tech) get the same button, but their proposal now lands in a new template_suggestions review queue instead — a lead on their team can approve it as-is, edit fields before approving, or discard it. Modeled on the existing ProcedureSuggestion review workflow.
146 lines
6.2 KiB
Python
146 lines
6.2 KiB
Python
"""End-to-end coverage for the operator template-proposal review workflow.
|
|
|
|
Leads create test templates directly via POST /test-templates. An operator
|
|
(red_tech/blue_tech) gets the same "propose a template" action, but their
|
|
submission lands in the review queue instead of the live catalog — a lead
|
|
on their team approves (optionally editing fields) or discards it.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
|
|
def _payload(**overrides):
|
|
base = {
|
|
"mitre_technique_id": "T1059.300",
|
|
"name": "Operator-proposed template",
|
|
"attack_procedure": "Run a discovery command.",
|
|
"expected_detection": "Check process creation logs.",
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
@pytest.fixture
|
|
def technique(api, auth_headers):
|
|
resp = api(
|
|
"post", "/api/v1/techniques", auth_headers,
|
|
json={"mitre_id": "T1059.300", "name": "Command Line Proposals"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
return resp.json()["id"]
|
|
|
|
|
|
def test_red_tech_can_propose_a_template(api, technique, red_tech_headers):
|
|
resp = api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
assert resp.status_code == 201, resp.text
|
|
body = resp.json()
|
|
assert body["status"] == "pending"
|
|
assert body["team"] == "red"
|
|
assert body["name"] == "Operator-proposed template"
|
|
|
|
|
|
def test_blue_tech_can_propose_a_template(api, technique, blue_tech_headers):
|
|
resp = api("post", "/api/v1/template-suggestions", blue_tech_headers, json=_payload())
|
|
assert resp.status_code == 201, resp.text
|
|
assert resp.json()["team"] == "blue"
|
|
|
|
|
|
def test_lead_cannot_propose_a_template_via_suggestion_endpoint(api, technique, red_lead_headers):
|
|
"""Leads use the direct POST /test-templates endpoint instead."""
|
|
resp = api("post", "/api/v1/template-suggestions", red_lead_headers, json=_payload())
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_proposed_template_does_not_appear_in_catalog(api, technique, red_tech_headers, red_lead_headers):
|
|
api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
|
|
catalog = api("get", "/api/v1/test-templates?search=Operator-proposed", red_lead_headers)
|
|
assert catalog.json() == []
|
|
|
|
|
|
def test_red_lead_sees_pending_red_suggestion(api, technique, red_tech_headers, red_lead_headers):
|
|
api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
|
|
listed = api("get", "/api/v1/template-suggestions", red_lead_headers)
|
|
assert listed.status_code == 200, listed.text
|
|
assert len(listed.json()) == 1
|
|
assert listed.json()[0]["status"] == "pending"
|
|
|
|
|
|
def test_blue_lead_does_not_see_red_suggestion(api, technique, red_tech_headers, blue_lead_headers):
|
|
api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
|
|
listed = api("get", "/api/v1/template-suggestions", blue_lead_headers)
|
|
assert listed.json() == []
|
|
|
|
|
|
def test_blue_lead_cannot_approve_a_red_suggestion(api, technique, red_tech_headers, blue_lead_headers):
|
|
created = api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
suggestion_id = created.json()["id"]
|
|
|
|
resp = api("post", f"/api/v1/template-suggestions/{suggestion_id}/approve", blue_lead_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_lead_approves_suggestion_as_is_creates_template(api, technique, red_tech_headers, red_lead_headers):
|
|
created = api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
suggestion_id = created.json()["id"]
|
|
|
|
resp = api("post", f"/api/v1/template-suggestions/{suggestion_id}/approve", red_lead_headers)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["status"] == "approved"
|
|
assert body["created_template_id"] is not None
|
|
|
|
template = api("get", f"/api/v1/test-templates/{body['created_template_id']}", red_lead_headers)
|
|
assert template.status_code == 200
|
|
assert template.json()["name"] == "Operator-proposed template"
|
|
|
|
listed = api("get", "/api/v1/template-suggestions", red_lead_headers)
|
|
assert listed.json() == []
|
|
|
|
|
|
def test_lead_approves_suggestion_with_edits(api, technique, red_tech_headers, red_lead_headers):
|
|
"""The lead can add/adjust information before it goes live — the created
|
|
template reflects the lead's edits, not just the operator's original text."""
|
|
created = api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
suggestion_id = created.json()["id"]
|
|
|
|
resp = api(
|
|
"post", f"/api/v1/template-suggestions/{suggestion_id}/approve", red_lead_headers,
|
|
json={"attack_procedure": "Run a discovery command, then pivot via WMI.", "severity": "high"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
template_id = resp.json()["created_template_id"]
|
|
|
|
template = api("get", f"/api/v1/test-templates/{template_id}", red_lead_headers)
|
|
assert template.json()["attack_procedure"] == "Run a discovery command, then pivot via WMI."
|
|
assert template.json()["severity"] == "high"
|
|
# Untouched fields still come from the operator's original submission.
|
|
assert template.json()["expected_detection"] == "Check process creation logs."
|
|
|
|
|
|
def test_lead_discards_suggestion_without_creating_template(api, technique, red_tech_headers, red_lead_headers):
|
|
created = api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
suggestion_id = created.json()["id"]
|
|
|
|
resp = api("post", f"/api/v1/template-suggestions/{suggestion_id}/reject", red_lead_headers)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["status"] == "rejected"
|
|
assert resp.json()["created_template_id"] is None
|
|
|
|
catalog = api("get", "/api/v1/test-templates?search=Operator-proposed", red_lead_headers)
|
|
assert catalog.json() == []
|
|
|
|
listed = api("get", "/api/v1/template-suggestions", red_lead_headers)
|
|
assert listed.json() == []
|
|
|
|
|
|
def test_cannot_review_an_already_reviewed_suggestion_twice(api, technique, red_tech_headers, red_lead_headers):
|
|
created = api("post", "/api/v1/template-suggestions", red_tech_headers, json=_payload())
|
|
suggestion_id = created.json()["id"]
|
|
|
|
api("post", f"/api/v1/template-suggestions/{suggestion_id}/approve", red_lead_headers)
|
|
resp = api("post", f"/api/v1/template-suggestions/{suggestion_id}/reject", red_lead_headers)
|
|
assert resp.status_code == 400
|