fix(web): Harden web fetch

ex_app/lib/all_tools/files.py

@@ -84,13 +84,15 @@ async def get_file_content(file_path: str):
 	@impulse(ImpulseRadius.SELF)
 	async def get_file_content_by_file_link(file_url: str):
 		"""
-		Get the content of a Nextcloud-internal file using its internal file link.
+		Get the content of a Nextcloud file with the internal link to that file.
 		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://<host>/f/<fileId> or https://<host>/index.php/f/<fileId>)
+		Use this tool only when the URL points to a file in Nextcloud, for example https://cloud.example.com/index.php/f/12345 or https://cloud.example.com/f/12345.
+		:param file_url: an internal Nextcloud file URL. The URL must agree with the pattern https://<host>/f/<fileId> or the pattern https://<host>/index.php/f/<fileId>.
 		:return: text content of the file
 		"""
-
+		# This tool does not check that the URL belongs to this instance. The check stays
+		# permissive, because the set of known hosts can omit some URLs from
+		# "trusted_domains". A similar check in web_fetch is more strict.
 		return await get_file_content_from_int_link(nc, file_url)
 
 	@tool

ex_app/lib/all_tools/lib/files.py

@@ -1,31 +1,128 @@
 # SPDX-FileCopyrightText: 2025 Nextcloud GmbH and Nextcloud contributors
 # SPDX-License-Identifier: AGPL-3.0-or-later
 import re
+from urllib.parse import urlparse
 
 from nc_py_api import AsyncNextcloudApp
+from nc_py_api.ex_app import LogLvl
 from nc_py_api.files.files_async import AsyncFilesAPI, FsNode
 
+from ex_app.lib.logger import log
+
+# todo: add multimodal support for the multimodal task type. That type needs a different
+# maximum size.
+# todo: https://github.com/nextcloud/context_agent/pull/211
 TEXT_LIKE_MIMETYPE_PARTS = ('text/', 'application/json', 'application/xml', 'application/xhtml', '+xml', '+json')
 MAX_FILE_SIZE = 100_000  # 100kB, approx. 25k tokens
 
+# Match only against the path of the URL. The path must end with the file ID. Thus a
+# `/f/<digits>` part in the middle of a path or in the query string is not a file link.
 _FILE_ID_SUFFIX_RE = re.compile(r'/(index\.php/)?f/(\d+)/?$')
 
+# The set of the known Nextcloud base URLs. The URLs are lowercase and have no scheme.
+# The code fills this set at the first use that gets the public URL from app_api.
+_nc_hosts: set[str] | None = None
+
 
 def _strip_scheme(url: str) -> str:
 	"""Remove the http(s):// scheme from a URL for scheme-agnostic comparison."""
