style(safety): 修复 YAPF 代码格式化问题

Author: Stelquis

scripts/tool_safety_check.py

@@ -24,29 +24,25 @@
 def detect_script_type(path: str) -> ScriptType:
     """Detect script type from file extension."""
     ext = os.path.splitext(path)[1].lower()
-    if ext in (".py",):
+    if ext in (".py", ):
         return ScriptType.PYTHON
     if ext in (".sh", ".bash", ".zsh", ".ksh"):
         return ScriptType.BASH
     return ScriptType.UNKNOWN
 
 
 def main():
-    parser = argparse.ArgumentParser(
-        description="Tool Script Safety Check — scan scripts for security risks",
-    )
+    parser = argparse.ArgumentParser(description="Tool Script Safety Check — scan scripts for security risks", )
     parser.add_argument("path", nargs="?", help="Path to script file")
-    parser.add_argument("--type", "-t", choices=["auto", "bash", "python"],
-                        default="auto", help="Script type (default: auto-detect)")
-    parser.add_argument("--stdin", action="store_true",
-                        help="Read script from stdin")
-    parser.add_argument("--json", action="store_true",
-                        help="Output as JSON")
-    parser.add_argument("--policy", "-p",
-                        default="tool_safety_policy.yaml",
-                        help="Path to policy file")
-    parser.add_argument("--version", "-v", action="store_true",
-                        help="Show version")
+    parser.add_argument("--type",
+                        "-t",
+                        choices=["auto", "bash", "python"],
+                        default="auto",
+                        help="Script type (default: auto-detect)")
+    parser.add_argument("--stdin", action="store_true", help="Read script from stdin")
+    parser.add_argument("--json", action="store_true", help="Output as JSON")
+    parser.add_argument("--policy", "-p", default="tool_safety_policy.yaml", help="Path to policy file")
+    parser.add_argument("--version", "-v", action="store_true", help="Show version")
 
     args = parser.parse_args()
 
@@ -126,4 +122,4 @@ def _print_report(report):
 
 
 if __name__ == "__main__":
-    main()
+    main()

tests/tools/safety/test_filter.py

@@ -27,7 +27,6 @@
 from trpc_agent_sdk.tools.safety._types import RiskLevel
 from trpc_agent_sdk.tools.safety._wrapper import SafetyWrapper
 
-
 # ── SafetyFilter Tests ────────────────────────────────────────────────────
 
 
@@ -123,7 +122,10 @@ def test_filter_skips_empty_command(self):
         import asyncio
         asyncio.run(safety_filter._before(
             ctx=None,
-            req={"name": "test", "not_a_command": "value"},
+            req={
+                "name": "test",
+                "not_a_command": "value"
+            },
             rsp=rsp,
         ))
         assert rsp.is_continue is True
@@ -133,8 +135,7 @@ def test_safety_blocked_error_message(self):
         from trpc_agent_sdk.tools.safety._types import RuleMatch
         from trpc_agent_sdk.tools.safety._types import RiskCategory
 
-        r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION,
-                       RiskLevel.CRITICAL, "rm -rf /", 1, "Remove")
+        r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION, RiskLevel.CRITICAL, "rm -rf /", 1, "Remove")
         report = SafetyReport(SafetyDecision.DENY, RiskLevel.CRITICAL, [r])
         err = SafetyBlockedError("Bash", report)
         assert "Bash" in str(err)
@@ -266,4 +267,4 @@ def test_scan_only_safe_script(self):
             "rules": {},
         }))
         result = wrapper.scan_only("Bash", "ls -la", "bash")
-        assert result["decision"] in ("ALLOW", "NEEDS_HUMAN_REVIEW")
+        assert result["decision"] in ("ALLOW", "NEEDS_HUMAN_REVIEW")

tests/tools/safety/test_policy.py

@@ -15,17 +15,18 @@
 from trpc_agent_sdk.tools.safety._types import RiskLevel
 from trpc_agent_sdk.tools.safety._types import SafetyDecision
 
-
 # ── Fixtures ──────────────────────────────────────────────────────────────
 
 
 @pytest.fixture(scope="module")
 def policy() -> SafetyPolicy:
     """Load the default policy file for testing."""
     policy_path = os.path.join(
-        os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(
-            os.path.abspath(__file__))))),
-        "trpc_agent_sdk", "tools", "safety", "tool_safety_policy.yaml",
+        os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))),
+        "trpc_agent_sdk",
+        "tools",
+        "safety",
+        "tool_safety_policy.yaml",
     )
     return SafetyPolicy.from_file(policy_path)
 
@@ -123,4 +124,4 @@ def test_load_from_dict(self):
 
     def test_rule_not_found_returns_none(self, policy: SafetyPolicy):
         """get_rule for non-existent rule should return None."""
-        assert policy.get_rule("non_existent_rule") is None
+        assert policy.get_rule("non_existent_rule") is None

tests/tools/safety/test_scanner.py

@@ -26,17 +26,18 @@
 from trpc_agent_sdk.tools.safety._types import ScanInput
 from trpc_agent_sdk.tools.safety._types import ScriptType
 
-
 # ── Fixtures ──────────────────────────────────────────────────────────────
 
 
 @pytest.fixture(scope="module")
 def policy() -> SafetyPolicy:
     """Load the default policy file."""
     policy_path = os.path.join(
-        os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(
-            os.path.abspath(__file__))))),
-        "trpc_agent_sdk", "tools", "safety", "tool_safety_policy.yaml",
+        os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))),
+        "trpc_agent_sdk",
+        "tools",
+        "safety",
+        "tool_safety_policy.yaml",
     )
     return SafetyPolicy.from_file(policy_path)
 
@@ -63,8 +64,7 @@ def test_safetyreport_properties(self):
         """SafetyReport properties should work correctly."""
         from trpc_agent_sdk.tools.safety._types import RuleMatch
 
-        r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION,
-                       RiskLevel.CRITICAL, "rm -rf /", 1, "Remove")
+        r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION, RiskLevel.CRITICAL, "rm -rf /", 1, "Remove")
         report = SafetyReport(SafetyDecision.DENY, RiskLevel.CRITICAL, [r])
         assert report.is_blocked is True
         assert report.is_allowed is False
@@ -75,10 +75,11 @@ def test_safetyreport_to_dict(self):
         """SafetyReport.to_dict() should produce correct JSON structure."""
         from trpc_agent_sdk.tools.safety._types import RuleMatch
 
-        r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION,
-                       RiskLevel.CRITICAL, "rm -rf /", 1, "Remove")
-        report = SafetyReport(SafetyDecision.DENY, RiskLevel.CRITICAL, [r],
-                              tool_name="Bash", script_type=ScriptType.BASH)
+        r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION, RiskLevel.CRITICAL, "rm -rf /", 1, "Remove")
+        report = SafetyReport(SafetyDecision.DENY,
+                              RiskLevel.CRITICAL, [r],
+                              tool_name="Bash",
+                              script_type=ScriptType.BASH)
         d = report.to_dict()
         assert d["decision"] == "DENY"
         assert d["risk_level"] == "CRITICAL"
@@ -92,8 +93,7 @@ def test_auditevent_otel_attributes(self):
         """AuditEvent should produce 6 OTel attributes."""
         from trpc_agent_sdk.tools.safety._types import AuditEvent
 
-        event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23,
-                           False, True, "2026-07-20T00:00:00")
+        event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23, False, True, "2026-07-20T00:00:00")
         otel = event.to_otel_attributes()
         assert len(otel) == 6
         assert otel["tool.safety.decision"] == "DENY"
@@ -154,7 +154,7 @@ def test_bash_pipe_sensitive(self, scanner: SafetyScanner):
         """cat /etc/passwd | grep root should be detected."""
         report = scanner.scan(ScanInput("cat /etc/passwd | grep root", ScriptType.BASH))
         # Should detect sensitive file read
-        sensitive = [m for m in report.matches if m.rule_id in ("R002",)]
+        sensitive = [m for m in report.matches if m.rule_id in ("R002", )]
         assert len(sensitive) > 0 or report.decision == SafetyDecision.DENY
 
     def test_fork_bomb_detected(self, scanner: SafetyScanner):
