From 8fcd733d4a79bffc6096ea37fb68b33c27794cca Mon Sep 17 00:00:00 2001 From: kitos Date: Tue, 14 Jul 2026 16:25:24 +0200 Subject: [PATCH] fix(tests): merge approved procedure suggestions instead of overwriting 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. --- .../services/procedure_suggestion_service.py | 48 +++++++++++++++++-- backend/tests/test_procedure_suggestions.py | 44 ++++++++++++++++- 2 files changed, 87 insertions(+), 5 deletions(-) diff --git a/backend/app/services/procedure_suggestion_service.py b/backend/app/services/procedure_suggestion_service.py index 3974ef6..77be721 100644 --- a/backend/app/services/procedure_suggestion_service.py +++ b/backend/app/services/procedure_suggestion_service.py @@ -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() diff --git a/backend/tests/test_procedure_suggestions.py b/backend/tests/test_procedure_suggestions.py index 3d8040c..f02e484 100644 --- a/backend/tests/test_procedure_suggestions.py +++ b/backend/tests/test_procedure_suggestions.py @@ -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, ):