fix: address CI lint and review feedback on safety module

- Remove unused typing.Any import in _wrapper.py (F401)

- _rules.py: parenthesize or/and precedence in decode-exec check

- _rules.py: simplify isinstance(...) if ... else False patterns

- _rules.py: extract is_process_call variable for readability

- _rules.py: use getattr chain instead of hasattr for ast.Subscript.slice

- _filter.py: drop redundant isinstance(rsp.rsp, dict) after the if not isinstance

- _policy.py: validate placeholder byte caps regardless of strict mode

- _wrapper.py: clarify SafeCodeExecutor alias rationale

- Wrap several >120-char lines to satisfy flake8 --max-line-length=120

flake8: clean. pytest tests/tool_safety/: 104 passed, 3 skipped

Author: Violet2314

trpc_agent_sdk/safety/_filter.py

@@ -76,8 +76,10 @@ async def _before(self, ctx: Any, req: Any, rsp: FilterResult) -> None:
         )
         report = self.scanner.scan(scan_input)
 
-        should_block = report.decision == Decision.DENY or (report.decision == Decision.NEEDS_HUMAN_REVIEW
-                                                            and self._block_on_review)
+        should_block = (
+            report.decision == Decision.DENY
+            or (report.decision == Decision.NEEDS_HUMAN_REVIEW and self._block_on_review)
+        )
         self.audit.log(report, intercepted=should_block)
 
         if should_block:
@@ -110,10 +112,9 @@ async def _before(self, ctx: Any, req: Any, rsp: FilterResult) -> None:
             # remain in audit / OTel.
             if not isinstance(rsp.rsp, dict):
                 rsp.rsp = {}
-            if isinstance(rsp.rsp, dict):
-                rsp.rsp["safety_warning"] = "TOOL_SAFETY_NEEDS_REVIEW"
-                rsp.rsp["safety_risk_level"] = report.risk_level.value
-                rsp.rsp["safety_rule_ids"] = list(report.rule_ids)
+            rsp.rsp["safety_warning"] = "TOOL_SAFETY_NEEDS_REVIEW"
+            rsp.rsp["safety_risk_level"] = report.risk_level.value
+            rsp.rsp["safety_rule_ids"] = list(report.rule_ids)
 
     def _resolve_tool_name(self, ctx: Any) -> str:
         get_tool_var = self._get_tool_var

trpc_agent_sdk/safety/_policy.py

@@ -74,9 +74,12 @@ def from_dict(cls, data: dict[str, Any]) -> "PolicyConfig":
             unknown = set(data.keys()) - cls._KNOWN_KEYS
             if unknown:
                 raise ValueError(f"unknown policy keys: {sorted(unknown)}")
-            for key in ("max_timeout_seconds", "max_output_bytes", "max_file_write_bytes"):
-                if key in data and data[key] is not None and int(data[key]) < 0:
-                    raise ValueError(f"{key} must be non-negative")
+        # max_output_bytes / max_file_write_bytes are runtime placeholders
+        # (static scanning cannot enforce byte caps). Validate as non-negative
+        # ints regardless of strict mode so malformed YAML fails fast.
+        for key in ("max_timeout_seconds", "max_output_bytes", "max_file_write_bytes"):
+            if key in data and data[key] is not None and int(data[key]) < 0:
+                raise ValueError(f"{key} must be non-negative")
 
         deny_lvl = _parse_risk_level(data.get("deny_risk_level"), RiskLevel.HIGH)
         review_lvl = _parse_risk_level(data.get("review_risk_level"), RiskLevel.MEDIUM)

trpc_agent_sdk/safety/_rules.py

@@ -188,7 +188,7 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
 
             if lname in {"open", "builtins.open"} or lname.endswith(".open"):
                 target = _first_str_or_path(node) or "<dynamic>"
-                if isinstance(node.args[0], ast.Name) if node.args else False:
+                if node.args and isinstance(node.args[0], ast.Name):
                     var = node.args[0].id
                     if var in path_vars:
                         target = path_vars[var]