-	return re.sub(r'^https?://', '', url)
+	return re.sub(r'^https?://', '', url).rstrip('/')
+
+
+def _split_authority_path(value: str) -> tuple[str, str]:
+	"""Divide *value* into an authority and a path with `urlparse`.
+
+	This function accepts a full URL, for example `https://host/p`. It also accepts a
+	Nextcloud base entry that has no scheme, for example `host[:port][/path]`. If the
+	value has no scheme, the function adds two slashes at the start. Then `urlparse`
+	reads the first part as the authority and not as the path.
+
+	The authority is `host[:port]` in lowercase. The function does not add a default
+	port. Thus a value with the port `:443` or `:80` does not match a value without
+	that port. This result is satisfactory, because both values come from the URLs in
+	the Nextcloud configuration.
+
+	The function compares the two hosts as exact strings. Thus an IDN host must have
+	the same form (Unicode or punycode) in both values.
+	"""
+	parsed = urlparse(value if '://' in value else '//' + value)
+	host = (parsed.hostname or '').lower()
+	if not host:
+		return '', ''
+	authority = f'{host}:{parsed.port}' if parsed.port is not None else host
+	return authority, parsed.path.rstrip('/')
+
+
+def _matches_nc_base(url: str, nc_base: str) -> bool:
+	"""Check if *url* is part of a known Nextcloud base. The two values can have no scheme.
+
+	The host and the port must match exactly. Thus `cloud.example.com` does not match a
+	similar host such as `cloud.example.com.evil.net` or `cloud.example.commercial.com`.
+	If the Nextcloud base contains a path, the URL must be at that path or below it.
+	A Nextcloud base contains a path when the server operates in a sub-path.
+	"""
+	nc_authority, nc_path = _split_authority_path(nc_base)
+	if not nc_authority:
+		return False
+	url_authority, url_path = _split_authority_path(url)
+	if url_authority != nc_authority:
+		return False
+	if not nc_path:
+		return True
+	return url_path == nc_path or url_path.startswith(nc_path + '/')
+
+
+async def _get_nc_hosts(nc: AsyncNextcloudApp) -> set[str]:
+	"""Make and cache the set of the known Nextcloud base URLs. These URLs have no scheme.
+
+	The sources are the endpoint of the app and the `overwrite.cli.url` configuration.
+	If the request to app_api fails, the result is not cached. The next call tries again.
+	"""
+	global _nc_hosts
+	if _nc_hosts is not None:
+		return _nc_hosts
+
+	hosts = set()
+
+	# 1. The NEXTCLOUD_URL environment variable. This is the URL of the deployment of
+	# the ExApp.
+	hosts.add(_strip_scheme(nc.app_cfg.endpoint).lower())
+
+	# 2. The public URL from `overwrite.cli.url`. The absolute-URL endpoint of app_api
+	# supplies this URL.
+	try:
+		info = await nc.ocs('GET', '/ocs/v1.php/apps/app_api/api/v1/info/nextcloud_url/absolute', params={'url': ''})
+		abs_url = info.get('absolute_url', '')
+		if abs_url:
+			hosts.add(_strip_scheme(abs_url).lower())
+	except Exception:
+		await log(nc, LogLvl.WARNING, 'Could not fetch Nextcloud absolute URL from app_api, using only NEXTCLOUD_URL for internal link detection')
+		return hosts
+
+	_nc_hosts = hosts
+	await log(nc, LogLvl.DEBUG, f'Known NC hosts for internal link detection: {_nc_hosts}')
+	return _nc_hosts
+
+
+async def is_known_nc_host(nc: AsyncNextcloudApp, url: str) -> bool:
+	"""Check whether the URL points to the host of this Nextcloud instance.
+
+	The scheme of the URL has no effect on the result.
+	"""
+	if not url:
+		return False
+	nc_hosts = await _get_nc_hosts(nc)
+	return any(_matches_nc_base(url, host) for host in nc_hosts)
 
 
-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):]))
+async def is_an_internal_file_link(nc: AsyncNextcloudApp, url: str) -> bool:
+	"""Check whether the URL is an internal file link of this Nextcloud instance."""
+	return bool(_FILE_ID_SUFFIX_RE.search(urlparse(url).path)) and await is_known_nc_host(nc, url)
 
 
 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)
+	match = _FILE_ID_SUFFIX_RE.search(urlparse(file_url).path)
 	if match:
 		return int(match.group(2))
 	raise Exception("Not a valid nextcloud file URL")

ex_app/lib/all_tools/lib/url_guard.py

@@ -0,0 +1,177 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""SSRF protection helpers for the `web_fetch` tool.
+
+The function :func:`assert_public_url` refuses these URLs:
+- A URL that does not use the HTTP or the HTTPS scheme.
+- A URL with a host that resolves to a loopback, a private, a link-local,
+  a reserved, or a different non-public IP address. This includes the cloud
+  metadata address `169.254.169.254` and the mDNS names that end with `.local`.
+
+The class :class:`PublicAddressResolver` applies the same address check when the
+HTTP client connects.
+
+This is a hard limit. It does not change with the content of the conversation.
+"""
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import socket
+import typing
+from urllib.parse import urlsplit
+
+from niquests.packages.urllib3.contrib.resolver._async.system import SystemResolver
+
+
+class BlockedUrlError(Exception):
+	"""The SSRF guard raises this error when it refuses a URL.
+
+	It is safe to show the message of this error to the agent.
+	"""
+
+
+# These ranges are not globally routable.
+_EXTRA_BLOCKED_NETWORKS = tuple(
+	ipaddress.ip_network(cidr)
+	for cidr in (
+		'100.64.0.0/10',       # RFC 6598 CGNAT and shared address space
+		'192.0.0.0/24',        # RFC 6890 IETF protocol assignments
+		'192.0.2.0/24',        # TEST-NET-1
+		'198.18.0.0/15',       # benchmark tests
+		'198.51.100.0/24',     # TEST-NET-2
+		'203.0.113.0/24',      # TEST-NET-3
+		'192.88.99.0/24',      # 6to4 relay anycast
+		'64:ff9b::/96',        # the well-known NAT64 prefix
+		'100::/64',            # IPv6 discard only
+	)
+)
+
+
+def _is_public_ip(ip: str) -> bool:
+	try:
+		addr = ipaddress.ip_address(ip)
+	except ValueError:
+		return False
+
+	if (
+		addr.is_private
+		or addr.is_loopback
+		or addr.is_link_local
+		or addr.is_multicast
+		or addr.is_reserved
+		or addr.is_unspecified
+	):
+		return False
+
+	# Use the global-reachability flag of the standard library if it is available. The
+	# meaning of this flag changes between Python versions. Thus do not use it alone.
+	if getattr(addr, 'is_global', True) is False:
+		return False
+
+	for network in _EXTRA_BLOCKED_NETWORKS:
+		if addr.version == network.version and addr in network:
+			return False
+
+	# An IPv4-mapped, a 6to4, or a Teredo address can hide a private IPv4 target.
+	# Extract the IPv4 addresses and examine them again.
+	if isinstance(addr, ipaddress.IPv6Address):
+		embedded = [addr.ipv4_mapped, addr.sixtofour, *(addr.teredo or ())]
+		return all(_is_public_ip(str(ip)) for ip in embedded if ip is not None)
+
+	return True
+
+
+async def _resolve_all(host: str) -> list[str]:
+	"""Find all A and AAAA addresses of a host name.
+
+	`socket.getaddrinfo` stops its thread until the DNS reply comes, and a hostile
+	server can delay that reply. Thus use `loop.getaddrinfo`, which does the same work
+	in a thread pool. The other tasks then continue. But the pool has a limited number
+	of threads, and it is not possible to cancel the call.
+	"""
+	loop = asyncio.get_running_loop()
+	infos = await loop.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)
+	return [info[4][0] for info in infos]
+
+
+class PublicAddressResolver(SystemResolver):
+	"""A niquests resolver that refuses the non-public addresses at connection time.
+
+	:func:`assert_public_url` resolves the host before the request. The HTTP client
+	then resolves the host again when it connects. A hostile DNS server can send a
+	public address for the first lookup and an internal address for the second lookup
+	(DNS rebinding). This resolver does the lookup for the connection itself. Thus the
+	client connects only to the addresses that this resolver examined.
+
+	The hosts in *allowed_hosts* skip the check. Use this for the hosts of this
+	Nextcloud instance, which can have an internal address.
+	"""
+
+	def __init__(self, allowed_hosts: typing.Iterable[str] = ()):
+		super().__init__()
+		self.allowed_hosts = {host.lower() for host in allowed_hosts}
+
+	async def getaddrinfo(self, host, port, family, type, proto=0, flags=0, *, quic_upgrade_via_dns_rr=False):
+		records = await super().getaddrinfo(
+			host, port, family, type, proto, flags, quic_upgrade_via_dns_rr=quic_upgrade_via_dns_rr,
+		)
+		name = host.decode('ascii') if isinstance(host, bytes) else (host or '')
+		if name.lower() in self.allowed_hosts:
+			return records
+		for record in records:
+			if not _is_public_ip(record[4][0]):
+				raise socket.gaierror(f"refused: '{name}' resolves to a non-public address")
+		return records
+
+
+async def assert_public_url(url: str) -> None:
+	"""Raise :class:`BlockedUrlError` if *url* is not an HTTP or HTTPS URL to a public host.
+
+	All the addresses of the host must be public. If one address of the host is
+	internal, this function refuses the URL.
+
+	This check alone does not prevent DNS rebinding, because the HTTP client resolves
+	the host again. Send the request with :class:`PublicAddressResolver` for that.
+	"""
+	parts = urlsplit(url)
+	if parts.scheme not in ('http', 'https'):
+		raise BlockedUrlError(f"(refused: only http(s) URLs are allowed, got '{parts.scheme or 'no scheme'}')")
+
+	host = (parts.hostname or '').strip()
+	if not host:
+		raise BlockedUrlError('(refused: URL has no host)')
+
+	# Refuse all mDNS and local names. These names resolve only on the local network.
+	if (
+		host == 'localhost'
+		or host.endswith('.local')
+		or host.endswith('.localhost')
+		or host.endswith('.internal')
+		or host.endswith('.lan')
+	):
+		raise BlockedUrlError(f"(refused: '{host}' is a local/internal hostname)")
+
+	# If the host is an IP address, check it directly.
+	try:
+		ipaddress.ip_address(host)
+		is_literal_ip = True
+	except ValueError:
+		is_literal_ip = False
+
+	if is_literal_ip:
+		if not _is_public_ip(host):
+			raise BlockedUrlError(f"(refused: '{host}' is a non-public IP address)")
+		return
+
+	try:
+		resolved = await _resolve_all(host)
+	except socket.gaierror:
+		raise BlockedUrlError(f"(refused: could not resolve host '{host}')")
+
+	if not resolved:
+		raise BlockedUrlError(f"(refused: host '{host}' did not resolve to any address)")
+
+	for ip in resolved:
+		if not _is_public_ip(ip):
+			raise BlockedUrlError(f"(refused: '{host}' resolves to a non-public address)")