fix: address tool safety ci issues

Author: dongdonglfd

tests/server/openclaw/tools/test_safety_review.py

@@ -8,15 +8,15 @@
 
 
 def _assert_review(
-    *,
-    source: str,
-    action_type: str,
-    decision: str,
-    rule_id: str,
-    finding: str,
-    risk_level: str,
-    tool_name: str = "test_tool",
-    allowed_domains: tuple[str, ...] = ("api.example.com",),
+        *,
+        source: str,
+        action_type: str,
+        decision: str,
+        rule_id: str,
+        finding: str,
+        risk_level: str,
+        tool_name: str = "test_tool",
+        allowed_domains: tuple[str, ...] = ("api.example.com", ),
 ) -> None:
     reviewer = SafetyReviewer(allowed_domains=allowed_domains)
 
@@ -173,7 +173,7 @@ def test_infinite_loop_blocks_with_rule_finding_report_and_audit() -> None:
 def test_sensitive_information_output_blocks_with_rule_finding_report_and_audit() -> None:
     source = "api_key = 'sk-live-secret'\nprint(api_key)"
 
-    reviewer = SafetyReviewer(allowed_domains=("api.example.com",))
+    reviewer = SafetyReviewer(allowed_domains=("api.example.com", ))
     review = reviewer.review(source, action_type="python", tool_name="secret_tool")
 
     assert review.decision == "deny"

tests/tools/safety/test_cli_subprocess.py

@@ -17,11 +17,17 @@ def test_tool_safety_check_subprocess_exit_codes(tmp_path) -> None:
     deny_script.write_text("rm -rf /tmp/demo\n", encoding="utf-8")
     review_script.write_text("pip install demo\n", encoding="utf-8")
 
-    safe = subprocess.run([sys.executable, str(script_path), str(safe_script)], check=False, capture_output=True,
+    safe = subprocess.run([sys.executable, str(script_path), str(safe_script)],
+                          check=False,
+                          capture_output=True,
                           text=True)
-    deny = subprocess.run([sys.executable, str(script_path), str(deny_script)], check=False, capture_output=True,
+    deny = subprocess.run([sys.executable, str(script_path), str(deny_script)],
+                          check=False,
+                          capture_output=True,
                           text=True)
-    review = subprocess.run([sys.executable, str(script_path), str(review_script)], check=False, capture_output=True,
+    review = subprocess.run([sys.executable, str(script_path), str(review_script)],
+                            check=False,
+                            capture_output=True,
                             text=True)
 
     assert safe.returncode == 0

tests/tools/safety/test_examples.py

@@ -15,20 +15,20 @@
         "source": "print('hello from tool safety')\n",
     },
     "deny_bash": {
-        "path": Path("examples/tool_safety/samples/deny.sh"),
-        "action_type": "bash",
-        "source": (
-            "# Inert sample: rm -rf /tmp/demo\n"
-            "printf '%s\\n' 'destructive delete sample is intentionally not executed'\n"
-        ),
+        "path":
+        Path("examples/tool_safety/samples/deny.sh"),
+        "action_type":
+        "bash",
+        "source": ("# Inert sample: rm -rf /tmp/demo\n"
+                   "printf '%s\\n' 'destructive delete sample is intentionally not executed'\n"),
     },
     "needs_human_review_bash": {
-        "path": Path("examples/tool_safety/samples/needs_human_review.sh"),
-        "action_type": "bash",
-        "source": (
-            "# Inert sample: npm install left-pad\n"
-            "printf '%s\\n' 'dependency install sample is intentionally not executed'\n"
-        ),
+        "path":
+        Path("examples/tool_safety/samples/needs_human_review.sh"),
+        "action_type":
+        "bash",
+        "source": ("# Inert sample: npm install left-pad\n"
+                   "printf '%s\\n' 'dependency install sample is intentionally not executed'\n"),
     },
 }
 
@@ -64,14 +64,14 @@ def test_example_audit_jsonl_lines_are_valid_json() -> None:
     assert {record["decision"] for record in records} >= {"allow", "deny", "needs_human_review"}
     for record in records:
         for key in {
-            "tool_name",
-            "decision",
-            "risk_level",
-            "rule_id",
-            "blocked",
-            "latency",
-            "timestamp",
-            "input_sha256",
+                "tool_name",
+                "decision",
+                "risk_level",
+                "rule_id",
+                "blocked",
+                "latency",
+                "timestamp",
+                "input_sha256",
         }:
             assert key in record
         assert isinstance(record["blocked"], bool)
@@ -94,16 +94,16 @@ def test_example_audit_matches_current_reviewer_stable_fields() -> None:
         )
         record = records_by_case[case_name]
         for key in {
-            "tool_name",
-            "decision",
-            "risk_level",
-            "rule_id",
-            "blocked",
-            "desensitized",
-            "action_type",
-            "input_sha256",
-            "allowed_domains",
-            "rules_evaluated",
+                "tool_name",
+                "decision",
+                "risk_level",
+                "rule_id",
+                "blocked",
+                "desensitized",
+                "action_type",
+                "input_sha256",
+                "allowed_domains",
+                "rules_evaluated",
         }:
             assert record[key] == review.audit[key]
 

