security: fix path traversal (CVE-2026-4307) and SSRF (CVE-2026-4308)

api/api_files_get.py

@@ -46,21 +46,28 @@ async def process(self, input: dict, request: Request) -> dict | Response:
 
             for path in paths:
                 try:
-                    # Convert internal paths to external paths
+                    # CVE-2026-4307: `path` comes from the request body, so it is never
+                    # trusted. Only /a0/... paths are addressable, each is resolved through
+                    # the containment check, and the old "assume it's already an
+                    # external/absolute path" branch -- which read ANY file on disk and
+                    # returned it base64-encoded -- is gone.
+                    if not isinstance(path, str) or not path.startswith("/a0/"):
+                        PrintStyle.warning(
+                            f"Refused path outside the Agent Zero directory: {path}"
+                        )
+                        continue
                     if path.startswith("/a0/tmp/uploads/"):
-                        # Internal path - convert to external
-                        filename = path.replace("/a0/tmp/uploads/", "")
-                        external_path = files.get_abs_path("usr/uploads", filename)
-                        filename = os.path.basename(external_path)
-                    elif path.startswith("/a0/"):
-                        # Other internal Agent Zero paths
-                        relative_path = path.replace("/a0/", "")
-                        external_path = files.get_abs_path(relative_path)
-                        filename = os.path.basename(external_path)
+                        relative_path = os.path.join(
+                            "usr/uploads", path.replace("/a0/tmp/uploads/", "", 1)
+                        )
                     else:
-                        # Assume it's already an external/absolute path
-                        external_path = path
-                        filename = os.path.basename(path)
+                        relative_path = path.replace("/a0/", "", 1)
+                    try:
+                        external_path = files.get_abs_path_contained(relative_path)
+                    except files.PathEscapesBaseDirError as exc:
+                        PrintStyle.warning(f"Refused traversal attempt: {exc}")
+                        continue
+                    filename = os.path.basename(external_path)
 
                     # Check if file exists
                     if not os.path.exists(external_path):

api/file_info.py

@@ -26,7 +26,19 @@ class FileInfo(TypedDict):
     message: str
 
 async def get_file_info(path: str) -> FileInfo:
-    abs_path = files.get_abs_path(path)
+    # CVE-2026-4307: `path` is request-supplied. Stat-ing an arbitrary absolute path leaks
+    # whether host files exist plus their size, mtime and permission bits, so the path is
+    # contained to the Agent Zero directory before anything touches the disk.
+    try:
+        abs_path = files.get_abs_path_contained(path)
+    except files.PathEscapesBaseDirError:
+        return {
+            "input_path": path, "abs_path": "", "exists": False,
+            "is_dir": False, "is_file": False, "is_link": False,
+            "size": 0, "modified": 0, "created": 0, "permissions": 0,
+            "dir_path": "", "file_name": "", "file_ext": "",
+            "message": "Path is outside the Agent Zero directory and was refused.",
+        }
     exists = os.path.exists(abs_path)
     message = ""
 

helpers/files.py

@@ -577,6 +577,28 @@ def get_abs_path(*relative_paths):
     return _resolve_path(*relative_paths)
 
 
+class PathEscapesBaseDirError(ValueError):
+    """A caller-supplied path resolved outside the Agent Zero base directory."""
+
+
+def get_abs_path_contained(*relative_paths):
+    """Resolve a path and REFUSE it if it escapes the base directory.
+
+    Patch for CVE-2026-4307. ``get_abs_path`` is used all over the codebase, including
+    with trusted absolute paths, so it cannot be made strict without breaking internals.
+    This is the variant every handler that accepts a path from an HTTP request must use:
+    it normalizes ``..`` and symlinks via ``realpath`` and then requires the result to sit
+    inside the base dir, so neither ``/etc/passwd`` nor ``../../etc/passwd`` survives.
+    """
+    resolved = os.path.realpath(_resolve_path(*relative_paths))
+    base = os.path.realpath(get_base_dir())
+    if not (resolved == base or resolved.startswith(base + os.sep)):
+        raise PathEscapesBaseDirError(
+            f"path escapes the Agent Zero directory: {os.path.join(*relative_paths)}"
+        )
+    return resolved
+
+
 def get_abs_path_dockerized(*relative_paths):
     "Ensures the abs path is dockerized (i.e. /a0/... path)"
     abs = get_abs_path(*relative_paths)

helpers/net_guard.py

