From 4b9259792f7250e9e9583aa21b760de62b42f6d8 Mon Sep 17 00:00:00 2001 From: FireCollector <220344011+FireCollector@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:47:04 +0000 Subject: [PATCH 1/8] fix(safety): scan all script-like filter arguments --- tests/tools/safety/test_wrapper.py | 21 +++++++++++++++++++++ trpc_agent_sdk/tools/safety/_filter.py | 8 +++----- 2 files changed, 24 insertions(+), 5 deletions(-) diff --git a/tests/tools/safety/test_wrapper.py b/tests/tools/safety/test_wrapper.py index 54e00646a..1266ade11 100644 --- a/tests/tools/safety/test_wrapper.py +++ b/tests/tools/safety/test_wrapper.py @@ -245,6 +245,27 @@ async def test_filter_extracts_command_as_bash(): assert result.rsp["language"] == "bash" +@pytest.mark.asyncio +async def test_filter_scans_all_script_like_fields(): + safety_filter = ToolSafetyFilter() + result = FilterResult() + + await safety_filter._before( + None, + { + "script": "echo ok", + "command": "rm -rf /", + "language": "bash", + "tool_name": "shell_tool", + }, + result, + ) + + assert result.is_continue is False + assert result.rsp["decision"] == "deny" + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) + + @pytest.mark.asyncio async def test_filter_extracts_python_code_language(): safety_filter = ToolSafetyFilter() diff --git a/trpc_agent_sdk/tools/safety/_filter.py b/trpc_agent_sdk/tools/safety/_filter.py index 3d2ef66f6..cc904b878 100644 --- a/trpc_agent_sdk/tools/safety/_filter.py +++ b/trpc_agent_sdk/tools/safety/_filter.py @@ -99,14 +99,14 @@ async def _after(self, ctx: AgentContext, req: Any, rsp: FilterResult): def _extract_script(req: dict[str, Any]) -> str: + parts: list[str] = [] for key in _SCRIPT_ARG_KEYS: value = req.get(key) if isinstance(value, str) and value.strip(): - return value + parts.append(value) code_blocks = req.get("code_blocks") if isinstance(code_blocks, list): - parts: list[str] = [] for block in code_blocks: if isinstance(block, dict): code = block.get("code", "") @@ -114,9 +114,7 @@ def _extract_script(req: dict[str, Any]) -> str: code = getattr(block, "code", "") if isinstance(code, str) and code: parts.append(code) - if parts: - return "\n".join(parts) - return "" + return "\n".join(parts) def _extract_tool_name(req: dict[str, Any]) -> str: From 686629ef7617dd0cf7d8178011b1e1279b11b6d1 Mon Sep 17 00:00:00 2001 From: FireCollector <220344011+FireCollector@users.noreply.github.com> Date: Fri, 31 Jul 2026 12:04:33 +0000 Subject: [PATCH 2/8] fix(safety): scan mixed-language tool arguments --- tests/tools/safety/test_wrapper.py | 21 +++++++++++++++++++++ trpc_agent_sdk/tools/safety/_filter.py | 9 +++++++-- 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/tests/tools/safety/test_wrapper.py b/tests/tools/safety/test_wrapper.py index 1266ade11..35db53eea 100644 --- a/tests/tools/safety/test_wrapper.py +++ b/tests/tools/safety/test_wrapper.py @@ -266,6 +266,27 @@ async def test_filter_scans_all_script_like_fields(): assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) +@pytest.mark.asyncio +async def test_filter_scans_mixed_language_fields_without_language(): + safety_filter = ToolSafetyFilter() + result = FilterResult() + + await safety_filter._before( + None, + { + "code": "print('ok')", + "command": "rm -rf /", + "tool_name": "mixed_tool", + }, + result, + ) + + assert result.is_continue is False + assert result.rsp["language"] == "unknown" + assert result.rsp["decision"] == "deny" + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) + + @pytest.mark.asyncio async def test_filter_extracts_python_code_language(): safety_filter = ToolSafetyFilter() diff --git a/trpc_agent_sdk/tools/safety/_filter.py b/trpc_agent_sdk/tools/safety/_filter.py index cc904b878..21448f308 100644 --- a/trpc_agent_sdk/tools/safety/_filter.py +++ b/trpc_agent_sdk/tools/safety/_filter.py @@ -129,13 +129,18 @@ def _extract_tool_name(req: dict[str, Any]) -> str: def _extract_language(req: dict[str, Any], tool_name: str) -> str: + has_python_source = any(isinstance(req.get(key), str) and req[key].strip() for key in ("code", "python_code")) + has_bash_source = any(isinstance(req.get(key), str) and req[key].strip() for key in ("command", "cmd", "bash_code")) + if has_python_source and has_bash_source: + return "unknown" + for key in _LANGUAGE_ARG_KEYS: value = req.get(key) if isinstance(value, str) and value.strip(): return value.strip().lower() - if isinstance(req.get("python_code"), str) or "code" in req: + if has_python_source: return "python" - if isinstance(req.get("bash_code"), str) or "command" in req or "cmd" in req: + if has_bash_source: return "bash" lowered_tool_name = tool_name.lower() if "python" in lowered_tool_name: From 150822f0cb4b63a2745f3be45bad8cf365b16420 Mon Sep 17 00:00:00 2001 From: FireCollector <220344011+FireCollector@users.noreply.github.com> Date: Fri, 31 Jul 2026 20:29:44 +0800 Subject: [PATCH 3/8] fix(safety): scan mixed-language fields separately --- tests/tools/safety/test_wrapper.py | 110 ++++++++++++++---- trpc_agent_sdk/tools/safety/_filter.py | 150 +++++++++++++++++++------ 2 files changed, 203 insertions(+), 57 deletions(-) diff --git a/tests/tools/safety/test_wrapper.py b/tests/tools/safety/test_wrapper.py index 35db53eea..e85793873 100644 --- a/tests/tools/safety/test_wrapper.py +++ b/tests/tools/safety/test_wrapper.py @@ -35,7 +35,10 @@ async def execute(): called = True return "executed" - result = await guard.run(ToolScriptScanRequest(script="rm -rf /", language="bash", tool_name="bash"), execute) + result = await guard.run( + ToolScriptScanRequest(script="rm -rf /", + language="bash", + tool_name="bash"), execute) assert result.blocked is True assert result.report.decision == Decision.DENY @@ -49,7 +52,9 @@ async def test_guard_allows_safe_execute(): async def execute(): return "executed" - result = await guard.run(ToolScriptScanRequest(script="print('ok')", language="python"), execute) + result = await guard.run( + ToolScriptScanRequest(script="print('ok')", language="python"), + execute) assert result.blocked is False assert result.result == "executed" @@ -65,7 +70,9 @@ async def execute(): called = True return "executed" - result = await guard.run(ToolScriptScanRequest(script="while True:\n pass", language="python"), execute) + result = await guard.run( + ToolScriptScanRequest(script="while True:\n pass", + language="python"), execute) assert result.report.decision == Decision.NEEDS_HUMAN_REVIEW assert result.blocked is False @@ -83,7 +90,9 @@ async def execute(): called = True return "executed" - result = await guard.run(ToolScriptScanRequest(script="while True:\n pass", language="python"), execute) + result = await guard.run( + ToolScriptScanRequest(script="while True:\n pass", + language="python"), execute) assert result.report.decision == Decision.NEEDS_HUMAN_REVIEW assert result.blocked is True @@ -95,13 +104,16 @@ def test_assert_allowed_raises_on_blocked_script(): guard = ToolSafetyGuard() with pytest.raises(ToolSafetyBlockedError): - guard.assert_allowed(ToolScriptScanRequest(script="rm -rf /", language="bash")) + guard.assert_allowed( + ToolScriptScanRequest(script="rm -rf /", language="bash")) def test_assert_allowed_allows_review_by_default(): guard = ToolSafetyGuard() - report = guard.assert_allowed(ToolScriptScanRequest(script="while True:\n pass", language="python")) + report = guard.assert_allowed( + ToolScriptScanRequest(script="while True:\n pass", + language="python")) assert report.decision == Decision.NEEDS_HUMAN_REVIEW assert report.blocked is False @@ -110,7 +122,8 @@ def test_assert_allowed_allows_review_by_default(): def test_assert_allowed_returns_report_for_safe_script(): guard = ToolSafetyGuard() - report = guard.assert_allowed(ToolScriptScanRequest(script="print('ok')", language="python")) + report = guard.assert_allowed( + ToolScriptScanRequest(script="print('ok')", language="python")) assert report.decision == Decision.ALLOW @@ -119,7 +132,10 @@ def test_guard_check_writes_audit_event(tmp_path): audit_path = tmp_path / "guard-audit.jsonl" guard = ToolSafetyGuard(audit_log_path=audit_path) - report = guard.check(ToolScriptScanRequest(script="print('ok')", language="python", tool_name="python")) + report = guard.check( + ToolScriptScanRequest(script="print('ok')", + language="python", + tool_name="python")) event = json.loads(audit_path.read_text(encoding="utf-8").splitlines()[0]) assert report.decision == Decision.ALLOW @@ -238,7 +254,10 @@ async def test_filter_extracts_command_as_bash(): safety_filter = ToolSafetyFilter() result = FilterResult() - await safety_filter._before(None, {"command": "echo ok", "tool_name": "shell_tool"}, result) + await safety_filter._before(None, { + "command": "echo ok", + "tool_name": "shell_tool" + }, result) assert result.is_continue is True assert result.rsp["decision"] == "allow" @@ -263,7 +282,8 @@ async def test_filter_scans_all_script_like_fields(): assert result.is_continue is False assert result.rsp["decision"] == "deny" - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" + for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -282,9 +302,34 @@ async def test_filter_scans_mixed_language_fields_without_language(): ) assert result.is_continue is False - assert result.rsp["language"] == "unknown" + assert result.rsp["language"] == "mixed" assert result.rsp["decision"] == "deny" - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" + for finding in result.rsp["findings"]) + assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" + for finding in result.rsp["findings"]) + + +@pytest.mark.asyncio +async def test_filter_allows_safe_mixed_language_fields_in_strict_mode(): + safety_filter = ToolSafetyFilter(block_on_review=True) + result = FilterResult() + + await safety_filter._before( + None, + { + "code": "print('ok')", + "command": "echo ok", + "tool_name": "mixed_tool", + }, + result, + ) + + assert result.is_continue is True + assert result.rsp["language"] == "mixed" + assert result.rsp["decision"] == "allow" + assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" + for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -292,7 +337,10 @@ async def test_filter_extracts_python_code_language(): safety_filter = ToolSafetyFilter() result = FilterResult() - await safety_filter._before(None, {"python_code": "print('ok')", "tool_name": "custom"}, result) + await safety_filter._before(None, { + "python_code": "print('ok')", + "tool_name": "custom" + }, result) assert result.is_continue is True assert result.rsp["decision"] == "allow" @@ -305,8 +353,14 @@ async def test_filter_infers_language_from_tool_name(): python_result = FilterResult() unknown_result = FilterResult() - await safety_filter._before(None, {"script": "print('ok')", "tool_name": "PythonRunner"}, python_result) - await safety_filter._before(None, {"script": "print('ok')", "tool_name": "custom"}, unknown_result) + await safety_filter._before(None, { + "script": "print('ok')", + "tool_name": "PythonRunner" + }, python_result) + await safety_filter._before(None, { + "script": "print('ok')", + "tool_name": "custom" + }, unknown_result) assert python_result.rsp["language"] == "python" assert unknown_result.rsp["language"] == "unknown" @@ -333,7 +387,8 @@ async def test_filter_extracts_code_blocks_from_dicts_and_objects(): assert result.is_continue is False assert result.rsp["decision"] == "deny" - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" + for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -360,7 +415,8 @@ async def test_filter_scans_command_args_and_context(): assert result.is_continue is False assert result.rsp["sanitized"] is True - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" + for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -368,8 +424,14 @@ async def test_filter_attaches_report_to_dict_response_after_execute(): safety_filter = ToolSafetyFilter() result = FilterResult(rsp={"stdout": "ok"}) - await safety_filter._before(None, {"command": "echo ok", "tool_name": "shell_tool"}, FilterResult()) - await safety_filter._after(None, {"command": "echo ok", "tool_name": "shell_tool"}, result) + await safety_filter._before(None, { + "command": "echo ok", + "tool_name": "shell_tool" + }, FilterResult()) + await safety_filter._after(None, { + "command": "echo ok", + "tool_name": "shell_tool" + }, result) assert result.rsp["stdout"] == "ok" assert result.rsp["safety_report"]["decision"] == "allow" @@ -381,8 +443,14 @@ async def test_filter_attaches_report_to_json_object_string_after_execute(): safety_filter = ToolSafetyFilter() result = FilterResult(rsp='{"stdout": "ok"}') - await safety_filter._before(None, {"command": "echo ok", "tool_name": "shell_tool"}, FilterResult()) - await safety_filter._after(None, {"command": "echo ok", "tool_name": "shell_tool"}, result) + await safety_filter._before(None, { + "command": "echo ok", + "tool_name": "shell_tool" + }, FilterResult()) + await safety_filter._after(None, { + "command": "echo ok", + "tool_name": "shell_tool" + }, result) parsed = json.loads(result.rsp) assert parsed["stdout"] == "ok" diff --git a/trpc_agent_sdk/tools/safety/_filter.py b/trpc_agent_sdk/tools/safety/_filter.py index 21448f308..a4a5963bf 100644 --- a/trpc_agent_sdk/tools/safety/_filter.py +++ b/trpc_agent_sdk/tools/safety/_filter.py @@ -24,8 +24,12 @@ from ._types import Decision from ._types import SafetyReport from ._types import ToolScriptScanRequest +from ._types import aggregate_decision +from ._types import max_risk_level -_SCRIPT_ARG_KEYS = ("script", "code", "command", "cmd", "python_code", "bash_code") +_PYTHON_ARG_KEYS = ("python_code", ) +_BASH_ARG_KEYS = ("command", "cmd", "bash_code") +_GENERIC_ARG_KEYS = ("script", ) _LANGUAGE_ARG_KEYS = ("language", "lang") _COMMAND_ARGS_KEYS = ("command_args", "args", "argv") @@ -60,22 +64,15 @@ async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult): self._current_report.set(None) if not isinstance(req, dict): return None - script = _extract_script(req) - if not script: - return None tool_name = _extract_tool_name(req) - request = ToolScriptScanRequest( - script=script, - language=_extract_language(req, tool_name), - command_args=_extract_command_args(req), - cwd=str(req.get("cwd", "")), - env=dict(req.get("env", {}) or {}), - tool_name=tool_name, - tool_metadata=dict(req.get("tool_metadata", {}) or {}), - ) - report = self.scanner.scan(request) - should_block = report.decision == Decision.DENY or (self.block_on_review - and report.decision == Decision.NEEDS_HUMAN_REVIEW) + requests = _extract_scan_requests(req, tool_name) + if not requests: + return None + report = _merge_reports( + [self.scanner.scan(request) for request in requests]) + should_block = report.decision == Decision.DENY or ( + self.block_on_review + and report.decision == Decision.NEEDS_HUMAN_REVIEW) report.set_blocked(should_block) record_safety_attributes(report) if self.audit_log_path: @@ -98,23 +95,97 @@ async def _after(self, ctx: AgentContext, req: Any, rsp: FilterResult): return None -def _extract_script(req: dict[str, Any]) -> str: - parts: list[str] = [] - for key in _SCRIPT_ARG_KEYS: - value = req.get(key) - if isinstance(value, str) and value.strip(): - parts.append(value) +def _extract_scan_requests(req: dict[str, Any], + tool_name: str) -> list[ToolScriptScanRequest]: + grouped_parts: dict[str, list[str]] = {} + + for key in _PYTHON_ARG_KEYS: + _add_script_part(grouped_parts, "python", req.get(key)) + for key in _BASH_ARG_KEYS: + _add_script_part(grouped_parts, "bash", req.get(key)) + + generic_language = _extract_language(req, tool_name) + for key in _GENERIC_ARG_KEYS: + _add_script_part(grouped_parts, generic_language, req.get(key)) + code_language = _extract_explicit_language(req) or "python" + _add_script_part(grouped_parts, code_language, req.get("code")) code_blocks = req.get("code_blocks") if isinstance(code_blocks, list): for block in code_blocks: if isinstance(block, dict): code = block.get("code", "") + language = block.get("language", "") else: code = getattr(block, "code", "") - if isinstance(code, str) and code: - parts.append(code) - return "\n".join(parts) + language = getattr(block, "language", "") + block_language = _canonical_language(language) if isinstance( + language, str) and language.strip() else generic_language + _add_script_part(grouped_parts, block_language, code) + + command_args = _extract_command_args(req) + cwd = str(req.get("cwd", "")) + env = dict(req.get("env", {}) or {}) + tool_metadata = dict(req.get("tool_metadata", {}) or {}) + requests: list[ToolScriptScanRequest] = [] + for index, (language, parts) in enumerate(grouped_parts.items()): + include_context = index == 0 + requests.append( + ToolScriptScanRequest( + script="\n".join(parts), + language=language, + command_args=command_args if include_context else [], + cwd=cwd if include_context else "", + env=env if include_context else {}, + tool_name=tool_name, + tool_metadata=tool_metadata if include_context else {}, + )) + return requests + + +def _add_script_part(grouped_parts: dict[str, list[str]], language: str, + value: Any) -> None: + if not isinstance(value, str) or not value.strip(): + return + parts = grouped_parts.setdefault(_canonical_language(language), []) + if value not in parts: + parts.append(value) + + +def _merge_reports(reports: list[SafetyReport]) -> SafetyReport: + report = reports[0] + if len(reports) == 1: + return report + + report.findings = [ + finding for item in reports for finding in item.findings + ] + report.decision = aggregate_decision(report.findings) + report.risk_level = max_risk_level(report.findings) + report.elapsed_ms = round(sum(item.elapsed_ms for item in reports), 3) + report.sanitized = any(item.sanitized for item in reports) + languages = list(dict.fromkeys(item.language for item in reports)) + report.language = languages[0] if len(languages) == 1 else "mixed" + rule_ids = [finding.rule_id for finding in report.findings] + if rule_ids: + report.summary = ( + f"Decision {report.decision.value} with {report.risk_level.value} risk from rules: " + f"{', '.join(rule_ids[:5])}.") + else: + report.summary = "No safety rules matched; execution is allowed by the current static policy." + report.telemetry_attributes.update({ + "tool.safety.decision": + report.decision.value, + "tool.safety.risk_level": + report.risk_level.value, + "tool.safety.rule_id": + ",".join(rule_ids[:10]), + "tool.safety.sanitized": + report.sanitized, + "tool.safety.duration_ms": + report.elapsed_ms, + }) + return report def _extract_tool_name(req: dict[str, Any]) -> str: @@ -128,20 +199,18 @@ def _extract_tool_name(req: dict[str, Any]) -> str: return "unknown_tool" -def _extract_language(req: dict[str, Any], tool_name: str) -> str: - has_python_source = any(isinstance(req.get(key), str) and req[key].strip() for key in ("code", "python_code")) - has_bash_source = any(isinstance(req.get(key), str) and req[key].strip() for key in ("command", "cmd", "bash_code")) - if has_python_source and has_bash_source: - return "unknown" - +def _extract_explicit_language(req: dict[str, Any]) -> str: for key in _LANGUAGE_ARG_KEYS: value = req.get(key) if isinstance(value, str) and value.strip(): - return value.strip().lower() - if has_python_source: - return "python" - if has_bash_source: - return "bash" + return _canonical_language(value) + return "" + + +def _extract_language(req: dict[str, Any], tool_name: str) -> str: + explicit_language = _extract_explicit_language(req) + if explicit_language: + return explicit_language lowered_tool_name = tool_name.lower() if "python" in lowered_tool_name: return "python" @@ -150,6 +219,15 @@ def _extract_language(req: dict[str, Any], tool_name: str) -> str: return "unknown" +def _canonical_language(language: str) -> str: + normalized = (language or "unknown").strip().lower() + if normalized in {"py", "python3"}: + return "python" + if normalized in {"shell", "sh"}: + return "bash" + return normalized + + def _extract_command_args(req: dict[str, Any]) -> list[str]: for key in _COMMAND_ARGS_KEYS: value = req.get(key) From 975992bc8132f88f6703fbf0714ea7ddb3888dc7 Mon Sep 17 00:00:00 2001 From: FireCollector <220344011+FireCollector@users.noreply.github.com> Date: Fri, 31 Jul 2026 20:33:00 +0800 Subject: [PATCH 4/8] style: format mixed-language safety fix --- tests/tools/safety/test_wrapper.py | 90 +++++++------------------- trpc_agent_sdk/tools/safety/_filter.py | 42 +++++------- 2 files changed, 37 insertions(+), 95 deletions(-) diff --git a/tests/tools/safety/test_wrapper.py b/tests/tools/safety/test_wrapper.py index e85793873..7608db65a 100644 --- a/tests/tools/safety/test_wrapper.py +++ b/tests/tools/safety/test_wrapper.py @@ -35,10 +35,7 @@ async def execute(): called = True return "executed" - result = await guard.run( - ToolScriptScanRequest(script="rm -rf /", - language="bash", - tool_name="bash"), execute) + result = await guard.run(ToolScriptScanRequest(script="rm -rf /", language="bash", tool_name="bash"), execute) assert result.blocked is True assert result.report.decision == Decision.DENY @@ -52,9 +49,7 @@ async def test_guard_allows_safe_execute(): async def execute(): return "executed" - result = await guard.run( - ToolScriptScanRequest(script="print('ok')", language="python"), - execute) + result = await guard.run(ToolScriptScanRequest(script="print('ok')", language="python"), execute) assert result.blocked is False assert result.result == "executed" @@ -70,9 +65,7 @@ async def execute(): called = True return "executed" - result = await guard.run( - ToolScriptScanRequest(script="while True:\n pass", - language="python"), execute) + result = await guard.run(ToolScriptScanRequest(script="while True:\n pass", language="python"), execute) assert result.report.decision == Decision.NEEDS_HUMAN_REVIEW assert result.blocked is False @@ -90,9 +83,7 @@ async def execute(): called = True return "executed" - result = await guard.run( - ToolScriptScanRequest(script="while True:\n pass", - language="python"), execute) + result = await guard.run(ToolScriptScanRequest(script="while True:\n pass", language="python"), execute) assert result.report.decision == Decision.NEEDS_HUMAN_REVIEW assert result.blocked is True @@ -104,16 +95,13 @@ def test_assert_allowed_raises_on_blocked_script(): guard = ToolSafetyGuard() with pytest.raises(ToolSafetyBlockedError): - guard.assert_allowed( - ToolScriptScanRequest(script="rm -rf /", language="bash")) + guard.assert_allowed(ToolScriptScanRequest(script="rm -rf /", language="bash")) def test_assert_allowed_allows_review_by_default(): guard = ToolSafetyGuard() - report = guard.assert_allowed( - ToolScriptScanRequest(script="while True:\n pass", - language="python")) + report = guard.assert_allowed(ToolScriptScanRequest(script="while True:\n pass", language="python")) assert report.decision == Decision.NEEDS_HUMAN_REVIEW assert report.blocked is False @@ -122,8 +110,7 @@ def test_assert_allowed_allows_review_by_default(): def test_assert_allowed_returns_report_for_safe_script(): guard = ToolSafetyGuard() - report = guard.assert_allowed( - ToolScriptScanRequest(script="print('ok')", language="python")) + report = guard.assert_allowed(ToolScriptScanRequest(script="print('ok')", language="python")) assert report.decision == Decision.ALLOW @@ -132,10 +119,7 @@ def test_guard_check_writes_audit_event(tmp_path): audit_path = tmp_path / "guard-audit.jsonl" guard = ToolSafetyGuard(audit_log_path=audit_path) - report = guard.check( - ToolScriptScanRequest(script="print('ok')", - language="python", - tool_name="python")) + report = guard.check(ToolScriptScanRequest(script="print('ok')", language="python", tool_name="python")) event = json.loads(audit_path.read_text(encoding="utf-8").splitlines()[0]) assert report.decision == Decision.ALLOW @@ -254,10 +238,7 @@ async def test_filter_extracts_command_as_bash(): safety_filter = ToolSafetyFilter() result = FilterResult() - await safety_filter._before(None, { - "command": "echo ok", - "tool_name": "shell_tool" - }, result) + await safety_filter._before(None, {"command": "echo ok", "tool_name": "shell_tool"}, result) assert result.is_continue is True assert result.rsp["decision"] == "allow" @@ -282,8 +263,7 @@ async def test_filter_scans_all_script_like_fields(): assert result.is_continue is False assert result.rsp["decision"] == "deny" - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" - for finding in result.rsp["findings"]) + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -304,10 +284,8 @@ async def test_filter_scans_mixed_language_fields_without_language(): assert result.is_continue is False assert result.rsp["language"] == "mixed" assert result.rsp["decision"] == "deny" - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" - for finding in result.rsp["findings"]) - assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" - for finding in result.rsp["findings"]) + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) + assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -328,8 +306,7 @@ async def test_filter_allows_safe_mixed_language_fields_in_strict_mode(): assert result.is_continue is True assert result.rsp["language"] == "mixed" assert result.rsp["decision"] == "allow" - assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" - for finding in result.rsp["findings"]) + assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -337,10 +314,7 @@ async def test_filter_extracts_python_code_language(): safety_filter = ToolSafetyFilter() result = FilterResult() - await safety_filter._before(None, { - "python_code": "print('ok')", - "tool_name": "custom" - }, result) + await safety_filter._before(None, {"python_code": "print('ok')", "tool_name": "custom"}, result) assert result.is_continue is True assert result.rsp["decision"] == "allow" @@ -353,14 +327,8 @@ async def test_filter_infers_language_from_tool_name(): python_result = FilterResult() unknown_result = FilterResult() - await safety_filter._before(None, { - "script": "print('ok')", - "tool_name": "PythonRunner" - }, python_result) - await safety_filter._before(None, { - "script": "print('ok')", - "tool_name": "custom" - }, unknown_result) + await safety_filter._before(None, {"script": "print('ok')", "tool_name": "PythonRunner"}, python_result) + await safety_filter._before(None, {"script": "print('ok')", "tool_name": "custom"}, unknown_result) assert python_result.rsp["language"] == "python" assert unknown_result.rsp["language"] == "unknown" @@ -387,8 +355,7 @@ async def test_filter_extracts_code_blocks_from_dicts_and_objects(): assert result.is_continue is False assert result.rsp["decision"] == "deny" - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" - for finding in result.rsp["findings"]) + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -415,8 +382,7 @@ async def test_filter_scans_command_args_and_context(): assert result.is_continue is False assert result.rsp["sanitized"] is True - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" - for finding in result.rsp["findings"]) + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) @pytest.mark.asyncio @@ -424,14 +390,8 @@ async def test_filter_attaches_report_to_dict_response_after_execute(): safety_filter = ToolSafetyFilter() result = FilterResult(rsp={"stdout": "ok"}) - await safety_filter._before(None, { - "command": "echo ok", - "tool_name": "shell_tool" - }, FilterResult()) - await safety_filter._after(None, { - "command": "echo ok", - "tool_name": "shell_tool" - }, result) + await safety_filter._before(None, {"command": "echo ok", "tool_name": "shell_tool"}, FilterResult()) + await safety_filter._after(None, {"command": "echo ok", "tool_name": "shell_tool"}, result) assert result.rsp["stdout"] == "ok" assert result.rsp["safety_report"]["decision"] == "allow" @@ -443,14 +403,8 @@ async def test_filter_attaches_report_to_json_object_string_after_execute(): safety_filter = ToolSafetyFilter() result = FilterResult(rsp='{"stdout": "ok"}') - await safety_filter._before(None, { - "command": "echo ok", - "tool_name": "shell_tool" - }, FilterResult()) - await safety_filter._after(None, { - "command": "echo ok", - "tool_name": "shell_tool" - }, result) + await safety_filter._before(None, {"command": "echo ok", "tool_name": "shell_tool"}, FilterResult()) + await safety_filter._after(None, {"command": "echo ok", "tool_name": "shell_tool"}, result) parsed = json.loads(result.rsp) assert parsed["stdout"] == "ok" diff --git a/trpc_agent_sdk/tools/safety/_filter.py b/trpc_agent_sdk/tools/safety/_filter.py index a4a5963bf..543a39679 100644 --- a/trpc_agent_sdk/tools/safety/_filter.py +++ b/trpc_agent_sdk/tools/safety/_filter.py @@ -68,11 +68,9 @@ async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult): requests = _extract_scan_requests(req, tool_name) if not requests: return None - report = _merge_reports( - [self.scanner.scan(request) for request in requests]) - should_block = report.decision == Decision.DENY or ( - self.block_on_review - and report.decision == Decision.NEEDS_HUMAN_REVIEW) + report = _merge_reports([self.scanner.scan(request) for request in requests]) + should_block = report.decision == Decision.DENY or (self.block_on_review + and report.decision == Decision.NEEDS_HUMAN_REVIEW) report.set_blocked(should_block) record_safety_attributes(report) if self.audit_log_path: @@ -95,8 +93,7 @@ async def _after(self, ctx: AgentContext, req: Any, rsp: FilterResult): return None -def _extract_scan_requests(req: dict[str, Any], - tool_name: str) -> list[ToolScriptScanRequest]: +def _extract_scan_requests(req: dict[str, Any], tool_name: str) -> list[ToolScriptScanRequest]: grouped_parts: dict[str, list[str]] = {} for key in _PYTHON_ARG_KEYS: @@ -119,8 +116,8 @@ def _extract_scan_requests(req: dict[str, Any], else: code = getattr(block, "code", "") language = getattr(block, "language", "") - block_language = _canonical_language(language) if isinstance( - language, str) and language.strip() else generic_language + block_language = _canonical_language(language) if isinstance(language, + str) and language.strip() else generic_language _add_script_part(grouped_parts, block_language, code) command_args = _extract_command_args(req) @@ -143,8 +140,7 @@ def _extract_scan_requests(req: dict[str, Any], return requests -def _add_script_part(grouped_parts: dict[str, list[str]], language: str, - value: Any) -> None: +def _add_script_part(grouped_parts: dict[str, list[str]], language: str, value: Any) -> None: if not isinstance(value, str) or not value.strip(): return parts = grouped_parts.setdefault(_canonical_language(language), []) @@ -157,9 +153,7 @@ def _merge_reports(reports: list[SafetyReport]) -> SafetyReport: if len(reports) == 1: return report - report.findings = [ - finding for item in reports for finding in item.findings - ] + report.findings = [finding for item in reports for finding in item.findings] report.decision = aggregate_decision(report.findings) report.risk_level = max_risk_level(report.findings) report.elapsed_ms = round(sum(item.elapsed_ms for item in reports), 3) @@ -168,22 +162,16 @@ def _merge_reports(reports: list[SafetyReport]) -> SafetyReport: report.language = languages[0] if len(languages) == 1 else "mixed" rule_ids = [finding.rule_id for finding in report.findings] if rule_ids: - report.summary = ( - f"Decision {report.decision.value} with {report.risk_level.value} risk from rules: " - f"{', '.join(rule_ids[:5])}.") + report.summary = (f"Decision {report.decision.value} with {report.risk_level.value} risk from rules: " + f"{', '.join(rule_ids[:5])}.") else: report.summary = "No safety rules matched; execution is allowed by the current static policy." report.telemetry_attributes.update({ - "tool.safety.decision": - report.decision.value, - "tool.safety.risk_level": - report.risk_level.value, - "tool.safety.rule_id": - ",".join(rule_ids[:10]), - "tool.safety.sanitized": - report.sanitized, - "tool.safety.duration_ms": - report.elapsed_ms, + "tool.safety.decision": report.decision.value, + "tool.safety.risk_level": report.risk_level.value, + "tool.safety.rule_id": ",".join(rule_ids[:10]), + "tool.safety.sanitized": report.sanitized, + "tool.safety.duration_ms": report.elapsed_ms, }) return report From a4c1d483be268e0263e4133d03deafafee1294d9 Mon Sep 17 00:00:00 2001 From: FireCollector <220344011+FireCollector@users.noreply.github.com> Date: Fri, 31 Jul 2026 20:42:30 +0800 Subject: [PATCH 5/8] refactor(safety): aggregate segmented scans in scanner --- tests/tools/safety/test_wrapper.py | 47 ++++++++++++++++++++++++ trpc_agent_sdk/tools/safety/_filter.py | 45 ++++------------------- trpc_agent_sdk/tools/safety/_scanner.py | 49 +++++++++++++++++-------- 3 files changed, 88 insertions(+), 53 deletions(-) diff --git a/tests/tools/safety/test_wrapper.py b/tests/tools/safety/test_wrapper.py index 7608db65a..7f2ca8699 100644 --- a/tests/tools/safety/test_wrapper.py +++ b/tests/tools/safety/test_wrapper.py @@ -309,6 +309,53 @@ async def test_filter_allows_safe_mixed_language_fields_in_strict_mode(): assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" for finding in result.rsp["findings"]) +@pytest.mark.asyncio +async def test_filter_scans_untyped_code_conservatively(): + safety_filter = ToolSafetyFilter() + result = FilterResult() + + await safety_filter._before( + None, + { + "code": "rm -rf /", + "tool_name": "custom", + }, + result, + ) + + assert result.is_continue is False + assert result.rsp["language"] == "unknown" + assert result.rsp["decision"] == "deny" + assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) + + +@pytest.mark.asyncio +async def test_filter_scans_execution_context_once_for_mixed_fields(): + safety_filter = ToolSafetyFilter() + result = FilterResult() + + await safety_filter._before( + None, + { + "python_code": "print('ok')", + "command": "echo ok", + "tool_metadata": { + "timeout": 301 + }, + "tool_name": "mixed_tool", + }, + result, + ) + + matching_findings = [ + finding for finding in result.rsp["findings"] if finding["rule_id"] == "RESOURCE_TIMEOUT_LIMIT_EXCEEDED" + ] + assert result.rsp["language"] == "mixed" + assert result.rsp["decision"] == "needs_human_review" + assert len(matching_findings) == 1 + assert result.rsp["telemetry_attributes"]["tool.safety.scan_id"] == result.rsp["scan_id"] + + @pytest.mark.asyncio async def test_filter_extracts_python_code_language(): safety_filter = ToolSafetyFilter() diff --git a/trpc_agent_sdk/tools/safety/_filter.py b/trpc_agent_sdk/tools/safety/_filter.py index 543a39679..cedbb92f7 100644 --- a/trpc_agent_sdk/tools/safety/_filter.py +++ b/trpc_agent_sdk/tools/safety/_filter.py @@ -24,8 +24,6 @@ from ._types import Decision from ._types import SafetyReport from ._types import ToolScriptScanRequest -from ._types import aggregate_decision -from ._types import max_risk_level _PYTHON_ARG_KEYS = ("python_code", ) _BASH_ARG_KEYS = ("command", "cmd", "bash_code") @@ -68,7 +66,7 @@ async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult): requests = _extract_scan_requests(req, tool_name) if not requests: return None - report = _merge_reports([self.scanner.scan(request) for request in requests]) + report = self.scanner.scan_segments(requests) should_block = report.decision == Decision.DENY or (self.block_on_review and report.decision == Decision.NEEDS_HUMAN_REVIEW) report.set_blocked(should_block) @@ -104,7 +102,7 @@ def _extract_scan_requests(req: dict[str, Any], tool_name: str) -> list[ToolScri generic_language = _extract_language(req, tool_name) for key in _GENERIC_ARG_KEYS: _add_script_part(grouped_parts, generic_language, req.get(key)) - code_language = _extract_explicit_language(req) or "python" + code_language = _extract_explicit_language(req) or generic_language _add_script_part(grouped_parts, code_language, req.get("code")) code_blocks = req.get("code_blocks") @@ -125,17 +123,16 @@ def _extract_scan_requests(req: dict[str, Any], tool_name: str) -> list[ToolScri env = dict(req.get("env", {}) or {}) tool_metadata = dict(req.get("tool_metadata", {}) or {}) requests: list[ToolScriptScanRequest] = [] - for index, (language, parts) in enumerate(grouped_parts.items()): - include_context = index == 0 + for language, parts in grouped_parts.items(): requests.append( ToolScriptScanRequest( script="\n".join(parts), language=language, - command_args=command_args if include_context else [], - cwd=cwd if include_context else "", - env=env if include_context else {}, + command_args=command_args, + cwd=cwd, + env=env, tool_name=tool_name, - tool_metadata=tool_metadata if include_context else {}, + tool_metadata=tool_metadata, )) return requests @@ -148,34 +145,6 @@ def _add_script_part(grouped_parts: dict[str, list[str]], language: str, value: parts.append(value) -def _merge_reports(reports: list[SafetyReport]) -> SafetyReport: - report = reports[0] - if len(reports) == 1: - return report - - report.findings = [finding for item in reports for finding in item.findings] - report.decision = aggregate_decision(report.findings) - report.risk_level = max_risk_level(report.findings) - report.elapsed_ms = round(sum(item.elapsed_ms for item in reports), 3) - report.sanitized = any(item.sanitized for item in reports) - languages = list(dict.fromkeys(item.language for item in reports)) - report.language = languages[0] if len(languages) == 1 else "mixed" - rule_ids = [finding.rule_id for finding in report.findings] - if rule_ids: - report.summary = (f"Decision {report.decision.value} with {report.risk_level.value} risk from rules: " - f"{', '.join(rule_ids[:5])}.") - else: - report.summary = "No safety rules matched; execution is allowed by the current static policy." - report.telemetry_attributes.update({ - "tool.safety.decision": report.decision.value, - "tool.safety.risk_level": report.risk_level.value, - "tool.safety.rule_id": ",".join(rule_ids[:10]), - "tool.safety.sanitized": report.sanitized, - "tool.safety.duration_ms": report.elapsed_ms, - }) - return report - - def _extract_tool_name(req: dict[str, Any]) -> str: explicit_name = req.get("tool_name") if isinstance(explicit_name, str) and explicit_name.strip(): diff --git a/trpc_agent_sdk/tools/safety/_scanner.py b/trpc_agent_sdk/tools/safety/_scanner.py index 5fde3002c..6672b7796 100644 --- a/trpc_agent_sdk/tools/safety/_scanner.py +++ b/trpc_agent_sdk/tools/safety/_scanner.py @@ -50,21 +50,38 @@ def register_rule(self, rule: SafetyRule) -> None: self.custom_rules.append(rule) def scan(self, request: ToolScriptScanRequest) -> SafetyReport: + return self.scan_segments([request]) + + def scan_segments(self, requests: Iterable[ToolScriptScanRequest]) -> SafetyReport: + """Scan language-specific script segments as one tool invocation.""" + request_list = list(requests) + if not request_list: + raise ValueError("At least one script segment is required.") + started = time.perf_counter() - language = self._normalize_language(request.language) - sanitized = self._env_contains_sensitive_keys(request.env) - _, script_sanitized = sanitize_text(request.script, limit=max(len(request.script), 1)) - sanitized = sanitized or script_sanitized + base_request = request_list[0] + languages: list[str] = [] + findings: list[RiskFinding] = [] + sanitized = False + + for request in request_list: + language = self._normalize_language(request.language) + languages.append(language) + sanitized = sanitized or self._env_contains_sensitive_keys(request.env) + _, script_sanitized = sanitize_text(request.script, limit=max(len(request.script), 1)) + sanitized = sanitized or script_sanitized + + if language == "python": + findings.extend(scan_python_script(request.script, self.policy)) + elif language in {"bash", "sh", "shell"}: + findings.extend(scan_bash_script(request.script, self.policy)) + else: + findings.extend(scan_bash_script(request.script, self.policy)) + findings.extend(scan_python_script(request.script, self.policy)) - if language == "python": - findings = scan_python_script(request.script, self.policy) - elif language in {"bash", "sh", "shell"}: - findings = scan_bash_script(request.script, self.policy) - else: - findings = scan_bash_script(request.script, self.policy) - findings.extend(scan_python_script(request.script, self.policy)) - findings.extend(self._scan_execution_context(request)) - findings.extend(self._scan_custom_rules(request)) + findings.extend(self._scan_execution_context(base_request)) + for request in request_list: + findings.extend(self._scan_custom_rules(request)) decision = aggregate_decision(findings) risk_level = max_risk_level(findings) @@ -74,6 +91,8 @@ def scan(self, request: ToolScriptScanRequest) -> SafetyReport: summary = self._build_summary(decision.value, risk_level.value, rule_ids) scan_id = str(uuid.uuid4()) timestamp = datetime.now(timezone.utc).isoformat() + normalized_languages = list(dict.fromkeys(languages)) + language = normalized_languages[0] if len(normalized_languages) == 1 else "mixed" telemetry_attributes = { "tool.safety.scan_id": scan_id, "tool.safety.decision": decision.value, @@ -81,7 +100,7 @@ def scan(self, request: ToolScriptScanRequest) -> SafetyReport: "tool.safety.rule_id": ",".join(rule_ids[:10]), "tool.safety.blocked": blocked, "tool.safety.sanitized": sanitized, - "tool.safety.tool_name": request.tool_name, + "tool.safety.tool_name": base_request.tool_name, "tool.safety.duration_ms": elapsed_ms, } return SafetyReport( @@ -90,7 +109,7 @@ def scan(self, request: ToolScriptScanRequest) -> SafetyReport: decision=decision, risk_level=risk_level, findings=findings, - tool_name=request.tool_name, + tool_name=base_request.tool_name, language=language, elapsed_ms=elapsed_ms, sanitized=sanitized, From 9c3bfdbe03d282adeb46aa691eb3866623356e23 Mon Sep 17 00:00:00 2001 From: FireCollector <220344011+FireCollector@users.noreply.github.com> Date: Fri, 31 Jul 2026 20:50:02 +0800 Subject: [PATCH 6/8] fix(safety): deduplicate segmented scan findings --- tests/tools/safety/test_wrapper.py | 20 ++++++++++++++++++++ trpc_agent_sdk/tools/safety/_scanner.py | 2 ++ 2 files changed, 22 insertions(+) diff --git a/tests/tools/safety/test_wrapper.py b/tests/tools/safety/test_wrapper.py index 7f2ca8699..6761aaf9e 100644 --- a/tests/tools/safety/test_wrapper.py +++ b/tests/tools/safety/test_wrapper.py @@ -356,6 +356,26 @@ async def test_filter_scans_execution_context_once_for_mixed_fields(): assert result.rsp["telemetry_attributes"]["tool.safety.scan_id"] == result.rsp["scan_id"] +@pytest.mark.asyncio +async def test_filter_deduplicates_findings_across_segments(): + safety_filter = ToolSafetyFilter() + result = FilterResult() + + await safety_filter._before( + None, + { + "command": "rm -rf /", + "script": "rm -rf /", + "tool_name": "custom", + }, + result, + ) + + matching_findings = [finding for finding in result.rsp["findings"] if finding["rule_id"] == "BASH_RECURSIVE_DELETE"] + assert result.rsp["decision"] == "deny" + assert len(matching_findings) == 1 + + @pytest.mark.asyncio async def test_filter_extracts_python_code_language(): safety_filter = ToolSafetyFilter() diff --git a/trpc_agent_sdk/tools/safety/_scanner.py b/trpc_agent_sdk/tools/safety/_scanner.py index 6672b7796..a68701516 100644 --- a/trpc_agent_sdk/tools/safety/_scanner.py +++ b/trpc_agent_sdk/tools/safety/_scanner.py @@ -17,6 +17,7 @@ from pathlib import Path from ._policy import ToolSafetyPolicy +from ._rules import _dedupe_findings from ._rules import _finding from ._rules import SENSITIVE_NAME_RE from ._rules import scan_bash_script @@ -82,6 +83,7 @@ def scan_segments(self, requests: Iterable[ToolScriptScanRequest]) -> SafetyRepo findings.extend(self._scan_execution_context(base_request)) for request in request_list: findings.extend(self._scan_custom_rules(request)) + findings = _dedupe_findings(findings) decision = aggregate_decision(findings) risk_level = max_risk_level(findings) From fb060f369c3b6a5dc05e3e0a062e6073a2e7b674 Mon Sep 17 00:00:00 2001 From: FireCollector <220344011+FireCollector@users.noreply.github.com> Date: Fri, 31 Jul 2026 20:59:18 +0800 Subject: [PATCH 7/8] fix(safety): preserve default code language --- tests/tools/safety/test_wrapper.py | 20 -------------------- trpc_agent_sdk/tools/safety/_filter.py | 2 +- 2 files changed, 1 insertion(+), 21 deletions(-) diff --git a/tests/tools/safety/test_wrapper.py b/tests/tools/safety/test_wrapper.py index 6761aaf9e..44be11fed 100644 --- a/tests/tools/safety/test_wrapper.py +++ b/tests/tools/safety/test_wrapper.py @@ -309,26 +309,6 @@ async def test_filter_allows_safe_mixed_language_fields_in_strict_mode(): assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" for finding in result.rsp["findings"]) -@pytest.mark.asyncio -async def test_filter_scans_untyped_code_conservatively(): - safety_filter = ToolSafetyFilter() - result = FilterResult() - - await safety_filter._before( - None, - { - "code": "rm -rf /", - "tool_name": "custom", - }, - result, - ) - - assert result.is_continue is False - assert result.rsp["language"] == "unknown" - assert result.rsp["decision"] == "deny" - assert any(finding["rule_id"] == "BASH_RECURSIVE_DELETE" for finding in result.rsp["findings"]) - - @pytest.mark.asyncio async def test_filter_scans_execution_context_once_for_mixed_fields(): safety_filter = ToolSafetyFilter() diff --git a/trpc_agent_sdk/tools/safety/_filter.py b/trpc_agent_sdk/tools/safety/_filter.py index cedbb92f7..4c2f39e38 100644 --- a/trpc_agent_sdk/tools/safety/_filter.py +++ b/trpc_agent_sdk/tools/safety/_filter.py @@ -102,7 +102,7 @@ def _extract_scan_requests(req: dict[str, Any], tool_name: str) -> list[ToolScri generic_language = _extract_language(req, tool_name) for key in _GENERIC_ARG_KEYS: _add_script_part(grouped_parts, generic_language, req.get(key)) - code_language = _extract_explicit_language(req) or generic_language + code_language = _extract_explicit_language(req) or "python" _add_script_part(grouped_parts, code_language, req.get("code")) code_blocks = req.get("code_blocks") From c46068897ffb6282db59f6711db7095780c2a655 Mon Sep 17 00:00:00 2001 From: FireCollector <220344011+FireCollector@users.noreply.github.com> Date: Sun, 2 Aug 2026 00:08:54 +0800 Subject: [PATCH 8/8] safety: suppress Python parse errors for unknown segments Unknown language segments may contain valid non-Python input. Continue Bash scanning, but merge Python findings only when AST parsing succeeds so strict review mode does not block safe Bash scripts. Updates #90 RELEASE NOTES: NONE --- tests/tools/safety/test_scanner.py | 9 ++++++++- tests/tools/safety/test_wrapper.py | 20 ++++++++++++++++++++ trpc_agent_sdk/tools/safety/_scanner.py | 6 +++++- 3 files changed, 33 insertions(+), 2 deletions(-) diff --git a/tests/tools/safety/test_scanner.py b/tests/tools/safety/test_scanner.py index d7e1af97d..67dfc7add 100644 --- a/tests/tools/safety/test_scanner.py +++ b/tests/tools/safety/test_scanner.py @@ -250,7 +250,14 @@ def test_unknown_language_scans_python_and_bash_rules(): assert report.language == "unknown" assert report.decision == Decision.DENY assert "BASH_RECURSIVE_DELETE" in _rule_ids(report) - assert "PY_PARSE_ERROR_REVIEW" in _rule_ids(report) + assert "PY_PARSE_ERROR_REVIEW" not in _rule_ids(report) + + +def test_unknown_language_keeps_python_findings_when_parse_succeeds(): + report = _scanner().scan_script("while True:\n pass\n", "unknown") + + assert "PY_INFINITE_LOOP" in _rule_ids(report) + assert "PY_PARSE_ERROR_REVIEW" not in _rule_ids(report) def test_language_aliases_are_normalized(): diff --git a/tests/tools/safety/test_wrapper.py b/tests/tools/safety/test_wrapper.py index 44be11fed..1bd17f206 100644 --- a/tests/tools/safety/test_wrapper.py +++ b/tests/tools/safety/test_wrapper.py @@ -354,6 +354,26 @@ async def test_filter_deduplicates_findings_across_segments(): matching_findings = [finding for finding in result.rsp["findings"] if finding["rule_id"] == "BASH_RECURSIVE_DELETE"] assert result.rsp["decision"] == "deny" assert len(matching_findings) == 1 + assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" for finding in result.rsp["findings"]) + + +@pytest.mark.asyncio +async def test_filter_allows_safe_unknown_bash_script_in_strict_mode(): + safety_filter = ToolSafetyFilter(block_on_review=True) + result = FilterResult() + + await safety_filter._before( + None, + { + "script": "git status", + "tool_name": "custom", + }, + result, + ) + + assert result.is_continue is True + assert result.rsp["decision"] == "allow" + assert all(finding["rule_id"] != "PY_PARSE_ERROR_REVIEW" for finding in result.rsp["findings"]) @pytest.mark.asyncio diff --git a/trpc_agent_sdk/tools/safety/_scanner.py b/trpc_agent_sdk/tools/safety/_scanner.py index a68701516..2fa7006fe 100644 --- a/trpc_agent_sdk/tools/safety/_scanner.py +++ b/trpc_agent_sdk/tools/safety/_scanner.py @@ -78,7 +78,11 @@ def scan_segments(self, requests: Iterable[ToolScriptScanRequest]) -> SafetyRepo findings.extend(scan_bash_script(request.script, self.policy)) else: findings.extend(scan_bash_script(request.script, self.policy)) - findings.extend(scan_python_script(request.script, self.policy)) + python_findings = scan_python_script(request.script, self.policy) + # An unknown segment may be valid non-Python input; keep Python + # findings only when AST parsing succeeds. + if not any(finding.rule_id == "PY_PARSE_ERROR_REVIEW" for finding in python_findings): + findings.extend(python_findings) findings.extend(self._scan_execution_context(base_request)) for request in request_list: