fix(safety): address CongkeChen review - stderr fallback, single-file delete, dd false-positive, path boundary

Critical:
- _filter._before: non-blocking NEEDS_HUMAN_REVIEW now also writes to
  stderr so the warning surfaces even if _after is skipped (handle()
  raised, earlier filter set is_continue=False, or non-dict tool return).

Warning:
- _rules.DangerousFilesRule: os.remove/os.unlink/Path.unlink on
  sensitive/forbidden/system paths now triggers R001 (previously only
  recursive delete was flagged, so os.remove("/etc/passwd") was missed
  while bash "rm /etc/passwd" was caught).
- _rules._DD_WRITE: tighten regex from \bdd\b to (?:^|\s|;|&&|\|\|)\s*dd
  so adduser/findd/xdd no longer false-positive HIGH/DENY.
- _rules._matches_forbidden/_matches_system_dir: switch from substring
  to path-boundary matching so '/etc' no longer matches '/etcetera' and
  '.env' no longer matches 'my.envrc'.
- _rules decode-exec: extract pipe_to_shell/decoder locals to make the
  or/and precedence explicit and avoid the nested-paren yapf ambiguity.
- test_tool_filter.test_filter_block_on_review: rename to
  test_filter_block_on_review_deny_still_blocks and tighten assertion to
  == "TOOL_SAFETY_DENY" (was loose `in (DENY, NEEDS_REVIEW)` that never
  exercised the review-block path).

Suggestion:
- safety.__init__: add _WRAPPER_AVAILABLE flag so callers can tell
  whether the wrap_tool/safety_wrapper group loaded independently of
  _SDK_AVAILABLE (ToolSafetyFilter).

Tests:
- 7 regression tests for os.remove/os.unlink sensitive delete, dd
  false-positive on adduser, dd real command still caught, forbidden
  path boundary (.env vs my.envrc), system dir boundary (/etc vs
  /etcetera).
- test_filter_sleep_at_threshold_is_allowed, test_filter_contextvar_
  does_not_leak_across_calls.

Author: Violet2314

tests/tool_safety/test_rules.py

@@ -203,3 +203,64 @@ def test_secret_env_access():
 def test_secret_redacts_evidence():
     from trpc_agent_sdk.safety._rules import redact
     assert redact("sk-abcdefghijklmnopqrstuvwxyz").endswith("***")
+
+
+def test_dangerous_files_python_remove_sensitive_path():
+    """Regression for CongkeChen review: os.remove('/etc/passwd') must trigger
+    R001 even without recursive delete. Previously only rmtree/unlink-with-r
+    were flagged, so single-file delete of sensitive paths was silently
+    allowed while the bash equivalent (rm /etc/passwd) was caught."""
+    rule = DangerousFilesRule()
+    inp = ScanInput(script="import os\nos.remove('/etc/passwd')\n", language="python")
+    findings = rule.check(inp, _policy())
+    assert findings and findings[0].rule_id == "R001_dangerous_files"
+
+
+def test_dangerous_files_python_unlink_ssh_key():
+    """os.unlink('~/.ssh/id_rsa') must trigger R001 (single-file sensitive delete)."""
+    rule = DangerousFilesRule()
+    inp = ScanInput(script="import os\nos.unlink('/home/u/.ssh/id_rsa')\n", language="python")
+    findings = rule.check(inp, _policy())
+    assert findings and findings[0].rule_id == "R001_dangerous_files"
+
+
+def test_resource_dd_not_false_positive_on_adduser():
+    """Regression for CongkeChen review: \\bdd\\b matched any 'dd' substring,
+    so 'adduser', 'findd', 'xdd' were false-positive HIGH/DENY. The regex now
+    requires dd as a command token (after start/space/;/&&/||)."""
+    rule = ResourceAbuseRule()
+    inp = ScanInput(script="adduser bob\nfind / -name xdd\n", language="bash")
+    findings = rule.check(inp, _policy())
+    # No dd finding should fire for adduser/findd/xdd.
+    assert not any("dd can write" in f.metadata.get("message", "") for f in findings)
+
+
+def test_resource_dd_real_command_still_caught():
+    """Real 'dd if=/dev/zero of=/tmp/big bs=1M count=100' must still be caught."""
+    rule = ResourceAbuseRule()
+    inp = ScanInput(script="dd if=/dev/zero of=/tmp/big bs=1M count=100\n", language="bash")
+    findings = rule.check(inp, _policy())
+    assert any("dd can write" in f.metadata.get("message", "") for f in findings)
+
+
+def test_dangerous_files_forbidden_path_boundary_not_substring():
+    """Regression for CongkeChen review: _matches_forbidden used substring
+    matching, so '.env' matched 'my.envrc'. Now uses path-boundary matching."""
+    rule = DangerousFilesRule()
+    policy = PolicyConfig(forbidden_paths=[".env"])
+    # 'my.envrc' must NOT match forbidden path '.env' (no path boundary before .env).
+    inp = ScanInput(script="open('my.envrc')\n", language="python")
+    findings = rule.check(inp, policy)
+    # No forbidden-path finding should fire for my.envrc.
+    assert not any("forbidden" in f.metadata.get("message", "").lower() for f in findings)
+
+
+def test_dangerous_files_system_dir_boundary_not_substring():
+    """Regression for CongkeChen review: _matches_system_dir used substring
+    matching, so '/etc' matched '/etcetera'. Now uses path-boundary matching."""
+    rule = DangerousFilesRule()
+    # '/etcetera/foo' must NOT match system dir '/etc' (no path boundary after /etc).
+    inp = ScanInput(script="open('/etcetera/foo')\n", language="python")
+    findings = rule.check(inp, _policy())
+    # No system-dir finding should fire for /etcetera.
+    assert not any("system" in f.metadata.get("message", "").lower() for f in findings)

