mirror of
https://github.com/jamiepine/voicebox.git
synced 2026-09-16 21:30:39 -07:00
fix(mcp): restrict voicebox.transcribe(audio_path=...) to loopback
audio_path mode took any absolute filesystem path and returned its decoded contents as transcribed text with no caller verification beyond the existence/size checks. The X-Voicebox-Client-Id middleware records the header but never rejects an absent or fake one, so a Voicebox bound to 0.0.0.0 (the documented "remote access" mode) was effectively an unauthenticated arbitrary-local-file read primitive. The middleware now stashes the request's remote address in a ContextVar alongside the existing client_id, and audio_path mode refuses anything that doesn't parse as a loopback address (IPv4 127.0.0.0/8, IPv6 ::1). audio_base64 mode is unchanged — that path was always bounded to bytes the caller already has. Loopback callers (the Tauri webview, local CLI scripts, MCP clients on the same machine) keep working. Remote callers now have to send the audio over the wire if they want it transcribed.
This commit is contained in:
@@ -7,6 +7,7 @@ ContextVar so tool implementations can read it without plumbing the request
|
||||
object through every service call.
|
||||
"""
|
||||
|
||||
import ipaddress
|
||||
import logging
|
||||
from contextvars import ContextVar
|
||||
from datetime import datetime
|
||||
@@ -26,6 +27,28 @@ current_client_id: ContextVar[str | None] = ContextVar(
|
||||
"current_client_id", default=None
|
||||
)
|
||||
|
||||
# Remote address of the in-flight request. Used by tools that gate
|
||||
# host-filesystem access to loopback callers (see voicebox.transcribe).
|
||||
current_remote_addr: ContextVar[str | None] = ContextVar(
|
||||
"current_remote_addr", default=None
|
||||
)
|
||||
|
||||
|
||||
def request_is_loopback() -> bool:
|
||||
"""True when the in-flight request originated on the loopback interface.
|
||||
|
||||
Returns False if no request is in flight or the remote address can't be
|
||||
parsed — callers gating filesystem reads on this should treat that as
|
||||
"deny".
|
||||
"""
|
||||
addr = current_remote_addr.get()
|
||||
if not addr:
|
||||
return False
|
||||
try:
|
||||
return ipaddress.ip_address(addr).is_loopback
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
# Endpoints that consume X-Voicebox-Client-Id for its MCP-semantic
|
||||
# meaning (per-client profile resolution + per-client default_personality).
|
||||
# These are the paths where a stamp into last_seen_at is accurate.
|
||||
@@ -53,11 +76,14 @@ class ClientIdMiddleware(BaseHTTPMiddleware):
|
||||
|
||||
async def dispatch(self, request: Request, call_next) -> Response:
|
||||
client_id = request.headers.get(CLIENT_ID_HEADER)
|
||||
token = current_client_id.set(client_id)
|
||||
remote_addr = request.client.host if request.client else None
|
||||
client_token = current_client_id.set(client_id)
|
||||
addr_token = current_remote_addr.set(remote_addr)
|
||||
try:
|
||||
response = await call_next(request)
|
||||
finally:
|
||||
current_client_id.reset(token)
|
||||
current_client_id.reset(client_token)
|
||||
current_remote_addr.reset(addr_token)
|
||||
|
||||
if client_id and _is_stamped_path(request.url.path):
|
||||
_stamp_last_seen(client_id)
|
||||
|
||||
@@ -21,7 +21,7 @@ from ..database import get_db
|
||||
from ..services import captures as captures_service
|
||||
from ..services import profiles as profiles_service
|
||||
from . import events as mcp_events
|
||||
from .context import current_client_id
|
||||
from .context import current_client_id, request_is_loopback
|
||||
from .resolve import resolve_profile
|
||||
|
||||
|
||||
@@ -103,7 +103,7 @@ def register_tools(mcp: FastMCP) -> None:
|
||||
description=(
|
||||
"Transcribe an audio clip to text using Voicebox's local Whisper. "
|
||||
"Pass exactly one of `audio_base64` (bytes as base64) or "
|
||||
"`audio_path` (absolute local file path)."
|
||||
"`audio_path` (absolute local file path — loopback callers only)."
|
||||
),
|
||||
)
|
||||
async def voicebox_transcribe(
|
||||
@@ -117,8 +117,15 @@ def register_tools(mcp: FastMCP) -> None:
|
||||
"Pass exactly one of `audio_base64` or `audio_path`."
|
||||
)
|
||||
|
||||
# Absolute-path mode: validate and transcribe in place.
|
||||
# Absolute-path mode: validate and transcribe in place. Restricted
|
||||
# to loopback callers so a Voicebox bound on 0.0.0.0 doesn't double
|
||||
# as an unauthenticated arbitrary-local-file read primitive.
|
||||
if audio_path is not None:
|
||||
if not request_is_loopback():
|
||||
raise ValueError(
|
||||
"`audio_path` is only available to loopback callers — "
|
||||
"remote callers must use `audio_base64`."
|
||||
)
|
||||
path = Path(audio_path)
|
||||
if not path.is_absolute():
|
||||
raise ValueError("`audio_path` must be absolute.")
|
||||
|
||||
Reference in New Issue
Block a user