@@ -283,8 +283,12 @@ def test_log_decision_creates_event(self):
         """AuditLogger.log_decision should create a valid event."""
         logger = AuditLogger()
         event = logger.log_decision(
-            tool_name="Bash", decision="DENY", risk_level="CRITICAL",
-            rule_id="R001", scan_duration_ms=1.23, blocked=True,
+            tool_name="Bash",
+            decision="DENY",
+            risk_level="CRITICAL",
+            rule_id="R001",
+            scan_duration_ms=1.23,
+            blocked=True,
         )
         assert event.tool_name == "Bash"
         assert event.decision == "DENY"
@@ -293,8 +297,7 @@ def test_log_decision_creates_event(self):
     def test_audit_event_to_dict(self):
         """AuditEvent.to_dict() should produce correct structure."""
         from trpc_agent_sdk.tools.safety._types import AuditEvent
-        event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23,
-                           False, True, "2026-07-20T00:00:00")
+        event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23, False, True, "2026-07-20T00:00:00")
         d = event.to_dict()
         assert d["tool_name"] == "Bash"
         assert d["decision"] == "DENY"
@@ -304,12 +307,14 @@ def test_audit_event_to_dict(self):
     def test_audit_event_otel_attributes(self):
         """AuditEvent should produce OTel-compatible attributes."""
         from trpc_agent_sdk.tools.safety._types import AuditEvent
-        event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23,
-                           False, True, "2026-07-20T00:00:00")
+        event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23, False, True, "2026-07-20T00:00:00")
         otel = event.to_otel_attributes()
         expected_keys = {
-            "tool.safety.decision", "tool.safety.risk_level",
-            "tool.safety.rule_id", "tool.safety.blocked",
-            "tool.safety.masked", "tool.safety.duration_ms",
+            "tool.safety.decision",
+            "tool.safety.risk_level",
+            "tool.safety.rule_id",
+            "tool.safety.blocked",
+            "tool.safety.masked",
+            "tool.safety.duration_ms",
         }
-        assert set(otel.keys()) == expected_keys
+        assert set(otel.keys()) == expected_keys

trpc_agent_sdk/tools/safety/_audit.py

@@ -79,14 +79,22 @@ def log_report(self, report: SafetyReport) -> AuditEvent:
         if report.is_blocked:
             logger.warning(
                 "Safety audit: tool=%s decision=%s risk=%s rules=%s duration=%.1fms blocked=%s",
-                event.tool_name, event.decision, event.risk_level,
-                event.rule_id, event.scan_duration_ms, event.blocked,
+                event.tool_name,
+                event.decision,
+                event.risk_level,
+                event.rule_id,
+                event.scan_duration_ms,
+                event.blocked,
             )
         else:
             logger.info(
                 "Safety audit: tool=%s decision=%s risk=%s rules=%s duration=%.1fms blocked=%s",
-                event.tool_name, event.decision, event.risk_level,
-                event.rule_id, event.scan_duration_ms, event.blocked,
+                event.tool_name,
+                event.decision,
+                event.risk_level,
+                event.rule_id,
+                event.scan_duration_ms,
+                event.blocked,
             )
 
         return event
