Release utcp-http 1.1.15, utcp-websocket 1.2.1

plugins/communication_protocols/http/README.md

@@ -118,6 +118,34 @@ client = await UtcpClient.create()
 await client.register_manual(manual)
 ```
 
+## Security
+
+Every outbound request must use HTTPS or address a literal loopback host over
+plain HTTP; plain HTTP to any other host is refused. Loopback is a trust
+boundary: it exists for local development, so a manual or OpenAPI spec
+fetched from a remote (non-loopback) origin may not aim requests at the
+agent's own interface, and nothing that happens later may move a request
+there.
+
+- A remote manual or spec may not declare a loopback tool URL, server URL or
+  OAuth2 `token_url` (`http://127.0.0.1:9200/...`, `https://localhost/token`,
+  any `127.0.0.0/8`, `0.0.0.0`, `::1` or IPv4-mapped spelling).
+- At call time a resolved URL may only be loopback when the call template
+  names a loopback host literally. `http://localhost:8080/items/{id}` works;
+  `http://{host}:8080/items` with `host=127.0.0.1` is refused, whoever wrote
+  the template. A template may still take its host from an argument for any
+  other destination, and OpenAPI server variables with a `default` are
+  resolved when a spec is converted.
+- A redirect is never followed into loopback from a non-loopback URL, and the
+  local-dev exemption is judged by the final URL a manual or spec was served
+  from, not the one first requested.
+
+The same rules apply to the HTTP, SSE and Streamable HTTP protocols. The
+OpenAPI rules can be lifted for a spec you trust: the text call template's
+`base_url` (or `OpenApiConverter(..., base_url=...)` when you drive the
+converter yourself) replaces the spec's servers with your own choice and is
+treated as your trust decision for the spec's OAuth token URL as well.
+
 ## Error Handling
 
 ```python

plugins/communication_protocols/http/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp-http"
-version = "1.1.14"
+version = "1.1.15"
 authors = [
   { name = "UTCP Contributors" },
 ]

plugins/communication_protocols/http/src/utcp_http/_security.py

@@ -162,6 +162,81 @@ def ensure_secure_url(url: str, *, context: Optional[str] = None) -> None:
     )
 
 
+# The scheme and authority of an absolute URL template, kept verbatim so a
+# placeholder in the host position survives for the caller to notice
+# (``urlparse`` lower-cases the host and may choke on a templated port).
+_ABSOLUTE_URL_ORIGIN = re.compile(r"^\s*([A-Za-z][A-Za-z0-9+.\-]*)://([^/?#]*)")
+
+
+def _host_of_authority(authority: str) -> str:
+    """The host inside a URL authority (``user:pw@host:port``), verbatim."""
+    hostport = authority.rsplit("@", 1)[-1]
+    if hostport.startswith("["):
+        end = hostport.find("]")
+        return hostport[1:end] if end != -1 else hostport[1:]
+    return hostport.split(":", 1)[0]
+
+
+def url_template_literal_host(url_template: str) -> Optional[str]:
+    """Return the host ``url_template`` names literally, or ``None``.
+
+    ``None`` when the template has no absolute origin of its own (a relative
+    reference, or something that is not a URL) and when the host position
+    holds a placeholder, so the host is only decided at call time.
+    """
+    if not isinstance(url_template, str):
+        return None
+    m = _ABSOLUTE_URL_ORIGIN.match(url_template)
+    if not m:
+        return None
+    host = _host_of_authority(m.group(2))
+    if not host or "{" in host:
+        return None
+    return host.lower()
+
+
+def url_template_host_is_loopback(url_template: str) -> bool:
+    """Return True if ``url_template`` literally names a loopback host.
+
+    The placeholder-aware twin of :func:`is_loopback_url`: a template whose
+    host is a placeholder is *not* loopback, whatever it resolves to later.
+    """
+    host = url_template_literal_host(url_template)
+    return host is not None and _is_loopback_host(host)
+
+
+def ensure_loopback_was_declared(
+    url_template: str, resolved_url: str, *, context: str
+) -> None:
+    """Raise ``ValueError`` if filling ``url_template`` produced a loopback
+    ``resolved_url`` that the template did not declare.
+
+    ``ensure_secure_url`` allows plain HTTP to loopback so local development
+    works, and ``reject_remote_loopback_tool_urls`` and the OpenAPI converter
+    judge a *template* at registration. All three see
+    ``http://{host}:7777/admin`` as harmless; the request it produces with
+    ``host=127.0.0.1`` is not. This is the one rule that covers a host chosen
+    by a tool argument, and it holds whoever wrote the template: a request may
+    target loopback only when the template itself names a loopback host
+    literally. Tool arguments may choose the rest of the URL, never move the
+    request onto the agent's own interface.
+
+    Call it on every tool invocation, right after ``ensure_secure_url`` on the
+    resolved URL.
+    """
+    if not is_loopback_url(resolved_url):
+        return
+    if url_template_host_is_loopback(url_template):
+        return
+    raise ValueError(
+        f"Security error during {context}: the resolved URL {resolved_url!r} "
+        "targets the agent's own loopback interface, but the call template's "
+        f"URL ({url_template!r}) does not name a loopback host literally. Tool "
+        "arguments substituted into a URL template may choose the path, never "
+        "the host; write the loopback host in the template itself."
+    )
+
+
 # HTTP statuses where the server expects the client to re-issue the request
 # against the URL given in the ``Location`` header. 303 forces a GET; the
 # rest preserve the original method.
@@ -531,17 +606,45 @@ def reject_remote_loopback_tool_urls(
     the URL originally requested: a loopback discovery URL that redirects to a
     remote origin is serving a remote manual and must not keep the local-dev
     exemption.
+
+    Two things a remote manual may not do, per tool:
+
+    * declare a loopback tool URL (the original rule);
+    * declare a loopback OAuth2 ``token_url``: the client-credentials POST is a
+      request the agent host makes on the manual's behalf, exactly like a tool
+      call, and it must not be aimed at loopback either.
+
+    A tool URL whose host is a placeholder (``http://{host}:7777/admin``) is
+    accepted here: the host is only known at call time, and that is where
+    ``ensure_loopback_was_declared`` refuses to let the arguments move the
+    request onto loopback. Judging the template here would only break manuals
+    whose tools legitimately take their destination from an argument.
     """
     if is_loopback_url(discovery_url):
         return
     for tool in getattr(manual, "tools", None) or []:
+        tool_name = getattr(tool, "name", "?")
         call_template = getattr(tool, "tool_call_template", None)
         url = getattr(call_template, "url", None)
-        if isinstance(url, str) and is_loopback_url(url):
+        if isinstance(url, str):
+            # Both classifiers: a template whose port is a placeholder
+            # (``http://localhost:{port}/x``) may not parse as a URL, yet it
+            # names a loopback host as literally as any other.
+            if is_loopback_url(url) or url_template_host_is_loopback(url):
+                raise ValueError(
+                    f"Security error during {context}: a manual fetched from "
+                    f"{discovery_url!r} declares a loopback tool URL ({url!r}) "
+                    f"for tool {tool_name!r}. A remote manual is not allowed to "
+                    "redirect tool calls at the agent's own loopback interface."
+                )
+        token_url = getattr(getattr(call_template, "auth", None), "token_url", None)
+        if isinstance(token_url, str) and (
+            is_loopback_url(token_url) or url_template_host_is_loopback(token_url)
+        ):
             raise ValueError(
                 f"Security error during {context}: a manual fetched from "
-                f"{discovery_url!r} declares a loopback tool URL ({url!r}) for "
-                f"tool {getattr(tool, 'name', '?')!r}. A remote manual is not "
-                "allowed to redirect tool calls at the agent's own loopback "
-                "interface."
+                f"{discovery_url!r} declares a loopback OAuth2 token URL "
+                f"({token_url!r}) for tool {tool_name!r}. A remote manual is not "
+                "allowed to route the client-credentials POST at the agent's "
+                "own loopback interface."
             )

plugins/communication_protocols/http/src/utcp_http/http_communication_protocol.py

@@ -33,7 +33,12 @@
 from utcp_http.http_call_template import HttpCallTemplate
 from aiohttp import ClientSession, BasicAuth as AiohttpBasicAuth
 from utcp_http.openapi_converter import OpenApiConverter
-from utcp_http._security import ensure_secure_url, safe_request_with_redirects, reject_remote_loopback_tool_urls
+from utcp_http._security import (
+    ensure_secure_url,
+    ensure_loopback_was_declared,
+    safe_request_with_redirects,
+    reject_remote_loopback_tool_urls,
+)
 from utcp_http._errors import raise_for_status_with_body
 import logging
 
@@ -232,7 +237,10 @@ async def register_manual(self, caller, manual_call_template: CallTemplate) -> R
                             reject_remote_loopback_tool_urls(str(response.url), utcp_manual)
                         else:
                             logger.info(f"Assuming OpenAPI spec from '{manual_call_template.name}'. Converting to UTCP manual.")
-                            converter = OpenApiConverter(response_data, spec_url=manual_call_template.url, call_template_name=manual_call_template.name, auth_tools=manual_call_template.auth_tools)
+                            # The final (post-redirect) URL is where the spec really
+                            # came from: it decides whether the spec is local or
+                            # remote, and relative ``tokenUrl`` values resolve against it.
+                            converter = OpenApiConverter(response_data, spec_url=str(response.url), call_template_name=manual_call_template.name, auth_tools=manual_call_template.auth_tools)
                             utcp_manual = converter.convert()
 
                         return RegisterManualResult(
@@ -317,6 +325,11 @@ async def call_tool(self, caller, tool_name: str, tool_args: Dict[str, Any], too
         # for an unauthenticated Elasticsearch). Without this re-check, tool
         # invocation is a blind SSRF primitive — see GHSA / issue #83.
         ensure_secure_url(url, context="tool invocation")
+        # A loopback destination must be written in the template itself: a
+        # placeholder in the host (``http://{host}:7777/admin``) would let the
+        # tool's arguments move the request onto the agent's own interface
+        # after every registration-time check has passed.
+        ensure_loopback_was_declared(tool_call_template.url, url, context="tool invocation")
 
         # The rest of the arguments are query parameters
         query_params = remaining_args

plugins/communication_protocols/http/src/utcp_http/openapi_converter.py

@@ -27,7 +27,7 @@
 from utcp.data.utcp_manual import UtcpManual
 from utcp.data.tool import Tool, JsonSchema
 from utcp_http.http_call_template import HttpCallTemplate
-from utcp_http._security import ensure_secure_url, is_loopback_url
+from utcp_http._security import ensure_secure_url, is_loopback_url, url_template_host_is_loopback
 
 # All HTTP methods that OpenAPI defines as operation fields on a Path Item
 # Object. The conversion loop uses this to tell operations apart from the other
@@ -164,7 +164,8 @@ def convert(self) -> UtcpManual:
             # trust decision, no further validation here.
             base_url = self._base_url_override
         elif self.spec.get("servers"):
-            base_url = self.spec["servers"][0].get("url", "/")
+            server = self.spec["servers"][0]
+            base_url = self._resolve_server_variables(server)
 
             # Rule: a spec fetched from a non-loopback source cannot declare
             # a loopback server URL. A user pointing the converter at their
@@ -174,7 +175,7 @@ def convert(self) -> UtcpManual:
             if (
                 self.spec_url
                 and not is_loopback_url(self.spec_url)
-                and is_loopback_url(base_url)
+                and (is_loopback_url(base_url) or url_template_host_is_loopback(base_url))
             ):
                 raise ValueError(
                     "Security error: OpenAPI spec fetched from "
@@ -186,6 +187,12 @@ def convert(self) -> UtcpManual:
                     "the call template's ``base_url`` override "
                     "explicitly to bypass this check."
                 )
+
+            # A server URL whose host is still a placeholder after that
+            # (``http://{host}:7777``) becomes a path parameter of every
+            # generated tool. The protocols refuse, at call time, to let such
+            # an argument move a request onto loopback
+            # (``ensure_loopback_was_declared``), so it is not judged here.
         elif self.spec_url:
             parsed_url = urlparse(self.spec_url)
             base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
@@ -203,13 +210,65 @@ def convert(self) -> UtcpManual:
 
         return UtcpManual(tools=tools)
 
+    @staticmethod
+    def _resolve_server_variables(server: Dict[str, Any]) -> str:
+        """Return the server's URL with its declared variables filled in.
+
+        OpenAPI 3.x lets ``servers[].url`` carry ``{name}`` placeholders whose
+        values come from ``servers[].variables``; every such variable must
+        declare a ``default``, and the default is what a client is meant to
+        use. A placeholder without a declared default is left in place.
+        """
+        url = server.get("url", "/")
+        if not isinstance(url, str):
+            return "/"
+        variables = server.get("variables")
+        if isinstance(variables, dict):
+            for name, variable in variables.items():
+                if not isinstance(variable, dict) or "default" not in variable:
+                    continue
+                default = variable["default"]
+                if default is None:
+                    continue
+                url = url.replace("{" + str(name) + "}", str(default))
+        return url
+
+    def _reject_remote_loopback_token_url(self, token_url: str) -> None:
+        """A remote spec may not route the token POST at loopback.
+
+        The same boundary ``convert`` enforces for ``servers[0].url``: the
+        client-credentials POST is a request the agent host makes on the
+        spec's behalf, so a spec fetched from a non-loopback origin must not
+        aim it at the agent's own interface. ``ensure_secure_url`` cannot
+        catch this -- it allows loopback by design, for local development.
+        A spec fetched from loopback keeps the local-dev exemption, and so
+        does a spec registered with an explicit ``base_url`` override: that
+        override is the caller's statement that they trust the spec, and it
+        lifts the server-URL rule for the same reason.
+        """
+        if self._base_url_override:
+            return
+        if (
+            self.spec_url
+            and not is_loopback_url(self.spec_url)
+            and is_loopback_url(token_url)
+        ):
+            raise ValueError(
+                "Security error: OpenAPI spec fetched from "
+                f"{self.spec_url!r} declares a loopback OAuth2 tokenUrl "
+                f"({token_url!r}). A remote spec is not allowed to route the "
+                "client-credentials POST at the agent's own loopback "
+                "interface."
+            )
+
     def _validate_token_url_eagerly(self, token_url: str) -> str:
         """Validate (and, when relative, resolve) an OpenAPI OAuth2
         ``tokenUrl`` at conversion time. Returns the absolute URL
         that should be embedded in the generated ``OAuth2Auth`` so
         the runtime check in ``_handle_oauth2`` sees a usable value
         instead of an unresolved relative reference. Backs
-        GHSA-8cp3-qxj6-px34.
+        GHSA-8cp3-qxj6-px34; the loopback rule
+        (``_reject_remote_loopback_token_url``) closes its residual.
 
         OpenAPI 3.0 / 3.1 explicitly allow ``tokenUrl`` to be a
         relative reference resolved against the spec's own location.
@@ -232,6 +291,7 @@ def _validate_token_url_eagerly(self, token_url: str) -> str:
 
         if is_absolute:
             ensure_secure_url(token_url, context="OAuth2 tokenUrl in OpenAPI spec")
+            self._reject_remote_loopback_token_url(token_url)
             return token_url
 
         if self.spec_url:
@@ -245,6 +305,7 @@ def _validate_token_url_eagerly(self, token_url: str) -> str:
                     resolved,
                     context="OAuth2 tokenUrl in OpenAPI spec (resolved from relative URL)",
                 )
+                self._reject_remote_loopback_token_url(resolved)
                 return resolved
 
         return token_url

plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py

@@ -19,7 +19,12 @@
 from utcp_http.sse_call_template import SseCallTemplate
 from aiohttp import ClientSession, BasicAuth as AiohttpBasicAuth
 from utcp_http._errors import raise_for_status_with_body
-from utcp_http._security import ensure_secure_url, safe_request_with_redirects, reject_remote_loopback_tool_urls
+from utcp_http._security import (
+    ensure_secure_url,
+    ensure_loopback_was_declared,
+    safe_request_with_redirects,
+    reject_remote_loopback_tool_urls,
+)
 import traceback
 import logging
 
@@ -234,6 +239,11 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str,
         # Defends against SSRF via attacker-controlled OpenAPI specs that point
         # ``servers[0].url`` at internal services. See issue #83.
         ensure_secure_url(url, context="tool invocation")
+        # A loopback destination must be written in the template itself: a
+        # placeholder in the host (``http://{host}:7777/admin``) would let the
+        # tool's arguments move the request onto the agent's own interface
+        # after every registration-time check has passed.
+        ensure_loopback_was_declared(tool_call_template.url, url, context="tool invocation")
 
         # The rest of the arguments are query parameters
         query_params = remaining_args

plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py

@@ -16,7 +16,12 @@
 from utcp_http.streamable_http_call_template import StreamableHttpCallTemplate
 from aiohttp import ClientSession, BasicAuth as AiohttpBasicAuth, ClientResponse
 from utcp_http._errors import raise_for_status_with_body
-from utcp_http._security import ensure_secure_url, safe_request_with_redirects, reject_remote_loopback_tool_urls
+from utcp_http._security import (
+    ensure_secure_url,
+    ensure_loopback_was_declared,
+    safe_request_with_redirects,
+    reject_remote_loopback_tool_urls,
+)
 import logging
 
 logging.basicConfig(
@@ -239,6 +244,11 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str,
         # Defends against SSRF via attacker-controlled OpenAPI specs that point
         # ``servers[0].url`` at internal services. See issue #83.
         ensure_secure_url(url, context="tool invocation")
+        # A loopback destination must be written in the template itself: a
+        # placeholder in the host (``http://{host}:7777/admin``) would let the
+        # tool's arguments move the request onto the agent's own interface
+        # after every registration-time check has passed.
+        ensure_loopback_was_declared(tool_call_template.url, url, context="tool invocation")
 
         # The rest of the arguments are query parameters
         query_params = remaining_args