test: harden tool safety guard acceptance coverage

Author: 张志鹏

scripts/tool_safety_check.py

@@ -7,6 +7,10 @@
 import sys
 from pathlib import Path
 
+ROOT_DIR = Path(__file__).resolve().parents[1]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
 from trpc_agent_sdk.tools.safety import ToolSafetyGuard
 from trpc_agent_sdk.tools.safety import ToolSafetyPolicy
 from trpc_agent_sdk.tools.safety import ToolSafetyScanRequest

tests/tools/safety/test_tool_safety_guard.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import json
 import time
 from pathlib import Path
 from unittest.mock import AsyncMock
@@ -27,6 +28,7 @@
 from trpc_agent_sdk.code_executors import CodeExecutionInput
 from trpc_agent_sdk.code_executors import BaseCodeExecutor
 from trpc_agent_sdk.code_executors import create_code_execution_result
+from scripts.tool_safety_check import main as tool_safety_check_main
 
 
 EXAMPLE_DIR = Path(__file__).resolve().parents[3] / "examples" / "tool_safety_guard"
@@ -86,6 +88,82 @@ def test_issue_required_samples(sample, decision, rule_id):
         assert first["risk_level"]
 
 
+def test_public_samples_emit_structured_reports_and_meet_acceptance_rates():
+    high_risk_samples = {
+        "02_dangerous_delete.sh",
+        "03_read_secret.py",
+        "04_network_external.py",
+        "07_shell_injection.py",
+        "08_dependency_install.sh",
+        "09_infinite_loop.py",
+        "10_sensitive_output.py",
+    }
+    safe_samples = {"01_safe_python.py", "05_network_allowlist.py"}
+    required_rule_by_sample = {
+        "02_dangerous_delete.sh": "TSG-FILE-RECURSIVE-DELETE",
+        "03_read_secret.py": "TSG-FILE-DENIED-PATH",
+        "04_network_external.py": "TSG-NETWORK-NONALLOWLIST",
+    }
+
+    sample_paths = sorted(SAMPLES.iterdir())
+    reports = {path.name: scan_sample(path.name) for path in sample_paths}
+
+    assert len(reports) == 12
+    for report in reports.values():
+        payload = json.loads(report.to_json())
+        assert payload["decision"]
+        assert payload["risk_level"]
+        if payload["decision"] != SafetyDecision.ALLOW.value:
+            for finding in payload["findings"]:
+                assert finding["rule_id"]
+                assert finding["evidence"]
+                assert finding["recommendation"]
+
+    detected_high_risk = sum(1 for name in high_risk_samples if reports[name].decision == SafetyDecision.DENY)
+    false_positive_safe = sum(1 for name in safe_samples if reports[name].decision != SafetyDecision.ALLOW)
+    assert detected_high_risk / len(high_risk_samples) >= 0.9
+    assert false_positive_safe / len(safe_samples) <= 0.1
+
+    for sample, rule_id in required_rule_by_sample.items():
+        report = reports[sample]
+        assert report.decision == SafetyDecision.DENY
+        assert any(finding.rule_id == rule_id for finding in report.findings)
+
+
+def test_cli_scans_public_samples_and_writes_structured_reports(tmp_path, monkeypatch):
+    for sample in sorted(SAMPLES.iterdir()):
+        report_path = tmp_path / f"{sample.name}.json"
+        audit_path = tmp_path / "audit.jsonl"
+        language = "bash" if sample.suffix == ".sh" else "python"
+        monkeypatch.setattr(
+            "sys.argv",
+            [
+                "tool_safety_check.py",
+                str(sample),
+                "--language",
+                language,
+                "--policy",
+                str(EXAMPLE_DIR / "tool_safety_policy.yaml"),
+                "--report-out",
+                str(report_path),
+                "--audit-out",
+                str(audit_path),
+            ],
+        )
+
+        exit_code = tool_safety_check_main()
+
+        payload = json.loads(report_path.read_text(encoding="utf-8"))
+        assert exit_code in {0, 2}
+        assert payload["decision"]
+        assert payload["risk_level"]
+        if payload["decision"] != SafetyDecision.ALLOW.value:
+            assert payload["findings"]
+            assert payload["findings"][0]["rule_id"]
+            assert payload["findings"][0]["evidence"]
+            assert payload["findings"][0]["recommendation"]
+
+
 def test_denied_path_policy_change_changes_result():
     script = 'open(".env", "r").read()'
     strict = ToolSafetyScanner(policy(denied_paths=[".env"]))
@@ -115,6 +193,49 @@ def test_allowed_command_policy_change_changes_result():
     assert relaxed_report.decision == SafetyDecision.ALLOW
 
 
+@pytest.mark.parametrize(
+    ("script", "language", "rule_id", "decision"),
+    [
+        ("echo data > /etc/app.conf", "bash", "TSG-FILE-SYSTEM-WRITE", SafetyDecision.DENY),
+        ("python worker.py &", "bash", "TSG-PROCESS-BACKGROUND", SafetyDecision.NEEDS_HUMAN_REVIEW),
+        ("printf '%s\n' a b | xargs -P 0 -n1 echo", "bash", "TSG-RESOURCE-PARALLELISM", SafetyDecision.DENY),
+        ("timeout 999 python job.py", "bash", "TSG-RESOURCE-LONG-TIMEOUT", SafetyDecision.NEEDS_HUMAN_REVIEW),
+        (
+            "import socket\ns = socket.socket()\ns.connect(('evil.example.net', 443))",
+            "python",
+            "TSG-NETWORK-NONALLOWLIST",
+            SafetyDecision.DENY,
+        ),
+        (
+            "import subprocess\nsubprocess.run(['python', 'job.py'], timeout=999, capture_output=True)",
+            "python",
+            "TSG-RESOURCE-LONG-TIMEOUT",
+            SafetyDecision.NEEDS_HUMAN_REVIEW,
+        ),
+        (
+            "import concurrent.futures\nconcurrent.futures.ThreadPoolExecutor(max_workers=999)",
+            "python",
+            "TSG-RESOURCE-PARALLELISM",
+            SafetyDecision.NEEDS_HUMAN_REVIEW,
+        ),
+    ],
+)
+def test_additional_required_risk_coverage(script, language, rule_id, decision):
+    report = ToolSafetyScanner(policy()).scan(ToolSafetyScanRequest(script=script, language=language))
+
+    assert report.decision == decision
+    assert any(finding.rule_id == rule_id for finding in report.findings)
+
+
+def test_subprocess_output_capture_has_output_size_recommendation():
+    script = "import subprocess\nsubprocess.run(['python', 'job.py'], stdout=subprocess.PIPE)"
+    report = ToolSafetyScanner(policy()).scan(ToolSafetyScanRequest(script=script, language="python"))
+
+    assert report.decision == SafetyDecision.NEEDS_HUMAN_REVIEW
+    finding = next(item for item in report.findings if item.rule_id == "TSG-RESOURCE-OUTPUT-CAPTURE")
+    assert "max_output_bytes" in finding.recommendation
+
+
 def test_audit_event_contains_required_fields(tmp_path):
     report = ToolSafetyScanner(policy()).scan(
         ToolSafetyScanRequest(script="rm -rf /tmp/demo", language="bash", tool_metadata={"name": "Bash"}))
@@ -149,6 +270,22 @@ async def test_tool_filter_blocks_before_handle():
     handle.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_tool_filter_block_writes_audit_event(tmp_path):
+    audit_path = tmp_path / "tool_safety_audit.jsonl"
+    filter_ = ToolSafetyFilter(guard=ToolSafetyGuard(policy=policy(), audit_log_path=audit_path))
+    handle = AsyncMock(return_value={"success": True})
+
+    result = await filter_.run(create_agent_context(), {"command": "rm -rf /tmp/demo"}, handle)
+
+    assert result.is_continue is False
+    handle.assert_not_called()
+    event = json.loads(audit_path.read_text(encoding="utf-8").strip())
+    assert event["decision"] == "deny"
+    assert event["rule_id"] == "TSG-FILE-RECURSIVE-DELETE"
+    assert event["blocked"] is True
+
+
 @pytest.mark.asyncio
 async def test_tool_filter_allows_safe_command():
     filter_ = ToolSafetyFilter(guard=ToolSafetyGuard(policy=policy()))