nanobot 源码解读 · 05 | 工具系统与安全边界：Agent 的手脚与护栏

# nanobot/agent/tools/schema.py:50
@staticmethod
def validate_json_schema_value(val: Any, schema: dict[str, Any], path: str = "") -> list[str]:
    """返回错误列表（空 = 合法）。"""
    raw_type = schema.get("type")
    nullable = (isinstance(raw_type, list) and "null" in raw_type) or schema.get("nullable", False)
    t = Schema.resolve_json_schema_type(raw_type)   # ['string','null'] -> 'string'
    label = path or "parameter"

    if nullable and val is None:
        return []
    if t == "integer" and (not isinstance(val, int) or isinstance(val, bool)):
        return [f"{label} should be integer"]
    # ... number / string(min/maxLength) / enum / object(required) / array(min/maxItems) ...
    return errors

# nanobot/agent/tools/base.py:141
class ToolResult(str):
    """带结构化状态的字符串型工具输出。"""
    is_error: bool
    def __new__(cls, content: str, *, is_error: bool = False) -> ToolResult:
        obj = str.__new__(cls, content)
        obj.is_error = is_error
        return obj
    @classmethod
    def error(cls, content: str) -> ToolResult:
        return cls(content, is_error=True)

# nanobot/agent/tools/base.py:186
@property
def read_only(self) -> bool:
    """是否无副作用、可安全并行。"""
    return False

@property
def concurrency_safe(self) -> bool:
    """是否可与其他 concurrency_safe 工具一起跑。"""
    return self.read_only and not self.exclusive

@property
def exclusive(self) -> bool:
    """即使开了并发，这个工具也必须单独跑。"""
    return False

# nanobot/agent/tools/base.py:248
def cast_params(self, params: dict[str, Any]) -> dict[str, Any]:
    """执行前做安全的 schema 驱动类型转换。"""
    ...
def validate_params(self, params: dict[str, Any]) -> list[str]:
    """按 JSON schema 校验；空列表 = 合法。"""
    ...

# nanobot/agent/tools/base.py:303
def to_schema(self) -> dict[str, Any]:
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        },
    }

# nanobot/agent/tools/registry.py:86
def get_definitions(self) -> list[dict[str, Any]]:
    """稳定排序，方便缓存友好地拼进 Prompt。"""
    if self._cached_definitions is not None:
        return self._cached_definitions
    definitions = [tool.to_schema() for tool in self._tools.values()]
    builtins: list[dict] = []
    mcp_tools: list[dict] = []
    for schema in definitions:
        name = self._schema_name(schema)
        if name.startswith("mcp_"):
            mcp_tools.append(schema)      # MCP 工具排在后面
        else:
            builtins.append(schema)        # 内置工具排前面，作为稳定前缀
    builtins.sort(key=self._schema_name)
    mcp_tools.sort(key=self._schema_name)
    self._cached_definitions = builtins + mcp_tools
    return self._cached_definitions

# nanobot/agent/tools/registry.py:111
def prepare_call(self, name, params) -> tuple[Tool | None, Any, str | None]:
    """解析、转换、校验一次工具调用。"""
    tool = self._tools.get(name)
    if not tool:
        suggestion = self._suggest_name(str(name))   # 名字拼错时给建议
        hint = f" Did you mean '{suggestion}'? ..." if suggestion else ""
        return None, params, ToolResult.error(f"Error: Tool '{name}' not found.{hint} ...")
    # 兼容老式 ContextAware 工具的上下文注入（内置工具直接读 ContextVar）
    if isinstance(tool, ContextAware) and (ctx := current_request_context()) is not None:
        tool.set_context(ctx)
    params = self._coerce_params(tool, params)       # 把 "{'a': 1}" 字符串解析成 dict
    if not isinstance(params, dict):
        return tool, params, ToolResult.error(f"Error: parameters must be a JSON object ...")
    cast_params = tool.cast_params(cast(dict, params))
    errors = tool.validate_params(cast_params)
    if errors:
        return tool, cast_params, ToolResult.error(f"Error: Invalid parameters for tool '{name}': " + "; ".join(errors))
    return tool, cast_params, None

# nanobot/agent/tools/loader.py:36
def discover(self) -> list[type[Tool]]:
    seen: set[int] = set()
    results: list[type[Tool]] = []
    for _importer, module_name, _ispkg in pkgutil.iter_modules(self._package.__path__):
        if module_name.startswith("_") or module_name in _SKIP_MODULES:   # 跳过 base/schema/...
            continue
        module = importlib.import_module(f".{module_name}", self._package.__name__)
        for attr_name in dir(module):
            attr = getattr(module, attr_name)
            if (isinstance(attr, type) and issubclass(attr, Tool) and attr is not Tool
                    and not attr_name.startswith("_")
                    and not getattr(attr, "__abstractmethods__", None)
                    and getattr(attr, "_plugin_discoverable", True)
                    and id(attr) not in seen):
                seen.add(id(attr))
                results.append(attr)
    results.sort(key=lambda cls: cls.__name__)
    return results

# nanobot/agent/tools/loader.py:68
def _discover_plugins(self) -> dict[str, type[Tool]]:
    """从 entry_points('nanobot.tools') 发现外部工具插件。"""
    eps = entry_points(group="nanobot.tools")
    for ep in eps:
        cls = ep.load()
        if isinstance(cls, type) and issubclass(cls, Tool) and ...:
            plugins[ep.name] = cls
    return plugins

# nanobot/agent/tools/shell.py:298
async def execute(self, command=None, cmd=None, working_dir=None, workdir=None,
                  timeout=None, shell=None, login=None, yield_time_ms=None,
                  max_output_chars=None, max_output_tokens=None, **kwargs) -> str:
    command = command or cmd
    working_dir = working_dir or workdir
    if not command:
        return ToolResult.error("Error: Missing command. Provide command or cmd.")
    prepared = self._prepare_command(command, working_dir, timeout, shell, login)
    if isinstance(prepared, str):
        return prepared                       # 准备阶段就报错（如护栏拦截）
    # ... 启动子进程、wait_for 超时、kill、输出截断 ...
    if len(result) > max_len:
        half = max_len // 2
        result = (result[:half] + f"\n\n... ({len(result)-max_len:,} chars truncated) ...\n\n" + result[-half:])
    return result

# nanobot/agent/tools/shell.py:785
segments = self._split_shell_segments(lower)        # 按 && / | / ; 切分顶层命令段
explicitly_allowed = bool(self.allow_patterns) and bool(segments) and all(
    any(re.fullmatch(pattern, segment) for pattern in self.allow_patterns)
    for segment in segments
)
if not explicitly_allowed:
    for pattern in self.deny_patterns:
        if re.search(pattern, lower):
            return ToolResult.error("Error: Command blocked by deny pattern filter")
    if self.allow_patterns:
        return ToolResult.error("Error: Command blocked by allowlist filter (not in allowlist)")

# nanobot/agent/tools/shell.py:798
from nanobot.security.network import contains_internal_url
if contains_internal_url(cmd, allow_loopback=current_scope_allows_loopback(...)):
    return ToolResult.error("Error: Command blocked by safety guard (internal/private URL detected)")

# nanobot/agent/tools/shell.py:810
if should_restrict:
    if "..\\" in cmd or "../" in cmd:
        return ToolResult.error("Error: Command blocked by safety guard (path traversal detected)" + _WORKSPACE_BOUNDARY_NOTE)

# nanobot/agent/tools/filesystem.py:86
agent_workspace = Path(ctx.workspace)
restrict = ctx.config.restrict_to_workspace or ctx.config.exec.sandbox
allowed_dir = agent_workspace if restrict else None
return cls(
    workspace=agent_workspace,
    allowed_dir=allowed_dir,
    extra_read_allowed_dirs=[BUILTIN_SKILLS_DIR, resolved_agent_workspace / "skills"],
    extra_read_allowed_files=[resolved_agent_workspace / "memory" / "history.jsonl"],
    restrict_to_workspace=ctx.config.restrict_to_workspace,
    sandbox_restricts_workspace=sandbox_restricts,
)

# nanobot/agent/tools/filesystem.py:150
def _resolve_read(self, path: str) -> Path:
    return self._resolve_with_extra(path, self._extra_read_allowed_dirs,
                                     self._extra_read_allowed_files,
                                     include_media_dir=True, extra_files_require_allowed_root=True)

def _resolve_write(self, path: str) -> Path:
    return self._resolve_with_extra(path, self._extra_write_allowed_dirs,
                                     self._extra_write_allowed_files, include_media_dir=False)

# nanobot/security/workspace_policy.py:96
def resolve_allowed_path(path, *, workspace=None, allowed_root=None,
                          extra_allowed_roots=None, extra_allowed_files=None, strict=False) -> Path:
    resolved = resolve_path(path, workspace, strict=False)
    roots = []
    if allowed_root is not None:
        roots.append(allowed_root)
    roots.extend(extra_allowed_roots or [])
    exact_allowed = bool(files) and _is_path_exactly_allowed(...)
    if not is_path_allowed(resolved, roots) and not exact_allowed:
        boundary = Path(allowed_root).expanduser() if allowed_root is not None else "allowed files"
        raise WorkspaceBoundaryError(
            f"Path {path} is outside allowed directory {boundary}" + WORKSPACE_BOUNDARY_NOTE
        )
    return resolved

_BLOCKED_DEVICE_PATHS = frozenset({
    "/dev/zero", "/dev/random", "/dev/urandom", "/dev/full",
    "/dev/stdin", "/dev/stdout", "/dev/stderr",
    "/dev/tty", "/dev/console", "/dev/fd/0", "/dev/fd/1", "/dev/fd/2",
})

# nanobot/agent/tools/web.py:111
def _validate_url_safe(url: str) -> tuple[bool, str]:
    """带 SSRF 防护的 URL 校验：scheme + 域名 + 解析后的 IP 检查。"""
    from nanobot.security.network import validate_url_target
    return validate_url_target(url)

_BLOCKED_NETWORKS = [
    ipaddress.ip_network("0.0.0.0/8"),
    ipaddress.ip_network("10.0.0.0/8"),          # 私有 A 类
    ipaddress.ip_network("100.64.0.0/10"),       # 运营商级 NAT
    ipaddress.ip_network("127.0.0.0/8"),         # 回环
    ipaddress.ip_network("169.254.0.0/16"),      # 链路本地 / 云元数据
    ipaddress.ip_network("172.16.0.0/12"),       # 私有 B 类
    ipaddress.ip_network("192.168.0.0/16"),      # 私有 C 类
    ipaddress.ip_network("::/128"), ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"), ipaddress.ip_network("fe80::/10"),
]

# nanobot/agent/tools/web.py:118
def _resolve_url_safe(url):
    """校验 URL 并返回要钉死的已解析 IP。"""
    return resolve_url_target(url)     # 返回 (ok, error, resolved_ips)

# network.py:264
class PinnedDNSAsyncTransport(httpx.AsyncBaseTransport):
    async def handle_async_request(self, request):
        url = str(request.url)
        ok, error, resolved_ips = resolve_url_target(url, allow_loopback=self._allow_loopback)
        if not ok:
            raise UnsafeURLRequestError(error, request=request)
        async with self._resolver_lock:
            with pin_resolved_url_dns(url, resolved_ips):   # 临时把该域名的 DNS 固定到已验证 IP
                return await self._inner.handle_async_request(request)

# nanobot/agent/tools/web.py:197
for _ in range(MAX_REDIRECTS + 1):
    is_valid, error_msg, _ = _resolve_url_safe(current_url)
    if not is_valid:
        return None, None, f"Redirect blocked: {error_msg}"
    stream = client.stream("GET", current_url, follow_redirects=False)   # 手动控制重定向
    response = await stream.__aenter__()
    is_redirect = 300 <= response.status_code < 400
    if not is_redirect:
        return response, stream, None
    next_url = urljoin(str(response.url), response.headers.get("location"))
    is_valid, error_msg = _validate_url_safe(next_url)                   # 下一跳也要过 SSRF
    if not is_valid:
        await stream.__aexit__(None, None, None)
        return None, None, f"Redirect blocked: {error_msg}"

# nanobot/agent/tools/runner.py:1556  (_execute_tools)
batches = self._partition_tool_batches(spec, tool_calls)
for batch in batches:
    if spec.concurrent_tools and len(batch) > 1:
        batch_results = await asyncio.gather(*(self._run_tool(spec, tc, ...) for tc in batch))
    else:
        for tool_call in batch:
            result = await self._run_tool(spec, tool_call, ...)
            tool_results.append(result)

# nanobot/agent/runner.py:1643
lookup_error = repeated_external_lookup_error(tool_call.name, tool_call.arguments, external_lookup_counts)
if lookup_error:
    # 模型反复查同一个外部资源（如反复 fetch 同一无效 URL），直接拦截防死循环
    ...

tool, params, prep_error = None, tool_call.arguments, None
if callable(prepare_call):
    prepared = prepare_call(tool_call.name, tool_call.arguments)   # registry 里解析+校验
    ...
if prep_error:
    handled = self._classify_violation(raw_text=prep_error, soft_payload=prep_error + hint,
                                       event=event, tool_call=tool_call,
                                       workspace_violation_counts=workspace_violation_counts)
    if handled is not None:
        return handled
    return prep_error + hint, event, (RuntimeError(prep_error) if spec.fail_on_tool_error else None)

await hook.before_execute_tool(context, tool_call, tool, params)
try:
    if tool is not None:
        result = await tool.execute(**params)
    else:
        result = await spec.tools.execute(tool_call.name, params)
except Exception as exc:
    handled = self._classify_violation(raw_text=str(exc), soft_payload=..., event=event, ...)
    if handled is not None:
        return handled
    ...

"""Workspace path boundary helpers.
These helpers are application-level guards.  They make path decisions
consistent across tools, but they are not a replacement for an OS sandbox.
"""