Coverage for open_webui/utils/terminals.py: 16%

1"""Shared routing helpers for admin-configured terminal servers."""

3import asyncio

4import logging

5import posixpath

6from urllib.parse import quote

8from open_webui.utils.chat_id import is_saved_chat_id

10TERMINAL_CONTEXT_HEADER = 'X-Terminal-Context-Id'

11TERMINAL_CONTEXT_DEFAULT = 'default'

12TERMINAL_CONTEXT_TYPES = {'chat', 'automation'}

13TERMINAL_CONTEXT_ID_SOURCES = {'chat': 'chat_id', 'automation': 'automation_id'}

14TERMINAL_CHAT_UPLOAD_MODES = {'default', 'filesystem'}

15MAX_AGENTS_MD_BYTES = 32 * 1024

17log = logging.getLogger(__name__)

20def is_terminal_orchestrator(connection: dict) -> bool:

21 """Return whether this connection points at Terminals, not raw Open Terminal."""

22 return connection.get('server_type') == 'orchestrator' or bool(connection.get('policy_id'))

25def get_terminal_server_url(connection: dict) -> str:

26 """Return the upstream base URL for a terminal connection.

28 An explicit policy uses the named-policy route. Connections without one

29 keep their existing root route.

30 """

31 base_url = str(connection.get('url') or '').rstrip('/')

32 policy_id = str(connection.get('policy_id') or '').strip()

33 if policy_id:

34 return f'{base_url}/p/{quote(policy_id, safe="")}'

35 return base_url

38def terminal_context_config(connection: dict, context: str) -> dict | bool:

39 """Return config for an OpenWebUI terminal context.

41 Missing config is legacy behavior: available, shared default terminal.

42 """

43 if not is_terminal_orchestrator(connection):

44 return {}

46 contexts = (connection.get('config') or {}).get('contexts')

47 if not isinstance(contexts, dict):

48 return {}

50 value = contexts.get(context, {})

51 if value is False:

52 return False

53 return value if isinstance(value, dict) else {}

56def terminal_context_available(connection: dict, context: str) -> bool:

57 """Return whether this terminal is exposed in an OpenWebUI context."""

58 if context not in TERMINAL_CONTEXT_TYPES:

59 return False

60 return terminal_context_config(connection, context) is not False

63def terminal_context_id(

64 connection: dict,

65 metadata: dict | None = None,

66 context: str = 'chat',

67) -> str | None:

68 """Return the terminal runtime context for trusted request metadata."""

69 if not is_terminal_orchestrator(connection) or not terminal_context_available(connection, context):

70 return None

72 config = terminal_context_config(connection, context)

73 context_id_source = config.get('context_id') if isinstance(config, dict) else None

74 if not context_id_source or context_id_source == TERMINAL_CONTEXT_DEFAULT:

75 return None

77 if context_id_source != TERMINAL_CONTEXT_ID_SOURCES.get(context):

78 return None

80 metadata = metadata or {}

82 if context == 'automation':

83 automation_id = metadata.get('automation_id')

84 return f'automation:{automation_id}' if automation_id else None

86 chat_id = metadata.get('chat_id')

87 if context == 'chat' and chat_id and is_saved_chat_id(chat_id):

88 return f'chat:{chat_id}'

89 return None

92def terminal_contexts(connection: dict) -> dict:

93 """Return normalized sparse context config for clients."""

94 if not is_terminal_orchestrator(connection):

95 return {}

97 contexts = (connection.get('config') or {}).get('contexts')

98 if not isinstance(contexts, dict): 98 ↛ 101line 98 didn't jump to line 101 because the condition on line 98 was always true

99 return {}

100

101 result = {}

102 for context, value in contexts.items():

103 if context not in TERMINAL_CONTEXT_TYPES:

104 continue

105 if value is False:

106 result[context] = False

107 elif isinstance(value, dict):

108 context_id_source = value.get('context_id')

109 if context_id_source in {TERMINAL_CONTEXT_DEFAULT, TERMINAL_CONTEXT_ID_SOURCES[context]}:

