From 551c3c7f692299ea90f17b6e7b16a5db24b26f80 Mon Sep 17 00:00:00 2001 From: 3baprinting <3baprinting@gmail.com> Date: Mon, 27 Jul 2026 15:58:35 +0800 Subject: [PATCH 1/6] security: guard path traversal and SSRF --- helpers/net_guard.py | 107 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 helpers/net_guard.py diff --git a/helpers/net_guard.py b/helpers/net_guard.py new file mode 100644 index 0000000000..f4390c8066 --- /dev/null +++ b/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 From 077c694d1e77895da1928c8b90ff4b9f799cdf53 Mon Sep 17 00:00:00 2001 From: 3baprinting <3baprinting@gmail.com> Date: Mon, 27 Jul 2026 15:58:47 +0800 Subject: [PATCH 2/6] security: guard path traversal and SSRF --- tests/test_path_traversal_and_ssrf_guards.py | 156 +++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 tests/test_path_traversal_and_ssrf_guards.py diff --git a/tests/test_path_traversal_and_ssrf_guards.py b/tests/test_path_traversal_and_ssrf_guards.py new file mode 100644 index 0000000000..b44c71be07 --- /dev/null +++ b/tests/test_path_traversal_and_ssrf_guards.py @@ -0,0 +1,156 @@ +"""Regression tests for the two patched CVEs. + +Each test states the exploit that worked against stock v2.6, then asserts it is refused. +Run: ./.venv-sec/bin/python -m pytest tests/test_security_patches.py -q +""" + +from __future__ import annotations + +import asyncio +import os + +import pytest + +from helpers import files +from helpers.net_guard import ( + BlockedRequestError, + assert_public_url, + is_public_url, +) + + +# --- CVE-2026-4307: path traversal / arbitrary file read ----------------------- + + +def test_absolute_path_outside_base_is_refused(): + """Stock v2.6: get_abs_path('/etc/passwd') returned '/etc/passwd' verbatim.""" + with pytest.raises(files.PathEscapesBaseDirError): + files.get_abs_path_contained("/etc/passwd") + + +def test_dotdot_traversal_is_refused(): + """Stock v2.6: os.path.join(base, '../../etc/passwd') walked straight out.""" + with pytest.raises(files.PathEscapesBaseDirError): + files.get_abs_path_contained("../../etc/passwd") + with pytest.raises(files.PathEscapesBaseDirError): + files.get_abs_path_contained("usr/../../../etc/shadow") + + +def test_nested_traversal_is_refused(): + for candidate in ( + "tmp/../../../../root/.ssh/id_rsa", + "./../../etc/hosts", + "usr/uploads/../../../../etc/passwd", + ): + with pytest.raises(files.PathEscapesBaseDirError): + files.get_abs_path_contained(candidate) + + +def test_legitimate_paths_still_resolve(): + """The guard must not simply break the feature.""" + inside = files.get_abs_path_contained("usr") + assert inside.startswith(os.path.realpath(files.get_base_dir())) + assert files.get_abs_path_contained(".") == os.path.realpath(files.get_base_dir()) + + +def test_symlink_out_of_base_is_refused(): + """realpath() resolution means a symlink cannot be used as a side door.""" + base = os.path.realpath(files.get_base_dir()) + link = os.path.join(base, "tmp", "_pytest_escape_link") + os.makedirs(os.path.dirname(link), exist_ok=True) + if os.path.islink(link): + os.unlink(link) + os.symlink("/etc", link) + try: + with pytest.raises(files.PathEscapesBaseDirError): + files.get_abs_path_contained("tmp/_pytest_escape_link/passwd") + finally: + os.unlink(link) + + +# --- CVE-2026-4308: SSRF -------------------------------------------------------- + + +@pytest.mark.parametrize("url", [ + "http://127.0.0.1:8080/admin", + "http://localhost/", + "http://169.254.169.254/latest/meta-data/", # cloud instance credentials + "http://[::1]/", + "http://0.0.0.0/", + "http://10.0.0.5/internal", + "http://192.168.1.1/router", + "http://172.16.0.10/", + "http://[::ffff:127.0.0.1]/", # IPv4-mapped IPv6 loopback +]) +def test_private_and_loopback_targets_are_refused(url): + """Stock v2.6 fetched every one of these with allow_redirects=True and no checks.""" + with pytest.raises(BlockedRequestError): + assert_public_url(url) + + +@pytest.mark.parametrize("url", [ + "file:///etc/passwd", + "gopher://127.0.0.1:11211/", + "ftp://internal.host/secrets", + "dict://127.0.0.1:11211/stat", +]) +def test_non_http_schemes_are_refused(url): + with pytest.raises(BlockedRequestError): + assert_public_url(url) + + +@pytest.mark.parametrize("url", [ + "https://93.184.216.34/doc.pdf", # public IPv4 literal + "https://8.8.8.8/", # public IPv4 literal + "https://[2606:4700:4700::1111]/", # public IPv6 literal +]) +def test_public_targets_are_allowed(url): + """The guard must not over-block: real public addresses still pass. + + IP literals are used deliberately so this does not depend on DNS. Some sandboxed or + proxied networks resolve every name into 198.18.0.0/15 (RFC 2544 benchmark space), + which is correctly *not* public — a hostname assertion would fail there for + environmental reasons rather than a logic error. + """ + assert is_public_url(url) is True + + +def test_hostname_resolution_path_is_exercised(): + """Whatever a name resolves to, the verdict must match that address's class.""" + import ipaddress + import socket + + try: + resolved = socket.getaddrinfo("example.com", None)[0][4][0] + except socket.gaierror: + pytest.skip("no DNS in this environment") + address = ipaddress.ip_address(resolved) + expected_public = not (address.is_private or address.is_loopback) + assert is_public_url("https://example.com/doc.pdf") is expected_public + + +def test_hostname_that_does_not_resolve_is_refused(): + with pytest.raises(BlockedRequestError): + assert_public_url("https://this-host-should-not-exist.invalid/x") + + +def test_url_without_host_is_refused(): + with pytest.raises(BlockedRequestError): + assert_public_url("http:///nohost") + + +def test_fetcher_refuses_a_blocked_target_without_retrying(): + """A refused target must raise immediately, not be retried and then masked as a + generic 'Document fetch error' (which is what the shared retry path would do).""" + from plugins._document_query.helpers.fetch import fetch_public_resource + + with pytest.raises(BlockedRequestError): + asyncio.run(fetch_public_resource("http://169.254.169.254/latest/meta-data/")) + + +def test_fetch_no_longer_delegates_redirects(): + """Guard against the patch being reverted: allow_redirects=True must not return.""" + source = open("plugins/_document_query/helpers/fetch.py").read() + assert "allow_redirects=True" not in source + assert "allow_redirects=False" in source + assert "assert_public_url" in source From 3b44c89f2fd09faa5b0512c4757da68b573bfcb6 Mon Sep 17 00:00:00 2001 From: 3baprinting <3baprinting@gmail.com> Date: Mon, 27 Jul 2026 15:59:04 +0800 Subject: [PATCH 3/6] security: guard path traversal and SSRF --- helpers/files.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/helpers/files.py b/helpers/files.py index c77ab54cf1..aa11269151 100644 --- a/helpers/files.py +++ b/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) From 27d9343fc11cea44c03930619278ff917e013bb8 Mon Sep 17 00:00:00 2001 From: 3baprinting <3baprinting@gmail.com> Date: Mon, 27 Jul 2026 15:59:09 +0800 Subject: [PATCH 4/6] security: guard path traversal and SSRF --- api/api_files_get.py | 33 ++++++++++++++++++++------------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/api/api_files_get.py b/api/api_files_get.py index b2533f4f4f..7f99a17db8 100644 --- a/api/api_files_get.py +++ b/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): From cc93fbf1f401ef10cdae90365ff73d81a58076d3 Mon Sep 17 00:00:00 2001 From: 3baprinting <3baprinting@gmail.com> Date: Mon, 27 Jul 2026 15:59:14 +0800 Subject: [PATCH 5/6] security: guard path traversal and SSRF --- api/file_info.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/api/file_info.py b/api/file_info.py index 9f2a9d6650..d3eea3f54d 100644 --- a/api/file_info.py +++ b/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 = "" From e52439c9577a8ad2387e0a4a14f06f3d4bdf9a42 Mon Sep 17 00:00:00 2001 From: 3baprinting <3baprinting@gmail.com> Date: Mon, 27 Jul 2026 15:59:18 +0800 Subject: [PATCH 6/6] security: guard path traversal and SSRF --- plugins/_document_query/helpers/fetch.py | 36 +++++++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/plugins/_document_query/helpers/fetch.py b/plugins/_document_query/helpers/fetch.py index 65b1a65675..288fe209ec 100644 --- a/plugins/_document_query/helpers/fetch.py +++ b/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: