Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 38 additions & 8 deletions cachibot/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -112,17 +112,35 @@ def _setup_sandbox(self) -> None:
"""Set up the Python sandbox with restrictions."""
workspace = str(self.config.workspace_path)

# Get timeout from tool_configs or fall back to config defaults
# Get per-tool config or fall back to config defaults
timeout = self.config.sandbox.timeout_seconds
py_cfg: dict[str, Any] = {}
if self.tool_configs and "python_execute" in self.tool_configs:
py_cfg = self.tool_configs["python_execute"]
timeout = py_cfg.get("timeoutSeconds", timeout)

# Build path lists: workspace is always included, configs can only add
blocked_paths = set(py_cfg.get("blockedPaths", []))
read_paths = [workspace] + [
p for p in py_cfg.get("allowedReadPaths", []) if p not in blocked_paths
]
write_paths = [workspace] + [
p for p in py_cfg.get("allowedWritePaths", []) if p not in blocked_paths
]

# Build import lists: base allowed + extra, with separate blocked set
allowed_imports = list(self.config.sandbox.allowed_imports)
for mod in py_cfg.get("extraImports", []):
if mod not in allowed_imports:
allowed_imports.append(mod)
blocked_imports = py_cfg.get("blockedImports", []) or None
Comment on lines 119 to +136

Copilot AI Feb 28, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

self.tool_configs ultimately comes from unvalidated JSON (e.g., websocket payload). Here py_cfg.get("blockedPaths"), allowedReadPaths, etc. are assumed to be list-like; if a client sends null or a non-list (string/object), set(...) / list comprehensions will raise and prevent the agent from starting. Consider normalizing each config field (treat None as [], enforce list[str], drop non-strings) before building blocked_paths, read_paths, and write_paths (and similarly for import lists).

Suggested change
py_cfg = self.tool_configs["python_execute"]
timeout = py_cfg.get("timeoutSeconds", timeout)
# Build path lists: workspace is always included, configs can only add
blocked_paths = set(py_cfg.get("blockedPaths", []))
read_paths = [workspace] + [
p for p in py_cfg.get("allowedReadPaths", []) if p not in blocked_paths
]
write_paths = [workspace] + [
p for p in py_cfg.get("allowedWritePaths", []) if p not in blocked_paths
]
# Build import lists: base allowed + extra, with separate blocked set
allowed_imports = list(self.config.sandbox.allowed_imports)
for mod in py_cfg.get("extraImports", []):
if mod not in allowed_imports:
allowed_imports.append(mod)
blocked_imports = py_cfg.get("blockedImports", []) or None
raw_py_cfg = self.tool_configs["python_execute"]
if isinstance(raw_py_cfg, dict):
py_cfg = raw_py_cfg
timeout = py_cfg.get("timeoutSeconds", timeout)
def _normalize_str_list(value: Any) -> list[str]:
"""Normalize a potentially untrusted list-like config field to list[str]."""
if not isinstance(value, list):
return []
return [item for item in value if isinstance(item, str)]
# Build path lists: workspace is always included, configs can only add
blocked_paths = set(_normalize_str_list(py_cfg.get("blockedPaths")))
read_paths = [workspace] + [
p for p in _normalize_str_list(py_cfg.get("allowedReadPaths")) if p not in blocked_paths
]
write_paths = [workspace] + [
p for p in _normalize_str_list(py_cfg.get("allowedWritePaths")) if p not in blocked_paths
]
# Build import lists: base allowed + extra, with separate blocked set
allowed_imports = list(self.config.sandbox.allowed_imports)
for mod in _normalize_str_list(py_cfg.get("extraImports")):
if mod not in allowed_imports:
allowed_imports.append(mod)
blocked_imports_list = _normalize_str_list(py_cfg.get("blockedImports"))
blocked_imports = blocked_imports_list or None

Copilot uses AI. Check for mistakes.

