commit e3448897eaef7fa743d878d3d42ad32e49ab1a68 Author: Jonas Bengtsson Date: Sun Aug 23 18:11:22 2026 +0200 🎉 Inital commit diff --git a/LICENSE.md b/LICENSE.md new file mode 100644 index 0000000..b43d74f --- /dev/null +++ b/LICENSE.md @@ -0,0 +1,52 @@ +The MIT License (MIT) + +===================== + + + +Copyright © `2026` `` + + + +Permission is hereby granted, free of charge, to any person + +obtaining a copy of this software and associated documentation + +files (the “Software”), to deal in the Software without + +restriction, including without limitation the rights to use, + +copy, modify, merge, publish, distribute, sublicense, and/or sell + +copies of the Software, and to permit persons to whom the + +Software is furnished to do so, subject to the following + +conditions: + + + +The above copyright notice and this permission notice shall be + +included in all copies or substantial portions of the Software. + + + +THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, + +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES + +OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + +NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + +HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + +WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + +FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR + +OTHER DEALINGS IN THE SOFTWARE. + + + diff --git a/README.md b/README.md new file mode 100644 index 0000000..e92982d --- /dev/null +++ b/README.md @@ -0,0 +1,1517 @@ +# Open WebUI SMB File Operations Tool + +An Open WebUI Workspace Tool for controlled SMB2/SMB3 file access. + +The tool allows an Open WebUI model to browse, read, create, update, copy, move, and delete files on an SMB share while keeping SMB connection details and permissions in per-user `UserValves`. + +Administrator `Valves` provide global policy controls such as server/share allow-lists, maximum file sizes, and whether write, delete, directory, and recursive operations are permitted. + +The implementation uses the Python `smbprotocol` package through its high-level `smbclient` API. + +## Features + +- SMB2/SMB3 access +- Per-user SMB server, share, username, password, and domain +- Optional SMB encryption +- Optional logical root directory inside a share +- Server and share allow-lists +- Path traversal protection +- Separate permissions for: + - creating files + - modifying existing files + - deleting files/directories + - directory operations + - move/copy operations +- Administrator-controlled read/write/search limits +- Text and Base64 file reads/writes +- Chunked file reads using byte offsets +- Exact text replacement for safer file updates +- Recursive wildcard file search +- Open WebUI status reporting through `EventEmitter` +- SMB session/cache isolation between Open WebUI users + +## Requirements + +The tool declares: + +```text +smbprotocol>=1.17.0 +``` + +The Open WebUI runtime hosting the tool must be able to reach the SMB server, normally on TCP port `445`. + +The SMB account must also have the filesystem/share permissions required for the operations you enable. + +## Installation + +Download or copy: + +```text +open_webui_smb_file_operations.py +``` + +into Open WebUI as a Workspace Tool. + +A typical workflow is: + +1. Open **Workspace** in Open WebUI. +2. Open **Tools**. +3. Create a new Tool. +4. Paste the contents of `open_webui_smb_file_operations.py`. +5. Save the Tool. +6. Configure the administrator `Valves`. +7. Give the intended users access to the Tool. +8. Each user configures their own `UserValves`. +9. Use `test_connection` before enabling write operations. + +The tool metadata specifies: + +```text +required_open_webui_version: 0.6.3 +requirements: smbprotocol>=1.17.0 +``` + +Ensure the Open WebUI Python environment is able to install/use the declared dependency. + +--- + +# Configuration + +## Administrator Valves + +`Valves` define the global policy for every user of the Tool. + +### `enabled` + +Default: + +```text +true +``` + +Global on/off switch. + +Setting this to `false` prevents SMB operations for all users. + +### `allowed_server_patterns` + +Default: + +```text +* +``` + +Comma-separated `fnmatch` patterns controlling which SMB servers users may configure. + +Examples: + +```text +nas01 +``` + +```text +nas01,fileserver01 +``` + +```text +*.internal.example +``` + +```text +nas*,10.20.* +``` + +For a production installation it is recommended to replace `*` with the specific SMB servers Open WebUI is supposed to access. + +Example: + +```text +nas01.internal.example,nas02.internal.example +``` + +### `allowed_share_patterns` + +Default: + +```text +* +``` + +Comma-separated patterns controlling which SMB shares may be used. + +Example: + +```text +Documents,Engineering,Projects-* +``` + +A production configuration should normally restrict this rather than leaving it as `*`. + +### `allow_create_files` + +Default: + +```text +true +``` + +Allows users to opt in to creating new files. + +A user must additionally enable: + +```text +allow_write_new_files +``` + +in their own `UserValves`. + +### `allow_modify_files` + +Default: + +```text +true +``` + +Allows users to opt in to changing existing files. + +This affects operations such as: + +- overwriting files +- appending data +- text replacement +- replacing an existing copy/move destination + +The user must additionally enable: + +```text +allow_modify_existing_files +``` + +### `allow_delete_paths` + +Default: + +```text +false +``` + +Global delete permission. + +Deletion is intentionally disabled by default. + +To allow deletion, both of these must be enabled: + +Administrator: + +```text +allow_delete_paths = true +``` + +User: + +```text +allow_delete_existing_paths = true +``` + +### `allow_directory_operations` + +Default: + +```text +true +``` + +Controls creation/manipulation of directories. + +Users must additionally enable: + +```text +allow_directory_operations +``` + +### `allow_move_copy` + +Default: + +```text +true +``` + +Allows users to opt in to copy and move operations. + +Users must additionally enable: + +```text +allow_move_copy_operations +``` + +### `allow_recursive_delete` + +Default: + +```text +false +``` + +Controls recursive directory deletion. + +This is a separate administrator protection in addition to normal delete permissions. + +Keeping this disabled is recommended unless recursive deletion is specifically required. + +### `allow_alternate_data_streams` + +Default: + +```text +false +``` + +Controls whether `:` is permitted in relative SMB paths. + +On Windows-compatible filesystems a colon can address an Alternate Data Stream. + +Normally leave this disabled. + +### `max_read_bytes` + +Default: + +```text +2097152 +``` + +2 MiB. + +Maximum amount returned by one `read_file` call. + +Large files can be read in chunks using `offset`. + +### `max_write_bytes` + +Default: + +```text +4194304 +``` + +4 MiB. + +Maximum payload accepted by a single write, append, or update operation. + +### `max_list_entries` + +Default: + +```text +500 +``` + +Maximum number of entries returned from a single directory listing. + +### `max_search_results` + +Default: + +```text +200 +``` + +Maximum results returned by `search_files`. + +### `max_search_depth` + +Default: + +```text +8 +``` + +Maximum recursive depth permitted for `search_files`. + +--- + +# UserValves + +Each user has their own SMB connection and permissions. + +## `server` + +SMB hostname or IP address. + +Example: + +```text +nas01.internal.example +``` + +Do not enter: + +```text +smb://nas01/Documents +``` + +or: + +```text +\\nas01\Documents +``` + +Only the server name/IP belongs here. + +## `port` + +Default: + +```text +445 +``` + +Normally this should remain `445`. + +## `share` + +SMB share name. + +Example: + +```text +Documents +``` + +For: + +```text +\\nas01\Documents +``` + +configure: + +```text +server = nas01 +share = Documents +``` + +## `username` + +SMB username. + +Example: + +```text +openwebui +``` + +## `password` + +Password for the SMB account. + +The field is presented as a password input in Open WebUI. + +## `domain` + +Optional Windows/Active Directory domain. + +Example: + +```text +CORP +``` + +With: + +```text +domain = CORP +username = jdoe +``` + +authentication is performed as: + +```text +CORP\jdoe +``` + +If the username already contains `\` or `@`, the tool leaves it unchanged. + +Examples: + +```text +CORP\jdoe +``` + +or: + +```text +jdoe@example.com +``` + +## `root_path` + +Optional directory inside the SMB share that becomes the logical filesystem root for this Tool. + +Example: + +```text +OpenWebUI +``` + +With: + +```text +server = nas01 +share = Documents +root_path = OpenWebUI +``` + +the actual SMB base becomes: + +```text +\\nas01\Documents\OpenWebUI +``` + +The model only supplies paths relative to this location. + +For example: + +```text +projects/report.md +``` + +maps to: + +```text +\\nas01\Documents\OpenWebUI\projects\report.md +``` + +Using `root_path` is strongly recommended because it limits the portion of the share exposed to the Tool. + +## `encrypt` + +Default: + +```text +false +``` + +When enabled, the SMB connection requests SMB encryption. + +The SMB server must support the required SMB3 encryption features. + +Consider enabling this when SMB traffic crosses networks where confidentiality cannot otherwise be guaranteed. + +## `connection_timeout` + +Default: + +```text +30 +``` + +SMB connection timeout in seconds. + +--- + +# User Write Permissions + +All mutation permissions default to disabled for users. + +This means the initial configuration is effectively read-only. + +## `allow_write_new_files` + +Allows creation of new files. + +Requires administrator: + +```text +allow_create_files = true +``` + +## `allow_modify_existing_files` + +Allows changes to existing files. + +This includes: + +- overwrite +- append +- `update_file` +- replacing existing copy/move destinations + +Requires administrator: + +```text +allow_modify_files = true +``` + +## `allow_delete_existing_paths` + +Allows deletion. + +Requires administrator: + +```text +allow_delete_paths = true +``` + +## `allow_directory_operations` + +Allows directory creation and directory-related move/delete operations. + +Requires administrator: + +```text +allow_directory_operations = true +``` + +## `allow_move_copy_operations` + +Allows copy and move operations. + +Requires administrator: + +```text +allow_move_copy = true +``` + +--- + +# Recommended Initial Configuration + +For initial testing, keep the Tool read-only. + +Administrator: + +```text +enabled = true + +allowed_server_patterns = nas01.internal.example +allowed_share_patterns = OpenWebUI + +allow_create_files = true +allow_modify_files = true +allow_delete_paths = false +allow_directory_operations = true +allow_move_copy = true +allow_recursive_delete = false +allow_alternate_data_streams = false + +max_read_bytes = 2097152 +max_write_bytes = 4194304 +max_list_entries = 500 +max_search_results = 200 +max_search_depth = 8 +``` + +User: + +```text +server = nas01.internal.example +port = 445 +share = OpenWebUI +username = openwebui +password = +domain = +root_path = +encrypt = true +connection_timeout = 30 + +allow_write_new_files = false +allow_modify_existing_files = false +allow_delete_existing_paths = false +allow_directory_operations = false +allow_move_copy_operations = false +``` + +Run: + +```text +test_connection +``` + +before enabling any write permissions. + +--- + +# Available Tool Functions + +## `test_connection` + +Tests: + +- SMB connectivity +- authentication +- share access +- logical `root_path` + +No credentials are returned in the result. + +Example request: + +```text +Test the SMB connection. +``` + +--- + +## `list_directory` + +Lists files and directories. + +Parameters include: + +```text +path +pattern +include_hidden +``` + +Example: + +```text +List the root directory. +``` + +Example: + +```text +List all PDF files in Reports. +``` + +Equivalent tool arguments could be: + +```text +path = Reports +pattern = *.pdf +``` + +The result includes: + +- name +- relative path +- file/directory type +- size +- created time +- modified time +- accessed time + +--- + +## `stat_path` + +Returns metadata for a file or directory. + +Example: + +```text +Show information about Projects/report.md. +``` + +--- + +## `read_file` + +Reads file data. + +Supports: + +- text +- configurable encoding +- Base64 +- byte offset +- maximum byte count + +Example: + +```text +Read Notes/todo.txt. +``` + +For a binary file: + +```text +Read image.png as Base64. +``` + +Large files can be processed incrementally: + +```text +offset = 0 +max_bytes = 1048576 +``` + +followed by another call with: + +```text +offset = 1048576 +``` + +The administrator `max_read_bytes` always takes precedence. + +--- + +## `search_files` + +Recursively searches for filenames using wildcard patterns. + +Examples: + +```text +*.pdf +``` + +```text +report* +``` + +```text +*.csv +``` + +Parameters include: + +```text +pattern +start_path +max_depth +include_directories +``` + +Example: + +```text +Find all CSV files below Data. +``` + +--- + +## `write_file` + +Creates a new file. + +It can also overwrite an existing file when: + +```text +overwrite = true +``` + +Creating a file requires: + +```text +Valves.allow_create_files = true +UserValves.allow_write_new_files = true +``` + +Overwriting requires: + +```text +Valves.allow_modify_files = true +UserValves.allow_modify_existing_files = true +``` + +Example: + +```text +Create Notes/example.txt containing "Hello world". +``` + +By default: + +```text +overwrite = false +``` + +which prevents accidental replacement of an existing file. + +The operation also supports Base64 content for binary data. + +--- + +## `append_file` + +Appends text or binary data to a file. + +Requires modification permission when the file exists. + +Example: + +```text +Append a line to Logs/manual.log. +``` + +A missing file can optionally be created using: + +```text +create_if_missing = true +``` + +Creating the missing file additionally requires create permission. + +--- + +## `update_file` + +Performs exact text replacement in a text file. + +This is useful when the model should make a small controlled change instead of rewriting the entire file. + +Example: + +Existing: + +```text +server=old-server +port=445 +``` + +Request: + +```text +Replace "server=old-server" with "server=new-server". +``` + +The operation supports: + +```text +replace_all +expected_replacements +``` + +`expected_replacements` can be used as an additional safety check. + +For example: + +```text +expected_replacements = 1 +``` + +causes the operation to fail rather than modify the file if the requested replacement would affect a different number of occurrences. + +For sensitive configuration files this is recommended. + +--- + +## `make_directory` + +Creates a directory. + +Supports: + +```text +parents +exist_ok +``` + +Example: + +```text +Create Projects/2026/Reports and create missing parent folders. +``` + +This requires directory-operation permission. + +--- + +## `copy_file` + +Copies a file within the configured SMB share/root. + +Example: + +```text +Copy Templates/report.md to Projects/report.md. +``` + +If the destination already exists, replacement requires: + +```text +overwrite = true +``` + +and modification permission. + +--- + +## `move_path` + +Moves or renames files and directories within the configured SMB share/root. + +Example: + +```text +Rename Drafts/report.md to Drafts/report-old.md. +``` + +or: + +```text +Move Drafts/report.md to Published/report.md. +``` + +Directory moves require directory-operation permission. + +--- + +## `delete_path` + +Deletes a file or directory. + +Deletion is disabled by default at both policy and user levels. + +A normal file delete requires: + +```text +Valves.allow_delete_paths = true +UserValves.allow_delete_existing_paths = true +``` + +Deleting an empty directory additionally requires directory operations. + +Recursive directory deletion additionally requires: + +```text +Valves.allow_recursive_delete = true +``` + +Example: + +```text +Delete Temp/file.txt. +``` + +Recursive delete should only be enabled when there is a clear operational requirement. + +--- + +# Path Security + +All model-supplied paths are relative paths. + +The Tool rejects: + +- UNC paths +- absolute paths +- drive-letter paths +- `..` traversal +- Alternate Data Stream syntax unless explicitly enabled + +For example, this is valid: + +```text +Projects/Report.txt +``` + +These are rejected: + +```text +..\..\secret.txt +``` + +```text +\\another-server\share\file.txt +``` + +```text +C:\Windows\file.txt +``` + +The configured SMB server/share cannot therefore be replaced through a normal file-operation argument. + +The `root_path` setting provides an additional confinement boundary within the configured share. + +--- + +# SMB Session Isolation + +The Python `smbclient` implementation maintains a process-wide connection/session cache. + +This matters in a multi-user Open WebUI instance because different users may configure different SMB credentials. + +The Tool therefore: + +1. serializes SMB access with a process-wide lock, +2. clears the `smbclient` connection cache before an operation, +3. registers only the current user's SMB session, +4. executes the operation, +5. clears the connection cache again afterward. + +This prioritizes credential/session isolation over parallel SMB operation throughput. + +If very high SMB concurrency is required, this design should be reviewed before changing the locking/session model. + +--- + +# EventEmitter Status Reporting + +All public operations use the included `EventEmitter`. + +The Tool reports: + +```text +in_progress +``` + +while an SMB operation is executing. + +On completion it reports either: + +```text +success +``` + +or: + +```text +error +``` + +with: + +```text +done = true +``` + +This provides visible operation status in Open WebUI while the model is working with the SMB share. + +--- + +# Security Recommendations + +## Use a dedicated SMB account + +Do not use a personal administrator account. + +Create an SMB/service account specifically for Open WebUI. + +Give that account only the filesystem permissions required for the intended use. + +For a read-only Tool, the SMB account itself should preferably also have read-only permissions. + +Application-level permission controls should not be the only protection. + +## Use a dedicated share or directory + +Prefer: + +```text +\\nas01\OpenWebUI +``` + +or configure: + +```text +root_path = OpenWebUI +``` + +instead of exposing a broad company/user share. + +## Restrict servers and shares + +Avoid production settings such as: + +```text +allowed_server_patterns = * +allowed_share_patterns = * +``` + +Prefer explicit values: + +```text +allowed_server_patterns = nas01.internal.example +allowed_share_patterns = AI-Documents +``` + +## Keep delete disabled unless required + +Recommended default: + +```text +allow_delete_paths = false +allow_recursive_delete = false +``` + +## Keep Alternate Data Streams disabled + +Recommended: + +```text +allow_alternate_data_streams = false +``` + +## Use SMB encryption where appropriate + +When supported by the server: + +```text +encrypt = true +``` + +SMB encryption protects SMB payloads in transit. + +## Protect stored credentials + +`UserValves.password` contains an SMB credential. + +Configure Open WebUI so sensitive Valve values are protected appropriately, and protect the Open WebUI database, backups, configuration, and secret keys. + +Do not give Tool-administration access to users who should not be able to change its security policy. + +## Use least privilege at both layers + +There are two distinct permission layers: + +1. SMB filesystem/share permissions +2. Tool `Valves` / `UserValves` + +Configure both. + +For example, if an account only needs read access, make it read-only on the SMB server even if the Tool's write switches are accidentally changed later. + +--- + +# Suggested Permission Profiles + +## Read-only + +Administrator: + +```text +allow_create_files = false +allow_modify_files = false +allow_delete_paths = false +allow_directory_operations = false +allow_move_copy = false +allow_recursive_delete = false +``` + +User mutation switches: + +```text +false +``` + +Use for knowledge/document retrieval. + +## Read + create + +Administrator: + +```text +allow_create_files = true +allow_modify_files = false +allow_delete_paths = false +``` + +User: + +```text +allow_write_new_files = true +allow_modify_existing_files = false +allow_delete_existing_paths = false +``` + +Useful when the model may generate new reports but must never alter existing documents. + +## Read + create + update + +Administrator: + +```text +allow_create_files = true +allow_modify_files = true +allow_delete_paths = false +``` + +User: + +```text +allow_write_new_files = true +allow_modify_existing_files = true +allow_delete_existing_paths = false +``` + +This is a reasonable general-purpose document-management profile. + +## Full file management + +Administrator: + +```text +allow_create_files = true +allow_modify_files = true +allow_delete_paths = true +allow_directory_operations = true +allow_move_copy = true +``` + +User: + +```text +allow_write_new_files = true +allow_modify_existing_files = true +allow_delete_existing_paths = true +allow_directory_operations = true +allow_move_copy_operations = true +``` + +Consider leaving: + +```text +allow_recursive_delete = false +``` + +unless recursive removal is truly needed. + +--- + +# Example Open WebUI Prompts + +Once the Tool is enabled for a model, typical prompts include: + +```text +List the files in the Reports directory. +``` + +```text +Find all PDF files under Projects. +``` + +```text +Read Projects/README.md and summarize it. +``` + +```text +Create Notes/meeting.txt with the following notes... +``` + +```text +Change "server01" to "server02" in Config/application.conf. +Only make the change if there is exactly one occurrence. +``` + +```text +Copy Templates/monthly-report.md to Reports/2026-08.md. +``` + +```text +Move Drafts/report.md to Approved/report.md. +``` + +The language model decides which Tool function to call based on the function descriptions and parameters. + +--- + +# Troubleshooting + +## Connection refused or timeout + +Verify: + +- SMB server hostname/IP +- TCP port `445` +- routing between Open WebUI and the SMB server +- host firewall +- container/network firewall +- SMB service availability + +If Open WebUI runs in Docker, test connectivity from the Open WebUI container/network rather than only from the Docker host. + +## Authentication failure + +Check: + +- username +- password +- domain +- SMB account status +- server authentication policy + +For Active Directory, try either: + +```text +domain = CORP +username = jdoe +``` + +or configure the username directly as: + +```text +CORP\jdoe +``` + +or, where supported: + +```text +jdoe@example.com +``` + +## Access denied + +Successful authentication does not imply filesystem access. + +Check both: + +- SMB share permissions +- underlying filesystem/NTFS permissions + +Also verify that the corresponding Tool/UserValve mutation permission is enabled. + +## Server not allowed + +If the Tool reports that the server is not allowed, check: + +```text +allowed_server_patterns +``` + +Example: + +```text +nas01.internal.example +``` + +must match the user's: + +```text +server +``` + +## Share not allowed + +Check: + +```text +allowed_share_patterns +``` + +against the configured: + +```text +share +``` + +## Cannot create or modify files + +Check both administrator and user controls. + +Creating: + +```text +Valves.allow_create_files +UserValves.allow_write_new_files +``` + +Modifying: + +```text +Valves.allow_modify_files +UserValves.allow_modify_existing_files +``` + +The SMB account itself must also have write permission. + +## Cannot delete files + +Deletion requires: + +```text +Valves.allow_delete_paths = true +UserValves.allow_delete_existing_paths = true +``` + +Recursive directory deletion additionally requires: + +```text +Valves.allow_recursive_delete = true +``` + +## File is too large + +Read and write operations are constrained by: + +```text +max_read_bytes +max_write_bytes +``` + +Use chunked reads for large files. + +For very large file transfer workloads, this Tool is intentionally not optimized as a bulk SMB transfer utility. + +## Binary file cannot be decoded + +Use: + +```text +as_base64 = true +``` + +with `read_file`. + +For binary writes, provide Base64 and use: + +```text +content_is_base64 = true +``` + +## Update text was not found + +`update_file` performs exact text replacement. + +Whitespace, line endings, capitalization, and encoding therefore matter. + +Read the relevant part of the file first and use the exact existing text. + +--- + +# Limitations + +- Operations are confined to one configured SMB server/share/root per user. +- Copy/move functions operate inside that configured logical SMB root. +- SMB operations are serialized to protect per-user session isolation. +- The Tool is intended for controlled model-driven file operations, not high-throughput file transfer. +- `update_file` performs exact text replacement rather than semantic document editing. +- Large files are constrained by administrator-configured limits. +- SMB permissions remain ultimately controlled by the SMB server and filesystem ACLs. +- The Tool cannot grant permissions that the configured SMB account does not possess. + +--- + +# File Layout + +A simple deployment/repository layout could be: + +```text +open-webui-smb-tool/ +├── open_webui_smb_file_operations.py +└── README.md +``` + +--- + +# Version + +Tool version: + +```text +0.1.0 +``` + +Minimum declared Open WebUI version: + +```text +0.6.3 +``` + +Python dependency: + +```text +smbprotocol>=1.17.0 +``` + +--- + +# License + +MIT diff --git a/smb_tool.py b/smb_tool.py new file mode 100644 index 0000000..ee65490 --- /dev/null +++ b/smb_tool.py @@ -0,0 +1,1449 @@ +""" +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 + + # --------------------------------------------------------------------- + # 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, + )