fix: enforce max_script_bytes, fix decorator silent skip, rm flag parsing, tee dead branch

  1. Enforce max_script_bytes: single-line giant payloads with no newlines
     are now caught by a byte-size check alongside the existing line-count
     check. Both trigger DENY with a blocklist pre-scan.

  2. Decorator silent skip: @safety_wrapper now logs a warning when the
     script_arg_name is not found in kwargs / positional dict args, so
     callers can detect missing protection.

  3. _check_blocklist_override now receives script_type — Python
     string-literal stripping is skipped for Bash to avoid false negatives
     on cat '/etc/shadow' etc.

  4. rm flag detection: parse short flags character-by-character instead of
     substring matching (rm -i no longer falsely triggers rm -rf).

  5. tee dead branch: tee writes to sensitive paths (tee /etc/shadow,
     tee ~/.aws/credentials) are now detected.

  6. Evidence truncation: <truncated:N> now reports len-300 consistently.

  7. Redundant ternary: Decision.DENY if x else Decision.DENY → Decision.DENY.

Author: lll-peanut

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -328,8 +328,25 @@ def _scan_lines(self) -> None:
                                 },
                             ))
                         break
-            elif cmd_lower in _FILE_WRITE_COMMANDS and cmd_lower == "dd":
+            elif cmd_lower == "dd":
                 self._check_dd(line_no, raw_line, args)
+            elif cmd_lower == "tee":
+                # tee writes to files; check for sensitive paths in args
+                for pa in args:
+                    if pa and not pa.startswith("-") and (_is_sensitive_path(pa) or pa.startswith("/dev/sd")):
+                        self._findings.append(
+                            BashScanFinding(
+                                kind="command",
+                                command="tee",
+                                args=args_str,
+                                line_number=line_no,
+                                evidence=raw_line.strip()[:300],
+                                extra={
+                                    "risk": "sensitive_file_write",
+                                    "path": pa
+                                },
+                            ))
+                        break
 
     # ------------------------------------------------------------------
     # Specific checks
@@ -338,17 +355,25 @@ def _scan_lines(self) -> None:
     def _check_rm(self, line_no: int, raw_line: str, tokens: List[str], args: List[str]) -> None:
         """Check for recursive-delete.  Flags both ``rm -rf`` and ``rm -r -f``."""
         all_tokens = set(tokens)
