1772 lines
64 KiB
Python
1772 lines
64 KiB
Python
"""
|
|
title: SMB File Operations Tool
|
|
author: Jonas Bengtsson
|
|
author_url: https://www.bengtsson.ws
|
|
description: SMB2/SMB3 file operations for Open WebUI with per-user credentials, write controls, path confinement, and status events.
|
|
required_open_webui_version: 0.10.0
|
|
requirements: smbprotocol>=1.17.0
|
|
version: 0.1.0
|
|
license: MIT
|
|
"""
|
|
|
|
import asyncio
|
|
import base64
|
|
import fnmatch
|
|
import json
|
|
import ntpath
|
|
import stat
|
|
import threading
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable, Optional
|
|
|
|
import smbclient
|
|
from pydantic import BaseModel, Field
|
|
from smbclient.shutil import rmtree
|
|
|
|
|
|
class EventEmitter:
|
|
"""Small status-event helper based on the supplied Open WebUI tool skeleton."""
|
|
|
|
def __init__(self, event_emitter: Callable[[dict], Any] = None):
|
|
self.event_emitter = event_emitter
|
|
|
|
async def progress_update(self, description: str):
|
|
await self.emit(description)
|
|
|
|
async def error_update(self, description: str):
|
|
await self.emit(description, "error", True)
|
|
|
|
async def success_update(self, description: str):
|
|
await self.emit(description, "success", True)
|
|
|
|
async def emit(
|
|
self,
|
|
description: str = "Unknown State",
|
|
status: str = "in_progress",
|
|
done: bool = False,
|
|
):
|
|
if self.event_emitter:
|
|
await self.event_emitter(
|
|
{
|
|
"type": "status",
|
|
"data": {
|
|
"status": status,
|
|
"description": description,
|
|
"done": done,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
class Tools:
|
|
"""
|
|
SMB file-management tools.
|
|
|
|
Security model:
|
|
- Valves are administrator-controlled global limits/policies.
|
|
- UserValves contain each user's SMB endpoint, credentials, and opt-in
|
|
mutation permissions.
|
|
- All caller-supplied paths are relative to the configured share/root_path.
|
|
- The smbclient connection cache is cleared before and after every operation
|
|
and access is serialized to prevent credentials/sessions being reused
|
|
across Open WebUI users in the same Python process.
|
|
"""
|
|
|
|
# smbclient keeps a process-wide connection/session cache. Serialize access
|
|
# so one user's registered SMB session cannot be accidentally reused by
|
|
# another user while an operation is running.
|
|
_smb_lock = threading.RLock()
|
|
|
|
class Valves(BaseModel):
|
|
enabled: bool = Field(
|
|
default=True,
|
|
description="Global administrator switch for this SMB tool.",
|
|
)
|
|
allowed_server_patterns: str = Field(
|
|
default="*",
|
|
description=(
|
|
"Comma-separated fnmatch patterns for SMB servers users may access. "
|
|
"Examples: 'fileserver01,*.corp.example,10.20.*'. '*' allows any server."
|
|
),
|
|
)
|
|
allowed_share_patterns: str = Field(
|
|
default="*",
|
|
description=(
|
|
"Comma-separated fnmatch patterns for SMB share names users may access. "
|
|
"Examples: 'documents,projects-*'. '*' allows any share."
|
|
),
|
|
)
|
|
|
|
allow_create_files: bool = Field(
|
|
default=True,
|
|
description="Allow users to opt in to creating new files.",
|
|
)
|
|
allow_modify_files: bool = Field(
|
|
default=True,
|
|
description="Allow users to opt in to modifying existing files.",
|
|
)
|
|
allow_delete_paths: bool = Field(
|
|
default=False,
|
|
description="Allow users to opt in to deleting files/directories.",
|
|
)
|
|
allow_directory_operations: bool = Field(
|
|
default=True,
|
|
description="Allow users to opt in to creating or moving directories.",
|
|
)
|
|
allow_move_copy: bool = Field(
|
|
default=True,
|
|
description="Allow users to opt in to move/copy operations.",
|
|
)
|
|
allow_recursive_delete: bool = Field(
|
|
default=False,
|
|
description="Allow recursive directory deletion. Keep disabled unless explicitly needed.",
|
|
)
|
|
allow_alternate_data_streams: bool = Field(
|
|
default=False,
|
|
description=(
|
|
"Allow ':' in relative paths (Windows Alternate Data Streams). "
|
|
"Normally this should stay disabled."
|
|
),
|
|
)
|
|
|
|
max_read_bytes: int = Field(
|
|
default=2 * 1024 * 1024,
|
|
ge=1,
|
|
le=100 * 1024 * 1024,
|
|
description="Maximum bytes returned by one read_file call.",
|
|
)
|
|
max_write_bytes: int = Field(
|
|
default=4 * 1024 * 1024,
|
|
ge=1,
|
|
le=100 * 1024 * 1024,
|
|
description="Maximum payload size accepted by one write/append/update call.",
|
|
)
|
|
max_list_entries: int = Field(
|
|
default=500,
|
|
ge=1,
|
|
le=10000,
|
|
description="Maximum entries returned from one directory listing.",
|
|
)
|
|
max_search_results: int = Field(
|
|
default=200,
|
|
ge=1,
|
|
le=5000,
|
|
description="Maximum results returned by search_files.",
|
|
)
|
|
max_search_depth: int = Field(
|
|
default=8,
|
|
ge=0,
|
|
le=64,
|
|
description="Maximum recursive depth allowed by search_files.",
|
|
)
|
|
|
|
class UserValves(BaseModel):
|
|
server: str = Field(
|
|
default="",
|
|
description="SMB server hostname or IP address, without smb:// or leading slashes.",
|
|
)
|
|
port: int = Field(
|
|
default=445,
|
|
ge=1,
|
|
le=65535,
|
|
description="SMB TCP port. Normally 445.",
|
|
)
|
|
share: str = Field(
|
|
default="",
|
|
description="SMB share name, for example 'Documents'.",
|
|
)
|
|
username: str = Field(
|
|
default="",
|
|
description="SMB username. May be empty for servers that permit guest/anonymous access.",
|
|
)
|
|
password: str = Field(
|
|
default="",
|
|
description="SMB password.",
|
|
json_schema_extra={"input": {"type": "password"}},
|
|
)
|
|
domain: str = Field(
|
|
default="",
|
|
description=(
|
|
"Optional Windows/AD domain. If set, authentication uses DOMAIN\\username."
|
|
),
|
|
)
|
|
root_path: str = Field(
|
|
default="",
|
|
description=(
|
|
"Optional directory inside the share that becomes this tool's logical root. "
|
|
"All tool paths are constrained beneath it."
|
|
),
|
|
)
|
|
encrypt: bool = Field(
|
|
default=False,
|
|
description="Require SMB encryption. The server must support SMB3 encryption.",
|
|
)
|
|
connection_timeout: int = Field(
|
|
default=30,
|
|
ge=1,
|
|
le=300,
|
|
description="SMB connection timeout in seconds.",
|
|
)
|
|
|
|
allow_write_new_files: bool = Field(
|
|
default=False,
|
|
description="Opt in to creating new files on the configured SMB share.",
|
|
)
|
|
allow_modify_existing_files: bool = Field(
|
|
default=False,
|
|
description="Opt in to overwriting, appending to, or updating existing files.",
|
|
)
|
|
allow_delete_existing_paths: bool = Field(
|
|
default=False,
|
|
description="Opt in to deleting existing files/directories.",
|
|
)
|
|
allow_directory_operations: bool = Field(
|
|
default=False,
|
|
description="Opt in to creating and manipulating directories.",
|
|
)
|
|
allow_move_copy_operations: bool = Field(
|
|
default=False,
|
|
description="Opt in to moving or copying files/directories.",
|
|
)
|
|
|
|
def __init__(self):
|
|
self.valves = self.Valves()
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Common helpers
|
|
# ---------------------------------------------------------------------
|
|
|
|
def _user_valves(self, __user__: dict) -> "Tools.UserValves":
|
|
if not __user__:
|
|
raise ValueError("Open WebUI user context is required.")
|
|
|
|
valves = __user__.get("valves")
|
|
if valves is None:
|
|
raise ValueError(
|
|
"SMB UserValves are not configured for this user. "
|
|
"Set the SMB server, share, and credentials in the tool settings."
|
|
)
|
|
|
|
if isinstance(valves, self.UserValves):
|
|
return valves
|
|
|
|
# Defensive compatibility for environments that provide a dict.
|
|
if isinstance(valves, dict):
|
|
return self.UserValves(**valves)
|
|
|
|
# Pydantic v1/v2 or another model-like object.
|
|
if hasattr(valves, "model_dump"):
|
|
return self.UserValves(**valves.model_dump())
|
|
if hasattr(valves, "dict"):
|
|
return self.UserValves(**valves.dict())
|
|
|
|
raise ValueError("Unsupported UserValves object supplied by Open WebUI.")
|
|
|
|
def _patterns_allow(self, value: str, patterns_csv: str) -> bool:
|
|
patterns = [p.strip() for p in patterns_csv.split(",") if p.strip()]
|
|
if not patterns:
|
|
return False
|
|
return any(fnmatch.fnmatch(value.lower(), p.lower()) for p in patterns)
|
|
|
|
def _validate_config(self, uv: "Tools.UserValves"):
|
|
if not self.valves.enabled:
|
|
raise PermissionError("The SMB tool has been disabled by the administrator.")
|
|
|
|
server = uv.server.strip()
|
|
share = uv.share.strip()
|
|
|
|
if not server:
|
|
raise ValueError("SMB server is not configured in UserValves.")
|
|
if not share:
|
|
raise ValueError("SMB share is not configured in UserValves.")
|
|
|
|
if "\\" in server or "/" in server:
|
|
raise ValueError(
|
|
"SMB server must be a hostname or IP address, not a URL or UNC path."
|
|
)
|
|
if "\\" in share or "/" in share or share in {".", ".."}:
|
|
raise ValueError("SMB share must be a single share name.")
|
|
|
|
if not self._patterns_allow(server, self.valves.allowed_server_patterns):
|
|
raise PermissionError(
|
|
f"SMB server '{server}' is not allowed by the administrator."
|
|
)
|
|
if not self._patterns_allow(share, self.valves.allowed_share_patterns):
|
|
raise PermissionError(
|
|
f"SMB share '{share}' is not allowed by the administrator."
|
|
)
|
|
|
|
# Validate root_path using the same rules used for operation paths.
|
|
self._normalize_relative_path(uv.root_path, allow_empty=True)
|
|
|
|
def _normalize_relative_path(self, path: Optional[str], allow_empty: bool = True) -> str:
|
|
raw = (path or "").strip().replace("/", "\\")
|
|
|
|
if not raw:
|
|
if allow_empty:
|
|
return ""
|
|
raise ValueError("A non-empty relative path is required.")
|
|
|
|
# Only relative paths are accepted. Users never supply \\server\share.
|
|
if raw.startswith("\\") or ntpath.isabs(raw) or ntpath.splitdrive(raw)[0]:
|
|
raise ValueError(
|
|
"Paths must be relative to the configured SMB share/root_path."
|
|
)
|
|
|
|
norm = ntpath.normpath(raw)
|
|
if norm in {"", "."}:
|
|
if allow_empty:
|
|
return ""
|
|
raise ValueError("A non-empty relative path is required.")
|
|
|
|
parts = [p for p in norm.split("\\") if p not in {"", "."}]
|
|
if any(p == ".." for p in parts) or norm == ".." or norm.startswith("..\\"):
|
|
raise ValueError("Path traversal using '..' is not allowed.")
|
|
|
|
if not self.valves.allow_alternate_data_streams and ":" in norm:
|
|
raise ValueError(
|
|
"':' is not allowed in SMB paths because Alternate Data Streams are disabled."
|
|
)
|
|
|
|
return "\\".join(parts)
|
|
|
|
def _base_unc(self, uv: "Tools.UserValves") -> str:
|
|
server = uv.server.strip()
|
|
share = uv.share.strip()
|
|
root = self._normalize_relative_path(uv.root_path, allow_empty=True)
|
|
|
|
base = rf"\\{server}\{share}"
|
|
if root:
|
|
base = ntpath.join(base, root)
|
|
return ntpath.normpath(base)
|
|
|
|
def _unc(self, uv: "Tools.UserValves", relative_path: Optional[str] = "") -> str:
|
|
rel = self._normalize_relative_path(relative_path, allow_empty=True)
|
|
base = self._base_unc(uv)
|
|
return ntpath.join(base, rel) if rel else base
|
|
|
|
def _display_path(self, relative_path: Optional[str]) -> str:
|
|
rel = self._normalize_relative_path(relative_path, allow_empty=True)
|
|
return rel.replace("\\", "/") if rel else "."
|
|
|
|
def _auth_username(self, uv: "Tools.UserValves") -> str:
|
|
username = uv.username.strip()
|
|
domain = uv.domain.strip()
|
|
if username and domain and "\\" not in username and "@" not in username:
|
|
return f"{domain}\\{username}"
|
|
return username
|
|
|
|
def _reset_cache_quietly(self):
|
|
try:
|
|
smbclient.reset_connection_cache()
|
|
except Exception:
|
|
# Cleanup should not mask the actual SMB operation result.
|
|
pass
|
|
|
|
@contextmanager
|
|
def _session(self, uv: "Tools.UserValves"):
|
|
"""
|
|
Register exactly one user's SMB session for the duration of an operation.
|
|
|
|
smbclient maintains a global connection cache. The lock + cache reset
|
|
prevents one Open WebUI user's session from being reused by another.
|
|
"""
|
|
self._validate_config(uv)
|
|
|
|
with self._smb_lock:
|
|
self._reset_cache_quietly()
|
|
try:
|
|
kwargs = {
|
|
"port": uv.port,
|
|
"encrypt": uv.encrypt,
|
|
"connection_timeout": uv.connection_timeout,
|
|
}
|
|
|
|
username = self._auth_username(uv)
|
|
if username or uv.password:
|
|
kwargs["username"] = username
|
|
kwargs["password"] = uv.password
|
|
|
|
smbclient.register_session(uv.server.strip(), **kwargs)
|
|
yield
|
|
finally:
|
|
self._reset_cache_quietly()
|
|
|
|
def _require_create(self, uv: "Tools.UserValves"):
|
|
if not self.valves.allow_create_files:
|
|
raise PermissionError("Creating files is disabled by the administrator.")
|
|
if not uv.allow_write_new_files:
|
|
raise PermissionError(
|
|
"Creating files is disabled in your SMB UserValves. "
|
|
"Enable 'allow_write_new_files' to permit it."
|
|
)
|
|
|
|
def _require_modify(self, uv: "Tools.UserValves"):
|
|
if not self.valves.allow_modify_files:
|
|
raise PermissionError("Modifying files is disabled by the administrator.")
|
|
if not uv.allow_modify_existing_files:
|
|
raise PermissionError(
|
|
"Modifying existing files is disabled in your SMB UserValves. "
|
|
"Enable 'allow_modify_existing_files' to permit it."
|
|
)
|
|
|
|
def _require_delete(self, uv: "Tools.UserValves"):
|
|
if not self.valves.allow_delete_paths:
|
|
raise PermissionError("Deleting paths is disabled by the administrator.")
|
|
if not uv.allow_delete_existing_paths:
|
|
raise PermissionError(
|
|
"Deleting existing paths is disabled in your SMB UserValves. "
|
|
"Enable 'allow_delete_existing_paths' to permit it."
|
|
)
|
|
|
|
def _require_directory_ops(self, uv: "Tools.UserValves"):
|
|
if not self.valves.allow_directory_operations:
|
|
raise PermissionError("Directory operations are disabled by the administrator.")
|
|
if not uv.allow_directory_operations:
|
|
raise PermissionError(
|
|
"Directory operations are disabled in your SMB UserValves. "
|
|
"Enable 'allow_directory_operations' to permit them."
|
|
)
|
|
|
|
def _require_move_copy(self, uv: "Tools.UserValves"):
|
|
if not self.valves.allow_move_copy:
|
|
raise PermissionError("Move/copy operations are disabled by the administrator.")
|
|
if not uv.allow_move_copy_operations:
|
|
raise PermissionError(
|
|
"Move/copy operations are disabled in your SMB UserValves. "
|
|
"Enable 'allow_move_copy_operations' to permit them."
|
|
)
|
|
|
|
def _json(self, data: Any) -> str:
|
|
return json.dumps(data, ensure_ascii=False, indent=2, default=str)
|
|
|
|
def _format_timestamp(self, timestamp: float) -> str:
|
|
return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()
|
|
|
|
def _stat_dict(self, relative_path: str, st: Any) -> dict:
|
|
is_dir = stat.S_ISDIR(st.st_mode)
|
|
return {
|
|
"path": self._display_path(relative_path),
|
|
"name": ntpath.basename(relative_path) if relative_path else ".",
|
|
"type": "directory" if is_dir else "file",
|
|
"size": 0 if is_dir else st.st_size,
|
|
"modified_utc": self._format_timestamp(st.st_mtime),
|
|
"created_utc": self._format_timestamp(st.st_ctime),
|
|
"accessed_utc": self._format_timestamp(st.st_atime),
|
|
}
|
|
|
|
def _exists(self, unc_path: str) -> tuple[bool, Any]:
|
|
try:
|
|
st = smbclient.stat(unc_path)
|
|
return True, st
|
|
except FileNotFoundError:
|
|
return False, None
|
|
except OSError as exc:
|
|
# SMB servers may map "not found" into a generic OSError.
|
|
text = str(exc).lower()
|
|
if "no such file" in text or "not found" in text or "0xc0000034" in text:
|
|
return False, None
|
|
raise
|
|
|
|
def _decode_payload(
|
|
self, content: str, encoding: str, content_is_base64: bool
|
|
) -> bytes:
|
|
if content_is_base64:
|
|
try:
|
|
payload = base64.b64decode(content, validate=True)
|
|
except Exception as exc:
|
|
raise ValueError(f"content is not valid base64: {exc}") from exc
|
|
else:
|
|
try:
|
|
payload = content.encode(encoding)
|
|
except LookupError as exc:
|
|
raise ValueError(f"Unknown text encoding '{encoding}'.") from exc
|
|
|
|
if len(payload) > self.valves.max_write_bytes:
|
|
raise ValueError(
|
|
f"Payload is {len(payload)} bytes; administrator limit is "
|
|
f"{self.valves.max_write_bytes} bytes."
|
|
)
|
|
return payload
|
|
|
|
def _parent_relative(self, relative_path: str) -> str:
|
|
parent = ntpath.dirname(relative_path)
|
|
return "" if parent in {"", "."} else parent
|
|
|
|
def _ensure_parent_directory(
|
|
self,
|
|
uv: "Tools.UserValves",
|
|
relative_path: str,
|
|
create_parents: bool,
|
|
):
|
|
parent_rel = self._parent_relative(relative_path)
|
|
if not parent_rel:
|
|
return
|
|
|
|
parent_unc = self._unc(uv, parent_rel)
|
|
exists, st = self._exists(parent_unc)
|
|
if exists:
|
|
if not stat.S_ISDIR(st.st_mode):
|
|
raise NotADirectoryError(
|
|
f"Parent path '{self._display_path(parent_rel)}' is not a directory."
|
|
)
|
|
return
|
|
|
|
if not create_parents:
|
|
raise FileNotFoundError(
|
|
f"Parent directory '{self._display_path(parent_rel)}' does not exist."
|
|
)
|
|
|
|
self._require_directory_ops(uv)
|
|
smbclient.makedirs(parent_unc, exist_ok=True)
|
|
|
|
def _error_text(self, exc: Exception) -> str:
|
|
return f"{exc.__class__.__name__}: {exc}"
|
|
|
|
async def _execute(
|
|
self,
|
|
emitter: EventEmitter,
|
|
in_progress: str,
|
|
success: str,
|
|
func: Callable[..., str],
|
|
*args,
|
|
) -> str:
|
|
await emitter.progress_update(in_progress)
|
|
try:
|
|
result = await asyncio.to_thread(func, *args)
|
|
except Exception as exc:
|
|
message = self._error_text(exc)
|
|
await emitter.error_update(message)
|
|
return f"Error: {message}"
|
|
|
|
await emitter.success_update(success)
|
|
return result
|
|
|
|
|
|
# =========================================================================
|
|
# Tool documentation
|
|
# =========================================================================
|
|
|
|
def get_tool_info(
|
|
self,
|
|
__user__: dict,
|
|
) -> str:
|
|
"""
|
|
Return the SMB tool operating guide, safety rules and current capabilities.
|
|
|
|
This function is intended to help Open WebUI instruct the model how to use
|
|
the tool safely and correctly.
|
|
|
|
No passwords, usernames, server names, share names, domains, credentials,
|
|
file contents, or other secret values are returned.
|
|
"""
|
|
|
|
try:
|
|
uv = self._user_valves(__user__)
|
|
|
|
create_allowed = bool(
|
|
self.valves.enabled
|
|
and self.valves.allow_create_files
|
|
and uv.allow_write_new_files
|
|
)
|
|
modify_allowed = bool(
|
|
self.valves.enabled
|
|
and self.valves.allow_modify_files
|
|
and uv.allow_modify_existing_files
|
|
)
|
|
delete_allowed = bool(
|
|
self.valves.enabled
|
|
and self.valves.allow_delete_paths
|
|
and uv.allow_delete_existing_paths
|
|
)
|
|
directory_ops_allowed = bool(
|
|
self.valves.enabled
|
|
and self.valves.allow_directory_operations
|
|
and uv.allow_directory_operations
|
|
)
|
|
move_copy_allowed = bool(
|
|
self.valves.enabled
|
|
and self.valves.allow_move_copy
|
|
and uv.allow_move_copy_operations
|
|
)
|
|
recursive_delete_allowed = bool(
|
|
delete_allowed
|
|
and directory_ops_allowed
|
|
and self.valves.allow_recursive_delete
|
|
)
|
|
|
|
info = {
|
|
"tool": "SMB File Operations Tool",
|
|
"version": "0.1.0",
|
|
"purpose": (
|
|
"Provide controlled SMB2/SMB3 file operations inside the "
|
|
"current user's configured SMB share and optional root_path."
|
|
),
|
|
"path_model": {
|
|
"paths_are_relative": True,
|
|
"logical_root": (
|
|
"All tool paths are relative to the configured SMB share "
|
|
"and optional UserValves.root_path."
|
|
),
|
|
"absolute_paths_allowed": False,
|
|
"unc_paths_allowed": False,
|
|
"parent_traversal_allowed": False,
|
|
"alternate_data_streams_allowed_by_admin": (
|
|
self.valves.allow_alternate_data_streams
|
|
),
|
|
},
|
|
"current_user_capabilities": {
|
|
"tool_enabled_by_admin": self.valves.enabled,
|
|
"smb_connection_configured": bool(
|
|
uv.server.strip() and uv.share.strip()
|
|
),
|
|
"authentication_identity_configured": bool(
|
|
uv.username.strip() or uv.password
|
|
),
|
|
"root_path_configured": bool(uv.root_path.strip()),
|
|
"smb_encryption_requested": uv.encrypt,
|
|
"create_new_files_allowed": create_allowed,
|
|
"modify_existing_files_allowed": modify_allowed,
|
|
"delete_existing_paths_allowed": delete_allowed,
|
|
"directory_operations_allowed": directory_ops_allowed,
|
|
"move_copy_operations_allowed": move_copy_allowed,
|
|
"recursive_delete_allowed": recursive_delete_allowed,
|
|
},
|
|
"administrator_limits": {
|
|
"max_read_bytes": self.valves.max_read_bytes,
|
|
"max_write_bytes": self.valves.max_write_bytes,
|
|
"max_list_entries": self.valves.max_list_entries,
|
|
"max_search_results": self.valves.max_search_results,
|
|
"max_search_depth": self.valves.max_search_depth,
|
|
},
|
|
"safety_rules": [
|
|
(
|
|
"Use get_tool_info when capabilities or safe operating "
|
|
"rules are unclear."
|
|
),
|
|
(
|
|
"Never invent a path when the target is uncertain. "
|
|
"Use list_directory, search_files or stat_path first."
|
|
),
|
|
(
|
|
"Treat all paths as relative to the configured SMB "
|
|
"share/root_path. Do not construct UNC, drive-letter or "
|
|
"absolute paths."
|
|
),
|
|
(
|
|
"Do not attempt '..' path traversal. The tool rejects "
|
|
"parent-directory traversal."
|
|
),
|
|
(
|
|
"Read a file before modifying it when the existing "
|
|
"contents or exact target text matter."
|
|
),
|
|
(
|
|
"Prefer update_file for small exact text changes instead "
|
|
"of rewriting an entire existing text file."
|
|
),
|
|
(
|
|
"When using update_file for a sensitive or targeted "
|
|
"change, set expected_replacements when the expected "
|
|
"number of matches is known."
|
|
),
|
|
(
|
|
"Do not set overwrite=true unless replacing an existing "
|
|
"destination is explicitly intended."
|
|
),
|
|
(
|
|
"Do not use create_if_missing=true with append_file "
|
|
"unless creating a missing file is explicitly intended."
|
|
),
|
|
(
|
|
"Use create_parents=true only when missing destination "
|
|
"directories should intentionally be created."
|
|
),
|
|
(
|
|
"Deletion requires both administrator permission and "
|
|
"current-user permission."
|
|
),
|
|
(
|
|
"Recursive directory deletion additionally requires the "
|
|
"administrator recursive-delete master switch."
|
|
),
|
|
(
|
|
"Before moving, copying, overwriting or deleting an "
|
|
"important path, use stat_path or list_directory when "
|
|
"necessary to verify the target."
|
|
),
|
|
(
|
|
"Large reads should be performed in chunks with offset "
|
|
"and max_bytes. Administrator limits always take precedence."
|
|
),
|
|
(
|
|
"Use as_base64=true for binary reads and "
|
|
"content_is_base64=true for binary writes."
|
|
),
|
|
(
|
|
"Credentials, usernames, passwords, domains, SMB server "
|
|
"names and share names are not returned by get_tool_info."
|
|
),
|
|
],
|
|
"recommended_workflow": [
|
|
(
|
|
"Call test_connection when SMB connectivity or the current "
|
|
"configuration needs to be verified."
|
|
),
|
|
(
|
|
"Use list_directory to inspect a known directory."
|
|
),
|
|
(
|
|
"Use search_files when the location of a file is unknown."
|
|
),
|
|
(
|
|
"Use stat_path to verify file/directory type, size and "
|
|
"timestamps before an operation when relevant."
|
|
),
|
|
(
|
|
"Use read_file before editing when current contents matter."
|
|
),
|
|
(
|
|
"Use write_file for new files; leave overwrite=false unless "
|
|
"replacement is intended."
|
|
),
|
|
(
|
|
"Use update_file for exact text replacement in existing "
|
|
"text files."
|
|
),
|
|
(
|
|
"Use append_file only when adding data to the end of a "
|
|
"file is semantically correct."
|
|
),
|
|
(
|
|
"Use make_directory for intentional directory creation."
|
|
),
|
|
(
|
|
"Use copy_file or move_path only when move/copy capability "
|
|
"is enabled for the current user."
|
|
),
|
|
(
|
|
"Use delete_path only when deletion is explicitly required "
|
|
"and permitted."
|
|
),
|
|
],
|
|
"functions": {
|
|
"get_tool_info": (
|
|
"Return operating instructions, safety rules, limits and "
|
|
"effective current-user capabilities."
|
|
),
|
|
"test_connection": (
|
|
"Test SMB connectivity, authentication and access to the "
|
|
"configured logical root."
|
|
),
|
|
"list_directory": (
|
|
"List files and directories in a relative SMB directory, "
|
|
"optionally using an SMB wildcard pattern."
|
|
),
|
|
"stat_path": (
|
|
"Return metadata for a relative file or directory."
|
|
),
|
|
"read_file": (
|
|
"Read text or binary/base64 file data, with offset and "
|
|
"chunk-size support."
|
|
),
|
|
"search_files": (
|
|
"Recursively search for file/directory names using "
|
|
"wildcard patterns within administrator limits."
|
|
),
|
|
"write_file": (
|
|
"Create a new file or explicitly overwrite an existing "
|
|
"file when permissions allow."
|
|
),
|
|
"append_file": (
|
|
"Append text or binary data to an existing file, or "
|
|
"optionally create it when permitted."
|
|
),
|
|
"update_file": (
|
|
"Perform controlled exact-text replacement in an existing "
|
|
"text file."
|
|
),
|
|
"make_directory": (
|
|
"Create a directory and optionally missing parent "
|
|
"directories when permitted."
|
|
),
|
|
"copy_file": (
|
|
"Copy a file inside the configured SMB logical root."
|
|
),
|
|
"move_path": (
|
|
"Rename or move a file/directory inside the configured "
|
|
"SMB logical root."
|
|
),
|
|
"delete_path": (
|
|
"Delete a file or directory when user/admin permissions "
|
|
"allow; recursive deletion has an additional admin gate."
|
|
),
|
|
},
|
|
"examples": [
|
|
{
|
|
"goal": "Find a report when its location is unknown",
|
|
"workflow": [
|
|
"search_files(pattern='report*')",
|
|
"stat_path(path='<returned path>')",
|
|
"read_file(path='<returned path>')",
|
|
],
|
|
},
|
|
{
|
|
"goal": "Safely change one configuration value",
|
|
"workflow": [
|
|
"read_file(path='config/app.conf')",
|
|
(
|
|
"update_file(path='config/app.conf', "
|
|
"old_text='<exact old text>', "
|
|
"new_text='<replacement>', "
|
|
"expected_replacements=1)"
|
|
),
|
|
],
|
|
},
|
|
{
|
|
"goal": "Create a new text document",
|
|
"workflow": [
|
|
(
|
|
"write_file(path='notes/new-note.txt', "
|
|
"content='<text>', overwrite=False)"
|
|
)
|
|
],
|
|
},
|
|
{
|
|
"goal": "Read a large file in chunks",
|
|
"workflow": [
|
|
(
|
|
"read_file(path='<file>', offset=0, "
|
|
"max_bytes=<allowed chunk size>)"
|
|
),
|
|
(
|
|
"Continue with offset increased by the returned "
|
|
"bytes_returned while truncated=true."
|
|
),
|
|
],
|
|
},
|
|
],
|
|
}
|
|
|
|
return self._json(info)
|
|
|
|
except Exception as exc:
|
|
return self._json(
|
|
{
|
|
"tool": "SMB File Operations Tool",
|
|
"version": "0.1.0",
|
|
"available": False,
|
|
"error": self._error_text(exc),
|
|
"note": (
|
|
"No credentials or secret configuration values are "
|
|
"included in this response."
|
|
),
|
|
}
|
|
)
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Read-only operations
|
|
# ---------------------------------------------------------------------
|
|
|
|
def _test_connection_sync(self, __user__: dict) -> str:
|
|
uv = self._user_valves(__user__)
|
|
with self._session(uv):
|
|
root = self._base_unc(uv)
|
|
st = smbclient.stat(root)
|
|
data = {
|
|
"connected": True,
|
|
"server": uv.server.strip(),
|
|
"port": uv.port,
|
|
"share": uv.share.strip(),
|
|
"root_path": self._display_path(uv.root_path),
|
|
"root_type": "directory" if stat.S_ISDIR(st.st_mode) else "file",
|
|
"encryption_requested": uv.encrypt,
|
|
}
|
|
return self._json(data)
|
|
|
|
async def test_connection(
|
|
self,
|
|
__user__: dict,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Tests the current user's SMB connection and configured logical root.
|
|
|
|
:return: JSON connection status. Credentials are never returned.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
"Testing SMB connection...",
|
|
"SMB connection successful.",
|
|
self._test_connection_sync,
|
|
__user__,
|
|
)
|
|
|
|
def _list_directory_sync(
|
|
self,
|
|
__user__: dict,
|
|
path: str,
|
|
pattern: str,
|
|
include_hidden: bool,
|
|
) -> str:
|
|
uv = self._user_valves(__user__)
|
|
rel = self._normalize_relative_path(path, allow_empty=True)
|
|
|
|
with self._session(uv):
|
|
unc = self._unc(uv, rel)
|
|
results = []
|
|
truncated = False
|
|
|
|
with smbclient.scandir(unc, search_pattern=pattern or "*") as entries:
|
|
for entry in entries:
|
|
name = entry.name
|
|
if not include_hidden and name.startswith("."):
|
|
continue
|
|
|
|
child_rel = ntpath.join(rel, name) if rel else name
|
|
entry_stat = entry.stat()
|
|
results.append(self._stat_dict(child_rel, entry_stat))
|
|
|
|
if len(results) >= self.valves.max_list_entries:
|
|
truncated = True
|
|
break
|
|
|
|
results.sort(key=lambda item: (item["type"] != "directory", item["name"].lower()))
|
|
return self._json(
|
|
{
|
|
"path": self._display_path(rel),
|
|
"pattern": pattern or "*",
|
|
"count": len(results),
|
|
"truncated": truncated,
|
|
"entries": results,
|
|
}
|
|
)
|
|
|
|
async def list_directory(
|
|
self,
|
|
path: str = "",
|
|
pattern: str = "*",
|
|
include_hidden: bool = True,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Lists files and directories under a directory.
|
|
|
|
Paths are relative to UserValves.root_path. Use an empty path for the root.
|
|
|
|
:param path: Relative directory path.
|
|
:param pattern: SMB wildcard pattern, for example '*.txt' or '*'.
|
|
:param include_hidden: Include dot-prefixed names in results.
|
|
:return: JSON directory listing with type, size, and timestamps.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Listing SMB directory '{path or '.'}'...",
|
|
"Directory listing complete.",
|
|
self._list_directory_sync,
|
|
__user__,
|
|
path,
|
|
pattern,
|
|
include_hidden,
|
|
)
|
|
|
|
def _stat_path_sync(self, __user__: dict, path: str) -> str:
|
|
uv = self._user_valves(__user__)
|
|
rel = self._normalize_relative_path(path, allow_empty=True)
|
|
|
|
with self._session(uv):
|
|
st = smbclient.stat(self._unc(uv, rel))
|
|
return self._json(self._stat_dict(rel, st))
|
|
|
|
async def stat_path(
|
|
self,
|
|
path: str,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Returns metadata for a file or directory.
|
|
|
|
:param path: Relative file/directory path.
|
|
:return: JSON metadata.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Reading metadata for '{path}'...",
|
|
"Metadata read complete.",
|
|
self._stat_path_sync,
|
|
__user__,
|
|
path,
|
|
)
|
|
|
|
def _read_file_sync(
|
|
self,
|
|
__user__: dict,
|
|
path: str,
|
|
encoding: str,
|
|
as_base64: bool,
|
|
offset: int,
|
|
max_bytes: int,
|
|
) -> str:
|
|
if offset < 0:
|
|
raise ValueError("offset must be >= 0.")
|
|
if max_bytes <= 0:
|
|
raise ValueError("max_bytes must be > 0.")
|
|
|
|
uv = self._user_valves(__user__)
|
|
rel = self._normalize_relative_path(path, allow_empty=False)
|
|
read_limit = min(max_bytes, self.valves.max_read_bytes)
|
|
|
|
with self._session(uv):
|
|
unc = self._unc(uv, rel)
|
|
st = smbclient.stat(unc)
|
|
if stat.S_ISDIR(st.st_mode):
|
|
raise IsADirectoryError(f"'{self._display_path(rel)}' is a directory.")
|
|
|
|
with smbclient.open_file(unc, mode="rb") as fd:
|
|
fd.seek(offset)
|
|
data = fd.read(read_limit + 1)
|
|
|
|
truncated = len(data) > read_limit
|
|
if truncated:
|
|
data = data[:read_limit]
|
|
|
|
if as_base64:
|
|
content = base64.b64encode(data).decode("ascii")
|
|
output_encoding = "base64"
|
|
else:
|
|
try:
|
|
content = data.decode(encoding)
|
|
except LookupError as exc:
|
|
raise ValueError(f"Unknown text encoding '{encoding}'.") from exc
|
|
except UnicodeDecodeError as exc:
|
|
raise ValueError(
|
|
f"File cannot be decoded as '{encoding}' at this offset. "
|
|
"Use as_base64=true for binary data or choose another encoding."
|
|
) from exc
|
|
output_encoding = encoding
|
|
|
|
return self._json(
|
|
{
|
|
"path": self._display_path(rel),
|
|
"file_size": st.st_size,
|
|
"offset": offset,
|
|
"bytes_returned": len(data),
|
|
"truncated": truncated or (offset + len(data) < st.st_size),
|
|
"encoding": output_encoding,
|
|
"content": content,
|
|
}
|
|
)
|
|
|
|
async def read_file(
|
|
self,
|
|
path: str,
|
|
encoding: str = "utf-8",
|
|
as_base64: bool = False,
|
|
offset: int = 0,
|
|
max_bytes: int = 2 * 1024 * 1024,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Reads a file as text or base64.
|
|
|
|
Large files can be read in chunks with offset/max_bytes. The administrator's
|
|
max_read_bytes limit always takes precedence.
|
|
|
|
:param path: Relative file path.
|
|
:param encoding: Text encoding when as_base64 is false.
|
|
:param as_base64: Return bytes encoded as base64 instead of text.
|
|
:param offset: Byte offset at which reading starts.
|
|
:param max_bytes: Maximum bytes requested in this call.
|
|
:return: JSON containing file data and chunk metadata.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Reading SMB file '{path}'...",
|
|
"File read complete.",
|
|
self._read_file_sync,
|
|
__user__,
|
|
path,
|
|
encoding,
|
|
as_base64,
|
|
offset,
|
|
max_bytes,
|
|
)
|
|
|
|
def _search_files_sync(
|
|
self,
|
|
__user__: dict,
|
|
start_path: str,
|
|
pattern: str,
|
|
max_depth: int,
|
|
include_directories: bool,
|
|
) -> str:
|
|
if max_depth < 0:
|
|
raise ValueError("max_depth must be >= 0.")
|
|
max_depth = min(max_depth, self.valves.max_search_depth)
|
|
|
|
uv = self._user_valves(__user__)
|
|
start_rel = self._normalize_relative_path(start_path, allow_empty=True)
|
|
results = []
|
|
truncated = False
|
|
|
|
with self._session(uv):
|
|
stack = [(start_rel, 0)]
|
|
|
|
while stack and len(results) < self.valves.max_search_results:
|
|
current_rel, depth = stack.pop()
|
|
current_unc = self._unc(uv, current_rel)
|
|
|
|
with smbclient.scandir(current_unc) as entries:
|
|
for entry in entries:
|
|
child_rel = (
|
|
ntpath.join(current_rel, entry.name)
|
|
if current_rel
|
|
else entry.name
|
|
)
|
|
child_stat = entry.stat()
|
|
is_dir = stat.S_ISDIR(child_stat.st_mode)
|
|
|
|
if fnmatch.fnmatch(entry.name.lower(), pattern.lower()):
|
|
if include_directories or not is_dir:
|
|
results.append(self._stat_dict(child_rel, child_stat))
|
|
if len(results) >= self.valves.max_search_results:
|
|
truncated = True
|
|
break
|
|
|
|
if is_dir and depth < max_depth:
|
|
stack.append((child_rel, depth + 1))
|
|
|
|
if stack:
|
|
truncated = True
|
|
|
|
results.sort(key=lambda item: item["path"].lower())
|
|
return self._json(
|
|
{
|
|
"start_path": self._display_path(start_rel),
|
|
"pattern": pattern,
|
|
"max_depth": max_depth,
|
|
"count": len(results),
|
|
"truncated": truncated,
|
|
"results": results,
|
|
}
|
|
)
|
|
|
|
async def search_files(
|
|
self,
|
|
pattern: str,
|
|
start_path: str = "",
|
|
max_depth: int = 4,
|
|
include_directories: bool = False,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Recursively searches for names matching a wildcard pattern.
|
|
|
|
:param pattern: Filename wildcard, for example '*.csv', 'report*', or '*'.
|
|
:param start_path: Relative directory at which to begin.
|
|
:param max_depth: Recursive depth, capped by the administrator.
|
|
:param include_directories: Include matching directory names as results.
|
|
:return: JSON search results.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Searching SMB files for '{pattern}'...",
|
|
"SMB search complete.",
|
|
self._search_files_sync,
|
|
__user__,
|
|
start_path,
|
|
pattern,
|
|
max_depth,
|
|
include_directories,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Write / mutation operations
|
|
# ---------------------------------------------------------------------
|
|
|
|
def _write_file_sync(
|
|
self,
|
|
__user__: dict,
|
|
path: str,
|
|
content: str,
|
|
encoding: str,
|
|
content_is_base64: bool,
|
|
overwrite: bool,
|
|
create_parents: bool,
|
|
) -> str:
|
|
uv = self._user_valves(__user__)
|
|
rel = self._normalize_relative_path(path, allow_empty=False)
|
|
payload = self._decode_payload(content, encoding, content_is_base64)
|
|
|
|
with self._session(uv):
|
|
unc = self._unc(uv, rel)
|
|
exists, st = self._exists(unc)
|
|
|
|
if exists:
|
|
if stat.S_ISDIR(st.st_mode):
|
|
raise IsADirectoryError(f"'{self._display_path(rel)}' is a directory.")
|
|
if not overwrite:
|
|
raise FileExistsError(
|
|
f"'{self._display_path(rel)}' already exists. "
|
|
"Set overwrite=true to replace it."
|
|
)
|
|
self._require_modify(uv)
|
|
action = "overwritten"
|
|
else:
|
|
self._require_create(uv)
|
|
self._ensure_parent_directory(uv, rel, create_parents)
|
|
action = "created"
|
|
|
|
mode = "wb" if exists else "xb"
|
|
with smbclient.open_file(unc, mode=mode) as fd:
|
|
fd.write(payload)
|
|
|
|
return self._json(
|
|
{
|
|
"path": self._display_path(rel),
|
|
"action": action,
|
|
"bytes_written": len(payload),
|
|
"encoding": "base64 input" if content_is_base64 else encoding,
|
|
}
|
|
)
|
|
|
|
async def write_file(
|
|
self,
|
|
path: str,
|
|
content: str,
|
|
encoding: str = "utf-8",
|
|
content_is_base64: bool = False,
|
|
overwrite: bool = False,
|
|
create_parents: bool = False,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Creates a new file or explicitly overwrites an existing file.
|
|
|
|
Creating and overwriting are controlled separately by UserValves and
|
|
administrator Valves. overwrite=false is the safer default.
|
|
|
|
:param path: Relative destination file path.
|
|
:param content: Text content, or base64 when content_is_base64=true.
|
|
:param encoding: Encoding for text content.
|
|
:param content_is_base64: Decode content from base64 before writing.
|
|
:param overwrite: Permit replacement when the file already exists.
|
|
:param create_parents: Create missing parent directories if allowed.
|
|
:return: JSON operation result.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Writing SMB file '{path}'...",
|
|
"File write complete.",
|
|
self._write_file_sync,
|
|
__user__,
|
|
path,
|
|
content,
|
|
encoding,
|
|
content_is_base64,
|
|
overwrite,
|
|
create_parents,
|
|
)
|
|
|
|
def _append_file_sync(
|
|
self,
|
|
__user__: dict,
|
|
path: str,
|
|
content: str,
|
|
encoding: str,
|
|
content_is_base64: bool,
|
|
create_if_missing: bool,
|
|
) -> str:
|
|
uv = self._user_valves(__user__)
|
|
rel = self._normalize_relative_path(path, allow_empty=False)
|
|
payload = self._decode_payload(content, encoding, content_is_base64)
|
|
|
|
with self._session(uv):
|
|
unc = self._unc(uv, rel)
|
|
exists, st = self._exists(unc)
|
|
|
|
if exists:
|
|
if stat.S_ISDIR(st.st_mode):
|
|
raise IsADirectoryError(f"'{self._display_path(rel)}' is a directory.")
|
|
self._require_modify(uv)
|
|
action = "appended"
|
|
else:
|
|
if not create_if_missing:
|
|
raise FileNotFoundError(
|
|
f"'{self._display_path(rel)}' does not exist."
|
|
)
|
|
self._require_create(uv)
|
|
self._ensure_parent_directory(uv, rel, create_parents=False)
|
|
action = "created_and_appended"
|
|
|
|
with smbclient.open_file(unc, mode="ab") as fd:
|
|
fd.write(payload)
|
|
|
|
new_size = smbclient.stat(unc).st_size
|
|
return self._json(
|
|
{
|
|
"path": self._display_path(rel),
|
|
"action": action,
|
|
"bytes_appended": len(payload),
|
|
"new_size": new_size,
|
|
}
|
|
)
|
|
|
|
async def append_file(
|
|
self,
|
|
path: str,
|
|
content: str,
|
|
encoding: str = "utf-8",
|
|
content_is_base64: bool = False,
|
|
create_if_missing: bool = False,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Appends bytes/text to an existing file.
|
|
|
|
:param path: Relative file path.
|
|
:param content: Text content, or base64 when content_is_base64=true.
|
|
:param encoding: Encoding for text input.
|
|
:param content_is_base64: Decode content from base64 before appending.
|
|
:param create_if_missing: Create the file if it does not exist.
|
|
:return: JSON operation result.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Appending to SMB file '{path}'...",
|
|
"Append complete.",
|
|
self._append_file_sync,
|
|
__user__,
|
|
path,
|
|
content,
|
|
encoding,
|
|
content_is_base64,
|
|
create_if_missing,
|
|
)
|
|
|
|
def _update_file_sync(
|
|
self,
|
|
__user__: dict,
|
|
path: str,
|
|
old_text: str,
|
|
new_text: str,
|
|
encoding: str,
|
|
replace_all: bool,
|
|
expected_replacements: int,
|
|
) -> str:
|
|
if old_text == "":
|
|
raise ValueError("old_text must not be empty.")
|
|
|
|
uv = self._user_valves(__user__)
|
|
self._require_modify(uv)
|
|
rel = self._normalize_relative_path(path, allow_empty=False)
|
|
|
|
with self._session(uv):
|
|
unc = self._unc(uv, rel)
|
|
st = smbclient.stat(unc)
|
|
if stat.S_ISDIR(st.st_mode):
|
|
raise IsADirectoryError(f"'{self._display_path(rel)}' is a directory.")
|
|
if st.st_size > self.valves.max_write_bytes:
|
|
raise ValueError(
|
|
f"File is {st.st_size} bytes; update limit is "
|
|
f"{self.valves.max_write_bytes} bytes."
|
|
)
|
|
|
|
with smbclient.open_file(unc, mode="rb") as fd:
|
|
raw = fd.read()
|
|
|
|
try:
|
|
text = raw.decode(encoding)
|
|
except LookupError as exc:
|
|
raise ValueError(f"Unknown text encoding '{encoding}'.") from exc
|
|
except UnicodeDecodeError as exc:
|
|
raise ValueError(
|
|
f"File cannot be decoded as '{encoding}'. update_file is text-only."
|
|
) from exc
|
|
|
|
occurrences = text.count(old_text)
|
|
if occurrences == 0:
|
|
raise ValueError("old_text was not found; file was not changed.")
|
|
|
|
replacements = occurrences if replace_all else 1
|
|
if expected_replacements >= 0 and replacements != expected_replacements:
|
|
raise ValueError(
|
|
f"Replacement count would be {replacements}, but "
|
|
f"expected_replacements={expected_replacements}. File was not changed."
|
|
)
|
|
|
|
updated = (
|
|
text.replace(old_text, new_text)
|
|
if replace_all
|
|
else text.replace(old_text, new_text, 1)
|
|
)
|
|
payload = updated.encode(encoding)
|
|
|
|
if len(payload) > self.valves.max_write_bytes:
|
|
raise ValueError(
|
|
f"Updated file would be {len(payload)} bytes; administrator limit is "
|
|
f"{self.valves.max_write_bytes} bytes."
|
|
)
|
|
|
|
with smbclient.open_file(unc, mode="wb") as fd:
|
|
fd.write(payload)
|
|
|
|
return self._json(
|
|
{
|
|
"path": self._display_path(rel),
|
|
"action": "updated",
|
|
"replacements": replacements,
|
|
"old_size": len(raw),
|
|
"new_size": len(payload),
|
|
"encoding": encoding,
|
|
}
|
|
)
|
|
|
|
async def update_file(
|
|
self,
|
|
path: str,
|
|
old_text: str,
|
|
new_text: str,
|
|
encoding: str = "utf-8",
|
|
replace_all: bool = False,
|
|
expected_replacements: int = -1,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Updates a text file by replacing exact text.
|
|
|
|
This is safer than asking a model to rewrite an entire file. Set
|
|
expected_replacements to a non-negative number when you want an extra
|
|
guard against changing more/fewer occurrences than intended.
|
|
|
|
:param path: Relative text file path.
|
|
:param old_text: Exact text to find.
|
|
:param new_text: Replacement text.
|
|
:param encoding: File text encoding.
|
|
:param replace_all: Replace every occurrence instead of only the first.
|
|
:param expected_replacements: -1 disables the count guard.
|
|
:return: JSON operation result.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Updating SMB file '{path}'...",
|
|
"File update complete.",
|
|
self._update_file_sync,
|
|
__user__,
|
|
path,
|
|
old_text,
|
|
new_text,
|
|
encoding,
|
|
replace_all,
|
|
expected_replacements,
|
|
)
|
|
|
|
def _make_directory_sync(
|
|
self,
|
|
__user__: dict,
|
|
path: str,
|
|
parents: bool,
|
|
exist_ok: bool,
|
|
) -> str:
|
|
uv = self._user_valves(__user__)
|
|
self._require_directory_ops(uv)
|
|
rel = self._normalize_relative_path(path, allow_empty=False)
|
|
|
|
with self._session(uv):
|
|
unc = self._unc(uv, rel)
|
|
if parents:
|
|
smbclient.makedirs(unc, exist_ok=exist_ok)
|
|
else:
|
|
if exist_ok:
|
|
exists, st = self._exists(unc)
|
|
if exists:
|
|
if not stat.S_ISDIR(st.st_mode):
|
|
raise FileExistsError(
|
|
f"'{self._display_path(rel)}' exists and is not a directory."
|
|
)
|
|
else:
|
|
smbclient.mkdir(unc)
|
|
else:
|
|
smbclient.mkdir(unc)
|
|
|
|
return self._json(
|
|
{
|
|
"path": self._display_path(rel),
|
|
"action": "directory_ready" if exist_ok else "directory_created",
|
|
"parents": parents,
|
|
}
|
|
)
|
|
|
|
async def make_directory(
|
|
self,
|
|
path: str,
|
|
parents: bool = False,
|
|
exist_ok: bool = False,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Creates a directory.
|
|
|
|
:param path: Relative directory path.
|
|
:param parents: Create missing parent directories.
|
|
:param exist_ok: Do not fail if the directory already exists.
|
|
:return: JSON operation result.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Creating SMB directory '{path}'...",
|
|
"Directory operation complete.",
|
|
self._make_directory_sync,
|
|
__user__,
|
|
path,
|
|
parents,
|
|
exist_ok,
|
|
)
|
|
|
|
def _copy_file_sync(
|
|
self,
|
|
__user__: dict,
|
|
source_path: str,
|
|
destination_path: str,
|
|
overwrite: bool,
|
|
create_parents: bool,
|
|
) -> str:
|
|
uv = self._user_valves(__user__)
|
|
self._require_move_copy(uv)
|
|
|
|
src_rel = self._normalize_relative_path(source_path, allow_empty=False)
|
|
dst_rel = self._normalize_relative_path(destination_path, allow_empty=False)
|
|
|
|
with self._session(uv):
|
|
src_unc = self._unc(uv, src_rel)
|
|
dst_unc = self._unc(uv, dst_rel)
|
|
|
|
src_st = smbclient.stat(src_unc)
|
|
if stat.S_ISDIR(src_st.st_mode):
|
|
raise IsADirectoryError("copy_file supports files only.")
|
|
|
|
dst_exists, dst_st = self._exists(dst_unc)
|
|
if dst_exists:
|
|
if stat.S_ISDIR(dst_st.st_mode):
|
|
raise IsADirectoryError(
|
|
f"Destination '{self._display_path(dst_rel)}' is a directory."
|
|
)
|
|
if not overwrite:
|
|
raise FileExistsError(
|
|
f"Destination '{self._display_path(dst_rel)}' already exists."
|
|
)
|
|
self._require_modify(uv)
|
|
else:
|
|
self._require_create(uv)
|
|
self._ensure_parent_directory(uv, dst_rel, create_parents)
|
|
|
|
smbclient.copyfile(src_unc, dst_unc)
|
|
return self._json(
|
|
{
|
|
"source": self._display_path(src_rel),
|
|
"destination": self._display_path(dst_rel),
|
|
"action": "copied",
|
|
"bytes": smbclient.stat(dst_unc).st_size,
|
|
"overwritten": dst_exists,
|
|
}
|
|
)
|
|
|
|
async def copy_file(
|
|
self,
|
|
source_path: str,
|
|
destination_path: str,
|
|
overwrite: bool = False,
|
|
create_parents: bool = False,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Copies a file within the configured SMB share/root.
|
|
|
|
:param source_path: Relative source file.
|
|
:param destination_path: Relative destination file.
|
|
:param overwrite: Permit replacing an existing destination.
|
|
:param create_parents: Create missing destination parent directories if allowed.
|
|
:return: JSON operation result.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Copying '{source_path}' to '{destination_path}'...",
|
|
"File copy complete.",
|
|
self._copy_file_sync,
|
|
__user__,
|
|
source_path,
|
|
destination_path,
|
|
overwrite,
|
|
create_parents,
|
|
)
|
|
|
|
def _move_path_sync(
|
|
self,
|
|
__user__: dict,
|
|
source_path: str,
|
|
destination_path: str,
|
|
overwrite: bool,
|
|
create_parents: bool,
|
|
) -> str:
|
|
uv = self._user_valves(__user__)
|
|
self._require_move_copy(uv)
|
|
|
|
src_rel = self._normalize_relative_path(source_path, allow_empty=False)
|
|
dst_rel = self._normalize_relative_path(destination_path, allow_empty=False)
|
|
|
|
with self._session(uv):
|
|
src_unc = self._unc(uv, src_rel)
|
|
dst_unc = self._unc(uv, dst_rel)
|
|
|
|
src_st = smbclient.stat(src_unc)
|
|
if stat.S_ISDIR(src_st.st_mode):
|
|
self._require_directory_ops(uv)
|
|
else:
|
|
self._require_modify(uv)
|
|
|
|
dst_exists, dst_st = self._exists(dst_unc)
|
|
if dst_exists:
|
|
if not overwrite:
|
|
raise FileExistsError(
|
|
f"Destination '{self._display_path(dst_rel)}' already exists."
|
|
)
|
|
# Overwriting an existing path is a modification of that target.
|
|
if stat.S_ISDIR(dst_st.st_mode):
|
|
self._require_directory_ops(uv)
|
|
else:
|
|
self._require_modify(uv)
|
|
else:
|
|
self._ensure_parent_directory(uv, dst_rel, create_parents)
|
|
|
|
if overwrite:
|
|
smbclient.replace(src_unc, dst_unc)
|
|
else:
|
|
smbclient.rename(src_unc, dst_unc)
|
|
|
|
return self._json(
|
|
{
|
|
"source": self._display_path(src_rel),
|
|
"destination": self._display_path(dst_rel),
|
|
"action": "moved",
|
|
"overwritten": dst_exists,
|
|
}
|
|
)
|
|
|
|
async def move_path(
|
|
self,
|
|
source_path: str,
|
|
destination_path: str,
|
|
overwrite: bool = False,
|
|
create_parents: bool = False,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Renames or moves a file/directory within the configured share/root.
|
|
|
|
:param source_path: Relative source path.
|
|
:param destination_path: Relative destination path.
|
|
:param overwrite: Replace an existing destination when supported.
|
|
:param create_parents: Create missing destination parents if allowed.
|
|
:return: JSON operation result.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Moving '{source_path}' to '{destination_path}'...",
|
|
"Move complete.",
|
|
self._move_path_sync,
|
|
__user__,
|
|
source_path,
|
|
destination_path,
|
|
overwrite,
|
|
create_parents,
|
|
)
|
|
|
|
def _delete_path_sync(
|
|
self,
|
|
__user__: dict,
|
|
path: str,
|
|
recursive: bool,
|
|
) -> str:
|
|
uv = self._user_valves(__user__)
|
|
self._require_delete(uv)
|
|
rel = self._normalize_relative_path(path, allow_empty=False)
|
|
|
|
with self._session(uv):
|
|
unc = self._unc(uv, rel)
|
|
st = smbclient.stat(unc)
|
|
is_dir = stat.S_ISDIR(st.st_mode)
|
|
|
|
if is_dir:
|
|
self._require_directory_ops(uv)
|
|
if recursive:
|
|
if not self.valves.allow_recursive_delete:
|
|
raise PermissionError(
|
|
"Recursive deletion is disabled by the administrator."
|
|
)
|
|
rmtree(unc)
|
|
action = "directory_tree_deleted"
|
|
else:
|
|
smbclient.rmdir(unc)
|
|
action = "empty_directory_deleted"
|
|
else:
|
|
smbclient.remove(unc)
|
|
action = "file_deleted"
|
|
|
|
return self._json(
|
|
{
|
|
"path": self._display_path(rel),
|
|
"action": action,
|
|
"recursive": bool(recursive and is_dir),
|
|
}
|
|
)
|
|
|
|
async def delete_path(
|
|
self,
|
|
path: str,
|
|
recursive: bool = False,
|
|
__user__: dict = None,
|
|
__event_emitter__: Callable[[dict], Any] = None,
|
|
) -> str:
|
|
"""
|
|
Deletes a file or directory.
|
|
|
|
Non-recursive directory deletion only removes empty directories. Recursive
|
|
deletion additionally requires the administrator's allow_recursive_delete.
|
|
|
|
:param path: Relative file/directory path. The configured root itself cannot be deleted.
|
|
:param recursive: Recursively delete a directory tree.
|
|
:return: JSON operation result.
|
|
"""
|
|
emitter = EventEmitter(__event_emitter__)
|
|
return await self._execute(
|
|
emitter,
|
|
f"Deleting SMB path '{path}'...",
|
|
"Delete operation complete.",
|
|
self._delete_path_sync,
|
|
__user__,
|
|
path,
|
|
recursive,
|
|
)
|