fix: pass tool safety lint checks

Author: impself

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -13,7 +13,6 @@
 from __future__ import annotations
 
 import re
-from typing import Iterator
 
 from trpc_agent_sdk.tools.safety._facts import (
     ConcurrencyFact,

trpc_agent_sdk/tools/safety/_cross_field_scanner.py

@@ -7,12 +7,9 @@
 
 from __future__ import annotations
 
-import os
 from typing import Iterable
 
-from trpc_agent_sdk.tools.safety._facts import Loc
 from trpc_agent_sdk.tools.safety._models import (
-    Evidence,
     RiskCategory,
     RiskLevel,
     SafetyDecision,

trpc_agent_sdk/tools/safety/_policy.py

@@ -24,7 +24,7 @@
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from trpc_agent_sdk.tools.safety._exceptions import SafetyPolicyError
-from trpc_agent_sdk.tools.safety._models import ScriptLanguage, ToolKind
+from trpc_agent_sdk.tools.safety._models import ScriptLanguage
 
 POLICY_VERSION = "1"
 

trpc_agent_sdk/tools/safety/_python_scanner.py

@@ -11,10 +11,7 @@
 from __future__ import annotations
 
 import ast
-import os
-from typing import Any, Iterator
 
-from trpc_agent_sdk.tools.safety._exceptions import SafetyScannerError
 from trpc_agent_sdk.tools.safety._facts import (
     ConcurrencyFact,
     DependencyInstallFact,
@@ -37,7 +34,6 @@
 )
 from trpc_agent_sdk.tools.safety._models import ScriptLanguage
 from trpc_agent_sdk.tools.safety._rules import _LanguageScannerRule, SafetyRule
-from trpc_agent_sdk.tools.safety._policy import is_sensitive_env_key
 from trpc_agent_sdk.tools.safety._redaction import contains_secret_literal
 
 # Networks libs and the attribute used to extract a host arg.
@@ -404,10 +400,6 @@ def _pathlib_write(self, node: ast.Call, canonical: str) -> FileWriteFact:
         if isinstance(node.func, ast.Attribute) \
                 and isinstance(node.func.value, ast.Call):
             target = _const_str(_first_arg(node.func.value)) or ""
-        size = None
-        if node.args:
-            data_arg = node.args[0] if node.args else None
-            size = self._static_size(data_arg)
         if not target:
             explicit = False
             target = "<dynamic>"
@@ -836,11 +828,10 @@ def _handle_sink_call(self, node: ast.Call, canonical: str) -> None:
         if sink_kind is None and isinstance(node.func, ast.Attribute):
             method = node.func.attr
             if method in {"info", "debug", "warning", "warn", "error", "critical", "exception", "log"}:
-                if isinstance(node.func.value, ast.Name) \
-                        and node.func.value.id.lower() in {"log", "logger",
-                                                              "logging"}:
-                    sink_kind = "output"
-                    sink_name = f"log.{method}"
+                if isinstance(node.func.value, ast.Name):
+                    if node.func.value.id.lower() in {"log", "logger", "logging"}:
+                        sink_kind = "output"
+                        sink_name = f"log.{method}"
         if sink_kind is None:
             return
         if not self._tainted_flows_into_call(node):

trpc_agent_sdk/tools/safety/_rules.py

@@ -12,29 +12,13 @@
 from __future__ import annotations
 
 import ipaddress
-from typing import Iterable, Protocol, Sequence, runtime_checkable
+from typing import Iterable, Protocol, runtime_checkable
 
 from trpc_agent_sdk.tools.safety._facts import (
     ConcurrencyFact,
-    DependencyInstallFact,
-    DynamicExecFact,
-    FileDeleteFact,
-    FileReadFact,
-    FileWriteFact,
-    ForkBombFact,
-    LargeWriteFact,
-    LongSleepFact,
-    NetworkFact,
-    ParseErrorFact,
-    PrivilegeFact,
-    ProcessFact,
     ScriptFacts,
-    SecretFlowFact,
-    ShellOperatorFact,
-    UnboundedLoopFact,
 )
 from trpc_agent_sdk.tools.safety._models import (
-    Evidence,
     RiskCategory,
     RiskLevel,
     SafetyDecision,

trpc_agent_sdk/tools/safety/_telemetry.py

@@ -171,8 +171,6 @@ def build_audit_event(
 ):
     """Build a :class:`SafetyAuditEvent` from a report."""
 
-    from trpc_agent_sdk.tools.safety._models import SafetyAuditEvent
-
     return SafetyAuditEvent(
         event_id=report.report_id,
         timestamp=timestamp,

trpc_agent_sdk/tools/safety/examples/samples/06_whitelist_network.py

@@ -10,7 +10,8 @@
 
 def fetch_user(handle: str) -> dict:
     response = requests.get(
-        f"https://api.github.com/users/{handle}", timeout=5,
+        f"https://api.github.com/users/{handle}",
+        timeout=5,
     )
     response.raise_for_status()
     return response.json()

trpc_agent_sdk/tools/safety/examples/samples/08_shell_injection.py

@@ -10,7 +10,9 @@
 
 def run_untrusted(user_input: str) -> int:
     return subprocess.run(
-        f"ls {user_input}", shell=True, check=False,
+        f"ls {user_input}",
+        shell=True,
+        check=False,
     ).returncode
 
 

trpc_agent_sdk/tools/safety/wrapper.py

@@ -19,12 +19,11 @@
 
 from __future__ import annotations
 
-import asyncio
 import inspect
 from typing import Any, Callable, Generic, Mapping, TypeVar
 
 from trpc_agent_sdk.tools.safety._audit import AuditSink, InMemoryAuditSink, NullAuditSink
-from trpc_agent_sdk.tools.safety._filter import BlockedExecutionError, ToolScriptSafetyFilter
+from trpc_agent_sdk.tools.safety._filter import ToolScriptSafetyFilter
 from trpc_agent_sdk.tools.safety._guard import ToolSafetyGuard
 from trpc_agent_sdk.tools.safety._models import (
     SafetyReport,
@@ -99,7 +98,7 @@ def __call__(self, *args: Any, **kwargs: Any) -> T:
             _ = report  # caller can inspect via last_report
 
     async def call_async(self, *args: Any, **kwargs: Any) -> T:
-        report = await self._enforce_async(args, kwargs)
+        await self._enforce_async(args, kwargs)
         result = self.delegate(*args, **kwargs)
         if inspect.isawaitable(result):
             return await result