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, ):