tests/tool_safety/test_tool_filter.py

@@ -76,17 +76,21 @@ def test_filter_dynamic_network_is_denied_by_default(tmp_path: Path):
     assert rsp.rsp["error"] == "TOOL_SAFETY_DENY"
 
 
-def test_filter_block_on_review(tmp_path: Path):
+def test_filter_block_on_review_deny_still_blocks(tmp_path: Path):
+    """When block_on_review=True, DENY (HIGH) findings must still block.
+
+    This is a sanity check that block_on_review does not weaken DENY handling.
+    The pure-MEDIUM review-block path is covered by
+    test_filter_block_on_review_pure_medium_returns_needs_review.
+    """
     flt = _make_filter(tmp_path, block_on_review=True)
     rsp = FilterResult()
-    # Medium-only finding that is not HIGH: still exercise block_on_review via
-    # a policy that treats medium as review-only when block_on_review is True.
-    # curl $URL is HIGH/deny under default policy; use empty script path via code
-    # that only triggers review is hard — assert deny still blocks.
+    # curl $URL is HIGH/deny under default policy; block_on_review must NOT
+    # downgrade a DENY to a non-blocking review.
     req = {"command": "curl $URL"}
     asyncio.run(flt._before(None, req, rsp))  # pylint: disable=protected-access
     assert rsp.is_continue is False
-    assert rsp.rsp["error"] in ("TOOL_SAFETY_DENY", "TOOL_SAFETY_NEEDS_REVIEW")
+    assert rsp.rsp["error"] == "TOOL_SAFETY_DENY"
     assert rsp.rsp.get("success") is False
 
 

trpc_agent_sdk/safety/__init__.py

@@ -47,6 +47,7 @@
 # original ImportError so callers can diagnose the cause instead of getting a
 # silent None + later TypeError when they try to use the symbol.
 _SDK_AVAILABLE = False
+_WRAPPER_AVAILABLE = False
 _IMPORT_ERRORS: dict[str, BaseException] = {}
 try:  # pragma: no cover
     from ._filter import ToolSafetyFilter
@@ -63,6 +64,7 @@
     from ._wrapper import safe_code_executor
     from ._wrapper import safety_wrapper
     from ._wrapper import wrap_tool
+    _WRAPPER_AVAILABLE = True
 except Exception as ex:  # pylint: disable=broad-except
     SafeCodeExecutor = None  # type: ignore[assignment]
     SafetyDeniedError = None  # type: ignore[assignment]
@@ -124,4 +126,5 @@ def import_error_for(name: str) -> BaseException | None:
     "SafetyReviewedSkillRunner",
     "import_error_for",
     "_SDK_AVAILABLE",
+    "_WRAPPER_AVAILABLE",
 ]

trpc_agent_sdk/safety/_filter.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import contextvars
+import sys
 from typing import Any
 from typing import Optional
 
@@ -128,6 +129,15 @@ async def _before(self, ctx: Any, req: Any, rsp: FilterResult) -> None:
             # tool's actual result. Setting rsp.rsp here is useless because
             # BaseFilter.run's handle() step overwrites result.rsp with the
             # tool's return value.