@@ -0,0 +1,107 @@
+"""SSRF guard — refuse outbound fetches aimed at the infrastructure, not the internet.
+
+Patch for CVE-2026-4308. The document fetcher accepted any URI and handed it straight to
+aiohttp with ``allow_redirects=True``, so a document URL (which an agent will happily take
+from a web page it just read) could reach:
+
+* the loopback interface — other services on the same host,
+* link-local 169.254.169.254 — cloud instance metadata, i.e. credentials,
+* RFC1918 ranges — anything else on the private network,
+* and via a redirect, all of the above starting from a perfectly public-looking URL.
+
+This module is the single decision point for "may we fetch this?". It is deliberately
+stdlib-only and deny-by-default: a hostname must resolve, and **every** address it
+resolves to must be a global unicast address, or the fetch is refused.
+
+Redirects are validated per hop by the caller (see fetch.py) rather than delegated to the
+HTTP client, because a redirect is just another attacker-chosen URL.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+ALLOWED_SCHEMES = frozenset({"http", "https"})
+MAX_REDIRECTS = 3
+
+
+class BlockedRequestError(ValueError):
+    """The requested URL points at non-public infrastructure and was refused."""
+
+
+def _addresses_for(host: str) -> list:
+    """Every address the hostname resolves to (A + AAAA). Raises if it resolves to none."""
+    try:
+        infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)
+    except socket.gaierror as exc:
+        raise BlockedRequestError(f"host does not resolve: {host}") from exc
+    addresses = []
+    for info in infos:
+        sockaddr = info[4]
+        try:
+            addresses.append(ipaddress.ip_address(sockaddr[0]))
+        except ValueError:
+            continue
+    if not addresses:
+        raise BlockedRequestError(f"host resolved to no usable address: {host}")
+    return addresses
+
+
+def _is_public(address) -> bool:
+    """Global unicast only. Rejects loopback, private, link-local, reserved, multicast.
+
+    ``is_global`` alone is not enough: it does not exclude every reserved range on all
+    Python versions, so the explicit checks stay as belt-and-braces.
+    """
+    if (
+        address.is_loopback
+        or address.is_private
+        or address.is_link_local
+        or address.is_reserved
+        or address.is_multicast
+        or address.is_unspecified
+    ):
+        return False
+    # IPv4-mapped/compatible IPv6 (e.g. ::ffff:127.0.0.1) must be judged on the v4 value.
+    mapped = getattr(address, "ipv4_mapped", None)
+    if mapped is not None:
+        return _is_public(mapped)
+    sixtofour = getattr(address, "sixtofour", None)
+    if sixtofour is not None:
+        return _is_public(sixtofour)
+    return bool(getattr(address, "is_global", True))
+
+
+def assert_public_url(url: str) -> str:
+    """Return the URL if it is safe to fetch; raise BlockedRequestError otherwise."""
+    parsed = urlparse(url)
+    scheme = (parsed.scheme or "").lower()
+    if scheme not in ALLOWED_SCHEMES:
+        raise BlockedRequestError(
+            f"refused scheme {scheme or '(none)'}: only http/https may be fetched"
+        )
+    host = parsed.hostname
+    if not host:
+        raise BlockedRequestError("refused URL with no host")
+    # A bare IP literal is checked directly; a name is checked on every resolved address.
+    try:
+        literal = ipaddress.ip_address(host)
+    except ValueError:
+        literal = None
+    addresses = [literal] if literal is not None else _addresses_for(host)
+    for address in addresses:
+        if not _is_public(address):
+            raise BlockedRequestError(
+                f"refused {host}: resolves to non-public address {address}"
+            )
+    return url
+
+
+def is_public_url(url: str) -> bool:
+    try:
+        assert_public_url(url)
+        return True
+    except BlockedRequestError:
+        return False

plugins/_document_query/helpers/fetch.py

@@ -13,8 +13,10 @@
 from urllib.parse import urlparse
 
 import aiohttp
+from yarl import URL
 
 from helpers import files
+from helpers.net_guard import MAX_REDIRECTS, BlockedRequestError, assert_public_url
 
 
 InterventionCallback = Callable[[], Awaitable[None]]
@@ -137,13 +139,40 @@ async def _fetch_http(
     if encoding:
         raise ValueError(f"Compressed documents are unsupported '{encoding}' ({uri})")
 
+    # CVE-2026-4308: validate the target before the first byte leaves the process. A
+    # blocked target is a hard stop -- never retried, never masked by the generic error
+    # path below (see the BlockedRequestError re-raise in the retry loop).
+    assert_public_url(uri)
+
     last_error = ""
     for attempt in range(retries):
         try:
             async with aiohttp.ClientSession(
                 timeout=aiohttp.ClientTimeout(total=timeout)
             ) as session:
-                async with session.get(uri, allow_redirects=True) as response:
+                # Redirects are followed manually so every hop is re-validated. Delegating
+                # to allow_redirects would let a public URL bounce us into the private
+                # network on hop two, which is the whole point of the CVE.
+                current_uri = uri
+                response = None
+                for _hop in range(MAX_REDIRECTS + 1):
+                    response = await session.get(current_uri, allow_redirects=False)
+                    if response.status not in (301, 302, 303, 307, 308):
+                        break
+                    location = response.headers.get("location", "")
+                    redirect_from = response.url
+                    response.release()
+                    if not location:
+                        raise ValueError("redirect without a location header")
+                    current_uri = str(redirect_from.join(URL(location)))
+                    assert_public_url(current_uri)
+                else:
+                    if response is not None:
+                        response.release()
+                    raise ValueError(f"too many redirects (> {MAX_REDIRECTS})")
+
+                assert response is not None
+                try:
                     if response.status > 399:
                         raise ValueError(f"HTTP {response.status}")
 
@@ -187,6 +216,11 @@ async def _fetch_http(
                         charset=charset,
                         content=b"".join(chunks),
                     )
+                finally:
+                    response.release()
+        except BlockedRequestError:
+            # A refused target is a security decision, not a transient failure.
+            raise
         except Exception as e:
             last_error = str(e)
             if attempt < retries - 1: