fix(benchmark): seed engage brief, close phase-skip loophole, evidence-backed coverage

This commit is contained in:
Violin
2026-08-11 17:09:35 +01:00
parent b8262d585d
commit 6b2bc26af3
14 changed files with 273 additions and 40 deletions
+17 -2
View File
@@ -203,14 +203,29 @@ def build_ffuf(args: dict) -> str:
return " ".join(parts)
def resolve_ffuf_wordlist(requested: object = "") -> str:
"""Resolve an ffuf wordlist across common Kali, Parrot, and custom installs."""
def resolve_ffuf_wordlist(requested: object = "", eng_dir: object = "") -> str:
"""Resolve an ffuf wordlist across common Kali, Parrot, custom installs, and engagement evidence."""
candidates: list[Path] = []
requested_text = os.path.expandvars(str(requested or "").strip())
if requested_text:
candidates.append(Path(requested_text).expanduser())
eng_text = os.path.expandvars(
str(
eng_dir or os.environ.get("ENG_DIR", "") or os.environ.get("VIOLIN_ENG_ROOT", "")
).strip()
)
if eng_text:
eng_path = Path(eng_text).expanduser()
candidates.extend(
(
eng_path / "evidence" / "recon" / "focused_wordlist.txt",
eng_path / "evidence" / "recon" / "wordlist.txt",
eng_path / "evidence" / "wordlist.txt",
)
)
seclists_root = os.environ.get("SECLISTS", "").strip()
if seclists_root:
candidates.append(
@@ -1,3 +1,4 @@
import os
import shlex
import sys
from pathlib import Path
@@ -51,7 +52,16 @@ def handle_ffuf(args, **kwargs):
values = dict(args or {})
try:
values["wordlist"] = resolve_ffuf_wordlist(values.get("wordlist"))
orig_eng_dir = os.environ.get("ENG_DIR")
if values.get("eng_dir"):
os.environ["ENG_DIR"] = str(values["eng_dir"])
try:
values["wordlist"] = resolve_ffuf_wordlist(values.get("wordlist"))
finally:
if orig_eng_dir is None:
os.environ.pop("ENG_DIR", None)
else:
os.environ["ENG_DIR"] = orig_eng_dir
token_file = str(values.get("auth_token_file") or "").strip()
if token_file:
engagement = _eng_path(str(values.get("eng_dir") or ""))
+46 -8
View File
@@ -73,14 +73,26 @@ def _validate_phase_exit(engagement: Path, task_id: str, status: str) -> None:
entries = matrix.get("coverage") if isinstance(matrix, dict) else None
if not isinstance(entries, dict) or not entries:
raise ValueError("coverage matrix must contain a non-empty coverage mapping")
unresolved_coverage = [
name
for name, entry in entries.items()
if not isinstance(entry, dict)
or str(entry.get("status") or "").strip().lower()
not in {"tested", "not_applicable", "blocked"}
or not str(entry.get("evidence_or_reason") or "").strip()
]
unresolved_coverage = []
for name, entry in entries.items():
if not isinstance(entry, dict):
unresolved_coverage.append(name)
continue
status = str(entry.get("status") or "").strip().lower()
reason = str(entry.get("evidence_or_reason") or "").strip()
if status not in {"tested", "not_applicable", "blocked"} or not reason:
unresolved_coverage.append(name)
continue
# not_applicable must be backed by an evidence file: a bare
# reason ("no rate-limit behavior observed") is a memory
# reconstruction, the exact false-positive factory. The
# 2026-08-11 run self-declared rate_limits N/A against a real
# challenge because it never probed. blocked must name the
# guard that prevented testing.
if status == "not_applicable" and "evidence/" not in reason:
unresolved_coverage.append(f"{name} (not_applicable without evidence file)")
elif status == "blocked" and "guard" not in reason.lower():
unresolved_coverage.append(f"{name} (blocked without guard reference)")
if unresolved_coverage:
raise ValueError(
"VULN_RESEARCH cannot close with undispositioned coverage: "
@@ -94,6 +106,32 @@ def _validate_phase_exit(engagement: Path, task_id: str, status: str) -> None:
"VULN_RESEARCH cannot close with unresolved hypotheses: " + ", ".join(unresolved)
)
if phase.value == "REPORTING":
scope_path = engagement / "scope" / "scope.yaml"
scope_data = (
yaml.safe_load(scope_path.read_text(encoding="utf-8")) if scope_path.is_file() else {}
)
# Benchmark mode: REPORTING may not close unless the run actually
# reached EXPLOITATION or later. The 2026-08-11 regression logged
# every command as phase=recon, declared RECON "coverage finalized",
# skipped PT-102/PT-103, and jumped straight to REPORTING — scoring
# 6/20 with 14 challenges never touched. The agent cannot fake this:
# evidence only accumulates by running commands in a later phase.
if isinstance(scope_data, dict) and (scope_data.get("benchmark") or {}).get("mode") is True:
history_path = engagement / "state" / "history.md"
reached_later_phase = False
if history_path.is_file():
history_text = history_path.read_text(encoding="utf-8", errors="replace")
for token in re.findall(r"phase=([a-z_]+)", history_text):
if token in {"exploitation", "post_exploitation", "privesc", "flags"}:
reached_later_phase = True
break
if not reached_later_phase:
raise ValueError(
"REPORTING cannot close: no commands were executed in EXPLOITATION or a "
"later phase (all history is phase=recon). Activate PT-103 and run "
"proof-verification commands under phase=exploitation before reporting; "
"skipping the exploitation phase produces an incomplete assessment."
)
missing: list[str] = []
for item in board:
if item.canonical_status() != "Validated":
+16 -2
View File
@@ -69,7 +69,15 @@ class RecordHypothesisArgsModel(BaseModel):
port: str = ""
id: str = ""
title: str = ""
status: str = ""
status: str = Field(
"",
description=(
"Canonical status: 'Candidate', 'Likely', 'Validated', or 'Rejected'. "
"When status='Validated', 'runtime_evidence' is required (path under evidence/). "
"When status='Rejected', 'verification_status' ('syntax_confirmed' or 'not_implemented'), "
"'test_command', 'test_response', and 'rejection_reason' are required."
),
)
confidence: str = Field("", description="0.1-1.0 guesstimate; escalate only with evidence")
timebox: str = Field("", description="e.g. 4 tool batches or 30 min — then re-evaluate")
cheapest_test: str = Field(
@@ -96,7 +104,13 @@ class RecordHypothesisArgsModel(BaseModel):
)
test_command: str = Field("", description="Exact syntax tested, including argument order")
test_response: str = Field("", description="Exact decisive response or error")
verification_status: str = ""
verification_status: str = Field(
"",
description=(
"Required when status='Rejected': must be 'syntax_confirmed' or 'not_implemented'. "
"Use 'syntax_uncertain' or 'not_tested' to keep hypothesis active for re-testing."
),
)
kill_criteria: str = Field(
"",
description="Evidence that contradicts, or no new info in N batches — then kill & log in Decoy Trail",
+21
View File
@@ -57,8 +57,21 @@ _NON_TARGET_DOTTED_TOKENS = frozenset(
"urllib.parse",
"urllib.error",
"http.client",
"http.server",
"json.decoder",
"json.encoder",
"json.tool",
"xml.etree",
"unittest.mock",
"importlib.util",
"asyncio.runner",
"wsgiref.simple_server",
"jwt.io",
"example.com",
"example.org",
"example.net",
"schema.org",
"w3.org",
}
)
_LOCAL_HOSTS = {"127.0.0.1", "0.0.0.0", "localhost", "::1"}
@@ -173,13 +186,21 @@ def extract_target_candidates(command: str) -> list[str]:
"""Return ordered, unique network targets found in a shell command."""
candidates: list[str] = []
skip_path_value = False
skip_next_token = False
for token in _command_tokens(command):
if skip_path_value:
skip_path_value = False
continue
if skip_next_token:
skip_next_token = False
continue
if token in _PATH_VALUE_FLAGS:
skip_path_value = True
continue
if token == "-m" or token.startswith("-m="):
if token == "-m":
skip_next_token = True
continue
if token in _REDIRECTION_OPERATORS or any(
token.startswith(f"{flag}=") for flag in _PATH_VALUE_FLAGS
):
+2 -2
View File
@@ -356,8 +356,8 @@ def block_terminal_command(command: str) -> str | None:
def _message(reason: str) -> str:
return (
"RAW TERMINAL TARGET EXECUTION BLOCKED by Violin: "
f"{reason}. Use `violin_exec` for one command or `violin_exec_burst` "
"for a bounded batch so scope, phase, PTT, hypotheses, history, "
f"{reason}. Use `violin_exec` or `violin_exec_burst` (or typed tools like `violin_ffuf`, `violin_httpx`) "
"for any target command so scope, phase, PTT, hypotheses, history, "
"evidence, and sync gates are enforced. The built-in terminal remains "
"available for host-local preparation, tests, builds, and bookkeeping."
)