+            #
+            # Belt-and-suspenders: also write to stderr so that if _after is
+            # skipped (handle() raised, earlier filter set is_continue=False,
+            # or the tool returned a non-dict that _after cannot annotate),
+            # the warning is still surfaced to the caller instead of being
+            # silently lost. The audit log already records the decision; this
+            # stderr line is the in-band fallback for human operators.
+            sys.stderr.write(f"TOOL_SAFETY_NEEDS_REVIEW: {list(report.rule_ids)} "
+                             f"(risk={report.risk_level.value})\n")
             _REVIEW_REPORT.set(report)
 
     async def _after(self, ctx: Any, req: Any, rsp: FilterResult) -> None:

trpc_agent_sdk/safety/_rules.py

@@ -129,13 +129,36 @@ def _matches_sensitive(target: str) -> bool:
 def _matches_system_dir(target: str) -> bool:
     if not target:
         return False
-    return any(sd in target for sd in _SYSTEM_DIRS)
+    # Use path-boundary matching so '/etc' matches '/etc/passwd' but NOT
+    # '/etcetera/foo'. We treat '/', '\\', start-of-string, and end-of-string
+    # as valid path boundaries around each system dir prefix.
+    for sd in _SYSTEM_DIRS:
+        # Normalize backslashes to forward slashes for uniform matching.
+        sd_norm = sd.replace("\\", "/")
+        tgt_norm = target.replace("\\", "/")
+        if tgt_norm == sd_norm or tgt_norm.startswith(sd_norm + "/"):
+            return True
+    return False
 
 
 def _matches_forbidden(target: str, policy: PolicyConfig) -> bool:
     if not target:
         return False
-    return any(fb in target for fb in policy.forbidden_paths)
+    # Path-boundary match so '.env' matches '/app/.env' but NOT 'my.envrc'.
+    # We accept the forbidden path appearing at start-of-string or after a
+    # path separator; a trailing separator or end-of-string is NOT required
+    # so that '.env' matches '.env.production' too (common convention).
+    tgt_norm = target.replace("\\", "/")
+    for fb in policy.forbidden_paths:
+        fb_norm = fb.replace("\\", "/")
+        if not fb_norm:
+            continue
+        if (tgt_norm == fb_norm
+                or tgt_norm.startswith(fb_norm + "/")
+                or tgt_norm.endswith("/" + fb_norm)
+                or ("/" not in tgt_norm and tgt_norm == fb_norm)):
+            return True
+    return False
 
 
 # ---------------------------------------------------------------------------
@@ -189,6 +212,18 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
                             "Avoid recursive deletion; restrict to known workspace paths.",
                             message=f"Recursive/forced delete via {name}({target!r})",
                         ))
+                elif (_matches_sensitive(target) or _matches_forbidden(target, policy) or _matches_system_dir(target)):
+                    # Single-file delete of a sensitive/forbidden/system path
+                    # (e.g. os.remove("/etc/passwd"), os.unlink("~/.ssh/id_rsa"))
+                    # is just as dangerous as recursive delete; flag it so the
+                    # Python side matches the bash rm /etc/passwd coverage.
+                    findings.append(
+                        self._finding(
+                            f"{name}({target})",
+                            node.lineno,
+                            f"Avoid deleting sensitive/forbidden path: {target}",
+                            message=f"Single-file delete of sensitive path via {name}({target!r})",
+                        ))
 
             if lname in {"open", "builtins.open"} or lname.endswith(".open"):
                 target = _first_str_or_path(node) or "<dynamic>"
@@ -902,8 +937,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):
+            pipe_to_shell = re.search(r"\|\s*(?:sh|bash|zsh)\b", line)
+            decoder = re.search(r"base64|xxd|openssl", line, re.IGNORECASE)
+            if _DECODE_EXEC_BASH.search(line) or (pipe_to_shell and decoder):
                 findings.append(
                     self._finding(
                         line,
@@ -1055,7 +1091,7 @@ def _subprocess_args_text(node: ast.Call) -> str:
     re.IGNORECASE,
 )
 _LONG_SLEEP_BASH = re.compile(r"\bsleep\s+(\d+)", re.IGNORECASE)
-_DD_WRITE = re.compile(r"\bdd\b", re.IGNORECASE)
+_DD_WRITE = re.compile(r"(?:^|\s|;|&&|\|\|)\s*dd\b", re.IGNORECASE)
 _BIG_WRITE = re.compile(r"(head|tail|yes|/dev/zero|/dev/urandom)", re.IGNORECASE)
 _HIGH_CONCURRENCY = {
     "concurrent.futures.threadpoolexecutor",