fix: address second-round code review issues in Tool Safety Guard

1. ENV-001 evidence leak: evidence field now stores ***REDACTED*** instead
   of the raw environment variable value (fixes secrets leaking into
   JSON reports, CLI output, and SafetyWrapper.last_report).

2. ENV-001 deduplication collapse: _deduplicate_findings now includes
   matched_pattern in the dedup key for line_number==0 findings, so
   multiple blocklisted env vars each produce independent findings
   instead of being collapsed into one.

3. _audit.py docstring: corrected misleading 'thread/process-safe' claim
   to 'thread-safe' — threading.Lock is not cross-process.

4. _python_scanner._handle_for: range() loop detection now handles
   2-arg range(start, stop) and 3-arg range(start, stop, step) forms
   in addition to the original 1-arg range(stop).

5. _bash_scanner._check_redirects: strip surrounding quotes from
   redirect targets to reduce false positives from bash test expression
   comparisons (e.g. [[ "a" > "/etc/passwd" ]]).

6. New tests: env value leak, env dedup collapse, allow_patterns vs
   blocklist conflict (both pattern and env-based).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: lll-peanut

tests/test_tool_safety.py

@@ -2347,3 +2347,82 @@ def test_set_safety_span_attributes_get_current_span_exception(monkeypatch):
 
     # Should not raise
     _telemetry.set_safety_span_attributes(report)
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# Tests for code-review fixes
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+def test_env_blocklist_no_value_leak_in_evidence():
+    """ENV-001 evidence must NOT contain the raw environment variable value."""
+    from trpc_agent_sdk.tools.safety._policy import SafetyPolicy
+    policy = SafetyPolicy(blocklist_env_vars=["AZURE_CLIENT_SECRET", "MY_SECRET_KEY"])
+    scanner = SafetyScanner(policy=policy)
+    report = scanner.scan(
+        SafetyScanInput(
+            script_content="echo hello",
+            script_type=ScriptType.BASH,
+            tool_name="env_leak_test",
+            environment_variables={
+                "AZURE_CLIENT_SECRET": "super-secret-real-value-12345",
+                "MY_SECRET_KEY": "another-secret-value-67890",
+            },
+        ))
+    env_findings = [f for f in report.findings if f.rule_id == "ENV-001"]
+    assert len(env_findings) >= 2, f"Expected ≥2 ENV-001 findings, got {len(env_findings)}"
+    for f in env_findings:
+        assert "super-secret-real-value-12345" not in f.evidence, \
+            f"Evidence leaked raw value: {f.evidence}"
+        assert "another-secret-value-67890" not in f.evidence, \
+            f"Evidence leaked raw value: {f.evidence}"
+        assert "***REDACTED***" in f.evidence, \
+            f"Evidence should contain REDACTED: {f.evidence}"
+
+
+def test_env_blocklist_multiple_not_collapsed():
+    """Multiple blocklist env vars must each produce an independent ENV-001 finding."""
+    from trpc_agent_sdk.tools.safety._policy import SafetyPolicy
+    policy = SafetyPolicy(blocklist_env_vars=[
+        "AWS_SECRET_ACCESS_KEY",
+        "GITHUB_TOKEN",
+        "DOCKER_PASSWORD",
+        "AZURE_CLIENT_ID",
+    ])
+    scanner = SafetyScanner(policy=policy)
+    report = scanner.scan(
+        SafetyScanInput(
+            script_content="echo hello",
+            script_type=ScriptType.BASH,
+            tool_name="env_collapse_test",
+            environment_variables={
+                "AWS_SECRET_ACCESS_KEY": "val1",
+                "GITHUB_TOKEN": "val2",
+                "DOCKER_PASSWORD": "val3",
+                "AZURE_CLIENT_ID": "val4",
+            },
+        ))
+    env_findings = [f for f in report.findings if f.rule_id == "ENV-001"]
+    assert len(env_findings) == 4, \
+        f"Expected 4 independent ENV-001 findings, got {len(env_findings)}. " \
+        f"matched_patterns: {[f.matched_pattern for f in env_findings]}"
+
+
+def test_allow_patterns_never_overrides_blocklist_deny():
+    """allow_patterns must NOT override DENY even when pattern matches the script."""
+    from trpc_agent_sdk.tools.safety._policy import SafetyPolicy
+    # Script matches BOTH a blocklist pattern AND an allow pattern
+    policy = SafetyPolicy(
+        blocklist_patterns=[r"rm\s+-rf\s+/tmp/x"],
+        allow_patterns=[r"rm\s+-rf\s+/tmp/x"],
+    )
+    scanner = SafetyScanner(policy=policy)
+    report = scanner.scan(
+        SafetyScanInput(
+            script_content="rm -rf /tmp/x",
+            script_type=ScriptType.BASH,
+            tool_name="blocklist_wins_test",
+        ))
+    # Blocklist must win — allow_patterns only upgrades NEEDS_HUMAN_REVIEW
+    assert report.decision == Decision.DENY, \
+        f"blocklist must win over allow_patterns, got {report.decision}"

