kitos
8fcd733d4a
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
Approving a suggestion was replacing the template's whole procedure field with just that submission's extracted text, silently dropping whatever an earlier approved round had already added. Now it appends only the genuinely new lines, so templates accumulate commands across rounds instead of losing them.
275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""End-to-end coverage for the procedure-suggestion review workflow.
|
|
|
|
An operator submitting a round with a filled-in procedure field that
|
|
contains an extractable command should produce a pending
|
|
ProcedureSuggestion against the originating template; a lead can then
|
|
approve it (writing it into the template) or reject it (leaving the
|
|
template untouched). Nothing is ever written to a template automatically.
|
|
"""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from app.models.enums import TeamSide
|
|
from app.models.evidence import Evidence
|
|
|
|
|
|
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 Suggestions"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
return resp.json()["id"]
|
|
|
|
|
|
@pytest.fixture
|
|
def template(api, red_lead_headers, technique):
|
|
resp = api(
|
|
"post", "/api/v1/test-templates", red_lead_headers,
|
|
json={
|
|
"mitre_technique_id": "T1059.200",
|
|
"name": "Suggestion source template",
|
|
"attack_procedure": "Run a discovery command on the target.",
|
|
"detect_suggested_procedure": "Check process creation logs.",
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
return resp.json()
|
|
|
|
|
|
@pytest.fixture
|
|
def test_from_template(api, red_lead_headers, technique, template):
|
|
resp = api(
|
|
"post", "/api/v1/tests/from-template", red_lead_headers,
|
|
json={"template_id": template["id"], "technique_id": technique},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
return resp.json()["id"]
|
|
|
|
|
|
def test_red_submission_with_command_creates_pending_suggestion(
|
|
client, db, api, test_from_template, red_tech_headers, red_lead_headers,
|
|
):
|
|
test_id = test_from_template
|
|
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Ran discovery.\nwhoami /priv\nGot back SeDebugPrivilege."},
|
|
)
|
|
_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
|
|
|
|
listed = api("get", "/api/v1/procedure-suggestions", red_lead_headers)
|
|
assert listed.status_code == 200, listed.text
|
|
suggestions = listed.json()
|
|
assert len(suggestions) == 1
|
|
assert suggestions[0]["team"] == "red"
|
|
assert suggestions[0]["suggested_text"] == "whoami /priv"
|
|
assert suggestions[0]["status"] == "pending"
|
|
|
|
|
|
def test_blue_submission_with_command_creates_pending_suggestion(
|
|
client, db, api, test_from_template, red_tech_headers, red_lead_headers,
|
|
blue_tech_headers, blue_lead_headers,
|
|
):
|
|
test_id = test_from_template
|
|
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)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/blue", blue_tech_headers,
|
|
json={
|
|
"detection_result": "detected",
|
|
"detect_procedure": "Checked logs.\nGet-WinEvent -LogName Security | where Id -eq 4688\nFound it.",
|
|
},
|
|
)
|
|
_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
|
|
|
|
listed = api("get", "/api/v1/procedure-suggestions", blue_lead_headers)
|
|
assert listed.status_code == 200, listed.text
|
|
suggestions = listed.json()
|
|
assert len(suggestions) == 1
|
|
assert suggestions[0]["team"] == "blue"
|
|
assert "Get-WinEvent" in suggestions[0]["suggested_text"]
|
|
|
|
|
|
def test_submission_without_extractable_command_creates_no_suggestion(
|
|
client, db, api, test_from_template, red_tech_headers, red_lead_headers,
|
|
):
|
|
test_id = test_from_template
|
|
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Just talked to the target and it agreed to be compromised."},
|
|
)
|
|
_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
|
|
|
|
listed = api("get", "/api/v1/procedure-suggestions", red_lead_headers)
|
|
assert listed.json() == []
|
|
|
|
|
|
def test_approving_a_suggestion_writes_it_into_the_template(
|
|
client, db, api, test_from_template, template, red_tech_headers, red_lead_headers,
|
|
):
|
|
test_id = test_from_template
|
|
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Ran discovery.\nwhoami /priv\nDone."},
|
|
)
|
|
_add_evidence(db, test_id, TeamSide.red)
|
|
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
|
|
|
suggestion = api("get", "/api/v1/procedure-suggestions", red_lead_headers).json()[0]
|
|
approve = api(
|
|
"post", f"/api/v1/procedure-suggestions/{suggestion['id']}/approve", red_lead_headers,
|
|
)
|
|
assert approve.status_code == 200, approve.text
|
|
assert approve.json()["status"] == "approved"
|
|
|
|
reloaded = api("get", f"/api/v1/test-templates/{template['id']}", red_lead_headers)
|
|
assert reloaded.json()["attack_procedure"] == "Run a discovery command on the target.\nwhoami /priv"
|
|
|
|
listed = api("get", "/api/v1/procedure-suggestions", red_lead_headers)
|
|
assert listed.json() == []
|
|
|
|
|
|
def test_rejecting_a_suggestion_leaves_the_template_untouched(
|
|
client, db, api, test_from_template, template, red_tech_headers, red_lead_headers,
|
|
):
|
|
test_id = test_from_template
|
|
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Ran discovery.\nwhoami /priv\nDone."},
|
|
)
|
|
_add_evidence(db, test_id, TeamSide.red)
|
|
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
|
|
|
suggestion = api("get", "/api/v1/procedure-suggestions", red_lead_headers).json()[0]
|
|
reject = api(
|
|
"post", f"/api/v1/procedure-suggestions/{suggestion['id']}/reject", red_lead_headers,
|
|
)
|
|
assert reject.status_code == 200, reject.text
|
|
assert reject.json()["status"] == "rejected"
|
|
|
|
reloaded = api("get", f"/api/v1/test-templates/{template['id']}", red_lead_headers)
|
|
assert reloaded.json()["attack_procedure"] == "Run a discovery command on the target."
|
|
|
|
|
|
def test_approving_a_later_suggestion_does_not_erase_an_earlier_approved_one(
|
|
client, db, api, test_from_template, template, red_tech_headers, red_lead_headers,
|
|
):
|
|
"""Two separate rounds, each contributing a different command, must
|
|
both survive in the template once their suggestions are approved —
|
|
approving round 2's suggestion must not wipe out what round 1 added."""
|
|
test_id = test_from_template
|
|
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Ran discovery.\nwhoami /priv\nDone."},
|
|
)
|
|
_add_evidence(db, test_id, TeamSide.red)
|
|
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
|
|
|
first_suggestion = api("get", "/api/v1/procedure-suggestions", red_lead_headers).json()[0]
|
|
api("post", f"/api/v1/procedure-suggestions/{first_suggestion['id']}/approve", red_lead_headers)
|
|
|
|
api(
|
|
"post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers,
|
|
json={"decision": "reopen", "notes": "one more pass"},
|
|
)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Ran a follow-up check.\nnltest /dsgetdc:corp\nDone."},
|
|
)
|
|
_add_evidence(db, test_id, TeamSide.red)
|
|
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
|
|
|
second_suggestion = api("get", "/api/v1/procedure-suggestions", red_lead_headers).json()[0]
|
|
assert second_suggestion["suggested_text"] == "nltest /dsgetdc:corp"
|
|
approve_second = api(
|
|
"post", f"/api/v1/procedure-suggestions/{second_suggestion['id']}/approve", red_lead_headers,
|
|
)
|
|
assert approve_second.status_code == 200, approve_second.text
|
|
|
|
reloaded = api("get", f"/api/v1/test-templates/{template['id']}", red_lead_headers)
|
|
assert reloaded.json()["attack_procedure"] == (
|
|
"Run a discovery command on the target.\nwhoami /priv\nnltest /dsgetdc:corp"
|
|
)
|
|
|
|
|
|
def test_blue_lead_cannot_review_a_red_suggestion(
|
|
client, db, api, test_from_template, red_tech_headers, red_lead_headers, blue_lead_headers,
|
|
):
|
|
test_id = test_from_template
|
|
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Ran discovery.\nwhoami /priv\nDone."},
|
|
)
|
|
_add_evidence(db, test_id, TeamSide.red)
|
|
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
|
|
|
suggestion = api("get", "/api/v1/procedure-suggestions", red_lead_headers).json()[0]
|
|
|
|
# A blue_lead sees no red suggestions in their own queue...
|
|
blue_view = api("get", "/api/v1/procedure-suggestions", blue_lead_headers)
|
|
assert blue_view.json() == []
|
|
|
|
# ...and is forbidden from acting on one directly by id.
|
|
forbidden = api(
|
|
"post", f"/api/v1/procedure-suggestions/{suggestion['id']}/approve", blue_lead_headers,
|
|
)
|
|
assert forbidden.status_code == 403
|
|
|
|
|
|
def test_duplicate_submission_does_not_create_a_second_pending_suggestion(
|
|
client, db, api, test_from_template, red_tech_headers, red_lead_headers,
|
|
):
|
|
test_id = test_from_template
|
|
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Ran discovery.\nwhoami /priv\nDone."},
|
|
)
|
|
_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": "reopen", "notes": "please redo this round"},
|
|
)
|
|
|
|
api(
|
|
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
|
json={"procedure_text": "Ran discovery again.\nwhoami /priv\nStill works."},
|
|
)
|
|
_add_evidence(db, test_id, TeamSide.red)
|
|
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
|
|
|
listed = api("get", "/api/v1/procedure-suggestions", red_lead_headers).json()
|
|
assert len(listed) == 1
|