@@ -366,8 +366,8 @@ def _collect_sensitive_path_vars(
         text = ""
         if isinstance(value, ast.Call):
             cname = resolve_name(value.func, aliases).lower()
-            if cname in join_names or cname.endswith(".join") or cname.endswith(".joinpath") or cname.endswith(
-                    ".expanduser"):
+            if (cname in join_names or cname.endswith(".join")
+                    or cname.endswith(".joinpath") or cname.endswith(".expanduser")):
                 text = path_expr_text(value)
             else:
                 text = path_expr_text(value)
@@ -736,10 +736,15 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
 
         for node, name in iter_python_calls(tree, aliases):
             lname = name.lower()
-            if lname in _PY_PROCESS_CALLS or any(
+            is_process_call = (
+                lname in _PY_PROCESS_CALLS
+                or lname in {c.lower() for c in _PY_PROCESS_CALLS}
+                or any(
                     lname.endswith("." + c.split(".")[-1]) and c.split(".")[0] in lname
-                    for c in _PY_PROCESS_CALLS) or lname in {c.lower()
-                                                             for c in _PY_PROCESS_CALLS}:
+                    for c in _PY_PROCESS_CALLS
+                )
+            )
+            if is_process_call:
                 shell_true = _has_shell_true(node)
                 findings.append(
                     self._finding(
@@ -892,8 +897,9 @@ def _check_bash(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Safet
                         level=RiskLevel.CRITICAL,
                         message="Use of eval in bash",
                     ))
-            if _DECODE_EXEC_BASH.search(line) or re.search(r"\|\s*(sh|bash|zsh)\b", line) and re.search(
-                    r"base64|xxd|openssl", line, re.IGNORECASE):
+            if (_DECODE_EXEC_BASH.search(line)
+                    or (re.search(r"\|\s*(sh|bash|zsh)\b", line)
+                        and re.search(r"base64|xxd|openssl", line, re.IGNORECASE))):
                 findings.append(
                     self._finding(
                         line,
@@ -1266,8 +1272,9 @@ def _const_int(node: ast.AST | None) -> int | None:
     "httpx.post",
 }
 
-_ENV_SECRET_KEYS = re.compile(r"(?i)(API[_-]?KEY|TOKEN|SECRET|PASSWORD|PASSWD|PRIVATE[_-]?KEY|ACCESS[_-]?KEY|"
-                              r"OPENAI|ANTHROPIC|AWS_SECRET|GITHUB_TOKEN|GH_TOKEN)")
+_ENV_SECRET_KEYS = re.compile(
+    r"(?i)(API[_-]?KEY|TOKEN|SECRET|PASSWORD|PASSWD|PRIVATE[_-]?KEY|"
+    r"ACCESS[_-]?KEY|OPENAI|ANTHROPIC|AWS_SECRET|GITHUB_TOKEN|GH_TOKEN)")
 
 
 class SecretLeakRule(SafetyRule):
@@ -1369,8 +1376,11 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
         # Direct print(os.environ["X"])
         for node in ast.walk(tree):
             if isinstance(node, ast.Subscript) and _is_environ_subscript(node, aliases):
-                key = get_string_literal(node.slice) if hasattr(node, "slice") else None
-                if key is None or _ENV_SECRET_KEYS.search(str(key) if key else "SECRET"):
+                # ast.Subscript always has .slice in py>=3.9; fall back to None
+                # for older interpreters or unusual AST shapes.
+                key = getattr(getattr(node, "slice", None), "value", None)
+                key_str = get_string_literal(key) if key is not None else None
+                if key_str is None or _ENV_SECRET_KEYS.search(str(key_str if key_str else "SECRET")):
                     # Flag any environ subscript used as expression; severity HIGH
                     findings.append(
                         self._finding(

trpc_agent_sdk/safety/_scanner.py

@@ -161,7 +161,10 @@ def scan(self, scan_input: ScanInput) -> SafetyReport:
 
         # blocked means "would be intercepted under this policy": DENY always,
         # and NEEDS_HUMAN_REVIEW when policy.block_on_review is enabled.
-        blocked = decision == Decision.DENY or (decision == Decision.NEEDS_HUMAN_REVIEW and self.policy.block_on_review)
+        blocked = (
+            decision == Decision.DENY
+            or (decision == Decision.NEEDS_HUMAN_REVIEW and self.policy.block_on_review)
+        )
         return SafetyReport(
             decision=decision,
             risk_level=agg_level,

trpc_agent_sdk/safety/_wrapper.py

@@ -8,7 +8,6 @@
 
 import asyncio
 import functools
-from typing import Any
 from typing import Optional
 
 from ._audit import AuditLogger
@@ -166,7 +165,11 @@ async def execute_code(self, invocation_context, input_data):
     return _SafeCodeExecutor()
 
 
-# Backwards-compatible factory alias (returns an object with execute_code).
+# Backwards-compatible alias. SafeCodeExecutor is exposed as a "class-like"
+# callable that returns an object with execute_code; it is intentionally a
+# factory function (not a class) so we can avoid importing the optional
+# code_executors package. Existing code that writes ``SafeCodeExecutor(inner,
+# policy)`` continues to work.
 SafeCodeExecutor = safe_code_executor