refactor(security): lift restrictToWorkspace to tools level

This commit is contained in:
Re-bin 2026-02-06 09:28:08 +00:00
parent b1782814fa
commit 943579b96a
6 changed files with 24 additions and 6 deletions

View File

@ -16,7 +16,7 @@
⚡️ Delivers core agent functionality in just **~4,000** lines of code — **99% smaller** than Clawdbot's 430k+ lines.
📏 Real-time line count: **3,412 lines** (run `bash core_agent_lines.sh` to verify anytime)
📏 Real-time line count: **3,428 lines** (run `bash core_agent_lines.sh` to verify anytime)
## 📢 News

View File

@ -44,6 +44,7 @@ class AgentLoop:
brave_api_key: str | None = None,
exec_config: "ExecToolConfig | None" = None,
cron_service: "CronService | None" = None,
restrict_to_workspace: bool = False,
):
from nanobot.config.schema import ExecToolConfig
from nanobot.cron.service import CronService
@ -55,6 +56,7 @@ class AgentLoop:
self.brave_api_key = brave_api_key
self.exec_config = exec_config or ExecToolConfig()
self.cron_service = cron_service
self.restrict_to_workspace = restrict_to_workspace
self.context = ContextBuilder(workspace)
self.sessions = SessionManager(workspace)
@ -66,6 +68,7 @@ class AgentLoop:
model=self.model,
brave_api_key=brave_api_key,
exec_config=self.exec_config,
restrict_to_workspace=restrict_to_workspace,
)
self._running = False
@ -74,7 +77,7 @@ class AgentLoop:
def _register_default_tools(self) -> None:
"""Register the default set of tools."""
# File tools (restrict to workspace if configured)
allowed_dir = self.workspace if self.exec_config.restrict_to_workspace else None
allowed_dir = self.workspace if self.restrict_to_workspace else None
self.tools.register(ReadFileTool(allowed_dir=allowed_dir))
self.tools.register(WriteFileTool(allowed_dir=allowed_dir))
self.tools.register(EditFileTool(allowed_dir=allowed_dir))
@ -84,7 +87,7 @@ class AgentLoop:
self.tools.register(ExecTool(
working_dir=str(self.workspace),
timeout=self.exec_config.timeout,
restrict_to_workspace=self.exec_config.restrict_to_workspace,
restrict_to_workspace=self.restrict_to_workspace,
))
# Web tools

View File

@ -34,6 +34,7 @@ class SubagentManager:
model: str | None = None,
brave_api_key: str | None = None,
exec_config: "ExecToolConfig | None" = None,
restrict_to_workspace: bool = False,
):
from nanobot.config.schema import ExecToolConfig
self.provider = provider
@ -42,6 +43,7 @@ class SubagentManager:
self.model = model or provider.get_default_model()
self.brave_api_key = brave_api_key
self.exec_config = exec_config or ExecToolConfig()
self.restrict_to_workspace = restrict_to_workspace
self._running_tasks: dict[str, asyncio.Task[None]] = {}
async def spawn(
@ -96,14 +98,14 @@ class SubagentManager:
try:
# Build subagent tools (no message tool, no spawn tool)
tools = ToolRegistry()
allowed_dir = self.workspace if self.exec_config.restrict_to_workspace else None
allowed_dir = self.workspace if self.restrict_to_workspace else None
tools.register(ReadFileTool(allowed_dir=allowed_dir))
tools.register(WriteFileTool(allowed_dir=allowed_dir))
tools.register(ListDirTool(allowed_dir=allowed_dir))
tools.register(ExecTool(
working_dir=str(self.workspace),
timeout=self.exec_config.timeout,
restrict_to_workspace=self.exec_config.restrict_to_workspace,
restrict_to_workspace=self.restrict_to_workspace,
))
tools.register(WebSearchTool(api_key=self.brave_api_key))
tools.register(WebFetchTool())

View File

@ -209,6 +209,7 @@ def gateway(
brave_api_key=config.tools.web.search.api_key or None,
exec_config=config.tools.exec,
cron_service=cron,
restrict_to_workspace=config.tools.restrict_to_workspace,
)
# Set cron callback (needs agent)
@ -316,6 +317,7 @@ def agent(
workspace=config.workspace_path,
brave_api_key=config.tools.web.search.api_key or None,
exec_config=config.tools.exec,
restrict_to_workspace=config.tools.restrict_to_workspace,
)
if message:

View File

@ -34,6 +34,7 @@ def load_config(config_path: Path | None = None) -> Config:
try:
with open(path) as f:
data = json.load(f)
data = _migrate_config(data)
return Config.model_validate(convert_keys(data))
except (json.JSONDecodeError, ValueError) as e:
print(f"Warning: Failed to load config from {path}: {e}")
@ -61,6 +62,16 @@ def save_config(config: Config, config_path: Path | None = None) -> None:
json.dump(data, f, indent=2)
def _migrate_config(data: dict) -> dict:
"""Migrate old config formats to current."""
# Move tools.exec.restrictToWorkspace → tools.restrictToWorkspace
tools = data.get("tools", {})
exec_cfg = tools.get("exec", {})
if "restrictToWorkspace" in exec_cfg and "restrictToWorkspace" not in tools:
tools["restrictToWorkspace"] = exec_cfg.pop("restrictToWorkspace")
return data
def convert_keys(data: Any) -> Any:
"""Convert camelCase keys to snake_case for Pydantic."""
if isinstance(data, dict):

View File

@ -100,13 +100,13 @@ class WebToolsConfig(BaseModel):
class ExecToolConfig(BaseModel):
"""Shell exec tool configuration."""
timeout: int = 60
restrict_to_workspace: bool = False # If true, block commands accessing paths outside workspace
class ToolsConfig(BaseModel):
"""Tools configuration."""
web: WebToolsConfig = Field(default_factory=WebToolsConfig)
exec: ExecToolConfig = Field(default_factory=ExecToolConfig)
restrict_to_workspace: bool = False # If true, restrict all tool access to workspace directory
class Config(BaseSettings):