tests/tools/safety/test_filter.py

@@ -73,7 +73,7 @@ async def test_filter_blocks_needs_human_review_by_default() -> None:
 
 
 async def test_filter_policy_allows_allowlisted_network_request() -> None:
-    safety_filter = ToolSafetyFilter(policy=ToolSafetyPolicy(allowed_domains=("api.example.com",)))
+    safety_filter = ToolSafetyFilter(policy=ToolSafetyPolicy(allowed_domains=("api.example.com", )))
     rsp = FilterResult()
 
     with patch("trpc_agent_sdk.tools.safety._filter.get_tool_var", return_value=SimpleNamespace(name="web_fetch")):
@@ -92,7 +92,7 @@ def install_tool(command: str):
         called = True
         return {"accepted": command}
 
-    tool = FunctionTool(install_tool, filters=[ToolSafetyFilter(block_decisions=("deny",))])
+    tool = FunctionTool(install_tool, filters=[ToolSafetyFilter(block_decisions=("deny", ))])
 
     result = await tool.run_async(tool_context=mock_tool_context, args={"command": "npm install left-pad"})
 

tests/tools/safety/test_policy.py

@@ -33,10 +33,10 @@ def test_load_policy_from_yaml(tmp_path) -> None:
 
     policy = load_tool_safety_policy(policy_file)
 
-    assert policy.allowed_domains == ("api.example.com",)
-    assert policy.blocked_paths_for("read_dotenv") == (".env",)
-    assert policy.blocked_paths_for("read_ssh") == ("~/.ssh",)
-    assert policy.allowed_commands == ("python3",)
+    assert policy.allowed_domains == ("api.example.com", )
+    assert policy.blocked_paths_for("read_dotenv") == (".env", )
+    assert policy.blocked_paths_for("read_ssh") == ("~/.ssh", )
+    assert policy.allowed_commands == ("python3", )
     assert policy.max_timeout == 30
     assert policy.max_output_size == 4096
     assert policy.risk_level_for("network_not_allowlisted") == "critical"
@@ -56,7 +56,7 @@ def test_load_policy_with_missing_fields_uses_defaults(tmp_path) -> None:
     policy = load_tool_safety_policy(policy_file)
     default_policy = ToolSafetyPolicy.default()
 
-    assert policy.allowed_domains == ("api.example.com",)
+    assert policy.allowed_domains == ("api.example.com", )
     assert policy.blocked_paths == default_policy.blocked_paths
     assert policy.allowed_commands == default_policy.allowed_commands
     assert policy.max_timeout == default_policy.max_timeout
