fix(safety): pathlib write/rmdir, open dynamic mode, wrapper intercept

Critical (CongkeChen review):
- R001: detect Path.write_text/write_bytes/mkdir/touch/rename/chmod and
  Path.rmdir/unlink receivers; treat non-constant open() mode as write
  (fail-closed) and include system_dir on open read branch
- safety_wrapper: should_block always intercepts and audits intercepted=True;
  raise_on_deny only controls raise vs deny-dict return (no silent execute)

Also: BashTool raises import_error_for when ToolSafetyFilter is None;
_deny_code_result stub Outcome gains .value for contract parity.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Violet2314

tests/tool_safety/test_review_p0_p1.py

@@ -218,3 +218,25 @@ def run(script: str):
         assert False, "expected SafetyDeniedError for review under block_on_review"
     except SafetyDeniedError:
         pass
+
+
+def test_safety_wrapper_raise_on_deny_false_blocks_without_raise():
+    """raise_on_deny=False must still prevent function execution on DENY."""
+    from trpc_agent_sdk.safety import safety_wrapper
+
+    executed = []
+
+    @safety_wrapper(script_arg="script", policy=PolicyConfig(), raise_on_deny=False)
+    def run(script: str):
+        executed.append(script)
+        return "ran"
+
+    result = run(script="rm -rf /")
+    assert executed == []
+    assert result["success"] is False
+    assert result["error"] == "TOOL_SAFETY_DENY"
+
+    # Safe script still runs.
+    result2 = run(script="echo hi")
+    assert executed == ["echo hi"]
+    assert result2 == "ran"

tests/tool_safety/test_rules.py

@@ -452,3 +452,28 @@ def test_dependency_executable_context_still_flagged():
         _policy(),
     )
     assert os_findings
+
+
+def test_dangerous_files_pathlib_write_system_dir():
+    """Path('/usr/bin/x').write_text must DENY (system-dir write, not just
+    sensitive-name path construction)."""
+    rule = DangerousFilesRule()
+    for script in (
+            "from pathlib import Path\nPath('/usr/bin/evil').write_text('x')\n",
+            "from pathlib import Path\nPath('/usr/bin/evil').write_bytes(b'x')\n",
+            "from pathlib import Path\nPath('/usr/local').rmdir()\n",
+    ):
+        findings = rule.check(ScanInput(script=script, language="python"), _policy())
+        assert any("R001" in f.rule_id for f in findings), f"missed: {script!r}"
+
+
+def test_dangerous_files_open_dynamic_mode_system_dir():
+    """open('/usr/bin/x', mode_var) must be treated as write (fail-closed)."""
+    rule = DangerousFilesRule()
+    for script in (
+            "mode='w'\nopen('/usr/bin/x', mode)\n",
+            "mode=mode_var\nopen('/usr/bin/x', mode)\n",
+            "open('/usr/bin/x', mode='a')\n",
+    ):
+        findings = rule.check(ScanInput(script=script, language="python"), _policy())
+        assert any("R001" in f.rule_id for f in findings), f"missed: {script!r}"

tests/tool_safety/test_wrapper.py

