diff --git a/backend/app/services/procedure_extraction_service.py b/backend/app/services/procedure_extraction_service.py new file mode 100644 index 0000000..c6ee344 --- /dev/null +++ b/backend/app/services/procedure_extraction_service.py @@ -0,0 +1,84 @@ +"""Heuristic extraction of the actual command(s)/query out of a free-text +procedure field (``procedure_text`` for Red, ``detect_procedure`` for +Blue), for proposing as a template's suggested procedure. + +Operators write these fields as free text mixing narrative ("I ran this +against the DC and it worked"), the actual command, and sometimes pasted +output. We only want the command — not the story around it, not the +output. There's no reliable general way to do this with NLP-level +accuracy, so this uses a deterministic, conservative regex approach: + +1. Fenced code blocks (```...``` or ~~~...~~~) are the strongest signal — + an operator who bothered to fence something almost always fenced the + command, not the narrative. If any exist, their contents are returned + verbatim (narrative outside the fence is ignored). +2. Otherwise, each line is checked against a curated set of patterns that + look like real commands or detection queries (shell prompts, known + binaries/cmdlets, SQL/KQL/SPL-style query syntax). Only matching lines + survive; narrative sentences and plain command output are discarded. + +This is deliberately conservative — false negatives (missing a real +command written in an unrecognized style) are far less harmful than false +positives (proposing narrative or output as "the command"), especially +since a lead reviews every suggestion before it reaches a template. +""" + +from __future__ import annotations + +import re + +_FENCE_RE = re.compile(r"```(?:\w*\n)?(.*?)```|~~~(?:\w*\n)?(.*?)~~~", re.DOTALL) + +# Leading shell/PowerShell prompt markers — stripped, not treated as +# evidence by themselves (the remainder still has to look like a command). +_PROMPT_RE = re.compile(r"^\s*(?:PS(?:\s+\S+)?>|[$#>])\s+") + +# Signatures that indicate "this line is a command or query", not prose. +_COMMAND_LINE_RE = re.compile( + r""" + ^\s*(?:PS(?:\s+\S+)?>|[$#>])\s*\S # explicit shell/PowerShell prompt with something after it + | + (?-i:\b[A-Z][a-zA-Z]+-[A-Z][a-zA-Z]+\b) # PowerShell Verb-Noun cmdlet, e.g. Get-Process + # (case-sensitive even though the rest of this + # pattern is IGNORECASE — otherwise lowercase + # hyphenated prose like "well-known" false-matches) + | + ^\s*(?:sudo|chmod|chown|curl|wget|nmap|ssh|scp|python3?|bash|sh|zsh| + powershell(?:\.exe)?|cmd(?:\.exe)?|reg(?:\.exe)?|net(?:\.exe)?| + netsh|wmic|schtasks|sc|rundll32|regsvr32|msiexec|certutil| + mimikatz\S*|cscript|wscript|osascript|openssl|systemctl|service| + reg\s+query|whoami|nslookup|dig|ping|tasklist|ps\s|kill| + Invoke-\S+|iex|dsquery|nltest)\b + | + \bSELECT\s+.+\s+FROM\b # SQL + | + \bsearch\s+index\s*= # SPL (explicit "search" keyword) + | + \bindex\s*=\S # SPL (bare "index=value") + | + \|\s*(?:where|project|summarize|extend|join|render)\b # KQL pipe syntax + """, + re.IGNORECASE | re.VERBOSE, +) + + +def extract_commands(text: str | None) -> str | None: + """Return the command/query lines found in *text*, or None if none found.""" + if not text or not text.strip(): + return None + + fence_matches = _FENCE_RE.findall(text) + if fence_matches: + blocks = [(a or b).strip() for a, b in fence_matches if (a or b).strip()] + if blocks: + return "\n\n".join(blocks) + + kept: list[str] = [] + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + if _COMMAND_LINE_RE.search(stripped): + kept.append(_PROMPT_RE.sub("", stripped)) + + return "\n".join(kept) if kept else None diff --git a/backend/tests/test_procedure_extraction_service.py b/backend/tests/test_procedure_extraction_service.py new file mode 100644 index 0000000..acef726 --- /dev/null +++ b/backend/tests/test_procedure_extraction_service.py @@ -0,0 +1,97 @@ +"""Tests for the regex-based command/query extraction heuristic used to +build procedure-improvement suggestions from free-text procedure fields.""" + +import pytest + +from app.services.procedure_extraction_service import extract_commands + + +@pytest.mark.parametrize("value", [None, "", " ", "\n\n"]) +def test_returns_none_for_empty_input(value): + assert extract_commands(value) is None + + +def test_returns_none_for_pure_narrative(): + text = ( + "I logged into the domain controller and tried a well-known credential " + "dumping tool. It worked well and I documented everything in the summary." + ) + assert extract_commands(text) is None + + +def test_fenced_code_block_takes_priority_over_surrounding_narrative(): + text = ( + "Ran the following from an elevated prompt:\n" + "```\n" + "Invoke-Mimikatz -DumpCreds\n" + "```\n" + "It successfully extracted plaintext passwords from LSASS memory." + ) + assert extract_commands(text) == "Invoke-Mimikatz -DumpCreds" + + +def test_tilde_fence_supported(): + text = "~~~\nGet-Process lsass\n~~~" + assert extract_commands(text) == "Get-Process lsass" + + +def test_multiple_fenced_blocks_are_joined(): + text = "```\nwhoami /priv\n```\nthen\n```\nGet-Process lsass\n```" + assert extract_commands(text) == "whoami /priv\n\nGet-Process lsass" + + +def test_extracts_command_lines_from_mixed_narrative_no_fence(): + text = ( + "First I checked what privileges I had.\n" + "whoami /priv\n" + "Then I dumped credentials from LSASS.\n" + "Invoke-Mimikatz -DumpCreds\n" + "This successfully retrieved plaintext passwords, as shown below.\n" + "Handles NPM(K) PM(K) WS(K) CPU(s) Id SI ProcessName\n" + " 542 27 23012 41564 0.45 1234 1 explorer\n" + ) + result = extract_commands(text) + assert result == "whoami /priv\nInvoke-Mimikatz -DumpCreds" + + +def test_strips_shell_prompt_markers(): + text = ( + "Ran this against the target host:\n" + "$ curl -X POST http://target/api/exfil -d @creds.txt\n" + "Got a 200 OK back.\n" + ) + assert extract_commands(text) == "curl -X POST http://target/api/exfil -d @creds.txt" + + +def test_strips_powershell_prompt_marker(): + text = "PS C:\\Users\\op> Get-Process lsass" + assert extract_commands(text) == "Get-Process lsass" + + +def test_extracts_kql_detection_query_mixed_with_narrative(): + text = ( + "I built a Sentinel query to catch this technique going forward.\n" + "DeviceProcessEvents\n" + "| where FileName == \"mimikatz.exe\"\n" + "That caught it within a minute of execution in testing.\n" + ) + result = extract_commands(text) + assert 'where FileName == "mimikatz.exe"' in result + assert "That caught it" not in result + + +def test_extracts_splunk_query(): + text = ( + "Search used for detection:\n" + "index=main sourcetype=WinEventLog:Security EventCode=4688\n" + "This surfaced the process creation event immediately.\n" + ) + result = extract_commands(text) + assert "index=main sourcetype=WinEventLog:Security EventCode=4688" in result + assert "immediately" not in result + + +def test_known_binary_at_start_of_line_is_recognized(): + text = "Notes: used the following.\nsudo tcpdump -i eth0 -w capture.pcap\nCaptured 500 packets." + result = extract_commands(text) + assert result == "sudo tcpdump -i eth0 -w capture.pcap"