trpc_agent_sdk/tools/safety/_audit.py

@@ -31,9 +31,9 @@
 
 _AUDIT_LOGGER = logging.getLogger("trpc_agent_sdk.tools.safety.audit")
 
-# Per-path locks for thread/process-safe concurrent writes.
-# Each JSONL file gets its own lock so different audit files can be written
-# concurrently without contention.
+# Per-path locks for thread-safe concurrent writes (threading.Lock is NOT
+# process-safe; multi-process deployments should use a dedicated audit daemon
+# or file-locking via fcntl/msvcrt).
 _FILE_LOCKS: dict[str, threading.Lock] = {}
 _FILE_LOCKS_LOCK = threading.Lock()
 

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -386,7 +386,9 @@ def _check_redirects(self, line_no: int, raw_line: str, tokens: List[str]) -> No
         # Look for > or >> patterns
         for i, t in enumerate(tokens):
             if t in (">", ">>", ">" + ">") and i + 1 < len(tokens):
-                target = tokens[i + 1]
+                # Strip surrounding quotes so that "/etc/passwd" in
+                # test expressions ([[ … ]]) is treated consistently.
+                target = tokens[i + 1].strip("'\"")
                 if _is_sensitive_path(target) or target.startswith("/dev/"):
                     self._findings.append(
                         BashScanFinding(
@@ -400,7 +402,7 @@ def _check_redirects(self, line_no: int, raw_line: str, tokens: List[str]) -> No
             # Also handle inline redirect: 2>/dev/null, >/etc/passwd
             if ">" in t and len(t) > 1:
                 parts = t.split(">", 1)
-                target = parts[1].strip()
+                target = parts[1].strip().strip("'\"")
                 if target and (_is_sensitive_path(target) or target.startswith("/dev/sd")):
                     self._findings.append(
                         BashScanFinding(

trpc_agent_sdk/tools/safety/_python_scanner.py

@@ -252,7 +252,6 @@ class PythonScanFinding:
     "importlib": "dynamic",
 }
 
-
 # ═══════════════════════════════════════════════════════════════════════════
 # Python AST Scanner
 # ═══════════════════════════════════════════════════════════════════════════
@@ -553,7 +552,7 @@ def _handle_assign(self, node: ast.Assign) -> None:
             # Reading HOME/USER/PATH is normal and should not trigger alerts.
             if isinstance(node.value, ast.Subscript):
                 canonical = self._resolve_canonical(node.value.value)
-                if canonical in ("os.environ",):
+                if canonical in ("os.environ", ):
                     key = self._get_subscript_key(node.value)
                     if key and _is_sensitive_env_key(key):
                         self._tainted[var_name] = f"os.environ:{key}"
@@ -600,8 +599,7 @@ def _handle_aug_assign(self, node: ast.AugAssign) -> None:
             if node.target.id in self._tainted:
                 pass  # already tainted
 
-    def _check_output_taint(self, node: ast.Call, canonical: str, line_no: int,
-                            evidence: str) -> None:
+    def _check_output_taint(self, node: ast.Call, canonical: str, line_no: int, evidence: str) -> None:
         """Check if a tainted variable is being printed/logged."""
         if len(node.args) < 1:
             return
@@ -654,20 +652,26 @@ def _handle_while(self, node: ast.While) -> None:
                 ))
 
     def _handle_for(self, node: ast.For) -> None:
-        """Detect range(very_large) patterns."""
+        """Detect range(very_large) patterns in 1-/2-/3-arg forms."""
         line_no = node.lineno or 0
         if isinstance(node.iter, ast.Call):
             canonical = self._resolve_canonical(node.iter.func)
-            if canonical == "range" and len(node.iter.args) == 1:
-                val = self._get_arg_value(node.iter, 0)
+            if canonical == "range" and 1 <= len(node.iter.args) <= 3:
+                # The stop-value argument index: arg 0 for range(stop),
+                # arg 1 for range(start, stop) or range(start, stop, step).
+                stop_idx = 0 if len(node.iter.args) == 1 else 1
+                val = self._get_arg_value(node.iter, stop_idx)
                 if isinstance(val, int) and val > 10_000_000:
                     self._findings.append(
                         PythonScanFinding(
                             kind="loop",
                             canonical_name="large_range",
                             line_number=line_no,
                             evidence=self._get_line(line_no),
-                            extra={"range_value": val},
+                            extra={
+                                "range_value": val,
+                                "arg_count": len(node.iter.args)
+                            },
                         ))
 
     def _handle_with(self, node: ast.With) -> None:
@@ -858,9 +862,7 @@ def has_python_call(findings: List[PythonScanFinding], canonical_prefix: str) ->
 
 def get_python_urls(findings: List[PythonScanFinding]) -> List[Tuple[str, str, int]]:
     """Return (url, domain, line_number) tuples for all URL findings."""
-    return [(f.extra.get("url", ""), f.extra.get("domain", ""), f.line_number)
-            for f in findings
-            if f.kind == "url"]
+    return [(f.extra.get("url", ""), f.extra.get("domain", ""), f.line_number) for f in findings if f.kind == "url"]
 
 
 def get_python_file_reads(findings: List[PythonScanFinding]) -> List[PythonScanFinding]:
@@ -900,8 +902,7 @@ def get_python_sleep(findings: List[PythonScanFinding]) -> List[PythonScanFindin
 
 def get_python_concurrency(findings: List[PythonScanFinding]) -> List[PythonScanFinding]:
     """Return all concurrency/fork findings."""
-    return [f for f in findings if f.kind == "fork" or
-            (f.kind == "call" and f.extra.get("risk") == "concurrency")]
+    return [f for f in findings if f.kind == "fork" or (f.kind == "call" and f.extra.get("risk") == "concurrency")]
 
 
 # ═══════════════════════════════════════════════════════════════════════════
@@ -926,8 +927,7 @@ def _is_credential_path(path: str) -> bool:
 _SECRET_LOOKALIKE_RE = re.compile(
     r"(?:sk-[a-zA-Z0-9]{20,}|ghp_[a-zA-Z0-9]{20,}|AKIA[0-9A-Z]{16}|"
     r"eyJ[a-zA-Z0-9_-]{10,}\.[a-zA-Z0-9_-]{10,}\.[a-zA-Z0-9_-]{10,}|"
-    r"xox[baprs]-[a-zA-Z0-9-]+|AIza[0-9A-Za-z\-_]{35})",
-)
+    r"xox[baprs]-[a-zA-Z0-9-]+|AIza[0-9A-Za-z\-_]{35})", )
 
 
 def _looks_like_secret(value: str) -> bool:

trpc_agent_sdk/tools/safety/_scanner.py

@@ -193,7 +193,7 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
                             rule_id="ENV-001",
                             category=RiskCategory.SENSITIVE_INFO_LEAK,
                             risk_level=RiskLevel.HIGH,
-                            evidence=f"env: {blocked_var}={scan_input.environment_variables[blocked_var][:50]}",
+                            evidence=f"env: {blocked_var}=***REDACTED***",
                             message=f"Blocklisted environment variable set: {blocked_var}",
                             recommendation="Do not pass sensitive environment variables to tools.",
                             line_number=0,
@@ -873,10 +873,18 @@ def quick_scan(
 
 
 def _deduplicate_findings(findings: List[SafetyFinding]) -> List[SafetyFinding]:
-    """Remove duplicates: keep only the highest-risk finding per (rule_id, line_number)."""
-    seen: dict[tuple[str, int], SafetyFinding] = {}
+    """Remove duplicates: keep only the highest-risk finding per (rule_id, line_number).
+
+    For line_number==0 findings (e.g. ENV-001 for blocklisted env vars), the
+    ``matched_pattern`` is included in the key so that multiple distinct
+    hits are not collapsed into a single record.
+    """
+    seen: dict[tuple[str, int, str], SafetyFinding] = {}
     for f in findings:
-        key = (f.rule_id, f.line_number)
+        # Include matched_pattern for line-0 findings to avoid collapsing
+        # distinct hits (e.g. multiple blocklisted environment variables).
+        discriminator = f.matched_pattern if f.line_number == 0 else ""
+        key = (f.rule_id, f.line_number, discriminator)
         if key not in seen or f.risk_level > seen[key].risk_level:
             seen[key] = f
     return list(seen.values())