fix: critical AST bypass + fail-open hardening + bash prefix coverage

  - _python_scanner: _resolve_canonical now checks _class_instances for
    Session().get() bypass (s=requests.Session(); s.get(url) now detected)
  - _python_scanner: _handle_assign propagates dangerous callable names
    (e=eval; e('x'), m=__import__('os'); m.system('id') now caught)
  - _scanner: rule exceptions now emit GLOBAL-003 sentinel finding instead
    of silent fail-open
  - _scanner: allow_patterns refuses upgrade when HIGH/CRITICAL findings exist
  - _scanner: reload_policy() also refreshes _rules
  - _bash_scanner: _PREFIX_CMDS extended with env/nohup/timeout/nice/xargs
  - _rules: _all_commands_whitelisted splits on ;/&&/||/& in addition to |

Author: lll-peanut

tests/test_tool_safety.py

@@ -1262,7 +1262,9 @@ def _bad_rule(script, scan_input, policy):
                 tool_name="rule_exc_test",
             ))
         # Should still complete despite the broken rule
-        assert report.decision == Decision.ALLOW
+        # Now produces GLOBAL-003 sentinel → MEDIUM → NEEDS_HUMAN_REVIEW
+        assert report.decision in (Decision.ALLOW, Decision.NEEDS_HUMAN_REVIEW)
+        assert any(f.rule_id == "GLOBAL-003" for f in report.findings)
     finally:
         # Clean up: remove the bad rule from registry
         from trpc_agent_sdk.tools.safety._rules import _EXTRA_RULES

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -276,9 +276,22 @@ def _analyse_one_command(self, line_no: int, raw_line: str, cmd_tokens: List[str
         if not cmd_tokens:
             return
         # Skip prefixes that don't change the real command:
-        #   VAR=val, export VAR=val, declare/readonly/local/typeset, command/builtin
+        #   VAR=val, export, env, nohup, timeout, nice, xargs, find, sudo variants
         idx = 0
-        _PREFIX_CMDS = frozenset({"export", "declare", "local", "readonly", "typeset", "command", "builtin"})
+        _PREFIX_CMDS = frozenset({
+            "export",
+            "declare",
+            "local",
+            "readonly",
+            "typeset",
+            "command",
+            "builtin",
+            "env",
+            "nohup",
+            "timeout",
+            "nice",
+            "xargs",
+        })
         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):

trpc_agent_sdk/tools/safety/_python_scanner.py

@@ -561,11 +561,26 @@ def _handle_assign(self, node: ast.Assign) -> None:
             if not var_name:
                 continue
 
-            # Check if RHS is an import
+            # Check if RHS is an import / class instantiation
             if isinstance(node.value, ast.Call):
                 canonical = self._resolve_canonical(node.value.func)
                 if canonical in _NETWORK_CALLS:
                     self._class_instances[var_name] = canonical
+                if canonical in _DYNAMIC_EXEC_CALLS or canonical in _PROCESS_CALLS:
+                    # e = eval; e("code") / s = os.system; s("id")
+                    self._aliases[var_name] = canonical
+
+            # Propagate bare-name assignments: e = eval; m = __import__
+            if isinstance(node.value, ast.Name):
+                src_canonical = self._resolve_canonical(node.value)
+                if src_canonical in _DYNAMIC_EXEC_CALLS or src_canonical in _PROCESS_CALLS:
+                    self._aliases[var_name] = src_canonical
+
+            # Propagate __import__ / importlib result
+            if isinstance(node.value, ast.Call):
+                inner = self._resolve_canonical(node.value.func)
+                if inner in ("__import__", "importlib.import_module"):
+                    self._aliases[var_name] = "__import__"
 
             # Check if RHS is from os.environ / os.getenv → taint
             # Only taint when the env key looks sensitive (KEY/TOKEN/SECRET/...)
@@ -720,7 +735,12 @@ def _resolve_canonical(self, node: ast.expr) -> str:
             ``getattr(x, 'y')`` → ``"getattr"``
         """
         if isinstance(node, ast.Name):
-            # Direct name lookup — consult alias table first
+            # Direct name lookup — consult class_instances first (so that
+            # s.get("http://evil.com") where s = requests.Session() resolves
+            # to "requests.Session.get"), then alias table, then bare name.
+            cls = self._class_instances.get(node.id)
+            if cls:
+                return cls
             return self._aliases.get(node.id, node.id)
 
         if isinstance(node, ast.Attribute):

trpc_agent_sdk/tools/safety/_rules.py

@@ -897,10 +897,15 @@ def _is_in_echo_string(line: str, pattern: str) -> bool:
 
 
 def _all_commands_whitelisted(line: str, policy: SafetyPolicy) -> bool:
-    """Return True if every command in a piped bash line is whitelisted."""
+    """Return True if every command in a line is whitelisted.
+
+    Splits on ``|``, ``;``, ``&&``, ``||``, and ``&`` so that commands
+    after non-pipe separators are also checked.
+    """
+    import re as _re
     cmds = []
-    for part in line.split("|"):
-        part = part.strip()
+    for part in _re.split(r"[|;&]", line):
+        part = part.strip().lstrip("&|")
         if part:
             first_word = part.split()[0] if part.split() else ""
             if first_word and not first_word.startswith("-"):

trpc_agent_sdk/tools/safety/_scanner.py

@@ -198,6 +198,17 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
                 all_findings.extend(findings)
             except Exception:  # pylint: disable=broad-except
                 logger.error("Safety rule raised an exception; skipping: %s", str(getattr(rule, "__class__", rule)))
+                all_findings.append(
+                    SafetyFinding(
+                        rule_id="GLOBAL-003",
+                        category=RiskCategory.RESOURCE_ABUSE,
+                        risk_level=RiskLevel.MEDIUM,
+                        evidence="A safety rule crashed during scanning.",
+                        message=f"Rule {getattr(rule, '__class__', rule)} failed — scan may be incomplete.",
+                        recommendation="Review the script manually; automated analysis was partial.",
+                        line_number=0,
+                        matched_pattern="",
+                    ))
 
         # Check environment variables against blocklist
         if scan_input.environment_variables:
@@ -270,8 +281,12 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
         # 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)
+        has_high_or_critical = any(f.risk_level in (RiskLevel.HIGH, RiskLevel.CRITICAL) for f in all_findings)
         allow_upgraded = False
-        if (decision == Decision.NEEDS_HUMAN_REVIEW and not has_critical and self._check_allow_patterns(script)):
+        # Only upgrade MEDIUM or lower (NEEDS_HUMAN_REVIEW) — never upgrade HIGH/CRITICAL
+        # even if the policy maps them to NEEDS_HUMAN_REVIEW.
+        if (decision == Decision.NEEDS_HUMAN_REVIEW and not has_high_or_critical
+                and self._check_allow_patterns(script)):
             logger.info("allow_patterns upgraded NEEDS_HUMAN_REVIEW → ALLOW for '%s'", scan_input.tool_name)
             decision = Decision.ALLOW
             allow_upgraded = True
@@ -321,8 +336,9 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
         )
 
     def reload_policy(self) -> None:
-        """Reload the policy from disk (useful for hot-reload)."""
+        """Reload the policy and rules from disk (useful for hot-reload)."""
         self._policy = reload_policy()
+        self._rules = get_all_rules()
 
     # ------------------------------------------------------------------
     # Layer 1: AST-based Python scanning