fix(safety): scan all tokens for privilege escalation; loop env-prefix in allowlist; localized redaction

Critical:
- _rules.ProcessRule._check_bash: privilege-escalation detection only
  checked tokens[0], so `FOO=bar sudo rm -rf /` and
  `echo x | sudo tee /etc/passwd` bypassed the CRITICAL privilege rule
  because the line-leading token was `FOO=bar` / `echo`, not `sudo`. The
  network-command side already scanned all tokens; privilege now does
  the same by collecting basenames of ALL tokens that hit
  _PRIVILEGE_CMDS. This closes a real CRITICAL-detection bypass.

Warning:
- _rules.ProcessRule._check_bash: strict_command_allowlist only stripped
  ONE leading KEY=VAL env-assignment token, so `FOO=bar A=b sudo ls`
  landed on `A=b` (has `=`) and stopped, missing `sudo`. Now loops
  through ALL consecutive leading KEY=VAL tokens (excluding subshell
  `(...)`). Same root cause as the privilege bypass above.

Suggestion:
- _scanner._redact_evidence: the `len > 80 AND 32+ char run` branch
  called redact(keep=8) on the WHOLE line, which truncated evidence to
  8 chars whenever a long identifier/URL-path happened to be present.
  This destroyed diagnostic context (e.g. requests.get('https://api.
  github.com/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa') became `requests`).
  Now applies re.sub(r"[A-Za-z0-9_\-]{32,}", "***", ...) so only the
  matched long token is replaced; the surrounding diagnostic context
  survives. Removed the now-unused `redact` import.

Tests:
- test_process_bash_sudo_with_env_prefix_is_caught: `FOO=bar sudo rm
  -rf /` must trigger privilege CRITICAL.
- test_process_bash_sudo_after_pipe_is_caught: `echo x | sudo tee
  /etc/passwd` must trigger privilege CRITICAL.
- test_process_bash_sudo_with_multiple_env_prefix_is_caught: `FOO=bar
  A=b sudo ls` must trigger privilege CRITICAL.
- test_strict_command_allowlist_multiple_env_prefix_is_fail_closed:
  `FOO=bar A=b ls` (whitelisted) passes; `FOO=bar A=b rm` (not
  whitelisted) is flagged.
- test_redact_evidence_preserves_diagnostic_context: long-URL evidence
  keeps `requests.get` + domain; only the 32-char token is ***.

Author: Violet2314

tests/tool_safety/test_rules.py

@@ -325,3 +325,57 @@ def test_resource_sleep_at_threshold_is_allowed_bash():
     inp = ScanInput(script="sleep 300\n", language="bash")
     findings = rule.check(inp, policy)
     assert not any("Long sleep" in f.metadata.get("message", "") for f in findings)
+
+
+def test_process_bash_sudo_with_env_prefix_is_caught():
+    """Regression for CongkeChen review: `FOO=bar sudo rm -rf /` used to bypass
+    the privilege rule because cmd=tokens[0]=`FOO=bar` was not in
+    _PRIVILEGE_CMDS. Now we scan ALL tokens for privilege-cmd basenames."""
+    rule = ProcessRule()
+    inp = ScanInput(script="FOO=bar sudo rm -rf /\n", language="bash")
+    findings = rule.check(inp, _policy())
+    priv = [f for f in findings if "Privilege escalation" in f.metadata.get("message", "")]
+    assert priv, "FOO=bar sudo ... must trigger privilege CRITICAL"
+    assert priv[0].risk_level == RiskLevel.CRITICAL
+
+
+def test_process_bash_sudo_after_pipe_is_caught():
+    """Regression for CongkeChen review: `echo x | sudo tee /etc/passwd` used
+    to bypass privilege because cmd=tokens[0]=`echo`. Now all tokens are
+    scanned."""
+    rule = ProcessRule()
+    inp = ScanInput(script="echo x | sudo tee /etc/passwd\n", language="bash")
+    findings = rule.check(inp, _policy())
+    priv = [f for f in findings if "Privilege escalation" in f.metadata.get("message", "")]
+    assert priv, "echo x | sudo ... must trigger privilege CRITICAL"
+
+
+def test_process_bash_sudo_with_multiple_env_prefix_is_caught():
+    """Regression for CongkeChen review: `FOO=bar A=b sudo ls` has TWO leading
+    env-assignment tokens; the old single-step skip landed on `A=b` and missed
+    `sudo`. Now strict allowlist loops through all KEY=VAL prefixes."""
+    rule = ProcessRule()
+    inp = ScanInput(script="FOO=bar A=b sudo ls\n", language="bash")
+    findings = rule.check(inp, _policy())
+    priv = [f for f in findings if "Privilege escalation" in f.metadata.get("message", "")]
+    assert priv, "FOO=bar A=b sudo ... must trigger privilege CRITICAL"
+
+
+def test_strict_command_allowlist_multiple_env_prefix_is_fail_closed():
+    """Regression for CongkeChen review: strict allowlist with multiple
+    leading env-assignment tokens must skip ALL of them and check the real
+    command, not stop at the first KEY=VAL token."""
+    from trpc_agent_sdk.safety._rules import ProcessRule
+    rule = ProcessRule()
+    policy = PolicyConfig(strict_command_allowlist=True, allowed_commands=["ls"])
+    # `FOO=bar A=b ls` should be ALLOWED (ls is whitelisted).
+    inp = ScanInput(script="FOO=bar A=b ls /tmp\n", language="bash")
+    findings = rule.check(inp, policy)
+    allow_list_findings = [f for f in findings if "allow-list" in f.metadata.get("message", "")]
+    assert not allow_list_findings, "ls is in allowed_commands; multi-env-prefix must still resolve to ls"
+
+    # `FOO=bar A=b rm` should be FLAGGED (rm not in allowed_commands).
+    inp2 = ScanInput(script="FOO=bar A=b rm /tmp\n", language="bash")
+    findings2 = rule.check(inp2, policy)
+    allow_list_findings2 = [f for f in findings2 if "allow-list" in f.metadata.get("message", "")]
+    assert allow_list_findings2, "rm is not in allowed_commands; multi-env-prefix must still flag rm"

tests/tool_safety/test_scanner.py

@@ -208,3 +208,19 @@ def test_register_custom_rule_is_picked_up_by_new_scanner():
         assert "TEST_CUSTOM_001" in rule_ids
     finally:
         unregister_custom_rule("TEST_CUSTOM_001")
+
+
+def test_redact_evidence_preserves_diagnostic_context():
+    """Regression for CongkeChen review: _redact_evidence used to truncate the
+    whole line to 8 chars when len>80 AND a 32+ char token was present, which
+    destroyed diagnostic context for evidence with long identifiers/URL paths.
+    Now only the matched 32+ token is replaced with ***, preserving the rest."""
+    from trpc_agent_sdk.safety._scanner import _redact_evidence
+    # Long URL with a 32-char path segment. The segment itself should be
+    # redacted to ***, but `requests.get('https://api.github.com/` and the
+    # closing `')` must survive so the operator can see what was flagged.
+    evidence = "requests.get('https://api.github.com/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')"
+    redacted = _redact_evidence(evidence)
+    assert "requests.get" in redacted, f"diagnostic context lost: {redacted!r}"
+    assert "api.github.com" in redacted, f"domain lost: {redacted!r}"
+    assert "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" not in redacted, f"long token not redacted: {redacted!r}"

trpc_agent_sdk/safety/_rules.py

@@ -852,17 +852,33 @@ def _check_bash(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Safet
             tokens = line.split()
             if not tokens:
                 continue
-            cmd = tokens[0].split("/")[-1]
+
+            # Scan ALL tokens for privilege-escalation commands, not just the
+            # first. `FOO=bar sudo rm -rf /` and `echo x | sudo tee /etc/passwd`
+            # would otherwise bypass the CRITICAL privilege rule because the
+            # line-leading token is `FOO=bar` / `echo`, not `sudo`. The network
+            # side already scans all tokens; privilege must too.
+            privilege_hits = [t.split("/")[-1].split("\\")[-1] for t in tokens
+                              if t.split("/")[-1].split("\\")[-1] in _PRIVILEGE_CMDS]
 
             if policy.strict_command_allowlist:
                 # Fail-closed: when strict mode is on but allowed_commands is
                 # empty, every non-builtin command is unknown and must be
                 # flagged HIGH. The previous `and policy.allowed_commands`
                 # guard turned strict mode into a no-op for empty allow-lists,
                 # letting rm/chmod slip through.
-                check_cmd = cmd
-                if "=" in check_cmd and len(tokens) > 1:
-                    check_cmd = tokens[1].split("/")[-1].split("\\")[-1]
+                #
+                # Skip ALL leading KEY=VAL env-assignment tokens, not just one.
+                # `A=b sudo ls` was caught, but `FOO=bar A=b sudo ls` slipped
+                # through because tokens[1]=="A=b" (has =) and the loop did not
+                # advance. Loop until the first non-assignment token.
+                check_idx = 0
+                while check_idx < len(tokens) - 1:
+                    tok = tokens[check_idx]
+                    if "=" not in tok or tok.startswith("("):
+                        break
+                    check_idx += 1
+                check_cmd = tokens[check_idx].split("/")[-1].split("\\")[-1] if tokens else ""
                 shell_builtins = {
                     "if",
                     "then",
@@ -884,7 +900,7 @@ def _check_bash(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Safet
                     "true",
                     "false",
                 }
-                if check_cmd not in shell_builtins and not policy.is_command_allowed(check_cmd):
+                if check_cmd and check_cmd not in shell_builtins and not policy.is_command_allowed(check_cmd):
                     findings.append(
                         self._finding(
                             line,
@@ -894,7 +910,7 @@ def _check_bash(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Safet
                             message=f"Command not in allow-list: {check_cmd}",
                         ))
 
-            if cmd in _PRIVILEGE_CMDS:
+            for cmd in privilege_hits:
                 findings.append(
                     self._finding(
                         line,

trpc_agent_sdk/safety/_scanner.py

@@ -16,7 +16,6 @@
 from ._policy import PolicyConfig
 from ._rules import SafetyRule
 from ._rules import default_rules
-from ._rules import redact
 from ._types import Decision
 from ._types import RiskLevel
 from ._types import SafetyFinding
@@ -209,7 +208,14 @@ def _run_rules(
 
 
 def _redact_evidence(text: str) -> str:
-    """Best-effort redaction of obvious secret tokens in evidence snippets."""
+    """Best-effort redaction of obvious secret tokens in evidence snippets.
+
+    Only redacts the matched secret substring, not the whole evidence line.
+    The previous `len > 80 and 32+ run` branch truncated the entire line to
+    8 chars, which destroyed diagnostic context for evidence that merely
+    contained a long identifier/URL path (e.g. requests.get('https://api.
+    github.com/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')).
+    """
     redacted = evidence_snippet(text)
     redacted = re.sub(r"(?i)bearer\s+[A-Za-z0-9_\-\.]{20,}", "bearer ***", redacted)
     redacted = re.sub(r"AKIA[0-9A-Z]{16}", "AKIA***", redacted)
@@ -219,7 +225,7 @@ def _redact_evidence(text: str) -> str:
         redacted,
         flags=re.IGNORECASE,
     )
-    # Also apply generic redact for long tokens
-    if len(redacted) > 80 and re.search(r"[A-Za-z0-9_\-]{32,}", redacted):
-        redacted = redact(redacted, keep=8)
+    # Localized replacement: only the 32+ char token itself becomes ***,
+    # not the whole line. Keeps the surrounding diagnostic context intact.
+    redacted = re.sub(r"[A-Za-z0-9_\-]{32,}", "***", redacted)
     return redacted