diff --git a/plugins/violin_guard/command.py b/plugins/violin_guard/command.py index 8670817..f0484a1 100644 --- a/plugins/violin_guard/command.py +++ b/plugins/violin_guard/command.py @@ -58,6 +58,7 @@ class CheckCommandArgs: target: str | None = None session_id: str | None = None account_sync: bool = True + hypothesis_id: str | None = None @dataclass @@ -530,10 +531,26 @@ def check_hypothesis_freshness( Phase.PRIVESC, Phase.FLAGS, }: - researched = [h for h in relevant if h.cve_research.strip() and h.exploit_research.strip()] - if not researched: + # Online research is required before exploit execution: the engagement + # brief may have no exploit path, so the first action for any real + # exploit is to look for prior work. When the command names a specific + # hypothesis, only that hypothesis must carry research rows; otherwise + # every candidate hypothesis must. 'no results' / 'not applicable' / + # 'source unavailable' are valid truthful outcomes. + if norm_hyp_id is not None: + research_targets = [ + h + for h in relevant + if (h.id.strip().upper().removeprefix("H-").lstrip("0") or "0") == norm_hyp_id + ] + else: + research_targets = relevant + researched = [h for h in research_targets if h.cve_research.strip() and h.exploit_research.strip()] + if len(researched) < len(research_targets): missing = [] - for h in relevant: + for h in research_targets: + if h.cve_research.strip() and h.exploit_research.strip(): + continue fields = [] if not h.cve_research.strip(): fields.append("CVE Research") @@ -543,8 +560,10 @@ def check_hypothesis_freshness( result.add_error( "online research must be attempted and recorded before exploit execution; " + "; ".join(missing) - + ". Record each query/source and outcome; 'no results', 'not applicable', " - "or 'source unavailable' are valid outcomes when truthful." + + ". Record each query/source/outcome via violin_record_hypothesis " + "id=H-00N cve_research='...' exploit_research='...' — 'no results', " + "'not applicable', or 'source unavailable' are valid outcomes when " + "truthful." ) return result @@ -746,7 +765,7 @@ def check_command(args: CheckCommandArgs) -> CheckResult: phase, args.command, args.target, - hypothesis_id=active_task_hyp_id, + hypothesis_id=args.hypothesis_id or active_task_hyp_id, match_command_target=not research_primary, ) result.errors.extend(hyp_result.errors) diff --git a/plugins/violin_guard/handlers/base.py b/plugins/violin_guard/handlers/base.py index 1e2b22f..030f59c 100644 --- a/plugins/violin_guard/handlers/base.py +++ b/plugins/violin_guard/handlers/base.py @@ -92,6 +92,7 @@ def _check_command_internal(a) -> cmd_module.CheckResult: scope=a.get("scope", ""), target=a.get("target"), session_id=a.get("session_id"), + hypothesis_id=a.get("hypothesis_id"), ) ) try: diff --git a/plugins/violin_guard/schemas.py b/plugins/violin_guard/schemas.py index 3ddf267..2bf35e3 100644 --- a/plugins/violin_guard/schemas.py +++ b/plugins/violin_guard/schemas.py @@ -30,6 +30,13 @@ class CheckCommandArgsModel(BaseModel): ), ) session_id: str = "" + hypothesis_id: str = Field( + "", + description=( + "Optional hypothesis the command targets (H-00N). When provided, exploit-phase " + "guards (e.g. online-research requirement) check only this hypothesis." + ), + ) class RecordPttArgsModel(BaseModel): diff --git a/tests/guard/integration/test_plugin_guard.py b/tests/guard/integration/test_plugin_guard.py index dedae59..7b2939c 100644 --- a/tests/guard/integration/test_plugin_guard.py +++ b/tests/guard/integration/test_plugin_guard.py @@ -252,6 +252,76 @@ def test_recon_does_not_require_hypothesis(tmp_path): assert any("hypothesis guard:" in warning for warning in research3.warnings) +def test_exploit_research_gate_scopes_to_named_hypothesis(tmp_path): + """Online-research gate: with hypothesis_id, only that hypothesis needs rows. + + A command that names H-001 must not be blocked because unrelated hypotheses + (H-002) lack CVE/Exploit Research — otherwise a single unresearched + candidate walls off the whole exploit phase. + """ + skill_file = tmp_path / ".skill-loaded-ts" + eng = _init_e2e(tmp_path, skill_file) + ts = datetime.now(UTC).strftime("%Y-%m-%d %H:%M") + (eng / "hypotheses.md").write_text( + (eng / "hypotheses.md").read_text(encoding="utf-8") + + ( + f"\n### H-001: JWT alg none\n- **Status:** Validated\n- **Phase:** EXPLOITATION\n" + f"- **Target:** duck-store.escape.tech\n- **CVE Research:** NVD queried; no CVE\n" + f"- **Exploit Research:** ExploitDB; none applicable\n- **Updated:** {ts} UTC\n" + f"\n### H-002: No research done\n- **Status:** Candidate\n- **Phase:** EXPLOITATION\n" + f"- **Target:** duck-store.escape.tech\n- **Updated:** {ts} UTC\n" + ), + encoding="utf-8", + ) + ptt_path = eng / "state" / "ptt.md" + ptt_path.write_text( + ptt_path.read_text(encoding="utf-8") + .replace("| PT-010 | [~] |", "| PT-010 | [x] |") + .replace("| PT-030 | [ ] |", "| PT-030 | [~] |"), + encoding="utf-8", + ) + bind_active_task(eng, "ts") + + # Named hypothesis has research rows -> passes even though H-002 is bare. + ok = command.check_command( + command.CheckCommandArgs( + command="curl -sk -i https://duck-store.escape.tech/api/v1/admin", + phase="exploitation", + eng_dir=str(eng), + scope=str(eng / "scope" / "scope.yaml"), + hypothesis_id="H-001", + session_id="ts", + ) + ) + assert not any("online research" in error.lower() for error in ok.errors) + + # H-002 itself is still blocked (no research rows). + blocked = command.check_command( + command.CheckCommandArgs( + command="curl -sk -i https://duck-store.escape.tech/api/v1/admin", + phase="exploitation", + eng_dir=str(eng), + scope=str(eng / "scope" / "scope.yaml"), + hypothesis_id="H-002", + session_id="ts", + ) + ) + assert any("online research must be attempted" in e.lower() for e in blocked.errors) + + # Without hypothesis_id, every candidate must carry research rows. + all_blocked = command.check_command( + command.CheckCommandArgs( + command="curl -sk -i https://duck-store.escape.tech/api/v1/admin", + phase="exploitation", + eng_dir=str(eng), + scope=str(eng / "scope" / "scope.yaml"), + session_id="ts", + ) + ) + assert any("H-002 missing" in e for e in all_blocked.errors) + assert not any("H-001 missing" in e for e in all_blocked.errors) + + def test_target_scanner_ignores_dotted_files_and_handles_dev_tcp_endpoint(): candidates = extract_target_candidates( "python3 server.py --output 01-nmap-full.txt "