@@ -140,7 +148,11 @@ def log_decision(
 
         logger.info(
             "Safety audit: tool=%s decision=%s risk=%s rules=%s blocked=%s",
-            tool_name, decision, risk_level, rule_id, blocked,
+            tool_name,
+            decision,
+            risk_level,
+            rule_id,
+            blocked,
         )
 
         return event
@@ -227,4 +239,4 @@ def _emit_otel_event(event: AuditEvent) -> None:
             if span and span.is_recording():
                 span.set_attributes(event.to_otel_attributes())
         except Exception:  # pylint: disable=broad-except
-            pass
+            pass

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -32,10 +32,8 @@ class BashScanner:
     """
 
     # ── Domain extraction patterns ──
-    _DOMAIN_RE = re.compile(
-        r"(?:https?://|ftp://)?([a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?"
-        r"(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+(?::\d+)?)"
-    )
+    _DOMAIN_RE = re.compile(r"(?:https?://|ftp://)?([a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?"
+                            r"(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+(?::\d+)?)")
 
     # ── Sensitive data patterns (for leak detection) ──
     _SENSITIVE_PATTERNS: list[tuple[str, str]] = [
@@ -109,15 +107,16 @@ def _scan_line(
                 if re.search(pattern, line, re.IGNORECASE):
                     # Determine risk category from rule name
                     category = self._rule_name_to_category(rule_name)
-                    matches.append(RuleMatch(
-                        rule_id=self._rule_name_to_id(rule_name),
-                        risk_category=category,
-                        risk_level=rule_cfg.risk_level,
-                        evidence=line[:200],
-                        line_number=line_no,
-                        recommendation=self._get_recommendation(category),
-                        masked=False,
-                    ))
+                    matches.append(
+                        RuleMatch(
+                            rule_id=self._rule_name_to_id(rule_name),
+                            risk_category=category,
+                            risk_level=rule_cfg.risk_level,
+                            evidence=line[:200],
+                            line_number=line_no,
+                            recommendation=self._get_recommendation(category),
+                            masked=False,
+                        ))
                     break  # One match per rule per line
 
             # Check domain whitelist for network egress rules
@@ -160,7 +159,7 @@ def _check_network_egress(
                     evidence=f"Non-whitelisted domain: {domain}",
                     line_number=0,
                     recommendation="Remove or replace with a whitelisted domain, "
-                                   "or add the domain to allowed_domains in the policy",
+                    "or add the domain to allowed_domains in the policy",
                     masked=False,
                 )
         return None
@@ -189,7 +188,7 @@ def _check_sensitive_leak(
                     evidence=f"Potential sensitive data leak: {name}",
                     line_number=line_no,
                     recommendation="Avoid writing secrets to files, logs, or network requests. "
-                                   "Use environment variables or a secrets manager instead.",
+                    "Use environment variables or a secrets manager instead.",
                     masked=True,
                 )
         return None
@@ -229,16 +228,13 @@ def _get_recommendation(category: RiskCategory) -> str:
         """Get a default recommendation for a risk category."""
         recommendations = {
             RiskCategory.DANGEROUS_FILE_OPERATION:
-                "Remove or replace this file operation. Avoid destructive commands like rm -rf.",
+            "Remove or replace this file operation. Avoid destructive commands like rm -rf.",
             RiskCategory.NETWORK_EGRESS:
-                "Remove network requests to non-whitelisted domains, or add the domain to the policy.",
-            RiskCategory.PROCESS_EXECUTION:
-                "Avoid executing system commands directly. Use safe APIs instead.",
+            "Remove network requests to non-whitelisted domains, or add the domain to the policy.",
+            RiskCategory.PROCESS_EXECUTION: "Avoid executing system commands directly. Use safe APIs instead.",
             RiskCategory.DEPENDENCY_INSTALLATION:
-                "Do not install packages during execution. Pre-install all dependencies.",
-            RiskCategory.RESOURCE_ABUSE:
-                "Avoid infinite loops, long sleeps, and resource-exhaustive patterns.",
-            RiskCategory.SENSITIVE_INFO_LEAK:
-                "Do not write secrets to files, logs, or network requests.",
+            "Do not install packages during execution. Pre-install all dependencies.",
+            RiskCategory.RESOURCE_ABUSE: "Avoid infinite loops, long sleeps, and resource-exhaustive patterns.",
+            RiskCategory.SENSITIVE_INFO_LEAK: "Do not write secrets to files, logs, or network requests.",
         }
-        return recommendations.get(category, "Review this line for potential security risks.")
+        return recommendations.get(category, "Review this line for potential security risks.")

trpc_agent_sdk/tools/safety/_policy.py

@@ -31,7 +31,6 @@
 from ._types import RiskLevel
 from ._types import SafetyDecision
 
-
 # ── Policy Data Classes ───────────────────────────────────────────────────
 
 
@@ -128,9 +127,7 @@ def from_dict(cls, raw: dict) -> SafetyPolicy:
         if global_settings:
             policy.max_timeout_seconds = int(global_settings.get("max_timeout_seconds", 300))
             policy.max_output_size_bytes = int(global_settings.get("max_output_size_bytes", 10 * 1024 * 1024))
-            policy.default_decision = _parse_decision(
-                global_settings.get("default_decision", "needs_human_review")
-            )
+            policy.default_decision = _parse_decision(global_settings.get("default_decision", "needs_human_review"))
 
         # ── Whitelists ──
         policy.allowed_domains = raw.get("allowed_domains", [])
@@ -280,4 +277,4 @@ def _glob_to_regex(pattern: str) -> str:
         else:
             regex += c
             i += 1
-    return regex
+    return regex