self.sandbox = PythonSandbox(
allowed_imports=self.config.sandbox.allowed_imports,
allowed_imports=allowed_imports,
blocked_imports=blocked_imports,
timeout_seconds=timeout,
allowed_read_paths=[workspace],
allowed_write_paths=[workspace],
allowed_read_paths=read_paths,
allowed_write_paths=write_paths,
)
Comment on lines +115 to 144

Copilot AI Feb 28, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The per-tool sandbox configuration logic (blocked/allowed paths, extra/blocked imports) is new behavior that directly affects runtime safety and can break agent initialization if mis-specified. There are existing security tests in tests/test_security.py, but nothing currently exercises the new _setup_sandbox config handling; adding focused tests for these config knobs (e.g., blockedPaths filtering, extraImports inclusion, blockedImports passthrough) would help prevent regressions.

Copilot uses AI. Check for mistakes.

def _build_registry_from_plugins(self) -> None:
Expand Down Expand Up @@ -313,12 +331,12 @@ def _on_tool_end_with_capture(tool_name: str, result: Any) -> None:

self._agent = PromptureAgent(**agent_kwargs)

@staticmethod
def _harden_security_context(ctx: object) -> None:
def _harden_security_context(self, ctx: object) -> None:
"""Add security restrictions to prevent secret leakage from bot agents.

Blocks .env file access and dangerous shell commands that could
expose environment variables or secrets.
Blocks .env file access, dangerous shell commands that could
expose environment variables or secrets, and user-configured
blocked paths.
"""
# Block .env files via ignore patterns (matched against filename)
ignore = getattr(ctx, "ignore_patterns", None)
Expand All @@ -334,6 +352,18 @@ def _harden_security_context(ctx: object) -> None:
if cmd not in blocked:
blocked.append(cmd)

# Apply user-configured blocked paths to the security context
py_cfg: dict[str, Any] = {}
if self.tool_configs and "python_execute" in self.tool_configs:
py_cfg = self.tool_configs["python_execute"]
blocked_paths_cfg = py_cfg.get("blockedPaths", [])
if blocked_paths_cfg:
ctx_blocked = getattr(ctx, "blocked_paths", None)
if ctx_blocked is not None:
for p in blocked_paths_cfg:
if p not in ctx_blocked:
ctx_blocked.append(p)

# ------------------------------------------------------------------
# Auto-capture: file_write → Asset
# ------------------------------------------------------------------
Expand Down
104 changes: 101 additions & 3 deletions cachibot/plugins/file_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,12 @@
from pathlib import Path

from tukuy.manifest import PluginManifest, PluginRequirements
from tukuy.skill import RiskLevel, Skill, skill
from tukuy.skill import ConfigParam, RiskLevel, Skill, skill

from cachibot.plugins.base import CachibotPlugin, PluginContext

_ENCODING_OPTIONS = ["utf-8", "ascii", "latin-1", "utf-16"]