110 result[context] = {'context_id': context_id_source}

111 else:

112 result[context] = {}

113 return result

114

115

116def terminal_chat_uploads(connection: dict) -> str:

117 """Return normalized main-chat upload behavior for this connection."""

118 value = (connection.get('config') or {}).get('chat_uploads')

119 return value if value in TERMINAL_CHAT_UPLOAD_MODES else 'default'

120

121

122async def get_terminal_request_info(request, user, metadata: dict, extra_params: dict | None = None):

123 from open_webui.models.config import Config

124 from open_webui.models.groups import Groups

125 from open_webui.models.users import UserModel

126 from open_webui.utils.access_control import has_connection_access

127 from open_webui.utils.headers import bearer_auth_header

128 from open_webui.utils.tools import build_tool_server_headers

129

130 metadata = metadata or {}

131 terminal_id = metadata.get('terminal_id')

132 if not terminal_id:

133 return None

134

135 user_model = user if isinstance(user, UserModel) else UserModel(**user)

136 connections = await Config.get('terminal_server.connections', []) or []

137 connection = next((item for item in connections if item.get('id') == terminal_id), None)

138

139 if connection:

140 terminal_context = 'automation' if metadata.get('automation_id') else 'chat'

141 context_id = terminal_context_id(connection, metadata, terminal_context)

142 config = terminal_context_config(connection, terminal_context)

143 if (

144 not connection.get('enabled', True)

145 or not terminal_context_available(connection, terminal_context)

146 or (config.get('context_id') in {'chat_id', 'automation_id'} and not context_id)

147 ):

148 return None

149 user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user_model.id)}

150 if not await has_connection_access(user_model, connection, user_group_ids):

151 return None

152

153 headers, cookies = await build_tool_server_headers(

154 connection,

155 request,

156 user_model,

157 metadata=metadata,

158 extra_params=extra_params,

159 )

160 headers['Accept'] = 'application/json'

161 headers['X-User-Id'] = user_model.id

162 if metadata.get('chat_id'):

163 headers['X-Session-Id'] = metadata['chat_id']

164 if context_id:

165 headers[TERMINAL_CONTEXT_HEADER] = context_id

166 return get_terminal_server_url(connection), headers, cookies

167

168 selector = str(terminal_id).rstrip('/')

169 direct_terminal = next(

170 (

171 server

172 for server in metadata.get('tool_servers') or []

173 if str(server.get('url') or '').rstrip('/') == selector

174 ),

175 None,

176 )

177 if not direct_terminal:

178 return None

179

180 headers = {'Accept': 'application/json'}

181 key = str(direct_terminal.get('key') or '').strip()

182 if key:

183 headers.update(bearer_auth_header(key))

184 if metadata.get('chat_id'):

185 headers['X-Session-Id'] = metadata['chat_id']

186 return selector, headers, {}

187

188

189async def get_terminal_agents_md(request, user, metadata: dict, extra_params: dict | None = None) -> str | None:

190 """Load the selected terminal user's home AGENTS.md afresh for this turn."""

191 import aiohttp

192 from open_webui.env import AIOHTTP_CLIENT_SESSION_TOOL_SERVER_SSL

193

194 try:

195 async with asyncio.timeout(5):

196 terminal_request = await get_terminal_request_info(request, user, metadata, extra_params)

197 if not terminal_request:

198 return None

199 base_url, headers, cookies = terminal_request

200 async with aiohttp.ClientSession(

201 headers=headers, cookies=cookies, timeout=aiohttp.ClientTimeout(total=5), trust_env=True

202 ) as session:

203 async with session.get(

204 f'{base_url.rstrip("/")}/files/cwd',

205 ssl=AIOHTTP_CLIENT_SESSION_TOOL_SERVER_SSL,

206 allow_redirects=False,

207 ) as response:

208 if response.status != 200:

209 log.debug('Skipping terminal AGENTS.md: home lookup returned HTTP %s', response.status)

210 return None