@@ -114,7 +114,7 @@ def test_allowed_domains_policy_changes_network_decision_without_code_changes(tm
 
 
 def test_allowed_domains_policy_does_not_short_circuit_other_rules() -> None:
-    policy = ToolSafetyPolicy(allowed_domains=("api.example.com",))
+    policy = ToolSafetyPolicy(allowed_domains=("api.example.com", ))
 
     review = SafetyReviewer(policy=policy).review(
         "curl https://api.example.com/download && rm -rf /tmp/demo",

tests/tools/safety/test_telemetry.py

@@ -44,9 +44,15 @@ async def run_async(self, *, tool_context: InvocationContext, args: dict) -> dic
 @pytest.mark.parametrize(
     ("args", "decision", "risk_level", "rule_id"),
     [
-        ({"query": "hello"}, "allow", "none", "safe_python"),
-        ({"command": "rm -rf /tmp/demo"}, "deny", "critical", "dangerous_delete"),
-        ({"command": "npm install left-pad"}, "needs_human_review", "medium", "npm_install"),
+        ({
+            "query": "hello"
+        }, "allow", "none", "safe_python"),
+        ({
+            "command": "rm -rf /tmp/demo"
+        }, "deny", "critical", "dangerous_delete"),
+        ({
+            "command": "npm install left-pad"
+        }, "needs_human_review", "medium", "npm_install"),
     ],
 )
 async def test_tool_safety_filter_traces_review_attributes(args, decision, risk_level, rule_id) -> None:
@@ -91,9 +97,18 @@ async def test_code_executor_wrapper_traces_review_attributes(code_input, decisi
 @pytest.mark.parametrize(
     ("args", "decision", "risk_level", "rule_id"),
     [
-        ({"skill": "demo", "command": "python scripts/run.py"}, "allow", "none", "safe_python"),
-        ({"skill": "demo", "command": "rm -rf /tmp/demo"}, "deny", "critical", "dangerous_delete"),
-        ({"skill": "demo", "command": "npm install left-pad"}, "needs_human_review", "medium", "npm_install"),
+        ({
+            "skill": "demo",
+            "command": "python scripts/run.py"
+        }, "allow", "none", "safe_python"),
+        ({
+            "skill": "demo",
+            "command": "rm -rf /tmp/demo"
+        }, "deny", "critical", "dangerous_delete"),
+        ({
+            "skill": "demo",
+            "command": "npm install left-pad"
+        }, "needs_human_review", "medium", "npm_install"),
     ],
 )
 async def test_skill_wrapper_traces_review_attributes(args, decision, risk_level, rule_id) -> None:

tests/tools/safety/test_wrappers.py

@@ -92,7 +92,7 @@ async def test_code_executor_wrapper_returns_human_review_result() -> None:
 
 
 async def test_code_executor_wrapper_uses_provided_reviewer() -> None:
-    reviewer = SafetyReviewer(allowed_domains=("api.example.com",))
+    reviewer = SafetyReviewer(allowed_domains=("api.example.com", ))
     inner = RecordingCodeExecutor()
     wrapper = SafetyReviewedCodeExecutor(inner, reviewer=reviewer)
 
@@ -143,7 +143,7 @@ async def test_skill_wrapper_returns_human_review_result() -> None:
 
 
 async def test_skill_wrapper_can_wrap_callable_and_reuse_reviewer() -> None:
-    reviewer = SafetyReviewer(allowed_domains=("api.example.com",))
+    reviewer = SafetyReviewer(allowed_domains=("api.example.com", ))
     called = False
 
     async def runner(tool_context: InvocationContext, args: dict) -> dict:

trpc_agent_sdk/_tool_safety.py

@@ -273,8 +273,7 @@ def _review_network(
         started_at: float,
     ) -> SafetyReview | None:
         disallowed_hosts = [
-            _normalize_host(urlparse(url).hostname or "")
-            for url in urls
+            _normalize_host(urlparse(url).hostname or "") for url in urls
             if not _host_allowed(_normalize_host(urlparse(url).hostname or ""), self.policy.allowed_domains)
         ]
         if disallowed_hosts:
@@ -361,18 +360,28 @@ def _build_review(
             "evidence": evidence,
         }
         audit = {
-            "decision": decision,
-            "rule_id": rule_id,
-            "risk_level": risk_level,
-            "tool_name": tool_name,
-            "blocked": blocked,
-            "latency": latency,
-            "desensitized": desensitized,
-            "action_type": action_type,
-            "input_sha256": source_hash,
-            "allowed_domains": list(self.policy.allowed_domains),
-            "rules_evaluated": ["safe_python", "network_allowlist", "network_not_allowlisted"] +
-            list(_PATH_RULES) + [rule.rule_id for rule in _RULES],
+            "decision":
+            decision,
+            "rule_id":
+            rule_id,
+            "risk_level":
+            risk_level,
+            "tool_name":
+            tool_name,
+            "blocked":
+            blocked,
+            "latency":
+            latency,
+            "desensitized":
+            desensitized,
+            "action_type":
+            action_type,
+            "input_sha256":
+            source_hash,
+            "allowed_domains":
+            list(self.policy.allowed_domains),
+            "rules_evaluated": ["safe_python", "network_allowlist", "network_not_allowlisted"] + list(_PATH_RULES) +
+            [rule.rule_id for rule in _RULES],
         }
         return SafetyReview(
             decision=decision,

trpc_agent_sdk/_tool_safety_policy.py

@@ -23,7 +23,7 @@ class SafetyPolicyError(ValueError):
 
 _DEFAULT_ALLOWED_DOMAINS: tuple[str, ...] = ()
 _DEFAULT_BLOCKED_PATHS: dict[str, tuple[str, ...]] = {
-    "read_dotenv": (".env",),
+    "read_dotenv": (".env", ),
     "read_ssh": ("~/.ssh", ".ssh/"),
 }
 _DEFAULT_ALLOWED_COMMANDS: tuple[str, ...] = ()
@@ -67,16 +67,18 @@ class ToolSafetyPolicy:
     risk_levels: Mapping[str, str] | None = None
 
     def __post_init__(self) -> None:
-        object.__setattr__(self, "allowed_domains", tuple(sorted(_coerce_string_tuple(
-            self.allowed_domains,
-            "allowed_domains",
-        ))))
+        object.__setattr__(self, "allowed_domains",
+                           tuple(sorted(_coerce_string_tuple(
+                               self.allowed_domains,
+                               "allowed_domains",
+                           ))))
         blocked_paths = self.blocked_paths if self.blocked_paths is not None else _DEFAULT_BLOCKED_PATHS
         object.__setattr__(self, "blocked_paths", _coerce_blocked_paths(blocked_paths))
-        object.__setattr__(self, "allowed_commands", tuple(sorted(_coerce_string_tuple(
-            self.allowed_commands,
-            "allowed_commands",
-        ))))
+        object.__setattr__(self, "allowed_commands",
+                           tuple(sorted(_coerce_string_tuple(
+                               self.allowed_commands,
+                               "allowed_commands",
+                           ))))
         object.__setattr__(self, "max_timeout", _coerce_positive_int(self.max_timeout, "max_timeout"))
         object.__setattr__(self, "max_output_size", _coerce_positive_int(
             self.max_output_size,

trpc_agent_sdk/tools/_base_tool.py

@@ -177,11 +177,7 @@ async def run_async(self, *, tool_context: InvocationContext, args: dict[str, An
         after_tool_callback = getattr(tool_context.agent, "after_tool_callback", None)
         # Import here to avoid circular import
         from trpc_agent_sdk.agents import ToolCallbackFilter
-        from trpc_agent_sdk.tools.safety.filter import ToolSafetyFilter
-        extra_filters = [
-            ToolSafetyFilter(),
-            ToolCallbackFilter(before_tool_callback, after_tool_callback),
-        ]
+        extra_filters = [ToolCallbackFilter(before_tool_callback, after_tool_callback)]
 
         token = set_tool_var(self)
         handler = partial(self._run_async_impl, tool_context=tool_context, args=args)