butler/tools/path_safety.py
Rama main · réplica de lectura del repositorio original.
"""Workspace and sensitive-path guards for Butler tools."""
from __future__ import annotations
import os
import re
import shlex
from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast
from butler.tools.path_safety_ops import (
workspace_from_project_safe,
workspace_for_session_key_safe,
default_project_workspace_safe,
audit_session_key_safe,
current_session_key_safe,
orchestrator_workspace_safe,
is_readable_session_tool_result_path_safe,
external_path_override_allowed_safe,
butlerignore_blocked_safe,
current_orchestrator_safe,
configured_safe_root_safe,
hooks_global_dir_blocks_write,
)
from butler.tools.tool_scope import (
environment_tool_scope_enabled,
project_relative_path_anchors_to_workspace,
resolve_environment_tool_root,
workspace_anchor_strict_for_paths,
)
from butler.defaults.env_defaults import (
CC_BRIDGE_DEFAULT,
TERMINAL_ALLOWLIST_EXTRA_DEFAULT,
TERMINAL_PROFILE_DEFAULT,
)
_BASE_TERMINAL_COMMANDS = {
"cat",
"echo",
"false",
"find",
"head",
"ls",
"pwd",
"sleep",
"tail",
"true",
}
# Pilot novel-factory scripts: enable with BUTLER_TERMINAL_ALLOWLIST_EXTRA=python3,bash
_EXTRA_TERMINAL_ENV = "BUTLER_TERMINAL_ALLOWLIST_EXTRA"
_PROFILE_ENV = "BUTLER_TERMINAL_PROFILE"
_TERMINAL_PROFILES: dict[str, frozenset[str]] = {
"pilot": frozenset({"python3", "bash"}),
"dev": frozenset({
"python3",
"bash",
"pytest",
"git",
"rg",
"pip",
"pip3",
"npm",
"npx",
"make",
}),
}
def _allowed_terminal_commands() -> set[str]:
allowed = set(_BASE_TERMINAL_COMMANDS)
profile = os.getenv(_PROFILE_ENV, TERMINAL_PROFILE_DEFAULT).strip().lower()
if profile in _TERMINAL_PROFILES:
allowed.update(_TERMINAL_PROFILES[profile])
from butler.execution_context import get_current_loop_role
loop_role = get_current_loop_role()
if loop_role == "butler":
allowed.update(_TERMINAL_PROFILES.get("dev", frozenset()))
raw = os.getenv(_EXTRA_TERMINAL_ENV, TERMINAL_ALLOWLIST_EXTRA_DEFAULT).strip()
if raw:
for part in raw.replace(";", ",").split(","):
name = part.strip()
if name:
allowed.add(name)
if os.getenv("BUTLER_CC_BRIDGE", CC_BRIDGE_DEFAULT).strip().lower() in ("1", "true", "yes", "on"):
allowed.add("claude")
return allowed
@dataclass(frozen=True)
class PathSafetyResult:
allowed: bool
path: Path
error: str = ""
@dataclass(frozen=True)
class CommandSafetyResult:
allowed: bool
argv: list[str]
error: str = ""
is_pipe: bool = False
def _workspace_from_project(project: object | None) -> Path | None:
return cast(Path | None, workspace_from_project_safe(project))
def _workspace_for_session_key(session_key: str) -> Path | None:
return cast(Path | None, workspace_for_session_key_safe(session_key))
def _default_project_workspace() -> Path | None:
return cast(Path | None, default_project_workspace_safe())
def _remap_projects_docs_trap(resolved: Path, *, session_ws: Path | None) -> Path:
"""Avoid ``projects/docs`` fixture dir when a project workspace is active."""
safe = _configured_safe_root()
if safe is None:
return resolved
trap = (safe / "docs").resolve(strict=False)
try:
if not resolved.is_relative_to(trap):
return resolved
except ValueError:
return resolved
ws = session_ws or _default_project_workspace()
if ws is None:
return resolved
try:
rel = resolved.relative_to(trap)
candidate = (ws / "docs" / rel).resolve(strict=False)
except ValueError:
return resolved
if candidate.exists():
return candidate
if not resolved.exists():
return candidate
return resolved
def current_workspace_root() -> Path | None:
"""Return the active Butler project workspace, if a turn has one."""
session_key = current_session_key_safe()
ws = orchestrator_workspace_safe(session_key)
if ws is not None:
return Path(ws)
if session_key:
ws = _workspace_for_session_key(session_key)
if ws is not None:
return ws
sk = audit_session_key_safe(fallback="")
if sk and sk != session_key:
return _workspace_for_session_key(sk)
ws = _default_project_workspace()
if ws is not None:
return ws
return None
def workspace_anchor_strict_enabled() -> bool:
return workspace_anchor_strict_for_paths()
def format_tool_workspace_line(session_key: str = "") -> str:
sk = str(session_key or "").strip()
if environment_tool_scope_enabled():
root = resolve_environment_tool_root()
hint = ""
ws = _workspace_for_session_key(sk) if sk else None
if ws is None and sk:
ws = current_workspace_root()
if ws is not None:
hint = f" · 上下文项目 {ws}"
return f"工具工作区: (环境) {root}{hint}"
ws = _workspace_for_session_key(sk) if sk else None
if ws is None:
ws = current_workspace_root()
if ws is not None:
return f"工具工作区: {ws}"
root = _configured_safe_root()
if root is not None:
return f"工具工作区: (safe_root) {root}"
return "工具工作区: (cwd) " + str(Path.cwd().resolve())
def check_tool_path(path: str | os.PathLike[str], *, for_write: bool = False) -> PathSafetyResult:
"""Resolve and validate a tool path against workspace and sensitive path rules."""
raw = str(path or ".")
# Sprint 27 P1-3.3: Windows 绝对路径 (C:/, C:\\, \\\\server\\...) 在 Linux 上
# 不被识别为绝对, resolve 后会变 relative 误判为 in-workspace.
# 显式 fail-closed 为 outside workspace.
if _is_windows_absolute_path(raw):
return PathSafetyResult(
False,
Path(raw),
f"Access denied: Windows path is outside workspace ({raw})",
)
session_ws = current_workspace_root()
rel = not Path(raw).expanduser().is_absolute()
if session_ws is not None and rel and (
workspace_anchor_strict_enabled() or project_relative_path_anchors_to_workspace()
):
root = session_ws
else:
root = tool_safe_root()
resolved = _resolve_tool_path(raw, root)
if session_ws is not None and rel and (
workspace_anchor_strict_enabled() or project_relative_path_anchors_to_workspace()
):
try:
if not resolved.is_relative_to(session_ws.resolve(strict=False)):
alt = _resolve_tool_path(raw, session_ws)
if alt.is_relative_to(session_ws.resolve(strict=False)):
resolved = alt
except (ValueError, AttributeError):
pass
resolved = _remap_projects_docs_trap(resolved, session_ws=session_ws)
sensitive_error = _sensitive_path_error(resolved, for_write=for_write)
if sensitive_error:
return PathSafetyResult(False, resolved, sensitive_error)
hooks_error = _hooks_config_write_error(
resolved, root, for_write=for_write
)
if hooks_error:
return PathSafetyResult(False, resolved, hooks_error)
if not for_write:
if is_readable_session_tool_result_path_safe(resolved):
return PathSafetyResult(True, resolved)
if root is not None and not _is_relative_to(resolved, root):
override = external_path_override_allowed_safe(resolved, for_write=for_write)
if override is True:
return PathSafetyResult(True, resolved)
return PathSafetyResult(
False,
resolved,
f"Access denied: path is outside workspace ({root})",
)
hardlink_error = _hardlink_error(resolved)
if hardlink_error:
return PathSafetyResult(False, resolved, hardlink_error)
ws = session_ws or root
if ws is not None:
blocked = butlerignore_blocked_safe(
resolved,
workspace=ws,
for_write=for_write,
)
if blocked:
return PathSafetyResult(False, resolved, blocked)
return PathSafetyResult(True, resolved)
def default_tool_workdir() -> PathSafetyResult:
"""Return the default cwd for shell tools."""
return PathSafetyResult(True, tool_safe_root())
def tool_safe_root() -> Path:
"""Return the active root that tools may access."""
if environment_tool_scope_enabled():
return resolve_environment_tool_root()
ws = current_workspace_root()
if ws is not None:
return ws
return _configured_safe_root() or Path.cwd().resolve()
def _pipe_mode_enabled() -> bool:
from butler.defaults.env_defaults import TERMINAL_PIPE_DEFAULT
return os.getenv("BUTLER_TERMINAL_PIPE", str(int(TERMINAL_PIPE_DEFAULT))).strip() == "1"
_PIPE_SAFE_COMMANDS = frozenset({
"grep", "rg", "wc", "head", "tail", "sort", "uniq", "tr", "cut",
"awk", "sed", "cat", "tee", "xargs", "find", "ls", "echo",
})
def _validate_pipe_segment(segment: str, allowed: set[str]) -> CommandSafetyResult:
"""Validate one pipe segment (no recursion into nested pipes)."""
seg = segment.strip()
if not seg:
return CommandSafetyResult(False, [], "Empty pipe segment")
try:
argv = shlex.split(seg, posix=True)
except ValueError as exc:
return CommandSafetyResult(False, [], f"Invalid pipe segment: {exc}")
if not argv:
return CommandSafetyResult(False, [], "Empty pipe segment")
cmd_name = Path(argv[0]).name
all_allowed = allowed | _PIPE_SAFE_COMMANDS
if "/" in argv[0] or cmd_name not in all_allowed:
return CommandSafetyResult(False, [], f"Pipe segment command '{cmd_name}' not in allowlist")
blocked = _dangerous_search_flags(cmd_name, argv)
if blocked:
return CommandSafetyResult(False, [], blocked)
if _uses_dynamic_interpreter(argv):
return CommandSafetyResult(False, [], "Dynamic interpreter in pipe segment")
if any("$" in token for token in argv):
return CommandSafetyResult(False, [], "Shell variables in pipe segment")
return CommandSafetyResult(True, argv)
def _dangerous_search_flags(command_name: str, argv: list[str]) -> str | None:
"""Block search tools flags that bypass workspace boundaries or run shell."""
if command_name == "find":
for tok in argv[1:]:
if tok in ("-exec", "-execdir", "-delete", "-ok", "-okdir"):
return f"find {tok} is not allowed in terminal commands"
if tok.startswith(("-exec=", "-execdir=", "-ok=", "-okdir=")):
return f"find {tok.split('=')[0]} is not allowed in terminal commands"
if command_name == "rg":
for tok in argv[1:]:
if tok == "--pre" or tok.startswith("--pre="):
return "rg --pre executes arbitrary commands and is not allowed in terminal commands"
if command_name == "grep":
for tok in argv[1:]:
if tok.startswith("--file="):
return "grep --file with embedded path is not allowed in terminal commands"
return None
def prepare_shell_command(command: str) -> CommandSafetyResult:
"""Validate a terminal command and return argv for ``shell=False`` execution.
When ``BUTLER_TERMINAL_PIPE=1``, simple pipes are allowed between
allowlisted commands. The result uses ``shell=True`` via ``bash -c``
in that case, with each segment validated individually.
"""
text = command or ""
has_pipe = "|" in text
has_other_meta = bool(re.search(r"[&;<>()`]", text))
if has_pipe and _pipe_mode_enabled() and not has_other_meta:
allowed = _allowed_terminal_commands()
segments = text.split("|")
if len(segments) > 5:
return CommandSafetyResult(False, [], "Too many pipe segments (max 5)")
for seg in segments:
result = _validate_pipe_segment(seg, allowed)
if not result.allowed:
return result
for candidate in _extract_command_paths(text):
path_result = check_tool_path(candidate, for_write=_looks_like_write_command(text, candidate))
if not path_result.allowed:
return CommandSafetyResult(False, [], path_result.error)
return CommandSafetyResult(True, ["bash", "-c", text], is_pipe=True)
try:
argv = shlex.split(text, posix=True)
except ValueError as exc:
return CommandSafetyResult(False, [], f"Invalid shell command: {exc}")
if not argv:
return CommandSafetyResult(False, [], "Command is empty")
command_name = Path(argv[0]).name
if "/" in argv[0] or command_name not in _allowed_terminal_commands():
return CommandSafetyResult(False, [], "Terminal command is not in the allowlist")
blocked = _dangerous_search_flags(command_name, argv)
if blocked:
return CommandSafetyResult(False, [], blocked)
if _has_shell_metacharacters(text):
return CommandSafetyResult(False, [], "Shell metacharacters are not allowed in terminal commands")
if _uses_dynamic_interpreter(argv):
return CommandSafetyResult(False, [], "Dynamic interpreter commands are not allowed")
if any("$" in token for token in argv):
return CommandSafetyResult(False, [], "Shell variables are not allowed in terminal commands")
for candidate in _extract_command_paths(text):
path_result = check_tool_path(candidate, for_write=_looks_like_write_command(text, candidate))
if not path_result.allowed:
return CommandSafetyResult(False, [], path_result.error)
for candidate in _existing_argv_paths(argv[1:]):
path_result = check_tool_path(candidate, for_write=_looks_like_write_command(text, candidate))
if not path_result.allowed:
return CommandSafetyResult(False, [], path_result.error)
return CommandSafetyResult(True, argv)
def safe_subprocess_env() -> dict[str, str]:
"""Return a minimal environment for tool subprocesses."""
env = {
"PATH": "/usr/bin:/bin",
"LANG": os.getenv("LANG", "C.UTF-8") or "C.UTF-8",
"LC_ALL": os.getenv("LC_ALL", "C.UTF-8") or "C.UTF-8",
}
return {key: value for key, value in env.items() if value}
def _current_orchestrator() -> Any:
return current_orchestrator_safe()
def _configured_safe_root() -> Path | None:
return cast(Path | None, configured_safe_root_safe())
def _resolve_tool_path(raw: str, root: Path | None) -> Path:
path = Path(raw).expanduser()
if not path.is_absolute() and root is not None:
path = root / path
resolved = path.resolve(strict=False)
if path.is_symlink():
target = resolved
if root is not None and not target.is_relative_to(root.resolve()):
import logging as _log
_log.getLogger(__name__).warning(
"Symlink %s resolves outside workspace: %s", path, target
)
raise PermissionError(
f"Symlink target {target} is outside workspace {root}"
)
return resolved
def _extract_command_paths(command: str) -> list[str]:
candidates: list[str] = []
try:
tokens = shlex.split(command, posix=True)
except ValueError:
tokens = command.split()
for token in tokens:
stripped = token.strip()
if "://" in stripped:
continue
stripped = re.sub(r"^\d?>+", "", stripped)
if stripped in {".", ".."} or stripped.startswith(("/", "~", "./", "../")) or "/" in stripped:
candidates.append(stripped)
for match in re.finditer(r"(?<![\w.-])(?:~?/[^ \t\n\r'\"`;|&<>$()]+)", command):
candidate = match.group(0)
if "://" not in candidate:
candidates.append(candidate)
seen: set[str] = set()
out: list[str] = []
for candidate in candidates:
candidate = candidate.strip(" '\"\t\n\r;|&<>`")
if candidate and candidate not in seen:
seen.add(candidate)
out.append(candidate)
return out
def _looks_like_write_command(command: str, candidate: str) -> bool:
del candidate
return bool(re.search(r"(^|\s)(>|>>|2>|tee|cp|mv|rm|touch|mkdir|chmod|chown)(\s|$)", command))
def _existing_argv_paths(tokens: list[str]) -> list[str]:
root = tool_safe_root()
out: list[str] = []
for token in tokens:
if not token or token.startswith("-") or "://" in token:
continue
if token.startswith(("/", "~", "./", "../")) or "/" in token:
continue
try:
if (root / token).exists():
out.append(token)
except OSError:
continue
return out
def _strip_safe_io_redirections(command: str) -> str:
"""Remove stderr/stdout null redirects before metacharacter scan."""
stripped = command
for pattern in (
r"\s+2>/dev/null",
r"\s+1>/dev/null",
r"\s+2>&1",
):
stripped = re.sub(pattern, "", stripped)
return stripped.strip()
def _has_shell_metacharacters(command: str) -> bool:
return bool(re.search(r"[|&;<>()`]", _strip_safe_io_redirections(command)))
def _uses_dynamic_interpreter(argv: list[str]) -> bool:
command = Path(argv[0]).name
if command in {"sh", "bash", "zsh", "fish"} and "-c" in argv[1:]:
return True
if command in {"python", "python3", "node", "ruby", "perl", "php"}:
return any(flag in argv[1:] for flag in {"-c", "-e"})
return False
def _is_relative_to(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
# Sprint 27 P1-3.3: Windows 绝对路径检测.
# 形态 1: <drive>:[\\/]<rest> (C:/..., C:\..., c:/..., Z:\foo\bar)
# 形态 2: UNC \\\\?\\<...> 或 \\\\<server>\\<share> (\\\\server\\share\\file)
# 形态 3: 盘符相对形式 C:foo (无分隔符, 罕见, 也按 Windows 绝对处理)
_WINDOWS_DRIVE_PATH = re.compile(r"^[A-Za-z]:[\\/]|^[A-Za-z]:[^\\/]")
_WINDOWS_UNC_PATH = re.compile(r"^\\\\[?]?[A-Za-z0-9_.$-]+[\\/]")
def _is_windows_absolute_path(path_str: str) -> bool:
"""Return True if path_str looks like a Windows absolute path (drive or UNC)."""
s = str(path_str or "").strip()
if not s:
return False
return bool(_WINDOWS_DRIVE_PATH.match(s) or _WINDOWS_UNC_PATH.match(s))
_HOOK_CONFIG_NAMES = frozenset({"hooks.yaml", "hooks.yml"})
def _hooks_config_write_error(
path: Path,
workspace_root: Path | None,
*,
for_write: bool,
) -> str:
"""R3-2: block LLM tool writes to hook rule files (persistent RCE chain)."""
if not for_write:
return ""
resolved = path.resolve(strict=False)
if hooks_global_dir_blocks_write(resolved):
return "Access denied: path targets hooks configuration"
if workspace_root is None:
return ""
butler_dir = (workspace_root / ".butler").resolve(strict=False)
if _is_relative_to(resolved, butler_dir) and resolved.name in _HOOK_CONFIG_NAMES:
return "Access denied: path targets hooks configuration"
return ""
def _sensitive_path_error(path: Path, *, for_write: bool) -> str:
home = Path.home().resolve()
sensitive_dirs = [
home / ".ssh",
home / ".aws",
home / ".gnupg",
home / ".kube",
home / ".docker",
home / ".azure",
home / ".config" / "gh",
]
sensitive_files = [
home / ".netrc",
home / ".pgpass",
home / ".npmrc",
home / ".pypirc",
Path("/etc/sudoers"),
Path("/etc/passwd"),
Path("/etc/shadow"),
]
if for_write:
sensitive_files.extend([
home / ".bashrc",
home / ".zshrc",
home / ".profile",
home / ".bash_profile",
home / ".zprofile",
])
sensitive_dirs.extend([
Path("/etc/sudoers.d"),
Path("/etc/systemd"),
])
for directory in sensitive_dirs:
if _is_relative_to(path, directory.resolve(strict=False)):
return "Access denied: path targets a sensitive location"
if path in {p.resolve(strict=False) for p in sensitive_files}:
return "Access denied: path targets a sensitive location"
return ""
def _hardlink_error(path: Path) -> str:
try:
if path.exists() and path.is_file() and path.stat().st_nlink > 1:
return "Access denied: hardlinked files are not allowed"
except OSError:
return ""
return ""