211 data = await response.json()

212 home = data.get('home') if isinstance(data, dict) else None

213 if not isinstance(home, str) or not posixpath.isabs(home):

214 return None

215 path = posixpath.join(home, 'AGENTS.md')

216 async with session.get(

217 f'{base_url.rstrip("/")}/files/read',

218 params={'path': path},

219 ssl=AIOHTTP_CLIENT_SESSION_TOOL_SERVER_SSL,

220 allow_redirects=False,

221 ) as response:

222 if response.status == 404:

223 return None

224 response.raise_for_status()

225 data = await response.json()

226

227 content = data.get('content') if isinstance(data, dict) else None

228 if not isinstance(content, str) or not content.strip():

229 return None

230 if len(content.encode('utf-8')) > MAX_AGENTS_MD_BYTES:

231 log.warning('Skipping terminal AGENTS.md: exceeds %s bytes', MAX_AGENTS_MD_BYTES)

232 return None

233 return f'# AGENTS.md\n\n{content}'

234 except Exception as e:

235 log.debug('Failed to load terminal AGENTS.md (%s)', type(e).__name__)

236 return None

237

238

239def add_terminal_agents_md(messages: list[dict], agents_md: str) -> list[dict]:

240 """Place file instructions before user requests without changing their content."""

241 for index, message in enumerate(messages):

242 if message.get('role') == 'user':

243 return [*messages[:index], {'role': 'user', 'content': agents_md}, *messages[index:]]

244 return messages

245

246

247async def get_terminal_skill(

248 request, user, metadata: dict, skill_name: str, extra_params: dict | None = None

249) -> dict | None:

250 from urllib.parse import quote

251

252 import aiohttp

253 from open_webui.env import AIOHTTP_CLIENT_SESSION_TOOL_SERVER_SSL, AIOHTTP_CLIENT_TIMEOUT_TOOL_SERVER_DATA

254

255 terminal_request = await get_terminal_request_info(request, user, metadata, extra_params)

256 if not terminal_request:

257 return None

258 base_url, headers, cookies = terminal_request

259

260 timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_TOOL_SERVER_DATA)

261 async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session:

262 async with session.get(

263 f'{base_url.rstrip("/")}/skills/{quote(skill_name, safe="")}',

264 headers=headers,

265 cookies=cookies,

266 ssl=AIOHTTP_CLIENT_SESSION_TOOL_SERVER_SSL,

267 ) as response:

268 skill = await response.json() if response.status == 200 else None

269

270 if not isinstance(skill, dict):

271 return None

272

273 location = skill.get('location') or skill.get('path') or ''

274 directory = location.rsplit('/', 1)[0] if '/' in location else location

275 resources = skill.get('resources') if isinstance(skill.get('resources'), list) else []

276 return {

277 'name': skill.get('name'),

278 'description': skill.get('description'),

279 'content': skill.get('content'),

280 'directory': directory,

281 'resources': resources,

282 }

283

284

285def format_terminal_skill_context(skill: dict) -> str:

286 resources = skill.get('resources') if isinstance(skill.get('resources'), list) else []

287 parts = [f'<skill name="{skill.get("name") or ""}">', skill.get('content') or '']

288 if skill.get('directory'):

289 parts.append(f'<directory>{skill["directory"]}</directory>')

290 if resources:

291 parts.append('<resources>')

292 parts.extend(f'<file>{resource}</file>' for resource in resources)

293 parts.append('</resources>')

294 parts.append('</skill>')

295 return '\n'.join(parts)

296

297

298def format_terminal_skill_manifest_entry(skill: dict) -> str:

299 location = skill.get('location') or skill.get('path') or ''

300 location_tag = f'<location>{location}</location>\n' if location else ''

301 return (

302 f'<skill>\n<id>{skill["id"]}</id>\n<name>{skill["name"]}</name>\n'

303 f'<description>{skill.get("description") or ""}</description>\n'

304 f'<source>terminal</source>\n'

305 f'{location_tag}'

306 f'</skill>\n'

307 )