class FileOpsPlugin(CachibotPlugin):
"""Provides workspace-restricted file operation tools."""
Expand Down Expand Up @@ -48,6 +50,26 @@ def _build_skills(self) -> dict[str, Skill]:
display_name="Read File",
icon="file-text",
risk_level=RiskLevel.SAFE,
config_params=[
ConfigParam(
name="maxFileSize",
display_name="Max File Size",
description="Maximum file size allowed for reading.",
type="number",
default=10485760,
min=1024,
max=104857600,
unit="bytes",
),
ConfigParam(
name="encoding",
display_name="Encoding",
description="Text encoding used when reading files.",
type="select",
default="utf-8",
options=_ENCODING_OPTIONS,
),
],
)
def file_read(path: str) -> str:
"""Read the contents of a file.
Expand All @@ -62,7 +84,18 @@ def file_read(path: str) -> str:
if not ctx.config.is_path_allowed(full_path):
return f"Error: Path '{path}' is outside the workspace"
try:
return ctx.sandbox.read_file(str(full_path)) # type: ignore[no-any-return]
cfg = ctx.tool_configs.get("file_read", {})
max_size: int = cfg.get("maxFileSize", 10485760)
encoding: str = cfg.get("encoding", "utf-8")

file_size = full_path.stat().st_size
if file_size > max_size:
return (
f"Error: File '{path}' is {file_size} bytes, "
f"exceeds limit of {max_size} bytes"
)

return ctx.sandbox.read_file(str(full_path), encoding=encoding) # type: ignore[no-any-return]
except Exception as e:
return f"Error reading file: {e}"

Expand All @@ -76,6 +109,24 @@ def file_read(path: str) -> str:
display_name="Write File",
icon="file-pen",
risk_level=RiskLevel.MODERATE,
config_params=[
ConfigParam(
name="encoding",
display_name="Encoding",
description="Text encoding used when writing files.",
type="select",
default="utf-8",
options=_ENCODING_OPTIONS,
),
ConfigParam(
name="allowedExtensions",
display_name="Allowed Extensions",
description="File extensions allowed for writing. Empty list allows all.",
type="string[]",
default=[],
item_placeholder=".txt, .json, .csv",
),
],
)
def file_write(path: str, content: str) -> str:
"""Write content to a file. Creates the file if it doesn't exist.
Expand All @@ -91,8 +142,18 @@ def file_write(path: str, content: str) -> str:
if not ctx.config.is_path_allowed(full_path):
return f"Error: Path '{path}' is outside the workspace"
try:
cfg = ctx.tool_configs.get("file_write", {})
encoding: str = cfg.get("encoding", "utf-8")
allowed_ext: list[str] = cfg.get("allowedExtensions", [])

