""" 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='')", "read_file(path='')", ], }, { "goal": "Safely change one configuration value", "workflow": [ "read_file(path='config/app.conf')", ( "update_file(path='config/app.conf', " "old_text='', " "new_text='', " "expected_replacements=1)" ), ], }, { "goal": "Create a new text document", "workflow": [ ( "write_file(path='notes/new-note.txt', " "content='', overwrite=False)" ) ], }, { "goal": "Read a large file in chunks", "workflow": [ ( "read_file(path='', offset=0, " "max_bytes=)" ), ( "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, )