feat(webui): add project workspaces and access controls (#4007)
* feat(webui): add project workspaces and access controls * feat(webui): add project workspaces and access controls * refactor(tools): centralize workspace access resolution * refactor(webui): remove unused workspace host state * fix(webui): hide estimated file edit label * fix(webui): clarify file edit deletion feedback * fix(webui): label deleted file activity * fix(webui): flatten file edit activity rows * fix(core): remove path-only patch deletion * fix(core): keep apply patch non-destructive * refactor(webui): trim workspace host plumbing * fix(tools): register exec with tools config
This commit is contained in:
Xubin Ren
@@ -42,9 +42,14 @@ def _is_private(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
|
||||
return any(addr in net for net in _BLOCKED_NETWORKS)
|
||||
|
||||
|
||||
def validate_url_target(url: str) -> tuple[bool, str]:
|
||||
def validate_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool, str]:
|
||||
"""Validate a URL is safe to fetch: scheme, hostname, and resolved IPs.
|
||||
|
||||
``allow_loopback`` is intentionally narrow: it only permits literal
|
||||
loopback hosts (localhost, 127.0.0.0/8, ::1) when every resolved address is
|
||||
loopback. It does not allow RFC1918, link-local, metadata, or public DNS
|
||||
names that happen to resolve to loopback.
|
||||
|
||||
Returns (ok, error_message). When ok is True, error_message is empty.
|
||||
"""
|
||||
try:
|
||||
@@ -66,11 +71,16 @@ def validate_url_target(url: str) -> tuple[bool, str]:
|
||||
except socket.gaierror:
|
||||
return False, f"Cannot resolve hostname: {hostname}"
|
||||
|
||||
addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
|
||||
for info in infos:
|
||||
try:
|
||||
addr = ipaddress.ip_address(info[4][0])
|
||||
except ValueError:
|
||||
continue
|
||||
addrs.append(addr)
|
||||
if allow_loopback and _is_allowed_loopback_target(hostname, addrs):
|
||||
return True, ""
|
||||
for addr in addrs:
|
||||
if _is_private(addr):
|
||||
return False, f"Blocked: {hostname} resolves to private/internal address {addr}"
|
||||
|
||||
@@ -109,11 +119,25 @@ def validate_resolved_url(url: str) -> tuple[bool, str]:
|
||||
return True, ""
|
||||
|
||||
|
||||
def contains_internal_url(command: str) -> bool:
|
||||
def contains_internal_url(command: str, *, allow_loopback: bool = False) -> bool:
|
||||
"""Return True if the command string contains a URL targeting an internal/private address."""
|
||||
for m in _URL_RE.finditer(command):
|
||||
url = m.group(0)
|
||||
ok, _ = validate_url_target(url)
|
||||
ok, _ = validate_url_target(url, allow_loopback=allow_loopback)
|
||||
if not ok:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_allowed_loopback_target(
|
||||
hostname: str,
|
||||
addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address],
|
||||
) -> bool:
|
||||
if not addrs or not all(addr.is_loopback for addr in addrs):
|
||||
return False
|
||||
normalized = hostname.rstrip(".").lower()
|
||||
if normalized == "localhost":
|
||||
return True
|
||||
with suppress(ValueError):
|
||||
return ipaddress.ip_address(hostname).is_loopback
|
||||
return False
|
||||
|
||||
@@ -0,0 +1,430 @@
|
||||
"""Workspace access scope and sandbox capability helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from contextvars import ContextVar, Token
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
WorkspaceAccessMode = Literal["restricted", "full"]
|
||||
WORKSPACE_SCOPE_METADATA_KEY = "workspace_scope"
|
||||
_ACCESS_MODES = {"restricted", "full"}
|
||||
|
||||
_TRUE_VALUES = {"1", "true", "yes", "on", "enabled"}
|
||||
_FALSE_VALUES = {"0", "false", "no", "off", "disabled", ""}
|
||||
_PROVIDER_LABELS = {
|
||||
"none": "None",
|
||||
"unknown": "Unknown system sandbox",
|
||||
"macos_app_sandbox": "macOS App Sandbox",
|
||||
"bwrap": "Bubblewrap",
|
||||
}
|
||||
|
||||
_CURRENT_WORKSPACE_SCOPE: ContextVar["WorkspaceScope | None"] = ContextVar(
|
||||
"nanobot_workspace_scope",
|
||||
default=None,
|
||||
)
|
||||
|
||||
|
||||
class WorkspaceScopeError(ValueError):
|
||||
"""Raised when a requested WebUI workspace scope is invalid."""
|
||||
|
||||
status = 400
|
||||
|
||||
def __init__(self, message: str, *, status: int = 400) -> None:
|
||||
super().__init__(message)
|
||||
self.message = message
|
||||
self.status = status
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WorkspaceSandboxStatus:
|
||||
"""Resolved workspace sandbox state for runtime display and tooling."""
|
||||
|
||||
restrict_to_workspace: bool
|
||||
workspace_root: str
|
||||
level: str
|
||||
enforced: bool
|
||||
provider: str
|
||||
provider_label: str
|
||||
summary: str
|
||||
|
||||
def as_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"restrict_to_workspace": self.restrict_to_workspace,
|
||||
"workspace_root": self.workspace_root,
|
||||
"level": self.level,
|
||||
"enforced": self.enforced,
|
||||
"provider": self.provider,
|
||||
"provider_label": self.provider_label,
|
||||
"summary": self.summary,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WorkspaceScope:
|
||||
"""Effective project root and access mode for one agent turn."""
|
||||
|
||||
project_path: Path
|
||||
access_mode: WorkspaceAccessMode
|
||||
restrict_to_workspace: bool
|
||||
sandbox_status: WorkspaceSandboxStatus
|
||||
source_channel: str | None = None
|
||||
|
||||
@property
|
||||
def project_name(self) -> str:
|
||||
return self.project_path.name or str(self.project_path)
|
||||
|
||||
def metadata(self) -> dict[str, str]:
|
||||
return {
|
||||
"project_path": str(self.project_path),
|
||||
"access_mode": self.access_mode,
|
||||
}
|
||||
|
||||
def payload(self) -> dict[str, Any]:
|
||||
return {
|
||||
**self.metadata(),
|
||||
"project_name": self.project_name,
|
||||
"restrict_to_workspace": self.restrict_to_workspace,
|
||||
"sandbox_status": self.sandbox_status.as_dict(),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ToolWorkspace:
|
||||
"""Workspace policy resolved for a tool call."""
|
||||
|
||||
project_path: Path | None
|
||||
restrict_to_workspace: bool
|
||||
scope: WorkspaceScope | None = None
|
||||
|
||||
@property
|
||||
def allowed_root(self) -> Path | None:
|
||||
if self.restrict_to_workspace and self.project_path is not None:
|
||||
return self.project_path
|
||||
return None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WorkspaceScopeResolver:
|
||||
"""Resolve the effective workspace scope at an agent turn boundary."""
|
||||
|
||||
default_workspace: str | Path
|
||||
default_restrict_to_workspace: bool
|
||||
scoped_channel: str = "websocket"
|
||||
|
||||
@property
|
||||
def sandbox_status(self) -> WorkspaceSandboxStatus:
|
||||
return self.default().sandbox_status
|
||||
|
||||
def default(self) -> WorkspaceScope:
|
||||
return default_workspace_scope(
|
||||
self.default_workspace,
|
||||
self.default_restrict_to_workspace,
|
||||
)
|
||||
|
||||
def for_message(
|
||||
self,
|
||||
msg: Any,
|
||||
session_metadata: Any,
|
||||
) -> WorkspaceScope:
|
||||
return self.for_turn(
|
||||
channel=getattr(msg, "channel", None),
|
||||
message_metadata=getattr(msg, "metadata", None),
|
||||
session_metadata=session_metadata,
|
||||
)
|
||||
|
||||
def for_turn(
|
||||
self,
|
||||
*,
|
||||
channel: str | None,
|
||||
message_metadata: Any,
|
||||
session_metadata: Any,
|
||||
) -> WorkspaceScope:
|
||||
if channel != self.scoped_channel:
|
||||
return self.default()
|
||||
return resolve_effective_workspace_scope(
|
||||
message_metadata=message_metadata,
|
||||
session_metadata=session_metadata,
|
||||
default_workspace=self.default_workspace,
|
||||
default_restrict_to_workspace=self.default_restrict_to_workspace,
|
||||
source_channel=channel,
|
||||
)
|
||||
|
||||
def persist_message_scope(self, session: Any, msg: Any) -> None:
|
||||
if getattr(msg, "channel", None) != self.scoped_channel:
|
||||
return
|
||||
metadata = getattr(msg, "metadata", None)
|
||||
if not isinstance(metadata, dict):
|
||||
return
|
||||
raw = metadata.get(WORKSPACE_SCOPE_METADATA_KEY)
|
||||
if isinstance(raw, dict):
|
||||
session.metadata[WORKSPACE_SCOPE_METADATA_KEY] = dict(raw)
|
||||
|
||||
|
||||
def workspace_sandbox_status(
|
||||
*,
|
||||
restrict_to_workspace: bool,
|
||||
workspace: str | Path,
|
||||
environ: dict[str, str] | None = None,
|
||||
) -> WorkspaceSandboxStatus:
|
||||
"""Return how workspace restriction is enforced in the current host."""
|
||||
|
||||
workspace_root = str(Path(workspace).expanduser().resolve(strict=False))
|
||||
provider = _env_system_provider(environ)
|
||||
if not restrict_to_workspace:
|
||||
return WorkspaceSandboxStatus(
|
||||
restrict_to_workspace=False,
|
||||
workspace_root=workspace_root,
|
||||
level="off",
|
||||
enforced=False,
|
||||
provider="none",
|
||||
provider_label=_provider_label("none"),
|
||||
summary="Workspace restriction is disabled.",
|
||||
)
|
||||
|
||||
if provider:
|
||||
label = _provider_label(provider)
|
||||
return WorkspaceSandboxStatus(
|
||||
restrict_to_workspace=True,
|
||||
workspace_root=workspace_root,
|
||||
level="system",
|
||||
enforced=True,
|
||||
provider=provider,
|
||||
provider_label=label,
|
||||
summary=f"Workspace restriction is system-enforced by {label}.",
|
||||
)
|
||||
|
||||
return WorkspaceSandboxStatus(
|
||||
restrict_to_workspace=True,
|
||||
workspace_root=workspace_root,
|
||||
level="application",
|
||||
enforced=False,
|
||||
provider="none",
|
||||
provider_label=_provider_label("none"),
|
||||
summary="Workspace restriction uses nanobot application-level guards.",
|
||||
)
|
||||
|
||||
|
||||
def default_access_mode(restrict_to_workspace: bool) -> WorkspaceAccessMode:
|
||||
return "restricted" if restrict_to_workspace else "full"
|
||||
|
||||
|
||||
def build_workspace_scope(
|
||||
project_path: str | Path,
|
||||
access_mode: str,
|
||||
*,
|
||||
source_channel: str | None = None,
|
||||
) -> WorkspaceScope:
|
||||
mode = _normalize_access_mode(access_mode)
|
||||
root = Path(project_path).expanduser().resolve(strict=False)
|
||||
restrict = mode == "restricted"
|
||||
return WorkspaceScope(
|
||||
project_path=root,
|
||||
access_mode=mode,
|
||||
restrict_to_workspace=restrict,
|
||||
sandbox_status=workspace_sandbox_status(
|
||||
restrict_to_workspace=restrict,
|
||||
workspace=root,
|
||||
),
|
||||
source_channel=source_channel,
|
||||
)
|
||||
|
||||
|
||||
def default_workspace_scope(
|
||||
workspace: str | Path,
|
||||
restrict_to_workspace: bool,
|
||||
*,
|
||||
source_channel: str | None = None,
|
||||
) -> WorkspaceScope:
|
||||
return build_workspace_scope(
|
||||
workspace,
|
||||
default_access_mode(restrict_to_workspace),
|
||||
source_channel=source_channel,
|
||||
)
|
||||
|
||||
|
||||
def validate_workspace_scope_payload(
|
||||
raw: Any,
|
||||
*,
|
||||
default_workspace: str | Path,
|
||||
default_restrict_to_workspace: bool,
|
||||
source_channel: str | None = None,
|
||||
) -> WorkspaceScope:
|
||||
"""Validate a client-requested workspace scope."""
|
||||
if raw is None:
|
||||
return default_workspace_scope(
|
||||
default_workspace,
|
||||
default_restrict_to_workspace,
|
||||
source_channel=source_channel,
|
||||
)
|
||||
if not isinstance(raw, dict):
|
||||
raise WorkspaceScopeError("workspace_scope must be an object")
|
||||
|
||||
raw_path = raw.get("project_path") or raw.get("path")
|
||||
if raw_path is None or raw_path == "":
|
||||
raw_path = str(Path(default_workspace).expanduser().resolve(strict=False))
|
||||
if not isinstance(raw_path, str):
|
||||
raise WorkspaceScopeError("project_path must be a string")
|
||||
if "\0" in raw_path:
|
||||
raise WorkspaceScopeError("project_path contains invalid characters")
|
||||
|
||||
project = Path(raw_path).expanduser()
|
||||
if not project.is_absolute():
|
||||
raise WorkspaceScopeError("project_path must be absolute")
|
||||
project = project.resolve(strict=False)
|
||||
if not project.is_dir():
|
||||
raise WorkspaceScopeError("project_path must be an existing directory")
|
||||
|
||||
raw_mode = raw.get("access_mode")
|
||||
if raw_mode is None:
|
||||
raw_mode = default_access_mode(default_restrict_to_workspace)
|
||||
if not isinstance(raw_mode, str):
|
||||
raise WorkspaceScopeError("access_mode must be a string")
|
||||
return build_workspace_scope(project, raw_mode, source_channel=source_channel)
|
||||
|
||||
|
||||
def workspace_scope_from_metadata(
|
||||
metadata: Any,
|
||||
*,
|
||||
default_workspace: str | Path,
|
||||
default_restrict_to_workspace: bool,
|
||||
source_channel: str | None = None,
|
||||
) -> WorkspaceScope:
|
||||
"""Resolve persisted metadata, falling back safely for old or stale sessions."""
|
||||
if not isinstance(metadata, dict):
|
||||
return default_workspace_scope(
|
||||
default_workspace,
|
||||
default_restrict_to_workspace,
|
||||
source_channel=source_channel,
|
||||
)
|
||||
try:
|
||||
return validate_workspace_scope_payload(
|
||||
metadata.get(WORKSPACE_SCOPE_METADATA_KEY),
|
||||
default_workspace=default_workspace,
|
||||
default_restrict_to_workspace=default_restrict_to_workspace,
|
||||
source_channel=source_channel,
|
||||
)
|
||||
except WorkspaceScopeError:
|
||||
return default_workspace_scope(
|
||||
default_workspace,
|
||||
default_restrict_to_workspace,
|
||||
source_channel=source_channel,
|
||||
)
|
||||
|
||||
|
||||
def resolve_effective_workspace_scope(
|
||||
*,
|
||||
message_metadata: Any,
|
||||
session_metadata: Any,
|
||||
default_workspace: str | Path,
|
||||
default_restrict_to_workspace: bool,
|
||||
source_channel: str | None = None,
|
||||
) -> WorkspaceScope:
|
||||
if isinstance(message_metadata, dict) and WORKSPACE_SCOPE_METADATA_KEY in message_metadata:
|
||||
return workspace_scope_from_metadata(
|
||||
message_metadata,
|
||||
default_workspace=default_workspace,
|
||||
default_restrict_to_workspace=default_restrict_to_workspace,
|
||||
source_channel=source_channel,
|
||||
)
|
||||
return workspace_scope_from_metadata(
|
||||
session_metadata,
|
||||
default_workspace=default_workspace,
|
||||
default_restrict_to_workspace=default_restrict_to_workspace,
|
||||
source_channel=source_channel,
|
||||
)
|
||||
|
||||
|
||||
def bind_workspace_scope(scope: WorkspaceScope) -> Token[WorkspaceScope | None]:
|
||||
return _CURRENT_WORKSPACE_SCOPE.set(scope)
|
||||
|
||||
|
||||
def reset_workspace_scope(token: Token[WorkspaceScope | None]) -> None:
|
||||
_CURRENT_WORKSPACE_SCOPE.reset(token)
|
||||
|
||||
|
||||
def current_workspace_scope() -> WorkspaceScope | None:
|
||||
return _CURRENT_WORKSPACE_SCOPE.get()
|
||||
|
||||
|
||||
def current_tool_workspace(
|
||||
default_workspace: str | Path | None,
|
||||
*,
|
||||
restrict_to_workspace: bool = False,
|
||||
sandbox_restricts_workspace: bool = False,
|
||||
) -> ToolWorkspace:
|
||||
"""Return the workspace/access policy for the current tool call."""
|
||||
|
||||
scope = current_workspace_scope()
|
||||
project_path = (
|
||||
scope.project_path
|
||||
if scope is not None
|
||||
else Path(default_workspace).expanduser() if default_workspace is not None else None
|
||||
)
|
||||
restrict = (
|
||||
scope.restrict_to_workspace
|
||||
if scope is not None
|
||||
else bool(restrict_to_workspace)
|
||||
) or sandbox_restricts_workspace
|
||||
return ToolWorkspace(
|
||||
project_path=project_path,
|
||||
restrict_to_workspace=restrict,
|
||||
scope=scope,
|
||||
)
|
||||
|
||||
|
||||
def current_scope_allows_loopback(*, enabled: bool) -> bool:
|
||||
"""Return True when the current WebUI Full Access turn may touch loopback URLs."""
|
||||
|
||||
scope = current_workspace_scope()
|
||||
return bool(
|
||||
enabled
|
||||
and scope is not None
|
||||
and scope.source_channel == "websocket"
|
||||
and scope.access_mode == "full"
|
||||
and not scope.restrict_to_workspace
|
||||
)
|
||||
|
||||
|
||||
def _env_system_provider(environ: dict[str, str] | None = None) -> str | None:
|
||||
env = environ if environ is not None else os.environ
|
||||
explicit_provider = env.get("NANOBOT_WORKSPACE_SANDBOX_PROVIDER")
|
||||
enforced = env.get("NANOBOT_WORKSPACE_SANDBOX_ENFORCED")
|
||||
compatibility = env.get("NANOBOT_SANDBOX_ENFORCED")
|
||||
|
||||
marker = enforced if enforced is not None else compatibility
|
||||
if marker is None:
|
||||
return None
|
||||
|
||||
normalized_marker = marker.strip().lower()
|
||||
if normalized_marker in _FALSE_VALUES:
|
||||
return None
|
||||
if normalized_marker in _TRUE_VALUES:
|
||||
return _normalize_provider(explicit_provider)
|
||||
return _normalize_provider(marker)
|
||||
|
||||
|
||||
def _normalize_provider(value: str | None) -> str:
|
||||
if not value:
|
||||
return "unknown"
|
||||
normalized = value.strip().lower().replace("-", "_").replace(" ", "_")
|
||||
return normalized or "unknown"
|
||||
|
||||
|
||||
def _provider_label(provider: str) -> str:
|
||||
if provider in _PROVIDER_LABELS:
|
||||
return _PROVIDER_LABELS[provider]
|
||||
return provider.replace("_", " ").title()
|
||||
|
||||
|
||||
def _normalize_access_mode(value: str) -> WorkspaceAccessMode:
|
||||
mode = value.strip().lower().replace("_", "-")
|
||||
if mode == "restrict":
|
||||
mode = "restricted"
|
||||
if mode == "full-access":
|
||||
mode = "full"
|
||||
if mode not in _ACCESS_MODES:
|
||||
raise WorkspaceScopeError("access_mode must be restricted or full")
|
||||
return mode # type: ignore[return-value]
|
||||
@@ -0,0 +1,85 @@
|
||||
"""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.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
WORKSPACE_BOUNDARY_NOTE = (
|
||||
" (this is a hard policy boundary, not a transient failure; "
|
||||
"do not retry with shell tricks or alternative tools, and ask "
|
||||
"the user how to proceed if the resource is genuinely required)"
|
||||
)
|
||||
|
||||
|
||||
class WorkspaceBoundaryError(PermissionError):
|
||||
"""Raised when a requested path escapes an allowed workspace boundary."""
|
||||
|
||||
|
||||
def resolve_path(path: str | Path, workspace: str | Path | None = None, *, strict: bool = False) -> Path:
|
||||
"""Resolve *path*, interpreting relative paths against *workspace* when set."""
|
||||
candidate = Path(path).expanduser()
|
||||
if not candidate.is_absolute() and workspace is not None:
|
||||
candidate = Path(workspace).expanduser() / candidate
|
||||
return candidate.resolve(strict=strict)
|
||||
|
||||
|
||||
def is_path_within(path: str | Path, root: str | Path) -> bool:
|
||||
"""Return True when *path* resolves to *root* or a descendant of *root*."""
|
||||
try:
|
||||
resolved_path = Path(path).expanduser().resolve(strict=False)
|
||||
resolved_root = Path(root).expanduser().resolve(strict=False)
|
||||
resolved_path.relative_to(resolved_root)
|
||||
return True
|
||||
except (OSError, RuntimeError, TypeError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def is_path_allowed(path: str | Path, roots: Iterable[str | Path]) -> bool:
|
||||
"""Return True when *path* is inside any allowed root."""
|
||||
return any(is_path_within(path, root) for root in roots)
|
||||
|
||||
|
||||
def require_path_within(
|
||||
path: str | Path,
|
||||
root: str | Path,
|
||||
*,
|
||||
message: str | None = None,
|
||||
) -> Path:
|
||||
"""Resolve *path* and require it to be inside *root*."""
|
||||
resolved = Path(path).expanduser().resolve(strict=False)
|
||||
if not is_path_within(resolved, root):
|
||||
raise WorkspaceBoundaryError(
|
||||
message
|
||||
or f"Path {path} is outside allowed directory {Path(root).expanduser()}"
|
||||
+ WORKSPACE_BOUNDARY_NOTE
|
||||
)
|
||||
return resolved
|
||||
|
||||
|
||||
def resolve_allowed_path(
|
||||
path: str | Path,
|
||||
*,
|
||||
workspace: str | Path | None = None,
|
||||
allowed_root: str | Path | None = None,
|
||||
extra_allowed_roots: Iterable[str | Path] | None = None,
|
||||
strict: bool = False,
|
||||
) -> Path:
|
||||
"""Resolve a path and enforce containment in allowed roots when configured."""
|
||||
resolved = resolve_path(path, workspace, strict=False)
|
||||
if allowed_root is None:
|
||||
return resolve_path(path, workspace, strict=strict) if strict else resolved
|
||||
|
||||
roots = [allowed_root, *(extra_allowed_roots or [])]
|
||||
if not is_path_allowed(resolved, roots):
|
||||
raise WorkspaceBoundaryError(
|
||||
f"Path {path} is outside allowed directory {Path(allowed_root).expanduser()}"
|
||||
+ WORKSPACE_BOUNDARY_NOTE
|
||||
)
|
||||
if strict:
|
||||
return resolve_path(path, workspace, strict=True)
|
||||
return resolved
|
||||
Reference in New Issue
Block a user