Coverage for open_webui/routers/terminals.py: 12%
263 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1"""Reverse proxy for admin-configured terminal servers.
3Routes:
4 GET / — list terminals the user has access to
5 * /{server_id}/{path:path} — proxy request to terminal server
6"""
8import asyncio
9import logging
10import posixpath
11from urllib.parse import unquote
13import aiohttp
14from fastapi import APIRouter, Depends, Request, Response, WebSocket
15from fastapi.responses import JSONResponse, StreamingResponse
16from open_webui.config import TERMINAL_PROXY_HEADERS
17from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL
18from open_webui.events import EVENTS, publish_event
19from open_webui.models.config import Config
20from open_webui.models.groups import Groups
21from open_webui.utils.access_control import has_connection_access
22from open_webui.utils.auth import get_verified_user, get_verified_user_by_token
23from open_webui.utils.headers import bearer_auth_header, normalize_bearer_token
24from open_webui.utils.json_codec import JSONCodec
25from open_webui.utils.terminals import (
26 TERMINAL_CONTEXT_HEADER,
27 get_terminal_server_url,
28 is_terminal_orchestrator,
29 terminal_chat_uploads,
30 terminal_context_available,
31 terminal_context_config,
32 terminal_context_id,
33 terminal_contexts,
34)
35from starlette.background import BackgroundTask
36from starlette.requests import ClientDisconnect
37from yarl import URL
39log = logging.getLogger(__name__)
41router = APIRouter()
43STREAMING_CONTENT_TYPES = ('application/octet-stream', 'image/', 'application/pdf')
44ADMIN_API_PATHS = ('api/v1/policies', 'api/v1/status', 'api/v1/terminals')
45# Drop the upstream's server and date: uvicorn adds its own and forwarding both duplicates them.
46STRIPPED_RESPONSE_HEADERS = frozenset(
47 ('transfer-encoding', 'connection', 'content-encoding', 'content-length', 'server', 'date')
48)
51def _sanitize_proxy_path(path: str) -> str | None:
52 """Sanitize a proxy path to prevent directory traversal / SSRF.
54 Returns the cleaned path, or None if the path is invalid.
55 Trailing slashes are preserved — many upstream frameworks treat
56 ``/path`` and ``/path/`` differently.
57 """
58 # Decode until stable: a single unquote pass leaves %252e%252e as %2e%2e,
59 # which the upstream then re-decodes into '..', bypassing the check below.
60 decoded = path
61 for _ in range(8):
62 once = unquote(decoded)
63 if once == decoded:
64 break
65 decoded = once
66 # Fail closed: still encoded after the cap means the upstream would decode further into traversal.
67 if unquote(decoded) != decoded:
68 return None
69 # posixpath splits on '/' only, so 'a/..\..\b' survives normpath as one component.
70 # Upstreams that treat '\' as a separator would resolve it, so reject outright.
71 # URL parsers also remove tabs/newlines, which can turn '.\t.' into '..'.
72 if any(char in decoded for char in '\\\t\r\n'):
73 return None
74 had_trailing_slash = decoded.endswith('/')
75 normalized = posixpath.normpath(decoded)
76 # Remove any leading slashes that would reset the base
77 cleaned = normalized.lstrip('/')
78 # Reject if normpath resolved to parent traversal or current-dir only
79 if cleaned.startswith('..') or cleaned == '.':
80 return None
81 # Restore trailing slash if the original path had one
82 if had_trailing_slash and cleaned and not cleaned.endswith('/'):
83 cleaned += '/'
84 return cleaned
87@router.get('/')
88async def list_terminal_servers(request: Request, user=Depends(get_verified_user)):
89 """Return terminal servers the authenticated user has access to."""
90 connections = await Config.get('terminal_server.connections', []) or []
91 user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id)}
93 return [
94 {
95 'id': connection.get('id', ''),
96 'url': connection.get('url', ''),
97 'name': connection.get('name', ''),
98 'contexts': terminal_contexts(connection),
99 'config': {'chat_uploads': terminal_chat_uploads(connection)},
100 }
101 for connection in connections
102 if connection.get('enabled', True) and await has_connection_access(user, connection, user_group_ids)
103 ]
106PROXY_METHODS = ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS']
109@router.api_route('/{server_id}/{path:path}', methods=PROXY_METHODS)
110async def proxy_terminal(
111 server_id: str,
112 path: str,
113 request: Request,
114 user=Depends(get_verified_user),
115):
116 """Proxy a request to the admin terminal server identified by *server_id*."""
117 connections = await Config.get('terminal_server.connections', []) or []
118 connection = next((c for c in connections if c.get('id') == server_id), None)
120 if connection is None: 120 ↛ 123line 120 didn't jump to line 123 because the condition on line 120 was always true
121 return JSONResponse({'error': f"Terminal server '{server_id}' not found"}, status_code=404)
123 if not connection.get('enabled', True):
124 return JSONResponse({'error': 'Terminal server disabled'}, status_code=403)
126 user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id)}
127 if not await has_connection_access(user, connection, user_group_ids):
128 return JSONResponse({'error': 'Access denied'}, status_code=403)
130 base_url = get_terminal_server_url(connection)
131 if not base_url:
132 return JSONResponse({'error': 'Terminal server URL not configured'}, status_code=503)
134 safe_path = _sanitize_proxy_path(path)
135 if safe_path is None:
136 return JSONResponse({'error': 'Invalid path'}, status_code=400)
138 target_url = f'{base_url}/{safe_path}'
140 # Check the path aiohttp will send, relative to the configured server root.
141 base_path = URL(str(connection.get('url') or '')).path.rstrip('/')
142 target_path = URL(target_url).path
143 if any(
144 target_path == f'{base_path}/{prefix}' or target_path.startswith(f'{base_path}/{prefix}/')
145 for prefix in ADMIN_API_PATHS
146 ):
147 return JSONResponse({'error': 'Path not allowed'}, status_code=403)
149 if request.query_params:
150 target_url += f'?{request.query_params}'
152 headers = {'X-User-Id': user.id}
153 # Forward per-session cwd tracking header
154 session_id = request.headers.get('x-session-id')
155 if session_id:
156 headers['X-Session-Id'] = session_id
157 if not terminal_context_available(connection, 'chat'):
158 return JSONResponse({'error': 'Terminal server is not available in chats'}, status_code=403)
159 context_id = terminal_context_id(connection, {'chat_id': session_id}, 'chat')
160 if terminal_context_config(connection, 'chat').get('context_id') == 'chat_id' and not context_id:
161 return JSONResponse({'error': 'A saved chat is required for this terminal'}, status_code=409)
162 if context_id:
163 headers[TERMINAL_CONTEXT_HEADER] = context_id
164 cookies = getattr(request, 'cookies', {}) if connection.get('forward_cookies', False) else {}
165 auth_type = connection.get('auth_type', 'bearer')
167 if auth_type == 'bearer':
168 headers.update(bearer_auth_header(connection.get('key', '')))
169 elif auth_type == 'session':
170 headers.update(bearer_auth_header(request.state.token.credentials))
171 elif auth_type == 'system_oauth':
172 # Resolve the token server-side from the caller's OAuth session; never trust a client header.
173 oauth_token = None
174 try:
175 if request.cookies.get('oauth_session_id', None):
176 oauth_token = await request.app.state.oauth_manager.get_oauth_token(
177 user.id,
178 request.cookies.get('oauth_session_id', None),
179 )
180 except Exception as e:
181 log.error(f'Error getting OAuth token: {e}')
182 if oauth_token:
183 headers.update(bearer_auth_header(oauth_token.get('access_token', '')))
184 # auth_type == "none": no Authorization header
186 content_type = request.headers.get('content-type')
187 if content_type:
188 headers['Content-Type'] = content_type
190 session = aiohttp.ClientSession(
191 timeout=aiohttp.ClientTimeout(total=300, connect=10),
192 trust_env=True,
193 )
195 try:
196 body = await request.body()
198 upstream_response = await session.request(
199 method=request.method,
200 url=target_url,
201 headers=headers,
202 cookies=cookies,
203 data=body or None,
204 ssl=AIOHTTP_CLIENT_SESSION_SSL,
205 allow_redirects=False,
206 )
208 upstream_content_type = upstream_response.headers.get('content-type', '')
209 filtered_headers = {
210 key: value
211 for key, value in upstream_response.headers.items()
212 if key.lower() not in STRIPPED_RESPONSE_HEADERS
213 }
214 if TERMINAL_PROXY_HEADERS:
215 filtered_headers.update(TERMINAL_PROXY_HEADERS)
217 # Stream binary responses directly
218 if any(t in upstream_content_type for t in STREAMING_CONTENT_TYPES):
220 async def cleanup():
221 await upstream_response.release()
222 await session.close()
224 return StreamingResponse(
225 content=upstream_response.content.iter_any(),
226 status_code=upstream_response.status,
227 headers=filtered_headers,
228 background=BackgroundTask(cleanup),
229 )
231 # Buffer text/JSON responses
232 response_body = await upstream_response.read()
233 status_code = upstream_response.status
234 await upstream_response.release()
235 await session.close()
237 return Response(content=response_body, status_code=status_code, headers=filtered_headers)
239 except ClientDisconnect:
240 await session.close()
241 return Response(status_code=499)
242 except (aiohttp.ClientConnectionError, TimeoutError) as error:
243 await session.close()
244 log.error('Terminal proxy error: %s', str(error) or type(error).__name__)
245 return JSONResponse({'error': f'Terminal proxy error: {error}'}, status_code=502)
246 except Exception as error:
247 await session.close()
248 log.exception('Terminal proxy error: %s', error)
249 return JSONResponse({'error': f'Terminal proxy error: {error}'}, status_code=502)
252# ---------------------------------------------------------------------------
253# WebSocket proxy for interactive terminal sessions
254# ---------------------------------------------------------------------------
257async def _resolve_authenticated_connection(ws: WebSocket, server_id: str):
258 """Authenticate a WebSocket via first-message auth and resolve the terminal server.
260 The client must send ``{"type": "auth", "token": "<jwt>"}`` as its first
261 message after connecting.
263 Returns ``(user, connection, chat_id, token)`` on success, or ``None`` after
264 closing *ws* with an appropriate error code.
265 """
266 # First-message authentication
267 try:
268 raw = await asyncio.wait_for(ws.receive_text(), timeout=10.0)
269 payload = JSONCodec.loads(raw)
270 if payload.get('type') != 'auth':
271 await ws.close(code=4001, reason='Expected auth message')
272 return None
273 token = payload.get('token', '')
274 except (TimeoutError, JSONCodec.JSONDecodeError):
275 await ws.close(code=4001, reason='Auth timeout or invalid payload')
276 return None
277 except Exception:
278 await ws.close(code=4001, reason='Invalid token')
279 return None
281 result = await _resolve_terminal_access(ws, server_id, token)
282 if result is None:
283 return None
284 user, connection = result
285 chat_id = payload.get('chat_id', '')
286 return user, connection, chat_id if isinstance(chat_id, str) else '', token
289async def _resolve_terminal_access(ws: WebSocket, server_id: str, token: str):
290 """Resolve current access for both the handshake and an open terminal session."""
291 try:
292 user = await get_verified_user_by_token(token, getattr(ws.app.state, 'redis', None))
293 if user is None:
294 await ws.close(code=4001, reason='Invalid token')
295 return None
296 except Exception:
297 await ws.close(code=4001, reason='Invalid token')
298 return None
300 # Resolve terminal server
301 connections = await Config.get('terminal_server.connections', []) or []
302 connection = next((c for c in connections if c.get('id') == server_id), None)
304 if connection is None:
305 await ws.close(code=4004, reason='Terminal server not found')
306 return None
308 if not connection.get('enabled', True):
309 await ws.close(code=4003, reason='Terminal server disabled')
310 return None
312 if not await has_connection_access(user, connection):
313 await ws.close(code=4003, reason='Access denied')
314 return None
316 if not terminal_context_available(connection, 'chat'):
317 await ws.close(code=4003, reason='Terminal server is not available in chats')
318 return None
319 return user, connection
322@router.websocket('/{server_id}/api/terminals/{session_id}')
323async def ws_terminal(
324 ws: WebSocket,
325 server_id: str,
326 session_id: str,
327):
328 """Proxy an interactive WebSocket terminal session to a terminal server.
330 Uses first-message auth: the client sends ``{"type": "auth", "token": "<jwt>"}``
331 as its first message. The proxy validates the JWT, then connects to the
332 upstream terminal server using the configured terminal auth mode.
333 """
334 await ws.accept()
336 result = await _resolve_authenticated_connection(ws, server_id)
337 if result is None:
338 return
339 user, connection, chat_id, token = result
341 base_url = get_terminal_server_url(connection)
342 if not base_url:
343 await ws.close(code=4003, reason='Terminal server URL not configured')
344 return
346 # Build upstream WebSocket URL (no token in URL)
347 ws_base = base_url.replace('https://', 'wss://').replace('http://', 'ws://')
349 upstream_params = {}
350 # For orchestrator-backed servers, pass user_id
351 upstream_params['user_id'] = user.id
352 context_id = terminal_context_id(connection, {'chat_id': chat_id}, 'chat')
353 upstream_headers = {'X-User-Id': user.id, 'X-Session-Id': chat_id}
354 if terminal_context_config(connection, 'chat').get('context_id') == 'chat_id' and not context_id:
355 await ws.close(code=4003, reason='A saved chat is required for this terminal')
356 return
357 if context_id:
358 upstream_headers[TERMINAL_CONTEXT_HEADER] = context_id
360 import urllib.parse
362 # Encode session_id as an opaque path segment so it cannot smuggle '?'/'#'/'&' (at any
363 # decode depth) and inject an attacker-chosen user_id ahead of the one appended below.
364 safe_session_id = urllib.parse.quote(session_id, safe='')
366 upstream_url = f'{ws_base}/api/terminals/{safe_session_id}'
367 if upstream_params:
368 upstream_url += f'?{urllib.parse.urlencode(upstream_params)}'
370 app = ws.scope.get('app')
371 opened = False
372 session = aiohttp.ClientSession()
373 try:
374 async with session.ws_connect(
375 upstream_url,
376 headers=upstream_headers,
377 ssl=AIOHTTP_CLIENT_SESSION_SSL,
378 ) as upstream:
379 import json as _json
381 # First-message auth to upstream terminal server
382 auth_type = connection.get('auth_type', 'bearer')
383 if auth_type == 'bearer':
384 key = normalize_bearer_token(connection.get('key', ''))
385 await upstream.send_str(_json.dumps({'type': 'auth', 'token': key}))
386 elif auth_type == 'session' and is_terminal_orchestrator(connection):
387 await upstream.send_str(_json.dumps({'type': 'auth', 'token': token}))
388 else:
389 await upstream.send_str(_json.dumps({'type': 'auth', 'token': ''}))
391 await publish_event(
392 app,
393 EVENTS.TERMINAL_SESSION_OPENED,
394 actor=user,
395 subject_id=session_id,
396 subject_type='terminal.session',
397 data={'server_id': server_id},
398 )
399 opened = True
401 async def _client_to_upstream():
402 """Forward client → upstream."""
403 try:
404 while True:
405 msg = await ws.receive()
406 if msg['type'] == 'websocket.disconnect':
407 break
408 elif 'bytes' in msg and msg['bytes']:
409 await upstream.send_bytes(msg['bytes'])
410 elif 'text' in msg and msg['text']:
411 await upstream.send_str(msg['text'])
412 except Exception:
413 pass
415 async def _upstream_to_client():
416 """Forward upstream → client."""
417 try:
418 async for msg in upstream:
419 if msg.type == aiohttp.WSMsgType.BINARY:
420 await ws.send_bytes(msg.data)
421 elif msg.type == aiohttp.WSMsgType.TEXT:
422 await ws.send_text(msg.data)
423 elif msg.type in (
424 aiohttp.WSMsgType.CLOSE,
425 aiohttp.WSMsgType.ERROR,
426 ):
427 break
428 except Exception:
429 pass
431 async def _watch_access():
432 try:
433 while True:
434 # Poll current state so revocation also works across workers.
435 await asyncio.sleep(10)
436 if await _resolve_terminal_access(ws, server_id, token) is None:
437 return
438 except Exception:
439 log.exception('Terminal access recheck failed')
441 # End the proxy as soon as any task finishes (e.g. a
442 # graceful upstream CLOSE) and cancel the rest, which would
443 # otherwise hang on a blocked ws.receive() until the browser leaves.
444 tasks = [
445 asyncio.create_task(_client_to_upstream()),
446 asyncio.create_task(_upstream_to_client()),
447 asyncio.create_task(_watch_access()),
448 ]
449 try:
450 await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
451 finally:
452 for task in tasks:
453 task.cancel()
454 await asyncio.gather(*tasks, return_exceptions=True)
455 except Exception as e:
456 log.exception('Terminal WebSocket proxy error: %s', e)
457 finally:
458 await session.close()
459 if opened:
460 await publish_event(
461 app,
462 EVENTS.TERMINAL_SESSION_CLOSED,
463 actor=user,
464 subject_id=session_id,
465 subject_type='terminal.session',
466 data={'server_id': server_id},
467 )
468 try:
469 await ws.close()
470 except Exception:
471 pass