if allowed_ext and full_path.suffix not in allowed_ext:
return (
f"Error: Extension '{full_path.suffix}' is not allowed. "
f"Allowed: {', '.join(allowed_ext)}"
Comment on lines +147 to +152

Copilot AI Feb 28, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

allowed_ext is assumed to be list[str], but tool configs are user-provided/unvalidated and could be a string or other type; in that case full_path.suffix not in allowed_ext becomes a substring check and can silently allow/deny unexpected extensions. Consider validating/coercing allowedExtensions to a list of strings (and normalizing to lowercase / trimming whitespace) before applying the restriction.

Suggested change
allowed_ext: list[str] = cfg.get("allowedExtensions", [])
if allowed_ext and full_path.suffix not in allowed_ext:
return (
f"Error: Extension '{full_path.suffix}' is not allowed. "
f"Allowed: {', '.join(allowed_ext)}"
# Normalize allowedExtensions config to a list of lowercase, dot-prefixed extensions.
raw_allowed_ext = cfg.get("allowedExtensions", [])
normalized_allowed_ext: list[str] = []
if isinstance(raw_allowed_ext, str):
raw_iter = [raw_allowed_ext]
elif isinstance(raw_allowed_ext, (list, tuple, set)):
raw_iter = raw_allowed_ext
else:
raw_iter = []
for ext in raw_iter:
if not isinstance(ext, str):
continue
cleaned = ext.strip().lower()
if not cleaned:
continue
if not cleaned.startswith("."):
cleaned = "." + cleaned
normalized_allowed_ext.append(cleaned)
suffix = full_path.suffix.lower()
if normalized_allowed_ext and suffix not in normalized_allowed_ext:
return (
f"Error: Extension '{full_path.suffix}' is not allowed. "
f"Allowed: {', '.join(normalized_allowed_ext)}"

Copilot uses AI. Check for mistakes.
)

full_path.parent.mkdir(parents=True, exist_ok=True)
ctx.sandbox.write_file(str(full_path), content)
ctx.sandbox.write_file(str(full_path), content, encoding=encoding)
return f"Successfully wrote to {path}"
except Exception as e:
return f"Error writing file: {e}"
Expand Down Expand Up @@ -154,6 +215,26 @@ def file_list(path: str = ".", pattern: str = "*", recursive: bool = False) -> s
display_name="Edit File",
icon="file-search",
risk_level=RiskLevel.MODERATE,
config_params=[
ConfigParam(
name="maxFileSize",
display_name="Max File Size",
description="Maximum file size allowed for editing.",
type="number",
default=10485760,
min=1024,
max=104857600,
unit="bytes",
),
ConfigParam(
name="allowedExtensions",
display_name="Allowed Extensions",
description="File extensions allowed for editing. Empty list allows all.",
type="string[]",
default=[],
item_placeholder=".txt, .json, .csv",
),
],
)
def file_edit(path: str, old_text: str, new_text: str) -> str:
"""Edit a file by replacing a specific string with new content.
Expand All @@ -170,6 +251,23 @@ def file_edit(path: str, old_text: str, new_text: str) -> str:
if not ctx.config.is_path_allowed(full_path):
return f"Error: Path '{path}' is outside the workspace"
try:
cfg = ctx.tool_configs.get("file_edit", {})
max_size: int = cfg.get("maxFileSize", 10485760)
allowed_ext: list[str] = cfg.get("allowedExtensions", [])

if allowed_ext and full_path.suffix not in allowed_ext:
return (
f"Error: Extension '{full_path.suffix}' is not allowed. "
f"Allowed: {', '.join(allowed_ext)}"
)

file_size = full_path.stat().st_size
if file_size > max_size:
return (
f"Error: File '{path}' is {file_size} bytes, "
f"exceeds limit of {max_size} bytes"
)

content = ctx.sandbox.read_file(str(full_path))
if old_text not in content:
return f"Error: Could not find the specified text in {path}"
Expand Down
47 changes: 47 additions & 0 deletions cachibot/plugins/python_sandbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,53 @@ def _build_skills(self) -> dict[str, Skill]:
step=1000,
unit="chars",
),
ConfigParam(
name="allowedReadPaths",
display_name="Extra Read Paths",
description="Additional directories the sandbox can read from "
"(workspace is always included).",
type="string[]",
default=[],
placeholder="/data/shared",
max_items=20,
),
ConfigParam(
name="allowedWritePaths",
display_name="Extra Write Paths",
description="Additional directories the sandbox can write to "
"(workspace is always included).",
type="string[]",
default=[],
placeholder="/data/output",
max_items=20,
),
ConfigParam(
name="blockedPaths",
display_name="Blocked Paths",
description="Directories forbidden from access, even if within allowed paths.",
type="string[]",
default=[],
placeholder="/etc/secrets",
max_items=20,
),
Comment on lines +68 to +96

Copilot AI Feb 28, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These array-valued ConfigParams set placeholder=..., but the frontend array editor uses itemPlaceholder (and ignores placeholder) for string[] / number[] inputs. As a result, the intended placeholder text won't render for these params. Consider using the array-specific field (item_placeholder / itemPlaceholder) instead of placeholder for string[] config params.

Copilot uses AI. Check for mistakes.
ConfigParam(
name="extraImports",
display_name="Extra Imports",
description="Additional Python modules to allow beyond the default safe set.",
type="string[]",
default=[],
placeholder="requests",
max_items=50,
),
ConfigParam(
name="blockedImports",
display_name="Blocked Imports",
description="Python modules to block, even if in the default safe set.",
type="string[]",
default=[],
placeholder="csv",
max_items=50,
),
],
)
def python_execute(code: str) -> str:
Expand Down
2 changes: 1 addition & 1 deletion cachibot/services/bot_creation_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ async def _extract_with_retry(
last_error: Exception | None = None
for attempt in range(max_retries):
try:
return await extract_with_model( # type: ignore[no-any-return]
return await extract_with_model(
model_cls,
text,
model_name,
Expand Down
Loading
Loading