diff --git a/src/agent-sec-core/cosh-extension/hooks/prompt_scanner_hook.py b/src/agent-sec-core/cosh-extension/hooks/prompt_scanner_hook.py index d8c97078d..ffae025dc 100644 --- a/src/agent-sec-core/cosh-extension/hooks/prompt_scanner_hook.py +++ b/src/agent-sec-core/cosh-extension/hooks/prompt_scanner_hook.py @@ -43,6 +43,11 @@ def _allow() -> str: return json.dumps({"decision": "allow"}) +def _ask(reason: str) -> str: + """Return an 'ask' cosh HookOutput so the user sees the issue and decides.""" + return json.dumps({"decision": "ask", "reason": reason}, ensure_ascii=False) + + def _build_detail_reason(scan_result: dict) -> str: """Build a detailed reason string from scan result for security operations.""" threat_type = scan_result.get("threat_type", "") @@ -71,9 +76,9 @@ def _format_cosh(scan_result: dict) -> str: verdict == "pass" -> decision "allow" verdict == "warn" -> decision "ask" (let user decide) verdict == "deny" -> decision "ask" (let user decide) - otherwise -> fail-open "allow" + otherwise -> decision "ask" (fail-ask: surface scan failure to user) """ - verdict = scan_result.get("verdict", "pass") + verdict = scan_result.get("verdict") if verdict == "pass": return json.dumps({"decision": "allow"}) @@ -92,8 +97,9 @@ def _format_cosh(scan_result: dict) -> str: {"decision": "ask", "reason": reason}, ensure_ascii=False, ) - # other error or unknown verdict -> fail-open - return json.dumps({"decision": "allow"}) + # missing, error, or unknown verdict -> fail-ask: let user decide + error_detail = scan_result.get("summary", verdict or "unknown") + return _ask(f"[prompt-scanner] 扫描异常 (verdict={verdict}): {error_detail}") # -- main ------------------------------------------------------------------ @@ -142,11 +148,11 @@ def main() -> None: f"[prompt-scanner] CLI timed out after {exc.timeout}s", file=sys.stderr, ) - print(_allow()) + print(_ask(f"[prompt-scanner] 安全扫描超时 ({exc.timeout}s)，未完成扫描")) return except Exception as exc: print(f"[prompt-scanner] CLI invocation failed: {exc}", file=sys.stderr) - print(_allow()) + print(_ask(f"[prompt-scanner] 安全扫描调用失败: {exc}")) return if proc.returncode != 0: @@ -156,7 +162,7 @@ def main() -> None: f" {'; '.join(stderr_tail)}", file=sys.stderr, ) - print(_allow()) + print(_ask(f"[prompt-scanner] 安全扫描异常退出 (code={proc.returncode})")) return # 4. Parse ScanResult JSON from stdout @@ -167,7 +173,7 @@ def main() -> None: f"[prompt-scanner] failed to parse CLI output: {exc}", file=sys.stderr, ) - print(_allow()) + print(_ask("[prompt-scanner] 安全扫描结果解析失败，未完成扫描")) return # 5. Format and print cosh output diff --git a/src/agent-sec-core/tests/unit-test/cosh_hooks/test_prompt_scanner_hook.py b/src/agent-sec-core/tests/unit-test/cosh_hooks/test_prompt_scanner_hook.py index ead7c10c8..53bf8bc27 100644 --- a/src/agent-sec-core/tests/unit-test/cosh_hooks/test_prompt_scanner_hook.py +++ b/src/agent-sec-core/tests/unit-test/cosh_hooks/test_prompt_scanner_hook.py @@ -6,7 +6,7 @@ Tests cover: 1. verdict → decision mapping (pass, warn, deny, error, unknown) -2. Error verdict fails open +2. Error/unknown verdicts fail-ask (not fail-open) 3. Subprocess integration: pipe JSON into the hook and verify stdout """ @@ -90,35 +90,39 @@ def test_deny_returns_ask(self): class TestFormatCoshError: - """verdict=error → fail-open allow.""" + """verdict=error → fail-ask: surface to user.""" - def test_error_returns_allow(self): + def test_error_returns_ask(self): result = json.loads( _format_cosh( { "verdict": "error", - "summary": "internal scanner failure", + "summary": "agent-sec daemon is unavailable", } ) ) - assert result["decision"] == "allow" + assert result["decision"] == "ask" + assert "扫描异常" in result["reason"] + assert "agent-sec daemon is unavailable" in result["reason"] - def test_error_with_empty_summary_returns_allow(self): + def test_error_with_no_detail_returns_ask(self): result = json.loads(_format_cosh({"verdict": "error"})) - assert result["decision"] == "allow" + assert result["decision"] == "ask" + assert "error" in result["reason"] class TestFormatCoshUnknown: - """Unknown verdict → fail-open allow.""" + """Unknown verdict → fail-ask: surface to user.""" - def test_unknown_verdict_returns_allow(self): + def test_unknown_verdict_returns_ask(self): result = json.loads(_format_cosh({"verdict": "unknown"})) - assert result["decision"] == "allow" + assert result["decision"] == "ask" - def test_missing_verdict_defaults_to_allow(self): - """When verdict key is missing, default is 'pass' → allow.""" + def test_missing_verdict_returns_ask(self): + """When verdict key is missing, fail-ask (not fail-open).""" result = json.loads(_format_cosh({})) - assert result["decision"] == "allow" + assert result["decision"] == "ask" + assert "扫描异常" in result["reason"] # --------------------------------------------------------------------------- @@ -220,3 +224,63 @@ def fake_run(args, **kwargs): "user_input", ] assert captured["kwargs"]["check"] is False + + def _mock_stdin(self, monkeypatch): + """Set stdin to a valid prompt so main() reaches the subprocess call.""" + monkeypatch.setattr( + prompt_scanner_hook.sys, + "stdin", + io.StringIO(json.dumps({"prompt": "test prompt"})), + ) + + def test_cli_timeout_returns_ask(self, monkeypatch, capsys): + self._mock_stdin(monkeypatch) + + def fake_run(args, **kwargs): + raise subprocess.TimeoutExpired(args, kwargs.get("timeout", 10)) + + monkeypatch.setattr(prompt_scanner_hook.subprocess, "run", fake_run) + prompt_scanner_hook.main() + output = json.loads(capsys.readouterr().out) + assert output["decision"] == "ask" + assert "超时" in output["reason"] + + def test_cli_exception_returns_ask(self, monkeypatch, capsys): + self._mock_stdin(monkeypatch) + + def fake_run(args, **kwargs): + raise OSError("No such file") + + monkeypatch.setattr(prompt_scanner_hook.subprocess, "run", fake_run) + prompt_scanner_hook.main() + output = json.loads(capsys.readouterr().out) + assert output["decision"] == "ask" + assert "调用失败" in output["reason"] + + def test_cli_nonzero_exit_returns_ask(self, monkeypatch, capsys): + self._mock_stdin(monkeypatch) + + def fake_run(args, **kwargs): + return subprocess.CompletedProcess( + args, returncode=1, stdout="", stderr="segfault" + ) + + monkeypatch.setattr(prompt_scanner_hook.subprocess, "run", fake_run) + prompt_scanner_hook.main() + output = json.loads(capsys.readouterr().out) + assert output["decision"] == "ask" + assert "异常退出" in output["reason"] + + def test_cli_bad_json_returns_ask(self, monkeypatch, capsys): + self._mock_stdin(monkeypatch) + + def fake_run(args, **kwargs): + return subprocess.CompletedProcess( + args, returncode=0, stdout="not-json", stderr="" + ) + + monkeypatch.setattr(prompt_scanner_hook.subprocess, "run", fake_run) + prompt_scanner_hook.main() + output = json.loads(capsys.readouterr().out) + assert output["decision"] == "ask" + assert "解析失败" in output["reason"]