fix(tests): merge approved procedure suggestions instead of overwriting
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
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.
This commit is contained in:
@@ -5,7 +5,9 @@ command(s) in it are extracted heuristically (see
|
||||
``procedure_extraction_service``) and proposed as an update to the
|
||||
originating template's suggested-procedure field. A suggestion is only
|
||||
ever created when there's a real, novel improvement to propose — nothing
|
||||
is written to the template until a lead reviews and approves it.
|
||||
is written to the template until a lead reviews and approves it, and
|
||||
approving one only ever appends new commands, never overwrites or drops
|
||||
whatever earlier approved rounds already contributed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -30,6 +32,43 @@ _TEMPLATE_FIELD = {"red": "attack_procedure", "blue": "detect_suggested_procedur
|
||||
_LEAD_ROLE = {"red": "red_lead", "blue": "blue_lead"}
|
||||
|
||||
|
||||
def _new_lines(current: str | None, suggested: str) -> list[str]:
|
||||
"""Lines in *suggested* not already present verbatim in *current*.
|
||||
|
||||
Line-level, not whole-text: a later round's extraction is usually a
|
||||
different subset/superset of commands than an earlier one, so an exact
|
||||
string comparison would treat almost every resubmission as "novel" and
|
||||
then, on approval, blow away whatever an earlier approved round already
|
||||
contributed.
|
||||
"""
|
||||
existing = {line.strip() for line in (current or "").splitlines() if line.strip()}
|
||||
seen: set[str] = set()
|
||||
new: list[str] = []
|
||||
for line in suggested.splitlines():
|
||||
stripped = line.strip()
|
||||
if not stripped or stripped in existing or stripped in seen:
|
||||
continue
|
||||
new.append(stripped)
|
||||
seen.add(stripped)
|
||||
return new
|
||||
|
||||
|
||||
def _merge_procedure_text(current: str | None, suggested: str) -> str:
|
||||
"""Append genuinely new lines from *suggested* onto *current*.
|
||||
|
||||
Approving a suggestion must only ever grow a template's procedure,
|
||||
never replace it — otherwise a narrower extraction from a later round
|
||||
would silently erase commands an earlier approved round already added.
|
||||
"""
|
||||
current_stripped = (current or "").strip()
|
||||
new = _new_lines(current, suggested)
|
||||
if not new:
|
||||
return current_stripped
|
||||
if not current_stripped:
|
||||
return "\n".join(new)
|
||||
return current_stripped + "\n" + "\n".join(new)
|
||||
|
||||
|
||||
def create_suggestion_from_submission(db: Session, test: Test, team: str) -> ProcedureSuggestion | None:
|
||||
"""Propose a template procedure improvement from *test*'s just-submitted round.
|
||||
|
||||
@@ -51,8 +90,8 @@ def create_suggestion_from_submission(db: Session, test: Test, team: str) -> Pro
|
||||
if template is None:
|
||||
return None
|
||||
|
||||
current = (getattr(template, _TEMPLATE_FIELD[team]) or "").strip()
|
||||
if extracted == current:
|
||||
current = getattr(template, _TEMPLATE_FIELD[team])
|
||||
if not _new_lines(current, extracted):
|
||||
return None
|
||||
|
||||
duplicate = (
|
||||
@@ -116,7 +155,8 @@ def approve_suggestion(db: Session, suggestion_id: uuid.UUID, user: User) -> Pro
|
||||
if template is None:
|
||||
raise EntityNotFoundError("TestTemplate", str(suggestion.template_id))
|
||||
|
||||
setattr(template, _TEMPLATE_FIELD[suggestion.team], suggestion.suggested_text)
|
||||
current_text = getattr(template, _TEMPLATE_FIELD[suggestion.team])
|
||||
setattr(template, _TEMPLATE_FIELD[suggestion.team], _merge_procedure_text(current_text, suggestion.suggested_text))
|
||||
suggestion.status = "approved"
|
||||
suggestion.reviewed_by = user.id
|
||||
suggestion.reviewed_at = datetime.utcnow()
|
||||
|
||||
@@ -151,7 +151,7 @@ def test_approving_a_suggestion_writes_it_into_the_template(
|
||||
assert approve.json()["status"] == "approved"
|
||||
|
||||
reloaded = api("get", f"/api/v1/test-templates/{template['id']}", red_lead_headers)
|
||||
assert reloaded.json()["attack_procedure"] == "whoami /priv"
|
||||
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() == []
|
||||
@@ -180,6 +180,48 @@ def test_rejecting_a_suggestion_leaves_the_template_untouched(
|
||||
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,
|
||||
):
|
||||
|
||||
Reference in New Issue
Block a user