From c29d561a21db9950259afe4b86312388167965c1 Mon Sep 17 00:00:00 2001 From: kyteinsky Date: Wed, 12 Aug 2026 16:32:34 +0530 Subject: [PATCH 1/2] fix: disambiguate between nextcloud internal file links tool and web_fetch Signed-off-by: kyteinsky Assisted-by: Github Copilot: claude-opus-4-6 --- ex_app/lib/all_tools/files.py | 7 +++++-- ex_app/lib/all_tools/web.py | 12 +++++++----- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/ex_app/lib/all_tools/files.py b/ex_app/lib/all_tools/files.py index 4c08c1f..61edf0e 100644 --- a/ex_app/lib/all_tools/files.py +++ b/ex_app/lib/all_tools/files.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import xml.etree.ElementTree as ET from urllib.parse import unquote + import niquests from langchain_core.tools import tool from nc_py_api import AsyncNextcloudApp @@ -45,8 +46,10 @@ async def get_file_content(file_path: str): @safe_tool async def get_file_content_by_file_link(file_url: str): """ - Get the content of a file given an internal Nextcloud link (e.g., https://host/index.php/f/12345) - :param file_url: the nextcloud-internal file URL + Get the content of a Nextcloud-internal file using its internal file link. + This is NOT for fetching arbitrary web URLs, use web_fetch for those. + Only use this tool when the URL points to a file stored in Nextcloud (e.g., https://cloud.example.com/index.php/f/12345 or https://cloud.example.com/f/12345). + :param file_url: a Nextcloud-internal file URL (must match the pattern https:///f/ or https:///index.php/f/) :return: text content of the file """ diff --git a/ex_app/lib/all_tools/web.py b/ex_app/lib/all_tools/web.py index 2d78930..c429ee9 100644 --- a/ex_app/lib/all_tools/web.py +++ b/ex_app/lib/all_tools/web.py @@ -13,12 +13,14 @@ async def get_tools(nc: AsyncNextcloudApp): @safe_tool async def web_fetch(url: str) -> str: """ - Get the contents of a web page via HTTP - :param url: The HTTP URL to the web page (e.g. https://nextcloud.com/team/ ) - :return: the web page content + Fetch the contents of an external web page via HTTP. + This is NOT for Nextcloud-internal file links (like https:///f/12345), use get_file_content_by_file_link for those. + Use this for any URL on the public internet or intranet (e.g., https://www.eff.org/). + :param url: the HTTP(S) URL of the web page to fetch + :return: the raw web page content (HTML, JSON, etc.) """ - res = await niquests.get(url) - return res.text() + res = await niquests.async_api.get(url) + return res.text or "(empty content)" return [ web_fetch, From 290a41cf0bc8584332baca2de5e44f977a1e0bd9 Mon Sep 17 00:00:00 2001 From: kyteinsky Date: Wed, 12 Aug 2026 19:01:29 +0530 Subject: [PATCH 2/2] fix: web_fetch redirect to internal file fetch for regex-matched links also more hardenings of the fetches: - don't fetch folders - limit file/content size to 100kB - limit mimetype to text-like Signed-off-by: kyteinsky Assisted-by: Github Copilot: claude-opus-4-6 --- ex_app/lib/all_tools/files.py | 53 +--------------- ex_app/lib/all_tools/lib/files.py | 100 ++++++++++++++++++++++++++++-- ex_app/lib/all_tools/web.py | 28 ++++++++- 3 files changed, 122 insertions(+), 59 deletions(-) diff --git a/ex_app/lib/all_tools/files.py b/ex_app/lib/all_tools/files.py index 61edf0e..5ff6cb4 100644 --- a/ex_app/lib/all_tools/files.py +++ b/ex_app/lib/all_tools/files.py @@ -3,13 +3,12 @@ import xml.etree.ElementTree as ET from urllib.parse import unquote -import niquests from langchain_core.tools import tool from nc_py_api import AsyncNextcloudApp from nc_py_api.files.files_async import AsyncFilesAPI, FsNode from ex_app.lib.all_tools.lib.decorator import dangerous_tool, safe_tool -from ex_app.lib.all_tools.lib.files import get_file_id_from_file_url +from ex_app.lib.all_tools.lib.files import format_fs_node, get_file_content_from_int_link, get_file_id_from_file_url def _validate_path(path: str) -> str: @@ -53,53 +52,7 @@ async def get_file_content_by_file_link(file_url: str): :return: text content of the file """ - file_id = get_file_id_from_file_url(file_url) - # Generate a direct download link using the fileId - info = await nc.ocs('POST', '/ocs/v2.php/apps/dav/api/v1/direct', json={'fileId': file_id}, response_type='json') - download_url = info.get('ocs', {}).get('data', {}).get('url', None) - - if not download_url: - raise Exception('Could not generate download URL from file id') - - # Download the file from the direct download URL - response = await niquests.async_api.get(download_url) - - return response.text - - def __format_fs_node(fsnode: FsNode) -> dict: - # todo: permissions info - return { - 'path': fsnode.user_path, - 'file_id': fsnode.info.fileid, - 'etag': fsnode.etag.replace('"', '').replace("'", ''), - 'bytes': fsnode.info.size, - 'creation_date': fsnode.info.creation_date.isoformat(), - 'last_modified': fsnode.info.last_modified.isoformat(), - 'mimetype': fsnode.info.mimetype, - 'is_shared': fsnode.is_shared, - 'is_favourite': fsnode.info.favorite, - 'is_version': fsnode.info.is_version, - 'trash_info': { - 'in_trash': fsnode.info.in_trash, - **({ - 'trashbin_filename': fsnode.info.trashbin_filename, - 'original_location': fsnode.info.trashbin_original_location, - 'deletion_time': fsnode.info.trashbin_deletion_time, - } if fsnode.info.in_trash else {}), - }, - 'lock_info': { - 'is_locked': fsnode.lock_info.is_locked, - **({ - 'owner': fsnode.lock_info.owner, - 'owner_display_name': fsnode.lock_info.owner_display_name, - 'type': fsnode.lock_info.type.name, - 'creation_time': fsnode.lock_info.lock_creation_time, - 'ttl': fsnode.lock_info.lock_ttl, - 'locked_by_app': fsnode.lock_info.owner_editor, - } if fsnode.lock_info.is_locked else {}), - }, - } - + return await get_file_content_from_int_link(nc, file_url) @tool @safe_tool @@ -115,7 +68,7 @@ async def get_file_tree(path: str = '/', include_metadata = False, depth: int = files_handle = AsyncFilesAPI(nc._session) fsnode_list = await files_handle.listdir(path, min(5, depth)) if include_metadata: - return [__format_fs_node(fsnode) for fsnode in fsnode_list] + return [format_fs_node(fsnode) for fsnode in fsnode_list] return [fsnode.user_path for fsnode in fsnode_list] diff --git a/ex_app/lib/all_tools/lib/files.py b/ex_app/lib/all_tools/lib/files.py index 5d04625..e21ed74 100644 --- a/ex_app/lib/all_tools/lib/files.py +++ b/ex_app/lib/all_tools/lib/files.py @@ -2,12 +2,100 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import re -def get_file_id_from_file_url(file_url) -> int: - # Define the regex pattern to capture only the digits - pattern = r"https?://[a-zA-Z-_.:0-9]+/(index\.php/)?f/(\d+)" - match = re.search(pattern, file_url) +from nc_py_api import AsyncNextcloudApp +from nc_py_api.files.files_async import AsyncFilesAPI, FsNode +TEXT_LIKE_MIMETYPE_PARTS = ('text/', 'application/json', 'application/xml', 'application/xhtml', '+xml', '+json') +MAX_FILE_SIZE = 100_000 # 100kB, approx. 25k tokens + +_FILE_ID_SUFFIX_RE = re.compile(r'/(index\.php/)?f/(\d+)/?$') + + +def _strip_scheme(url: str) -> str: + """Remove the http(s):// scheme from a URL for scheme-agnostic comparison.""" + return re.sub(r'^https?://', '', url) + + +def is_an_internal_file_link(nc: AsyncNextcloudApp, url: str) -> bool: + """Check whether a URL is an internal file link for this Nextcloud instance (scheme-agnostic).""" + nc_host = _strip_scheme(nc.app_cfg.endpoint.rstrip('/')) + url_without_scheme = _strip_scheme(url) + return url_without_scheme.startswith(nc_host) and bool(_FILE_ID_SUFFIX_RE.search(url_without_scheme[len(nc_host):])) + + +def get_file_id_from_file_url(file_url: str) -> int: + """Extract the numeric file ID from a Nextcloud-internal file URL.""" + match = _FILE_ID_SUFFIX_RE.search(file_url) if match: return int(match.group(2)) - else: - raise Exception("Not a valid nextcloud file URL") \ No newline at end of file + raise Exception("Not a valid nextcloud file URL") + + +def format_fs_node(fsnode: FsNode) -> dict: + # todo: permissions info + return { + 'path': fsnode.user_path, + 'file_id': fsnode.info.fileid, + 'etag': fsnode.etag.replace('"', '').replace("'", ''), + 'bytes': fsnode.info.size, + 'creation_date': fsnode.info.creation_date.isoformat(), + 'last_modified': fsnode.info.last_modified.isoformat(), + 'mimetype': fsnode.info.mimetype, + 'is_shared': fsnode.is_shared, + 'is_favourite': fsnode.info.favorite, + 'is_version': fsnode.info.is_version, + 'trash_info': { + 'in_trash': fsnode.info.in_trash, + **({ + 'trashbin_filename': fsnode.info.trashbin_filename, + 'original_location': fsnode.info.trashbin_original_location, + 'deletion_time': fsnode.info.trashbin_deletion_time, + } if fsnode.info.in_trash else {}), + }, + 'lock_info': { + 'is_locked': fsnode.lock_info.is_locked, + **({ + 'owner': fsnode.lock_info.owner, + 'owner_display_name': fsnode.lock_info.owner_display_name, + 'type': fsnode.lock_info.type.name, + 'creation_time': fsnode.lock_info.lock_creation_time, + 'ttl': fsnode.lock_info.lock_ttl, + 'locked_by_app': fsnode.lock_info.owner_editor, + } if fsnode.lock_info.is_locked else {}), + }, + } + + +async def get_file_node(nc: AsyncNextcloudApp, file_id: int) -> FsNode: + files_handle = AsyncFilesAPI(nc._session) + node = await files_handle.by_id(file_id) + if not node: + raise RuntimeError(f'No file/folder found with id: {file_id}') + return node + + +async def get_file_contents(nc: AsyncNextcloudApp, fsnode: FsNode) -> str: + """ + RuntimeError: just return the metadata to the model since the node is one of the following: + - a folder + - very large in size + - non-text mimetype + """ + if fsnode.is_dir: + raise RuntimeError('Folder found at the given file id, skipping download') + if fsnode.info.content_length > MAX_FILE_SIZE: + raise RuntimeError(f'File id {fsnode.info.fileid} is too large to download at {fsnode.info.content_length} bytes') + if not any(t in fsnode.info.mimetype for t in TEXT_LIKE_MIMETYPE_PARTS): + raise RuntimeError(f'File id {fsnode.info.fileid} is of content type {fsnode.info.mimetype} so cannot be displayed as text') + files_handle = AsyncFilesAPI(nc._session) + return (await files_handle.download(fsnode)).decode(encoding='utf-8', errors='ignore') + + +async def get_file_content_from_int_link(nc: AsyncNextcloudApp, url: str) -> str: + file_id = get_file_id_from_file_url(url) + fsnode = await get_file_node(nc, file_id) + + try: + return await get_file_contents(nc, fsnode) + except RuntimeError as e: + return f'Failed to download the file/folder: {e}.\nMore info about the node:{format_fs_node(fsnode)}' diff --git a/ex_app/lib/all_tools/web.py b/ex_app/lib/all_tools/web.py index c429ee9..3965922 100644 --- a/ex_app/lib/all_tools/web.py +++ b/ex_app/lib/all_tools/web.py @@ -5,6 +5,12 @@ from nc_py_api import AsyncNextcloudApp from ex_app.lib.all_tools.lib.decorator import safe_tool +from ex_app.lib.all_tools.lib.files import ( + MAX_FILE_SIZE, + TEXT_LIKE_MIMETYPE_PARTS, + get_file_content_from_int_link, + is_an_internal_file_link, +) async def get_tools(nc: AsyncNextcloudApp): @@ -14,13 +20,29 @@ async def get_tools(nc: AsyncNextcloudApp): async def web_fetch(url: str) -> str: """ Fetch the contents of an external web page via HTTP. - This is NOT for Nextcloud-internal file links (like https:///f/12345), use get_file_content_by_file_link for those. Use this for any URL on the public internet or intranet (e.g., https://www.eff.org/). :param url: the HTTP(S) URL of the web page to fetch :return: the raw web page content (HTML, JSON, etc.) """ - res = await niquests.async_api.get(url) - return res.text or "(empty content)" + # Detect Nextcloud-internal file links and fetch via the internal API, for the models that would still call this tool. + if is_an_internal_file_link(nc, url): + return await get_file_content_from_int_link(nc, url) + + # Pre-flight HEAD request to check content type and size before downloading + head_res = await niquests.async_api.head(url, allow_redirects=True) + + content_type = head_res.headers.get('Content-Type', '') + if not any(t in content_type for t in TEXT_LIKE_MIMETYPE_PARTS): + return f"(binary or unknown content detected: {content_type or 'no Content-Type header'}. Cannot display as text.)" + + # download first 100kB of the file + res = await niquests.async_api.get(url, headers={'Range': f'bytes=0-{MAX_FILE_SIZE - 1}'}) + text = res.text or "(empty content)" + + if res.status_code == 206: + text += "\n\n(truncated: only the first 100 kB of content was fetched.)" + + return text return [ web_fetch,