fix: echo bypass, VAR=val extraction, multi-cmd analysis

Author: lll-peanut

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -237,115 +237,118 @@ def _scan_lines(self) -> None:
             # Check pipes and background
             self._check_operators(line_no, raw_line, tokens)
 
-            # Command analysis
-            cmd = tokens[0]
-            # Skip variable assignments (FOO=bar cmd)
-            if "=" in cmd and "=" == cmd.split(None, 1)[0][-1:]:
-                # VAR=val — skip to next token
-                if len(tokens) > 1:
-                    cmd = tokens[1]
-                    tokens = tokens[1:]
-                else:
-                    continue
+            # Command analysis — process each sub-command in the token stream
+            # (splits on ; && || so that "echo x; rm -rf /" is fully analysed)
+            self._dispatch_commands(line_no, raw_line, tokens)
 
-            cmd_lower = cmd.lower()
-            args = tokens[1:]
-            args_str = " ".join(args)
+    # ------------------------------------------------------------------
+    # Sub-command dispatcher (handles ; && || on a line)
+    # ------------------------------------------------------------------
 
-            # --- Dispatch ---
-            if cmd_lower == "rm":
-                self._check_rm(line_no, raw_line, tokens, args)
-            elif cmd_lower in _NETWORK_COMMANDS:
-                self._findings.append(
-                    BashScanFinding(
-                        kind=cmd_lower,
-                        command=cmd_lower,
-                        args=args_str,
-                        line_number=line_no,
-                        evidence=raw_line.strip()[:300],
-                    ))
-            elif cmd_lower in _PRIVILEGE_COMMANDS:
-                self._findings.append(
-                    BashScanFinding(
-                        kind="sudo" if cmd_lower != "chroot" else "sudo",
-                        command=cmd_lower,
-                        args=args_str,
-                        line_number=line_no,
-                        evidence=raw_line.strip()[:300],
-                        extra={"privilege_command": cmd_lower},
-                    ))
-            elif cmd_lower in _INSTALL_COMMANDS:
-                sub = args[0].lower() if args else ""
-                is_install = sub in _SUBCOMMAND_MAP.get(cmd_lower, set())
-                if is_install or cmd_lower in ("pip", "pip3", "pipx"):
-                    self._findings.append(
-                        BashScanFinding(
-                            kind="install",
-                            command=cmd_lower,
-                            args=args_str,
-                            line_number=line_no,
-                            evidence=raw_line.strip()[:300],
-                            extra={
-                                "package_manager": cmd_lower,
-                                "subcommand": sub if is_install else "",
-                            },
-                        ))
-            elif cmd_lower in _DESTRUCTIVE_COMMANDS:
-                self._findings.append(
-                    BashScanFinding(
-                        kind="command",
-                        command=cmd_lower,
-                        args=args_str,
-                        line_number=line_no,
-                        evidence=raw_line.strip()[:300],
-                        extra={"risk": "destructive"},
-                    ))
-            elif cmd_lower in _DYNAMIC_COMMANDS:
-                self._findings.append(
-                    BashScanFinding(
-                        kind="eval",
-                        command=cmd_lower,
-                        args=args_str,
-                        line_number=line_no,
-                        evidence=raw_line.strip()[:300],
-                    ))
-            elif cmd_lower in _FILE_READ_COMMANDS:
-                # Check if reads a sensitive path
-                path_args = [a for a in args if not a.startswith("-")]
-                for pa in path_args:
-                    if _is_sensitive_path(pa):
-                        self._findings.append(
-                            BashScanFinding(
-                                kind="command",
+    def _dispatch_commands(self, line_no: int, raw_line: str, tokens: List[str]) -> None:
+        """Analyse each sub-command in *tokens*, splitting on ``;`` ``&&`` ``||``.
+
+        This ensures that ``echo x; rm -rf /`` and ``FOO=bar rm -rf /`` are
+        both fully analysed, not just the first token.
+        """
+        seg_start = 0
+        for i, t in enumerate(tokens):
+            if t in (";", "&&", "||"):
+                self._analyse_one_command(line_no, raw_line, tokens[seg_start:i])
+                seg_start = i + 1
+        # Tail segment (or the whole line if no separators)
+        if seg_start < len(tokens):
+            self._analyse_one_command(line_no, raw_line, tokens[seg_start:])
+
+    def _analyse_one_command(self, line_no: int, raw_line: str, cmd_tokens: List[str]) -> None:
+        """Dispatch a single command segment to the appropriate checker."""
+        if not cmd_tokens:
+            return
+        # Skip variable assignments (FOO=bar cmd ...)
+        idx = 0
+        while idx < len(cmd_tokens) and re.match(r"[A-Za-z_]\w*=", cmd_tokens[idx]):
+            idx += 1
+        if idx >= len(cmd_tokens):
+            return  # pure assignment, no command
+        cmd = cmd_tokens[idx]
+        args = cmd_tokens[idx + 1:]
+        cmd_lower = cmd.lower()
+        args_str = " ".join(args)
+        evidence = " ".join(cmd_tokens)[:300]
+
+        if cmd_lower == "rm":
+            self._check_rm(line_no, raw_line, cmd_tokens[idx:], args)
+        elif cmd_lower in _NETWORK_COMMANDS:
+            self._findings.append(
+                BashScanFinding(kind=cmd_lower,
                                 command=cmd_lower,
                                 args=args_str,
                                 line_number=line_no,
-                                evidence=raw_line.strip()[:300],
-                                extra={
-                                    "risk": "sensitive_file_read",
-                                    "path": pa
-                                },
-                            ))
-                        break
-            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",
+                                evidence=evidence))
+        elif cmd_lower in _PRIVILEGE_COMMANDS:
+            self._findings.append(
+                BashScanFinding(kind="sudo",
+                                command=cmd_lower,
+                                args=args_str,
+                                line_number=line_no,
+                                evidence=evidence,
+                                extra={"privilege_command": cmd_lower}))
+        elif cmd_lower in _INSTALL_COMMANDS:
+            sub = args[0].lower() if args else ""
+            is_install = sub in _SUBCOMMAND_MAP.get(cmd_lower, set())
+            if is_install or cmd_lower in ("pip", "pip3", "pipx"):
+                self._findings.append(
+                    BashScanFinding(kind="install",
+                                    command=cmd_lower,
+                                    args=args_str,
+                                    line_number=line_no,
+                                    evidence=evidence,
+                                    extra={
+                                        "package_manager": cmd_lower,
+                                        "subcommand": sub if is_install else ""
+                                    }))
+        elif cmd_lower in _DESTRUCTIVE_COMMANDS:
+            self._findings.append(
+                BashScanFinding(kind="command",
+                                command=cmd_lower,
                                 args=args_str,
                                 line_number=line_no,
-                                evidence=raw_line.strip()[:300],
-                                extra={
-                                    "risk": "sensitive_file_write",
-                                    "path": pa
-                                },
-                            ))
-                        break
+                                evidence=evidence,
+                                extra={"risk": "destructive"}))
+        elif cmd_lower in _DYNAMIC_COMMANDS:
+            self._findings.append(
+                BashScanFinding(kind="eval", command=cmd_lower, args=args_str, line_number=line_no, evidence=evidence))
+        elif cmd_lower in _FILE_READ_COMMANDS:
+            path_args = [a for a in args if not a.startswith("-")]
+            for pa in path_args:
+                if _is_sensitive_path(pa):
+                    self._findings.append(
+                        BashScanFinding(kind="command",
+                                        command=cmd_lower,
+                                        args=args_str,
+                                        line_number=line_no,
+                                        evidence=evidence,
+                                        extra={
+                                            "risk": "sensitive_file_read",
+                                            "path": pa
+                                        }))
+                    break
+        elif cmd_lower == "dd":
+            self._check_dd(line_no, raw_line, args)
+        elif cmd_lower == "tee":
+            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=evidence,
+                                        extra={
+                                            "risk": "sensitive_file_write",
+                                            "path": pa
+                                        }))
+                    break
 
     # ------------------------------------------------------------------
     # Specific checks