@@ -265,13 +265,16 @@ def test_skill_runner_block_review_overrides_policy_records_intercepted(tmp_path
     assert rec["intercepted"] is True
 
 
-def test_safety_wrapper_raise_on_deny_false_records_intercepted_false(tmp_path: Path):
-    """When raise_on_deny=False, a DENY hit must not be recorded as intercepted.
-
-    Regression for CongkeChen's review: intercepted=report.blocked used
-    policy.block_on_review, so a DENY with raise_on_deny=False was falsely
-    recorded as intercepted=True. The actual interception is raise_on_deny.
+def test_safety_wrapper_raise_on_deny_false_still_blocks_and_audits(tmp_path: Path):
+    """raise_on_deny=False must still intercept DENY (not run the function).
+
+    CongkeChen review: intercepted = should_block and raise_on_deny made
+    raise_on_deny=False silently execute DENY scripts and audit
+    intercepted=False. Correct semantics:
+    - should_block always intercepts (function not called, audit True)
+    - raise_on_deny only controls raise vs return deny dict
     """
+    executed = []
 
     @safety_wrapper(
         tool_name="denied_tool",
@@ -280,14 +283,18 @@ def test_safety_wrapper_raise_on_deny_false_records_intercepted_false(tmp_path:
         raise_on_deny=False,
     )
     async def run_tool(*, script):
+        executed.append(script)
         return "ran"
 
     result = asyncio.run(run_tool(script="rm -rf /"))
 
-    assert result == "ran"
+    assert executed == [], "DENY must not run the wrapped function"
+    assert isinstance(result, dict)
+    assert result.get("success") is False
+    assert result.get("error") == "TOOL_SAFETY_DENY"
 
     audit_path = tmp_path / "audit.jsonl"
     assert audit_path.exists()
     rec = json.loads(audit_path.read_text(encoding="utf-8").strip().splitlines()[-1])
     assert rec["decision"] == "deny"
-    assert rec["intercepted"] is False
+    assert rec["intercepted"] is True

trpc_agent_sdk/safety/_rules.py

@@ -238,10 +238,13 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
                     "os.unlink",
                     "pathlib.path.unlink",
                     "pathlib.Path.unlink",
-            } or lname.endswith(".unlink") or lname.endswith(".rmtree"):
+            } or lname.endswith(".unlink") or lname.endswith(".rmtree") or lname.endswith(".rmdir"):
                 arg = node.args[0] if node.args else None
-                target = get_string_literal(arg) or path_expr_text(arg) if arg else "<dynamic>"
-                target = target or "<dynamic>"
+                if arg is not None:
+                    target = get_string_literal(arg) or path_expr_text(arg) or "<dynamic>"
+                else:
+                    # Path(...).unlink() / Path(...).rmdir() — path is the receiver.
+                    target = _path_target_from_attr(node, aliases, path_vars) or "<dynamic>"
                 if "rmtree" in lname or _is_recursive_delete(name, node):
                     findings.append(
                         self._finding(
@@ -269,17 +272,20 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
                     var = node.args[0].id
                     if var in path_vars:
                         target = path_vars[var]
+                # Non-constant mode is treated as potentially writable (fail-closed)
+                # so open("/usr/bin/x", mode_var) cannot bypass system-dir writes.
                 if _is_write_open(node):
                     if _matches_sensitive(target) or _matches_forbidden(target, policy) or _matches_system_dir(target):
                         findings.append(
                             self._finding(
-                                f"open({target!r}, 'w')",
+                                f"open({target!r}, write)",
                                 node.lineno,
                                 "Do not write to system or credential paths.",
                                 message=f"Write to sensitive path {target!r}",
                             ))
                 else:
-                    if _matches_sensitive(target) or _matches_forbidden(target, policy):
+                    if (_matches_sensitive(target) or _matches_forbidden(target, policy)
+                            or _matches_system_dir(target)):
                         findings.append(
                             self._finding(
                                 f"{name}({target!r})",
@@ -290,8 +296,8 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
 
             # pathlib Path.read_text / read_bytes with sensitive target
             if lname.endswith(".read_text") or lname.endswith(".read_bytes"):
-                target = _path_from_attr_call(node, aliases)
-                if _matches_sensitive(target) or _matches_forbidden(target, policy):
+                target = _path_target_from_attr(node, aliases, path_vars)
+                if (_matches_sensitive(target) or _matches_forbidden(target, policy) or _matches_system_dir(target)):
                     findings.append(
                         self._finding(
                             f"{name}(...)",
@@ -300,6 +306,29 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
                             message=f"Read sensitive file via pathlib {target!r}",
                         ))
 
+            # pathlib write / mkdir / touch / rename on sensitive, forbidden, or system
+            # paths. Missing write_* left Path("/usr/bin/x").write_text(...) as
+            # ALLOW (fail-open) because only path-construction of sensitive names
+            # was covered, not system-dir writes.
+            if (lname.endswith(".write_text") or lname.endswith(".write_bytes") or lname.endswith(".mkdir")
+                    or lname.endswith(".touch") or lname.endswith(".replace") or lname.endswith(".rename")
+                    or lname.endswith(".chmod")):
+                target = _path_target_from_attr(node, aliases, path_vars)
+                if target and (_matches_sensitive(target) or _matches_forbidden(target, policy)
+                               or _matches_system_dir(target)):
+                    if lname.endswith(".write_text") or lname.endswith(".write_bytes"):
+                        msg = f"Write to sensitive path via pathlib {target!r}"
+                        rec = "Do not write to system or credential paths."
+                    else:
+                        msg = f"Mutate sensitive path via pathlib {name}({target!r})"
+                        rec = "Do not mutate system or credential paths via pathlib."
+                    findings.append(self._finding(
+                        f"{name}(...)",
+                        node.lineno,
+                        rec,
+                        message=msg,
+                    ))
+
             # Path(...).joinpath(...).read_text pattern: also inspect path construction
             if "pathlib" in lname or lname.endswith("path") or lname.endswith("joinpath") or lname in {
                     "os.path.join",
@@ -391,16 +420,30 @@ def _is_recursive_delete(name: str, node: ast.Call) -> bool:
 
 
 def _is_write_open(node: ast.Call) -> bool:
+    """True when open() may write.
+
+    Constant modes containing w/a/x/+ are writes. A present but non-constant
+    mode (Name/Call/…) is treated as write (fail-closed) so
+    ``open("/usr/bin/x", mode_var)`` cannot bypass system-dir checks.
+    Missing mode defaults to read-only.
+    """
     mode_val = None
+    mode_dynamic = False
     for kw in node.keywords:
         if kw.arg == "mode":
             if isinstance(kw.value, ast.Constant) and isinstance(kw.value.value, str):
                 mode_val = kw.value.value
+            else:
+                mode_dynamic = True
             break
-    if mode_val is None and len(node.args) >= 2:
+    if mode_val is None and not mode_dynamic and len(node.args) >= 2:
         arg = node.args[1]
         if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
             mode_val = arg.value
+        else:
+            mode_dynamic = True
+    if mode_dynamic:
+        return True
     if not mode_val:
         return False
     return any(m in mode_val for m in ("w", "a", "x", "+"))
@@ -424,6 +467,21 @@ def _path_from_attr_call(node: ast.Call, aliases: dict[str, str]) -> str:
     return ""
 
 
+def _path_target_from_attr(
+    node: ast.Call,
+    aliases: dict[str, str],
+    path_vars: dict[str, str],
+) -> str:
+    """Path string for Path(...).method() / p.method() calls."""
+    target = _path_from_attr_call(node, aliases)
+    func = node.func
+    if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name):
+        var = func.value.id
+        if var in path_vars:
+            target = path_vars[var] or target
+    return target or ""
+
+
 def _collect_sensitive_path_vars(
     tree: ast.AST,
     aliases: dict[str, str],

trpc_agent_sdk/safety/_wrapper.py

@@ -74,6 +74,8 @@ class _Outcome:
 
         def __init__(self, name: str):
             self.name = name
+            # Match real Outcome enum contract (downstream reads .value).
+            self.value = name
 
     class _CodeExecutionResult:
 
@@ -251,6 +253,15 @@ def safety_wrapper(
     scanned for simple callables. A positional dict containing *script_arg*
     is also accepted. Variadic ``*args`` callables without a named parameter
     still require a keyword.
+
+    Blocking semantics match ToolSafetyFilter / SafetyGuardedCodeExecutor:
+
+    - ``should_block`` (DENY, or NEEDS_HUMAN_REVIEW with block_on_review) always
+      prevents the wrapped function from running and records
+      ``intercepted=True`` in the audit log.
+    - ``raise_on_deny`` only controls whether a :class:`SafetyDeniedError` is
+      raised (True) or a deny dict is returned (False). It must **not** allow
+      the wrapped function to run when the decision is block.
     """
     import inspect
 
@@ -279,28 +290,43 @@ def _extract_script(func, args, kwargs):
         return None
 
     def _guard(func, args, kwargs):
+        """Return (blocked, report_or_none). blocked=True means do not call func."""
         script = _extract_script(func, args, kwargs)
         if not script or not isinstance(script, str):
-            return
+            return False, None
         report = _scanner.scan(ScanInput(script=script, tool_name=tool_name))
-        # Honor block_on_review (policy or explicit) so decorator semantics
-        # match ToolSafetyFilter / SafetyGuardedCodeExecutor.
         should_block = _should_block_report(report, _block_review)
-        intercepted = should_block and raise_on_deny
-        _audit.log(report, intercepted=intercepted)
-        if intercepted:
-            raise SafetyDeniedError(report)
+        # Audit reflects real interception (should_block), not whether we raise.
+        _audit.log(report, intercepted=should_block)
+        if should_block:
+            if raise_on_deny:
+                raise SafetyDeniedError(report)
+            return True, report
+        return False, report
+
+    def _deny_payload(report):
+        return {
+            "success": False,
+            "error": ("TOOL_SAFETY_DENY" if report.decision == Decision.DENY else "TOOL_SAFETY_NEEDS_REVIEW"),
+            "decision": report.decision.value,
+            "risk_level": report.risk_level.value,
+            "rule_ids": list(report.rule_ids),
+        }
 
     def decorator(func):
 
         @functools.wraps(func)
         async def async_wrapper(*args, **kwargs):
-            _guard(func, args, kwargs)
+            blocked, report = _guard(func, args, kwargs)
+            if blocked:
+                return _deny_payload(report)
             return await func(*args, **kwargs)
 
         @functools.wraps(func)
         def sync_wrapper(*args, **kwargs):
-            _guard(func, args, kwargs)
+            blocked, report = _guard(func, args, kwargs)
+            if blocked:
+                return _deny_payload(report)
             return func(*args, **kwargs)
 
         return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper

trpc_agent_sdk/tools/file_tools/_bash_tool.py

@@ -68,9 +68,19 @@ def _attach_safety_guard(
         block_on_review: bool,
     ) -> None:
         """Attach ToolSafetyFilter when enable_safety_guard is True."""
+        from trpc_agent_sdk import safety as safety_pkg
         from trpc_agent_sdk.safety import PolicyConfig
         from trpc_agent_sdk.safety import ToolSafetyFilter
 
+        # When optional filter deps fail to import, ToolSafetyFilter is None and
+        # import_error_for holds the original ImportError. Raise that instead of
+        # a confusing TypeError: NoneType is not callable at construction time.
+        if ToolSafetyFilter is None:
+            err = safety_pkg.import_error_for("ToolSafetyFilter")
+            if err is not None:
+                raise err
+            raise ImportError("ToolSafetyFilter is unavailable")
+
         if policy_path:
             policy = PolicyConfig.from_yaml(policy_path)
         else: