Skip to content

Commit dcce3b1

Browse files
jfeng18claude
andcommitted
fix(sec-core): prompt scanner fail-ask on scan error
Convert 4 error paths in the cosh prompt_scanner_hook from fail-open (_allow) to fail-ask (_ask), so users are informed when scanning fails rather than silently passing unscanned prompts. Changed paths: CLI timeout, CLI exception, non-zero exit code, unparseable CLI output. Also fix _format_cosh: missing/unknown verdict now triggers fail-ask instead of fail-open, and use "summary" key (matching CLI output schema) instead of "error". Empty/missing prompt input still correctly returns allow (nothing to scan). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 7068938 commit dcce3b1

2 files changed

Lines changed: 87 additions & 21 deletions

File tree

src/agent-sec-core/cosh-extension/hooks/prompt_scanner_hook.py

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,11 @@ def _allow() -> str:
4343
return json.dumps({"decision": "allow"})
4444

4545

46+
def _ask(reason: str) -> str:
47+
"""Return an 'ask' cosh HookOutput so the user sees the issue and decides."""
48+
return json.dumps({"decision": "ask", "reason": reason}, ensure_ascii=False)
49+
50+
4651
def _build_detail_reason(scan_result: dict) -> str:
4752
"""Build a detailed reason string from scan result for security operations."""
4853
threat_type = scan_result.get("threat_type", "")
@@ -71,9 +76,9 @@ def _format_cosh(scan_result: dict) -> str:
7176
verdict == "pass" -> decision "allow"
7277
verdict == "warn" -> decision "ask" (let user decide)
7378
verdict == "deny" -> decision "ask" (let user decide)
74-
otherwise -> fail-open "allow"
79+
otherwise -> decision "ask" (fail-ask: surface scan failure to user)
7580
"""
76-
verdict = scan_result.get("verdict", "pass")
81+
verdict = scan_result.get("verdict")
7782

7883
if verdict == "pass":
7984
return json.dumps({"decision": "allow"})
@@ -92,8 +97,9 @@ def _format_cosh(scan_result: dict) -> str:
9297
{"decision": "ask", "reason": reason},
9398
ensure_ascii=False,
9499
)
95-
# other error or unknown verdict -> fail-open
96-
return json.dumps({"decision": "allow"})
100+
# missing, error, or unknown verdict -> fail-ask: let user decide
101+
error_detail = scan_result.get("summary", verdict or "unknown")
102+
return _ask(f"[prompt-scanner] 扫描异常 (verdict={verdict}): {error_detail}")
97103

98104

99105
# -- main ------------------------------------------------------------------
@@ -142,11 +148,11 @@ def main() -> None:
142148
f"[prompt-scanner] CLI timed out after {exc.timeout}s",
143149
file=sys.stderr,
144150
)
145-
print(_allow())
151+
print(_ask(f"[prompt-scanner] 安全扫描超时 ({exc.timeout}s),未完成扫描"))
146152
return
147153
except Exception as exc:
148154
print(f"[prompt-scanner] CLI invocation failed: {exc}", file=sys.stderr)
149-
print(_allow())
155+
print(_ask(f"[prompt-scanner] 安全扫描调用失败: {exc}"))
150156
return
151157

152158
if proc.returncode != 0:
@@ -156,7 +162,7 @@ def main() -> None:
156162
f" {'; '.join(stderr_tail)}",
157163
file=sys.stderr,
158164
)
159-
print(_allow())
165+
print(_ask(f"[prompt-scanner] 安全扫描异常退出 (code={proc.returncode})"))
160166
return
161167

162168
# 4. Parse ScanResult JSON from stdout
@@ -167,7 +173,7 @@ def main() -> None:
167173
f"[prompt-scanner] failed to parse CLI output: {exc}",
168174
file=sys.stderr,
169175
)
170-
print(_allow())
176+
print(_ask("[prompt-scanner] 安全扫描结果解析失败,未完成扫描"))
171177
return
172178

173179
# 5. Format and print cosh output

src/agent-sec-core/tests/unit-test/cosh_hooks/test_prompt_scanner_hook.py

Lines changed: 73 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
77
Tests cover:
88
1. verdict → decision mapping (pass, warn, deny, error, unknown)
9-
2. Error verdict fails open
9+
2. Error/unknown verdicts fail-ask (not fail-open)
1010
3. Subprocess integration: pipe JSON into the hook and verify stdout
1111
"""
1212

@@ -90,35 +90,39 @@ def test_deny_returns_ask(self):
9090

9191

9292
class TestFormatCoshError:
93-
"""verdict=error → fail-open allow."""
93+
"""verdict=error → fail-ask: surface to user."""
9494

