From 2139673bbd46a28b6d4b3a6d5b1be4154b5ea60c Mon Sep 17 00:00:00 2001 From: eastmadc Date: Tue, 4 Aug 2026 20:49:52 -0600 Subject: [PATCH] test: add static weak-assertion lint for attacker procedures The attacker agent writes the exploit and the assertions that grade it, and nothing checks that those assertions can tell a working exploit from a broken one. An expect of status: 200, or a body_contains that also matches the error page, passes a procedure that proves nothing -- green, in CI, forever. Reports assertion shapes that cannot discriminate: errors no_expect on the final step, status_only_2xx across a whole procedure, exec_target_in_procedure, unparsed warnings generic_text, short_text, tautological, no_dynamic_evidence, unused_output Severities were calibrated against tests/fixtures/procedures/. The first pass errored on 4 of 5 fixtures, almost all false positives: stdout_contains: root after whoami is a strong assertion, and a step that posts a payload and asserts 201 is fine when a later step verifies it fired. So the generic-word rule now skips command output, and status-only is judged per procedure rather than per step. Six of eight existing fixtures are clean; the two that error both use exec_target, which their own headers describe as god-view -- correct for an executor fixture, and a real bug in a generated procedure, so the rule stays. No Docker and no model calls, so it runs under the standard contributor check. 16 tests, every rule covered in both directions. Standalone: python -m goe.eval.procedure_lint tests/fixtures/procedures/ The dynamic half -- run each procedure against a baseline with the vulnerability absent and require failure -- needs the executor and is not included here. --- goe/eval/procedure_lint.py | 261 ++++++++++++++++++++++++++ tests/fixtures/procedures/strong.yaml | 30 +++ tests/fixtures/procedures/weak.yaml | 26 +++ tests/test_procedure_lint.py | 176 +++++++++++++++++ 4 files changed, 493 insertions(+) create mode 100644 goe/eval/procedure_lint.py create mode 100644 tests/fixtures/procedures/strong.yaml create mode 100644 tests/fixtures/procedures/weak.yaml create mode 100644 tests/test_procedure_lint.py diff --git a/goe/eval/procedure_lint.py b/goe/eval/procedure_lint.py new file mode 100644 index 0000000..dd93736 --- /dev/null +++ b/goe/eval/procedure_lint.py @@ -0,0 +1,261 @@ +"""Static lint for attacker procedures: find assertions that can't discriminate. + +The attacker agent writes the exploit and the assertions that grade it. Nothing +checks that those assertions can tell a working exploit from a broken one. An +`expect` of `status: 200`, or a `body_contains` that also matches the error page, +passes a procedure that proves nothing — and passes it green in CI forever. + +The full answer is a dynamic negative control: run the procedure against a target +with the vulnerability absent and require failure. That needs the executor and a +baseline fixture. This module is the part that needs neither, so it runs under +`pytest -m 'not docker and not llm'`. + +Reads Procedure YAML (goe/models/procedure.py) without importing pydantic, so it +can report on documents that fail strict validation too. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from dataclasses import dataclass, asdict +from pathlib import Path +from typing import Any, Iterator + +import yaml + +# Assertion keys whose value is content the exploit is meant to have produced. +TEXT_KEYS = frozenset({ + "stdout_contains", "received_contains", "body_contains", "url_contains", + "url_equals", "evaluate_result_contains", "extracted_contains", + "title_contains", "contains", +}) +REGEX_KEYS = frozenset({ + "stdout_regex", "received_regex", "body_regex", "evaluate_result_regex", + "extracted_regex", +}) +# Keys that assert structure, not content. Fine on their own; just not proof of +# a value, so the generic/short/tautology rules don't apply to them. +STRUCTURAL_KEYS = frozenset({ + "status", "exit_code", "selector", "selector_visible", + "selector_not_visible", "cookie_exists", "count", "name", "key", +}) + +# Action fields that carry what the step transmits. +SENT_KEYS = ("url", "body", "headers", "command", "script", "path", "fields", + "filename", "file_content", "selector") + +# So common that matching them proves nothing: an error page, a login form and a +# 404 body all routinely contain several. +GENERIC = frozenset({ + "ok", "error", "true", "false", "success", "failed", "failure", "login", + "admin", "user", "users", "password", "welcome", "home", "index", "page", + "server", "http", "html", "", "null", "none", "data", "result", + "status", "message", "root", "test", "flag", +}) + +MIN_LEN = 4 +DYNAMIC = re.compile(r"\$\{\s*(edge|steps)\.") + + +@dataclass(frozen=True) +class Finding: + rule: str + severity: str # "error" | "warn" + step: str + detail: str + + def line(self) -> str: + return f" {'ERROR' if self.severity == 'error' else 'warn '} " \ + f"[{self.rule}] {self.step}: {self.detail}" + + +def _walk_assertions(expect: Any) -> Iterator[dict]: + """Yield leaf assertions, flattening `all:` nesting.""" + if not isinstance(expect, dict): + return + if "all" in expect: + for sub in expect.get("all") or []: + yield from _walk_assertions(sub) + return + yield expect + + +def _sent_blob(action: Any) -> str: + if not isinstance(action, dict): + return "" + parts = [] + for k in SENT_KEYS: + v = action.get(k) + if v is not None: + parts.append(v if isinstance(v, str) else json.dumps(v, default=str)) + return " ".join(parts).lower() + + +def lint_procedure(doc: Any) -> list[Finding]: + out: list[Finding] = [] + + steps = doc.get("procedure") if isinstance(doc, dict) else None + if not isinstance(steps, list) or not steps: + return [Finding("unparsed", "error", "", + "no 'procedure' list found — an unreadable procedure is " + "not a clean one")] + + dynamic_seen = False + content_proof_anywhere = False + status_only_steps: list[tuple[str, list[int]]] = [] + declared_outputs: dict[str, str] = {} # output name -> step it came from + all_text = [] + + for i, step in enumerate(steps): + last = i == len(steps) - 1 + if not isinstance(step, dict): + out.append(Finding("unparsed", "error", f"#{i}", + "step is not a mapping")) + continue + + sid = str(step.get("step_id") or f"#{i}") + action = step.get("action") or {} + atype = action.get("type") if isinstance(action, dict) else None + expect = step.get("expect") + + # exec_target is god-view only, per goe/models/procedure.py. + if atype == "exec_target": + out.append(Finding("exec_target_in_procedure", "error", sid, + "exec_target is L1-diagnostic only and must not " + "appear in an L2 attack procedure — it runs " + "inside the target, so it proves nothing about " + "what an attacker can reach")) + + for name in (step.get("outputs") or {}): + declared_outputs[str(name)] = sid + + leaves = list(_walk_assertions(expect)) + if not leaves: + out.append(Finding( + "no_expect", "error" if last else "warn", sid, + "no expect" + (" — final step, so the exploit's payoff is " + "unverified" if last else ""))) + continue + + texts: list[tuple[str, str]] = [] # (assertion key, value) + structural_only = True + for a in leaves: + for k, v in a.items(): + if k in TEXT_KEYS and isinstance(v, str): + texts.append((k, v)) + structural_only = False + elif k in REGEX_KEYS: + structural_only = False + elif k not in STRUCTURAL_KEYS: + structural_only = False + + all_text.extend(v for _, v in texts) + if any(DYNAMIC.search(v) for _, v in texts): + dynamic_seen = True + if not structural_only: + content_proof_anywhere = True + + # Success-status-only. Recorded per step but judged for the PROCEDURE: + # a step that posts a payload and asserts 201, followed by a step that + # verifies the payload fired, is a correct pattern. Only a procedure in + # which NO step ever asserts content is proving nothing. + statuses = [a["status"] for a in leaves + if isinstance(a.get("status"), int)] + if structural_only and statuses and all(200 <= s < 400 for s in statuses) \ + and len(leaves) == len(statuses): + status_only_steps.append((sid, statuses)) + + sent = _sent_blob(action) + for key, t in texts: + from_stdout = key in ("stdout_contains", "received_contains") + s = t.strip() + if not s or DYNAMIC.search(s): + continue + low = s.lower() + # Context matters: `stdout_contains: root` after `whoami` is a + # strong assertion; `body_contains: root` on a web page is not. + # Calibrated against tests/fixtures/procedures/*.yaml. + if low in GENERIC and not from_stdout: + out.append(Finding("generic_text", "warn", sid, + f"asserts on {s!r}, which appears on error " + "and login pages too")) + elif len(s) < MIN_LEN: + out.append(Finding("short_text", "warn", sid, + f"asserts on {len(s)} characters ({s!r}) — " + "likely to match incidentally")) + if len(low) >= MIN_LEN and sent and low in sent: + out.append(Finding( + "tautological", "warn", sid, + f"asserts on {s!r}, which this step transmits — check this is " + "a stored-payload round-trip and not a bare reflection")) + + # Only an error if the whole procedure never asserts content anywhere. + if status_only_steps and not content_proof_anywhere: + names = ", ".join(sid for sid, _ in status_only_steps) + out.append(Finding( + "status_only_2xx", "error", "", + f"no step asserts on content; only success statuses ({names}) — an " + "error page, a redirect and a stack trace all satisfy this")) + + if not dynamic_seen: + out.append(Finding( + "no_dynamic_evidence", "warn", "", + "no assertion references ${edge.*} or ${steps.*} — nothing ties the " + "proof to this scenario's values")) + + joined = " ".join(all_text) + for name, sid in declared_outputs.items(): + if f"steps.{sid}.{name}" not in joined and \ + not any(f"steps.{sid}.{name}" in _sent_blob(s.get("action") or {}) + for s in steps if isinstance(s, dict)): + out.append(Finding("unused_output", "warn", sid, + f"captures output {name!r} that no later step " + "uses — the chain may not be chained")) + + out.sort(key=lambda f: (0 if f.severity == "error" else 1, f.step, f.rule)) + return out + + +def lint_file(path: Path) -> list[Finding]: + try: + doc = yaml.safe_load(path.read_text()) + except yaml.YAMLError as exc: + return [Finding("unparsed", "error", "", f"invalid YAML: {exc}")] + return lint_procedure(doc) + + +def has_errors(findings: list[Finding]) -> bool: + return any(f.severity == "error" for f in findings) + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(prog="procedure-lint") + ap.add_argument("paths", nargs="+", type=Path) + ap.add_argument("--json", action="store_true", dest="as_json") + ap.add_argument("--warn-as-error", action="store_true") + args = ap.parse_args(argv) + + files: list[Path] = [] + for p in args.paths: + files.extend(sorted(p.rglob("*.y*ml")) if p.is_dir() else [p]) + + report, failed = {}, False + for f in files: + findings = lint_file(f) + report[str(f)] = [asdict(x) for x in findings] + bad = has_errors(findings) or (args.warn_as_error and findings) + failed |= bad + if not args.as_json: + print(f"{'FAIL' if bad else 'ok':4} {f}") + for x in findings: + print(x.line()) + if args.as_json: + print(json.dumps(report, indent=2)) + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/fixtures/procedures/strong.yaml b/tests/fixtures/procedures/strong.yaml new file mode 100644 index 0000000..c223b83 --- /dev/null +++ b/tests/fixtures/procedures/strong.yaml @@ -0,0 +1,30 @@ +procedure: + - step_id: reach + action: + type: http_request + method: GET + url: "http://${target_host}:${target_port}/" + expect: + all: + - status: 200 + - body_contains: "St Dismas Staff Intranet" + - step_id: read_policy + action: + type: http_request + method: GET + url: "http://${target_host}:${target_port}/policies/it-sec-014" + expect: + body_contains: "capitalised department word followed by the year" + outputs: + rule: "body" + - step_id: authenticate + action: + type: http_request + method: GET + url: "http://${target_host}:${target_port}/portal" + headers: + Authorization: "Basic ${steps.read_policy.rule}" + expect: + all: + - status: 200 + - body_contains: "${edge.portal_flag.value}" diff --git a/tests/fixtures/procedures/weak.yaml b/tests/fixtures/procedures/weak.yaml new file mode 100644 index 0000000..0f5c8b5 --- /dev/null +++ b/tests/fixtures/procedures/weak.yaml @@ -0,0 +1,26 @@ +procedure: + - step_id: reach + action: + type: http_request + method: GET + url: "http://${target_host}:${target_port}/" + expect: + status: 200 + - step_id: inject + action: + type: http_request + method: GET + url: "http://${target_host}:${target_port}/search?q=phantom_ride" + expect: + body_contains: "phantom_ride" + - step_id: read + action: + type: http_request + method: GET + url: "http://${target_host}:${target_port}/backup.bak" + expect: + body_contains: "admin" + - step_id: shell + action: + type: exec_attacker + command: "curl -s http://${target_host}/cgi-bin/x" diff --git a/tests/test_procedure_lint.py b/tests/test_procedure_lint.py new file mode 100644 index 0000000..257eaef --- /dev/null +++ b/tests/test_procedure_lint.py @@ -0,0 +1,176 @@ +"""Tests for the attacker-procedure assertion lint. + +Every rule is checked in both directions — it fires on the shape it targets, and +stays quiet on a correct procedure. Severities were calibrated against the real +fixtures in tests/fixtures/procedures/, which is why several rules are warnings: +`stdout_contains: root` after `whoami` is a strong assertion, and a step that +posts a payload and asserts 201 is fine when a later step verifies it fired. +""" + +from pathlib import Path + +from goe.eval.procedure_lint import Finding, has_errors, lint_file, lint_procedure + +FIXTURES = Path(__file__).parent / "fixtures" / "procedures" + + +def rules(findings: list[Finding]) -> set[str]: + return {f.rule for f in findings} + + +def errors(findings: list[Finding]) -> set[str]: + return {f.rule for f in findings if f.severity == "error"} + + +def _step(step_id="s", action=None, expect=None, **kw): + step = {"step_id": step_id, "action": action or {"type": "http_request", + "method": "GET", + "url": "http://h/"}} + if expect is not None: + step["expect"] = expect + step.update(kw) + return step + + +# --- errors --------------------------------------------------------------- + +def test_missing_expect_on_final_step_is_an_error(): + doc = {"procedure": [_step()]} + assert "no_expect" in errors(lint_procedure(doc)) + + +def test_missing_expect_mid_chain_is_a_warning(): + doc = {"procedure": [ + _step("a"), + _step("b", expect={"body_contains": "${edge.flag.value}"}), + ]} + findings = lint_procedure(doc) + assert "no_expect" in rules(findings) + assert not has_errors(findings) + + +def test_status_only_across_whole_procedure_is_an_error(): + doc = {"procedure": [ + _step("a", expect={"status": 200}), + _step("b", expect={"status": 201}), + ]} + assert "status_only_2xx" in errors(lint_procedure(doc)) + + +def test_status_only_step_is_fine_when_another_step_proves_content(): + # Posting a payload and asserting 201, then verifying it fired, is correct. + doc = {"procedure": [ + _step("inject", expect={"status": 201}), + _step("verify", expect={"body_contains": "${edge.xss.marker}"}), + ]} + assert "status_only_2xx" not in rules(lint_procedure(doc)) + + +def test_exec_target_in_a_procedure_is_an_error(): + doc = {"procedure": [_step("s", action={"type": "exec_target", + "command": "id"}, + expect={"stdout_contains": "uid=0"})]} + assert "exec_target_in_procedure" in errors(lint_procedure(doc)) + + +def test_document_without_a_procedure_list_is_an_error(): + assert "unparsed" in errors(lint_procedure({})) + assert "unparsed" in errors(lint_procedure({"procedure": []})) + + +# --- warnings ------------------------------------------------------------- + +def test_generic_body_text_warns(): + doc = {"procedure": [_step(expect={"body_contains": "success"})]} + assert "generic_text" in rules(lint_procedure(doc)) + + +def test_generic_word_from_stdout_does_not_warn(): + # `whoami` returning root is exactly the right assertion. + doc = {"procedure": [_step("s", action={"type": "exec_attacker", + "command": "whoami"}, + expect={"stdout_contains": "root"})]} + assert "generic_text" not in rules(lint_procedure(doc)) + + +def test_reflected_value_warns_but_does_not_fail(): + doc = {"procedure": [_step( + "s", + action={"type": "http_request", "method": "GET", + "url": "http://h/search?q=phantom_ride"}, + expect={"body_contains": "phantom_ride"})]} + findings = lint_procedure(doc) + assert "tautological" in rules(findings) + assert "tautological" not in errors(findings) + + +def test_interpolated_value_is_not_treated_as_reflection(): + doc = {"procedure": [_step( + "s", + action={"type": "http_request", "method": "GET", + "url": "http://h/?q=${edge.token.value}"}, + expect={"body_contains": "${edge.token.value}"})]} + assert "tautological" not in rules(lint_procedure(doc)) + + +def test_missing_dynamic_reference_warns_once(): + doc = {"procedure": [_step(expect={"body_contains": "St Dismas Intranet"})]} + findings = lint_procedure(doc) + assert [f.rule for f in findings].count("no_dynamic_evidence") == 1 + + +def test_unused_output_warns(): + doc = {"procedure": [ + _step("grab", expect={"body_contains": "${edge.a.v}"}, + outputs={"token": "body"}), + _step("next", expect={"body_contains": "${edge.b.v}"}), + ]} + assert "unused_output" in rules(lint_procedure(doc)) + + +def test_used_output_does_not_warn(): + doc = {"procedure": [ + _step("grab", expect={"body_contains": "${edge.a.v}"}, + outputs={"token": "body"}), + _step("next", + action={"type": "http_request", "method": "GET", + "url": "http://h/?t=${steps.grab.token}"}, + expect={"body_contains": "${edge.b.v}"}), + ]} + assert "unused_output" not in rules(lint_procedure(doc)) + + +# --- nesting -------------------------------------------------------------- + +def test_all_assertions_are_walked(): + doc = {"procedure": [_step(expect={"all": [{"status": 200}, + {"body_contains": "success"}]})]} + findings = lint_procedure(doc) + assert "generic_text" in rules(findings) + assert "status_only_2xx" not in rules(findings) + + +# --- the project's own fixtures ------------------------------------------- + +def test_existing_fixtures_do_not_error_except_exec_target(): + """A noisy lint is one people learn to skip, so this pins the false-positive + rate against the procedures already in the repo.""" + offenders = {} + for path in sorted(FIXTURES.glob("*.yaml")): + if path.name in ("weak.yaml",): + continue + errs = errors(lint_file(path)) + if errs: + offenders[path.name] = errs + # Only the two fixtures that deliberately exercise the god-view action. + # Their own headers say so ("Tests: exec_target (god-view)") — they are + # executor fixtures, not generated attacker procedures. + assert offenders == { + "exec_and_listen.yaml": {"exec_target_in_procedure"}, + "step_chaining.yaml": {"exec_target_in_procedure"}, + }, offenders + + +def test_weak_fixture_fails_and_strong_fixture_is_clean(): + assert has_errors(lint_file(FIXTURES / "weak.yaml")) + assert not has_errors(lint_file(FIXTURES / "strong.yaml"))