-        # Handles combined flags: -rf, -fr, -r, -R, --recursive
-        has_r = any(("r" in t.replace("-", "").lower() and not t.startswith("--")) or t in ("-r", "-R", "--recursive")
-                    for t in tokens if t.startswith("-"))
-        # Handles combined flags: -rf, -fr, -f, --force
-        has_f = any(("f" in t.replace("-", "").lower() and not t.startswith("--")) or t in ("-f", "--force")
-                    for t in tokens if t.startswith("-"))
-
-        # Check recursive via long option
-        has_recursive = "--recursive" in all_tokens
-
-        if (has_r or has_recursive) and has_f:
+        # Parse short flags character-by-character (avoids substring false
+        # positives: "-i" has no "r", "-v" has no "f", etc.).
+        short_flags: set[str] = set()
+        has_long_recursive = False
+        has_long_force = False
+        for t in tokens:
+            if t.startswith("--"):
+                if t == "--recursive":
+                    has_long_recursive = True
+                if t == "--force":
+                    has_long_force = True
+            elif t.startswith("-") and not t.startswith("--"):
+                for ch in t.lstrip("-"):
+                    short_flags.add(ch.lower())
+
+        has_r = "r" in short_flags or has_long_recursive
+        has_f = "f" in short_flags or has_long_force
+
+        if has_r and has_f:
             target = args[-1] if args else "?"
             self._findings.append(
                 BashScanFinding(
@@ -363,7 +388,7 @@ def _check_rm(self, line_no: int, raw_line: str, tokens: List[str], args: List[s
                         "force": True
                     },
                 ))
-        elif has_r or has_recursive:
+        elif has_r:
             target = args[-1] if args else "?"
             if target and (target.startswith("/") or _is_sensitive_path(target)):
                 self._findings.append(

trpc_agent_sdk/tools/safety/_safety_wrapper.py

@@ -31,12 +31,15 @@ async def my_tool_run(tool_context, args):
 from __future__ import annotations
 
 import functools
+import logging
 from contextlib import asynccontextmanager
 from typing import Any
 from typing import AsyncIterator
 from typing import Callable
 from typing import Optional
 
+_logger = logging.getLogger("trpc_agent_sdk.tools.safety.wrapper")
+
 from ._audit import AuditLogger
 from ._policy import SafetyPolicy
 from ._policy import get_policy
@@ -217,7 +220,7 @@ def decorator(func: Callable) -> Callable:
 
         @functools.wraps(func)
         async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
-            script = kwargs.get(script_arg_name)
+            script: Optional[str] = kwargs.get(script_arg_name)
             if script is None:
                 # Try to find it in positional args (e.g. tool_context, args)
                 for arg in args:
@@ -226,18 +229,32 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                         break
             if script and isinstance(script, str):
                 wrapper_inst.check(script)
+            elif script is not None and not isinstance(script, str):
+                _logger.warning("safety_wrapper: '%s' value is not a string (type=%s) — scan skipped.", script_arg_name,
+                                type(script).__name__)
+            else:
+                _logger.warning(
+                    "safety_wrapper: could not find '%s' in kwargs or positional dict args "
+                    "for %s — scan skipped.", script_arg_name, func.__name__)
             return await func(*args, **kwargs)
 
         @functools.wraps(func)
         def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
-            script = kwargs.get(script_arg_name)
+            script: Optional[str] = kwargs.get(script_arg_name)
             if script is None:
                 for arg in args:
                     if isinstance(arg, dict) and script_arg_name in arg:
                         script = arg[script_arg_name]
                         break
             if script and isinstance(script, str):
                 wrapper_inst.check(script)
+            elif script is not None and not isinstance(script, str):
+                _logger.warning("safety_wrapper: '%s' value is not a string (type=%s) — scan skipped.", script_arg_name,
+                                type(script).__name__)
+            else:
+                _logger.warning(
+                    "safety_wrapper: could not find '%s' in kwargs or positional dict args "
+                    "for %s — scan skipped.", script_arg_name, func.__name__)
             return func(*args, **kwargs)
 
         import asyncio

trpc_agent_sdk/tools/safety/_scanner.py

@@ -103,14 +103,23 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
                 script = script + "\n" + args_text
 
         script_lines = script.count("\n") + (1 if script else 0)
+        script_bytes = len(script.encode("utf-8"))
 
         # ══════════════════════════════════════════════════════════════
-        # EARLY RETURN: script too large → skip expensive scanning
-        # BUT still run the lightweight blocklist-pattern check first.
-        # Without this an attacker can pad a malicious script with empty
-        # lines / comments past max_script_lines and bypass all detection.
+        # EARLY RETURN: script too large (lines OR bytes) → skip
+        # expensive scanning.  Runs a lightweight blocklist-pattern
+        # pre-check first so that an attacker cannot pad a malicious
+        # script past the limit and bypass all detection.
         # ══════════════════════════════════════════════════════════════
+        oversized = script_lines > self._policy.max_script_lines
+        oversized_reason = ""
         if script_lines > self._policy.max_script_lines:
+            oversized_reason = (f"Script is {script_lines} lines (max {self._policy.max_script_lines})")
+        elif script_bytes > self._policy.max_script_bytes:
+            oversized = True
+            oversized_reason = (f"Script is {script_bytes} bytes (max {self._policy.max_script_bytes})")
+
+        if oversized:
             # Fast pre-check: scan blocklist patterns against the full text
             blocklist_hit = None
             for pattern in self._policy.blocklist_patterns:
@@ -126,9 +135,9 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
                     rule_id="GLOBAL-001",
                     category=RiskCategory.RESOURCE_ABUSE,
                     risk_level=RiskLevel.MEDIUM,
-                    evidence=f"Script is {script_lines} lines (max {self._policy.max_script_lines})",
-                    message="Script exceeds maximum line count.",
-                    recommendation="Split the script or increase max_script_lines in policy.",
+                    evidence=oversized_reason,
+                    message="Script exceeds maximum size limit.",
+                    recommendation="Split the script or increase the limit in policy.",
                     line_number=0,
                     matched_pattern="",
                 )
@@ -152,10 +161,10 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
                 tool_name=scan_input.tool_name,
                 script_type=scan_input.script_type,
                 script_size_lines=script_lines,
-                decision=Decision.DENY if blocklist_hit else Decision.DENY,
+                decision=Decision.DENY,
                 risk_level=RiskLevel.CRITICAL if blocklist_hit else RiskLevel.HIGH,
                 findings=oversized_findings,
-                summary=f"Script is too large: {script_lines} lines (max {self._policy.max_script_lines})." +
+                summary=f"{oversized_reason}." +
                 (" Blocklist pattern matched — denied." if blocklist_hit else " Denied for safety."),
                 scan_duration_ms=round(duration_ms, 2),
                 policy_version=self._policy.content_hash,
@@ -214,7 +223,7 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
 
         # Apply blocklist override — blocklist patterns always → deny
         if decision != Decision.DENY:
-            decision = self._check_blocklist_override(script, decision)
+            decision = self._check_blocklist_override(script, decision, scan_input.script_type)
 
         # Apply allow-pattern override — allow patterns → allow
         # Only upgrades NEEDS_HUMAN_REVIEW; never overrides DENY (blocklist wins).
@@ -745,12 +754,17 @@ def _detect_type(script: str) -> ScriptType:
             return ScriptType.BASH
         return ScriptType.UNKNOWN
 
-    def _check_blocklist_override(self, script: str, current_decision: Decision) -> Decision:
+    def _check_blocklist_override(self,
+                                  script: str,
+                                  current_decision: Decision,
+                                  script_type: ScriptType = ScriptType.UNKNOWN) -> Decision:
         """If any blocklist pattern matches, escalate to DENY.
 
         Per-line matching is used so that patterns appearing inside
         ``echo`` / ``printf`` string literals do not trigger false
-        positives.
+        positives.  Python string-literal stripping is only applied for
+        Python scripts — applying it to Bash would turn ``cat
+        '/etc/shadow'`` into a false negative.
         """
         for pattern in self._policy.blocklist_patterns:
             try:
@@ -762,9 +776,13 @@ def _check_blocklist_override(self, script: str, current_decision: Decision) ->
                 stripped = line.lstrip()
                 if stripped.startswith("#"):
                     continue
-                # Strip Python string-literal content so that patterns
-                # like ``rm -rf /`` inside ``r'...'`` don't match.
-                search_line = _strip_python_comment_line(line)
+                # Strip Python string-literal content only for Python
+                # scripts so that Bash single-quoted paths like
+                # cat '/etc/shadow' are not masked.
+                if script_type in (ScriptType.PYTHON, ScriptType.UNKNOWN):
+                    search_line = _strip_python_comment_line(line)
+                else:
+                    search_line = line
                 if pat.search(search_line):
                     # Skip if the match is inside an echo/printf string literal
                     if _is_in_echo_string(line, pattern):
@@ -824,7 +842,7 @@ def _redact_evidence(self, findings: list[SafetyFinding]) -> list[SafetyFinding]
                 f.evidence = re.sub(pat, repl, f.evidence, flags=re.IGNORECASE)
             # Truncate very long evidence to prevent data leakage
             if len(f.evidence) > 320:
-                f.evidence = f.evidence[:300] + f"...<truncated:{len(f.evidence) - 320}>"
+                f.evidence = f.evidence[:300] + f"...<truncated:{len(f.evidence) - 300}>"
         return findings