95-
def test_error_returns_allow(self):
95+
def test_error_returns_ask(self):
9696
result = json.loads(
9797
_format_cosh(
9898
{
9999
"verdict": "error",
100-
"summary": "internal scanner failure",
100+
"summary": "agent-sec daemon is unavailable",
101101
}
102102
)
103103
)
104-
assert result["decision"] == "allow"
104+
assert result["decision"] == "ask"
105+
assert "扫描异常" in result["reason"]
106+
assert "agent-sec daemon is unavailable" in result["reason"]
105107

106-
def test_error_with_empty_summary_returns_allow(self):
108+
def test_error_with_no_detail_returns_ask(self):
107109
result = json.loads(_format_cosh({"verdict": "error"}))
108-
assert result["decision"] == "allow"
110+
assert result["decision"] == "ask"
111+
assert "error" in result["reason"]
109112

110113

111114
class TestFormatCoshUnknown:
112-
"""Unknown verdict → fail-open allow."""
115+
"""Unknown verdict → fail-ask: surface to user."""
113116

114-
def test_unknown_verdict_returns_allow(self):
117+
def test_unknown_verdict_returns_ask(self):
115118
result = json.loads(_format_cosh({"verdict": "unknown"}))
116-
assert result["decision"] == "allow"
119+
assert result["decision"] == "ask"
117120

118-
def test_missing_verdict_defaults_to_allow(self):
119-
"""When verdict key is missing, default is 'pass' → allow."""
121+
def test_missing_verdict_returns_ask(self):
122+
"""When verdict key is missing, fail-ask (not fail-open)."""
120123
result = json.loads(_format_cosh({}))
121-
assert result["decision"] == "allow"
124+
assert result["decision"] == "ask"
125+
assert "扫描异常" in result["reason"]
122126

123127

124128
# ---------------------------------------------------------------------------
@@ -220,3 +224,59 @@ def fake_run(args, **kwargs):
220224
"user_input",
221225
]
222226
assert captured["kwargs"]["check"] is False
227+
228+
def _mock_stdin(self, monkeypatch):
229+
"""Set stdin to a valid prompt so main() reaches the subprocess call."""
230+
monkeypatch.setattr(
231+
prompt_scanner_hook.sys,
232+
"stdin",
233+
io.StringIO(json.dumps({"prompt": "test prompt"})),
234+
)
235+
236+
def test_cli_timeout_returns_ask(self, monkeypatch, capsys):
237+
self._mock_stdin(monkeypatch)
238+
239+
def fake_run(args, **kwargs):
240+
raise subprocess.TimeoutExpired(args, kwargs.get("timeout", 10))
241+
242+
monkeypatch.setattr(prompt_scanner_hook.subprocess, "run", fake_run)
243+
prompt_scanner_hook.main()
244+
output = json.loads(capsys.readouterr().out)
245+
assert output["decision"] == "ask"
246+
assert "超时" in output["reason"]
247+
248+
def test_cli_exception_returns_ask(self, monkeypatch, capsys):
249+
self._mock_stdin(monkeypatch)
250+
251+
def fake_run(args, **kwargs):
252+
raise OSError("No such file")
253+
254+
monkeypatch.setattr(prompt_scanner_hook.subprocess, "run", fake_run)
255+
prompt_scanner_hook.main()
256+
output = json.loads(capsys.readouterr().out)
257+
assert output["decision"] == "ask"
258+
assert "调用失败" in output["reason"]
259+
260+
def test_cli_nonzero_exit_returns_ask(self, monkeypatch, capsys):
261+
self._mock_stdin(monkeypatch)
262+
263+
def fake_run(args, **kwargs):
264+
return subprocess.CompletedProcess(args, returncode=1, stdout="", stderr="segfault")
265+
266+
monkeypatch.setattr(prompt_scanner_hook.subprocess, "run", fake_run)
267+
prompt_scanner_hook.main()
268+
output = json.loads(capsys.readouterr().out)
269+
assert output["decision"] == "ask"
270+
assert "异常退出" in output["reason"]
271+
272+
def test_cli_bad_json_returns_ask(self, monkeypatch, capsys):
273+
self._mock_stdin(monkeypatch)
274+
275+
def fake_run(args, **kwargs):
276+
return subprocess.CompletedProcess(args, returncode=0, stdout="not-json", stderr="")
277+
278+
monkeypatch.setattr(prompt_scanner_hook.subprocess, "run", fake_run)
279+
prompt_scanner_hook.main()
280+
output = json.loads(capsys.readouterr().out)
281+
assert output["decision"] == "ask"
282+
assert "解析失败" in output["reason"]

0 commit comments

Comments
 (0)