trpc_agent_sdk/tools/safety/_rules.py

@@ -852,29 +852,46 @@ def __call__(
 
 
 def _is_in_echo_string(line: str, pattern: str) -> bool:
-    """Return True if *pattern* match is inside an echo/printf string literal.
+    """Return True if *pattern* matches are ALL inside echo/printf string literals.
 
-    In Bash, ``echo 'rm -rf /'`` is harmless — the dangerous command is just
-    printed, not executed.  This helper avoids flagging such lines.
+    In Bash, ``echo 'rm -rf /'`` is harmless.  But ``echo "rm -rf /"; rm -rf /``
+    is dangerous — the first ``rm`` is harmless but the second is real.  This
+    helper only suppresses a finding when the pattern appears **nowhere** outside
+    echo/printf quoted strings on the line.
     """
     stripped = line.strip()
     # Only applies to echo / printf commands
     if not (stripped.startswith("echo ") or stripped.startswith("echo\t") or stripped.startswith("printf ")
             or stripped.startswith("printf\t") or stripped.startswith("/bin/echo ")
             or stripped.startswith("/usr/bin/echo ")):
         return False
-    # Check if the pattern *matches* inside single or double quotes
+    # Check if the pattern matches inside any quoted string
     try:
         pat = re.compile(pattern, re.IGNORECASE)
     except re.error:
         return False
+
+    in_quotes = False
     for m in re.finditer(r"'[^']*'", stripped):
         if pat.search(m.group(0)):
-            return True
-    for m in re.finditer(r'"[^"]*"', stripped):
-        if pat.search(m.group(0)):
-            return True
-    return False
+            in_quotes = True
+            break
+    if not in_quotes:
+        for m in re.finditer(r'"[^"]*"', stripped):
+            if pat.search(m.group(0)):
+                in_quotes = True
+                break
+    if not in_quotes:
+        return False  # pattern not in any quoted string — normal danger report
+
+    # Pattern found inside quotes.  Now check whether it ALSO appears
+    # outside quotes (e.g. after ; / &&).  If so, it is a real danger.
+    outside = re.sub(r"'[^']*'", " ", stripped)
+    outside = re.sub(r'"[^"]*"', " ", outside)
+    if pat.search(outside):
+        return False  # match outside quotes → real danger
+
+    return True
 
 
 def _all_commands_whitelisted(line: str, policy: SafetyPolicy) -> bool:

trpc_agent_sdk/tools/safety/_safety_filter.py

@@ -127,6 +127,9 @@ async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult) -> None:
         # Always set OTel attributes (no-op if OTel not installed)
         set_safety_span_attributes(report)
 
+        # Always expose the report for downstream inspection
+        setattr(rsp, "safety_report", report)
+
         if report.decision == Decision.DENY:
             logger.warning(
                 "ToolSafetyFilter BLOCKED tool '%s': %s",
@@ -143,14 +146,12 @@ async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult) -> None:
                 tool_name,
                 report.summary,
             )
-            setattr(rsp, "safety_report", report)
             if self._block_on_review:
                 rsp.error = ToolSafetyDeniedError(report)
                 rsp.is_continue = False
 
         else:
             logger.debug("ToolSafetyFilter allowed tool '%s'.", tool_name)
-            setattr(rsp, "safety_report", report)
 
 
 class ToolSafetyDeniedError(RuntimeError):

trpc_agent_sdk/tools/safety/_scanner.py

@@ -1014,10 +1014,10 @@ def _strip_python_comment_line(line: str) -> str:
 
 
 def _is_in_echo_string(line: str, pattern: str) -> bool:
-    """Return True if *pattern* matches inside an echo/printf string literal.
+    """Return True if *pattern* matches are ALL inside echo/printf string literals.
 
-    In Bash, ``echo 'rm -rf /'`` is harmless — the dangerous command is just
-    printed, not executed.
+    In Bash, ``echo 'rm -rf /'`` is harmless.  But ``echo "rm -rf /"; rm -rf /``
+    is dangerous — the first match is harmless but the second is real.
     """
     stripped = line.strip()
     if not (stripped.startswith("echo ") or stripped.startswith("echo\t") or stripped.startswith("printf ")
@@ -1028,10 +1028,21 @@ def _is_in_echo_string(line: str, pattern: str) -> bool:
         pat = re.compile(pattern, re.IGNORECASE)
     except re.error:
         return False
+    in_quotes = False
     for m in re.finditer(r"'[^']*'", stripped):
         if pat.search(m.group(0)):
-            return True
-    for m in re.finditer(r'"[^"]*"', stripped):
-        if pat.search(m.group(0)):
-            return True
-    return False
+            in_quotes = True
+            break
+    if not in_quotes:
+        for m in re.finditer(r'"[^"]*"', stripped):
+            if pat.search(m.group(0)):
+                in_quotes = True
+                break
+    if not in_quotes:
+        return False
+    # Pattern found inside quotes — also check if it appears outside.
+    outside = re.sub(r"'[^']*'", " ", stripped)
+    outside = re.sub(r'"[^"]*"', " ", outside)
+    if pat.search(outside):
+        return False
+    return True