diff --git a/backend/open_webui/main.py b/backend/open_webui/main.py
index 7fc9304bce..863d8be4a0 100644
--- a/backend/open_webui/main.py
+++ b/backend/open_webui/main.py
@@ -411,6 +411,9 @@ async def lifespan(app: FastAPI):
await install_tool_and_function_dependencies()
app.state.redis = get_redis_client(async_mode=True)
+ from open_webui.socket.apps import connections as app_connections
+
+ await app_connections.start(app.state.redis)
if app.state.redis is not None:
app.state.redis_task_command_listener = asyncio.create_task(redis_task_command_listener(app))
@@ -503,6 +506,7 @@ async def lifespan(app: FastAPI):
yield
await publish_event(app, EVENTS.SYSTEM_SHUTDOWN_STARTED, source='system')
+ await app_connections.close()
# Shutdown: clean up shared resources
from open_webui.utils.session_pool import close_session
diff --git a/backend/open_webui/models/apps.py b/backend/open_webui/models/apps.py
index 4329fe5cc9..8dd8581a62 100644
--- a/backend/open_webui/models/apps.py
+++ b/backend/open_webui/models/apps.py
@@ -159,6 +159,7 @@ class AppConnectionForm(StrictModel):
class AppResponse(BaseModel):
+ connection_status: Literal['connecting', 'connected', 'disconnected'] = 'disconnected'
id: str
slug: str
name: str
@@ -217,6 +218,7 @@ class AppsTable:
return None
app.is_active = False
app.api_key = None
+ app.data = {k: v for k, v in app.data.items() if k not in ('connection_token', 'installation_id', 'webui_url')}
app.updated_at = max(time.time_ns(), app.updated_at + 1)
await db.execute(
update(ChannelMember).where(ChannelMember.app_id == id).values(is_active=False, left_at=time.time_ns())
diff --git a/backend/open_webui/routers/apps.py b/backend/open_webui/routers/apps.py
index c735e36995..b82356a485 100644
--- a/backend/open_webui/routers/apps.py
+++ b/backend/open_webui/routers/apps.py
@@ -2,51 +2,37 @@
import time
import uuid
-from urllib.parse import urlsplit
-import jwt
-from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status
+from fastapi import APIRouter, Depends, Header, HTTPException, status
from open_webui.internal.db import get_async_session
from open_webui.models.access_grants import AccessGrant, AccessGrants
from open_webui.models.apps import (
App,
AppAccessResponse,
- AppChannelMessage,
AppConnectionForm,
AppForm,
AppInteractionForm,
AppManifestForm,
AppResponse,
Apps,
- AppTokenForm,
AppUpdateForm,
Manifest,
)
-from open_webui.models.channels import Channel, ChannelMember, Channels
from open_webui.models.config import Config
from open_webui.models.messages import Message
-from open_webui.models.users import Users
from open_webui.utils.apps import (
- check_app_channel_membership,
check_app_invocation_access,
- create_app_message,
decode_app_context_token,
generate_app_channel_reply,
get_app_by_id,
get_app_encryption_cipher,
- get_app_message_data,
get_app_response,
- get_verified_app,
has_app_access,
- require_scope,
- send_app_event,
send_app_request,
- update_app_message,
validate_app_url,
- verify_app_token,
)
from open_webui.utils.auth import get_admin_user, get_verified_user
-from sqlalchemy import and_, or_, select, update
+from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
@@ -63,8 +49,10 @@ async def get_apps(user=Depends(get_verified_user), db: AsyncSession = Depends(g
if not await Config.get('apps.enable'):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Apps are disabled')
apps = (await db.scalars(select(App).order_by(App.name))).all()
+ from open_webui.socket.apps import connections
+
return [
- get_app_response(app)
+ {**get_app_response(app), 'connection_status': await connections.status(app)}
for app in apps
if user.role == 'admin' or (app.is_active and await has_app_access(app, user, db=db))
]
@@ -91,9 +79,7 @@ async def preview_app_manifest(
if not api_key:
raise HTTPException(400, 'An app API key is required')
try:
- status_code, raw = await send_app_request(
- form_data.url, headers={'Authorization': f'Bearer {api_key}'}
- )
+ status_code, raw = await send_app_request(form_data.url, headers={'Authorization': f'Bearer {api_key}'})
if status_code in (401, 403):
raise HTTPException(400, 'App rejected the API key. Check the API key configured on the app server.')
if status_code != 200:
@@ -155,161 +141,6 @@ async def create_new_app(
return get_app_response(app)
-# Fixed runtime routes must precede /{app_id}.
-############################
-# GetCurrentApp
-############################
-
-
-@router.get('/runtime/me', response_model=AppResponse)
-async def get_current_app(app=Depends(get_verified_app)):
- return get_app_response(app)
-
-
-############################
-# GetAppChannels
-############################
-
-
-@router.get('/runtime/channels')
-async def get_app_channels(app=Depends(get_verified_app), db: AsyncSession = Depends(get_async_session)):
- if not await Config.get('channels.enable'):
- raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='Channels are disabled')
- await get_app_by_id(app.id, require_active=True, db=db)
- require_scope(app, 'channels:read')
- channels = (
- await db.scalars(
- select(Channel)
- .join(ChannelMember, Channel.id == ChannelMember.channel_id)
- .where(
- ChannelMember.app_id == app.id,
- ChannelMember.is_active.is_(True),
- Channel.deleted_at.is_(None),
- Channel.archived_at.is_(None),
- )
- )
- ).all()
- result = []
- for channel in channels:
- channel_data = await Channels.get_channel_by_id(channel.id, db=db)
- is_private = (
- channel_data.is_private is True
- if channel_data.type == 'group'
- else not any(grant.principal_id == '*' for grant in channel_data.access_grants)
- )
- result.append({'id': channel.id, 'name': channel.name, 'type': channel.type, 'is_private': is_private})
- return result
-
-
-############################
-# AuthorizeAppExecution
-############################
-
-
-@router.post('/runtime/authorize')
-async def authorize_app_execution(
- form_data: AppTokenForm, app=Depends(get_verified_app), db: AsyncSession = Depends(get_async_session)
-):
- value = decode_app_context_token(app, 'execution', form_data.token)
- user = await Users.get_user_by_id(value['actor_id'], db=db)
- if not user:
- raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='Actor is unavailable')
- await check_app_invocation_access(app, value['channel_id'], user, db=db)
- require_scope(app, value['scope'])
- return {
- 'authorized': True,
- 'operation_id': value['operation_id'],
- 'actor': {'id': user.id, 'is_admin': user.role == 'admin'},
- 'channel_id': value['channel_id'],
- 'type': value.get('type'),
- 'data': value.get('data'),
- }
-
-
-############################
-# VerifyAppEvent
-############################
-
-
-@router.post('/runtime/events/verify')
-async def verify_app_event(
- form_data: AppTokenForm, app=Depends(get_verified_app), db: AsyncSession = Depends(get_async_session)
-):
- value = decode_app_context_token(app, 'event', form_data.token)
- await check_app_channel_membership(app, value['channel']['id'], db=db)
- require_scope(app, 'channels:messages:read')
- return {k: v for k, v in value.items() if k not in ('purpose', 'exp', 'nonce')}
-
-
-############################
-# GetAppChannelMessages
-############################
-
-
-@router.get('/runtime/channels/{channel_id}/messages')
-async def get_app_channel_messages(
- channel_id: str,
- before: str | None = None,
- parent_id: str | None = None,
- limit: int = Query(default=50, ge=1, le=100),
- app=Depends(get_verified_app),
- db: AsyncSession = Depends(get_async_session),
-):
- await check_app_channel_membership(app, channel_id, db=db)
- require_scope(app, 'channels:messages:read')
- query = select(Message).where(Message.channel_id == channel_id, Message.parent_id == parent_id)
- if parent_id:
- parent = await db.get(Message, parent_id)
- if not parent or parent.channel_id != channel_id:
- raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Thread not found')
- if before:
- cursor = await db.get(Message, before)
- if not cursor or cursor.channel_id != channel_id or cursor.parent_id != parent_id:
- raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Invalid message cursor')
- query = query.where(
- or_(
- Message.created_at < cursor.created_at,
- and_(Message.created_at == cursor.created_at, Message.id < cursor.id),
- )
- )
- rows = (await db.scalars(query.order_by(Message.created_at.desc(), Message.id.desc()).limit(limit + 1))).all()
- return {
- 'messages': [get_app_message_data(row) for row in rows[:limit]],
- 'next_cursor': rows[limit - 1].id if len(rows) > limit else None,
- }
-
-
-############################
-# PostAppMessage
-############################
-
-
-@router.post('/runtime/channels/{channel_id}/messages', status_code=status.HTTP_201_CREATED)
-async def post_app_message(
- channel_id: str,
- form_data: AppChannelMessage,
- app=Depends(get_verified_app),
- db: AsyncSession = Depends(get_async_session),
-):
- return await create_app_message(db, app, channel_id, form_data)
-
-
-############################
-# UpdateAppMessageById
-############################
-
-
-@router.patch('/runtime/messages/{message_id}')
-async def update_app_message_by_id(
- message_id: str,
- form_data: AppChannelMessage,
- if_match: str | None = Header(default=None),
- app=Depends(get_verified_app),
- db: AsyncSession = Depends(get_async_session),
-):
- return await update_app_message(db, app, message_id, form_data, if_match)
-
-
############################
# GetAppDetailsById
############################
@@ -319,7 +150,10 @@ async def update_app_message_by_id(
async def get_app_details_by_id(
app_id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)
):
- result = get_app_response(await get_app_by_id(app_id, db=db))
+ from open_webui.socket.apps import connections
+
+ app = await get_app_by_id(app_id, db=db)
+ result = {**get_app_response(app), 'connection_status': await connections.status(app)}
result['access_grants'] = [
{'principal_type': g.principal_type, 'principal_id': g.principal_id, 'permission': g.permission}
for g in (
@@ -362,6 +196,9 @@ async def update_app_by_id(
'manifest': form_data.manifest.model_dump(),
'granted_scopes': form_data.granted_scopes,
}
+ if form_data.manifest.public_key != app.data['manifest']['public_key']:
+ data['data'].pop('connection_token', None)
+ data['data'].pop('installation_id', None)
data['is_active'] = False
elif form_data.granted_scopes is not None:
if not set(form_data.granted_scopes) <= set(app.data['manifest']['requested_scopes']):
@@ -380,90 +217,33 @@ async def update_app_by_id(
async def connect_app_by_id(
app_id: str,
form_data: AppConnectionForm,
- request: Request,
user=Depends(get_admin_user),
db: AsyncSession = Depends(get_async_session),
):
app = await get_app_by_id(app_id, db=db)
- webui_url = str(await Config.get('webui.url') or request.base_url).strip().rstrip('/')
- try:
- origin = urlsplit(webui_url)
- if (
- origin.scheme not in ('http', 'https')
- or not origin.hostname
- or origin.username is not None
- or origin.password is not None
- or origin.path
- or origin.query
- or origin.fragment
- or '\\' in webui_url
- or len(webui_url) > 2048
- or any(c.isspace() for c in webui_url)
- ):
- raise ValueError('Invalid WebUI origin')
- host = f'[{origin.hostname}]' if ':' in origin.hostname else origin.hostname.lower()
- port = origin.port or (443 if origin.scheme == 'https' else 80)
- suffix = f':{port}' if port != (443 if origin.scheme == 'https' else 80) else ''
- webui_url = f'{origin.scheme}://{host}{suffix}'
- except ValueError:
- raise HTTPException(
- status_code=status.HTTP_400_BAD_REQUEST,
- detail='Invalid WebUI URL. Set WebUI URL in Settings > Admin > General to an HTTP(S) origin '
- 'without a path, query, fragment or credentials.',
- ) from None
+ from open_webui.socket.apps import connections
+
if form_data.api_key is not None:
- app.api_key = get_app_encryption_cipher().encrypt(form_data.api_key.get_secret_value().encode()).decode()
+ cipher = get_app_encryption_cipher()
+ value = form_data.api_key.get_secret_value()
+ old = cipher.decrypt(app.api_key.encode()).decode() if app.api_key else None
+ if value != old:
+ app.data = {k: v for k, v in app.data.items() if k not in ('connection_token', 'installation_id')}
+ app.api_key = cipher.encrypt(value.encode()).decode()
if not app.api_key:
- raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='An app API key is required')
- app.data = {**app.data, 'webui_url': webui_url}
- app.is_active = False
- app.updated_at = max(time.time_ns(), app.updated_at + 1)
- checked_at, api_key = app.updated_at, app.api_key
+ raise HTTPException(400, 'An app API key is required')
app = await Apps.update_app_by_id(
- app_id, {'api_key': api_key, 'data': app.data, 'is_active': False, 'updated_at': checked_at}, db=db
+ app_id,
+ {
+ 'api_key': app.api_key,
+ 'data': app.data,
+ 'is_active': False,
+ 'updated_at': max(time.time_ns(), app.updated_at + 1),
+ },
+ db=db,
)
- await db.commit()
- challenge = str(uuid.uuid4())
- status_code, value = await send_app_event(
- app,
- {'id': str(uuid.uuid4()), 'type': 'connect', 'data': {'challenge': challenge}},
- timeout=20,
- )
- if status_code != 200:
- detail = {
- 401: 'App rejected the API key. Enter the API key configured on the app server.',
- 403: 'App refused the connection. Check the app server access rules and logs.',
- 400: 'App rejected the connection request. Check the WebUI origin and app server configuration.',
- 404: 'App event endpoint was not found. Check the URL in app settings.',
- 405: 'App endpoint does not accept POST requests. Use its event endpoint URL.',
- 503: 'App is busy or shutting down. Connect again when it is ready.',
- 504: 'App connection request timed out. Check the app server logs.',
- }.get(status_code, f'App returned HTTP {status_code}. Check the app server logs.')
- raise HTTPException(504 if status_code == 504 else 502, detail)
- if not isinstance(value, dict) or not isinstance(value.get('proof'), str):
- raise HTTPException(502, 'App did not return an identity proof. Check its event endpoint and SDK version.')
- try:
- claims = verify_app_token(app, value['proof'], 'webui-app-connect+jwt')
- if claims.get('challenge') != challenge:
- raise ValueError()
- except jwt.InvalidSignatureError:
- raise HTTPException(
- 502, 'App public key does not match the installed key. Review the app public key in settings.'
- ) from None
- except (jwt.InvalidTokenError, ValueError, TypeError, KeyError):
- raise HTTPException(
- 502, 'App identity proof is invalid or expired. Check the installed public key and server clocks.'
- ) from None
- result = await db.execute(
- update(App)
- .where(App.id == app_id, App.updated_at == checked_at, App.api_key == api_key)
- .values(is_active=True, updated_at=max(time.time_ns(), checked_at + 1))
- )
- if result.rowcount != 1:
- await db.rollback()
- raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail='App changed while connecting; connect again')
- await db.commit()
- return get_app_response(await get_app_by_id(app_id, db=db))
+ app = await connections.connect(app)
+ return {**get_app_response(app), 'connection_status': 'connected'}
############################
diff --git a/backend/open_webui/socket/apps.py b/backend/open_webui/socket/apps.py
new file mode 100644
index 0000000000..be3b308c12
--- /dev/null
+++ b/backend/open_webui/socket/apps.py
@@ -0,0 +1,446 @@
+"""Outbound app sockets. Each WebUI worker owns its own authenticated connections."""
+
+import asyncio
+import hashlib
+import json
+import logging
+import random
+import re
+import time
+import uuid
+from urllib.parse import urlsplit
+
+import aiohttp
+import jwt
+import socketio
+from fastapi import HTTPException
+from open_webui.env import REDIS_KEY_PREFIX
+from open_webui.internal.db import get_async_db
+from open_webui.models.apps import App, AppChannelMessage, Apps, AppTokenForm
+from open_webui.models.config import Config
+from open_webui.utils.apps import (
+ AppTCPConnector,
+ get_app_by_id,
+ get_app_encryption_cipher,
+ get_app_response,
+ validate_app_url,
+)
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+from socketio.packet import Packet
+from sqlalchemy import select, update
+
+log = logging.getLogger(__name__)
+NAMESPACE = '/apps'
+MAX_PACKET = 1048576
+
+
+class JSON:
+ @staticmethod
+ def dumps(value, **kwargs):
+ return json.dumps(value, **{**kwargs, 'ensure_ascii': False, 'separators': (',', ':')})
+
+ loads = staticmethod(json.loads)
+
+
+class JSONPacket(Packet):
+ def decode(self, encoded_packet):
+ if not isinstance(encoded_packet, str) or len(encoded_packet.encode()) > MAX_PACKET:
+ raise ValueError('Socket packet exceeds 1 MiB or is not JSON')
+ if encoded_packet[:1] in ('5', '6'):
+ raise ValueError('Binary app packets are not supported')
+ return super().decode(encoded_packet)
+
+
+class AppHistoryRequest(BaseModel):
+ model_config = ConfigDict(extra='forbid')
+ channel_id: str
+ parent_id: str | None = None
+ before: str | None = None
+ limit: int = Field(default=50, ge=1, le=100)
+
+
+def connection_config(app):
+ return (app.api_key, app.data.get('connection_token'), app.data['manifest']['public_key'], app.data['url'])
+
+
+class AppConnection:
+ def __init__(self, app):
+ self.app = app
+ self.session = None
+ self.sio = None
+ self.ready = False
+ self.verified = False
+ self.requests = set()
+ self.connected = asyncio.Event()
+
+ async def open(self, *, enroll=False):
+ allow_private = validate_app_url(self.app.data['url'])
+ trace = aiohttp.TraceConfig()
+
+ async def reject_redirect(*args):
+ raise ValueError('App endpoint must not redirect')
+
+ trace.on_request_redirect.append(reject_redirect)
+ self.session = aiohttp.ClientSession(
+ connector=AppTCPConnector(allow_private=allow_private),
+ trust_env=False,
+ trace_configs=[trace],
+ timeout=aiohttp.ClientTimeout(total=20),
+ )
+ self.sio = socketio.AsyncClient(
+ http_session=self.session,
+ reconnection=False,
+ json=JSON,
+ serializer=JSONPacket,
+ websocket_extra_options={'max_msg_size': MAX_PACKET},
+ )
+ self.sio.on('app:request', self.request, namespace=NAMESPACE)
+
+ async def disconnected(*args):
+ self.ready = False
+ self.connected.clear()
+
+ self.sio.on('disconnect', disconnected, namespace=NAMESPACE)
+ cipher = get_app_encryption_cipher()
+ key = cipher.decrypt(self.app.api_key.encode()).decode()
+ saved = self.app.data.get('connection_token')
+ if not saved and not enroll:
+ raise HTTPException(409, 'Reconnect this app to upgrade its transport')
+ token = cipher.decrypt(saved.encode()).decode() if saved else None
+ challenge = str(uuid.uuid4())
+ url = urlsplit(self.app.data['url'])
+ await self.sio.connect(
+ f'{url.scheme}://{url.netloc}',
+ socketio_path=url.path.rstrip('/') + '/socket.io',
+ headers={'Authorization': 'Bearer ' + key},
+ namespaces=[NAMESPACE],
+ transports=['websocket'],
+ wait_timeout=20,
+ )
+ result = await self.sio.call(
+ 'app:connect',
+ {
+ 'app_id': self.app.id,
+ 'challenge': challenge,
+ 'connection_token': token,
+ },
+ namespace=NAMESPACE,
+ timeout=10,
+ )
+ if not isinstance(result, dict) or result.get('status') != 200:
+ raise HTTPException(403, 'App rejected its connection credential. Review credentials before reconnecting.')
+ token, installation_id = self.verify_connection(result['data'], challenge, saved)
+ self.verified = True
+ if enroll:
+ async with get_async_db() as db:
+ values = {k: v for k, v in self.app.data.items() if k != 'webui_url'}
+ values.update(installation_id=installation_id, connection_token=cipher.encrypt(token.encode()).decode())
+ result = await db.execute(
+ update(App)
+ .where(
+ App.id == self.app.id,
+ App.updated_at == self.app.updated_at,
+ )
+ .values(data=values, is_active=False, updated_at=max(time.time_ns(), self.app.updated_at + 1))
+ )
+ if result.rowcount != 1:
+ raise HTTPException(409, 'App changed while connecting; connect again')
+ await db.commit()
+ self.app = await Apps.get_app_by_id(self.app.id, db=db)
+ result = await self.sio.call('app:ready', {}, namespace=NAMESPACE, timeout=10)
+ if result != {'status': 200, 'data': {'received': True}}:
+ raise HTTPException(502, 'App could not complete its bidirectional connection check')
+ if enroll:
+ async with get_async_db() as db:
+ result = await db.execute(
+ update(App)
+ .where(
+ App.id == self.app.id,
+ App.updated_at == self.app.updated_at,
+ )
+ .values(is_active=True, updated_at=max(time.time_ns(), self.app.updated_at + 1))
+ )
+ if result.rowcount != 1:
+ raise HTTPException(409, 'App changed while connecting; connect again')
+ await db.commit()
+ self.app = await Apps.get_app_by_id(self.app.id, db=db)
+ self.ready = True
+ self.connected.set()
+
+ def verify_connection(self, data, challenge, saved):
+ token = data['connection_token']
+ if not isinstance(token, str) or len(token) > 8192:
+ raise ValueError('Invalid app connection token')
+ proof = data['proof']
+ if jwt.get_unverified_header(proof).get('typ') != 'oix-connect+jwt':
+ raise ValueError('Invalid app identity proof')
+ claims = jwt.decode(
+ proof,
+ self.app.data['manifest']['public_key'],
+ algorithms=['EdDSA'],
+ audience='oix:connect',
+ options={'require': ['sub', 'app_id', 'manifest_id', 'iat', 'exp', 'challenge', 'token_hash']},
+ )
+ installation_id = str(uuid.UUID(data['installation_id']))
+ if (
+ claims['sub'] != installation_id
+ or claims['app_id'] != self.app.id
+ or claims['manifest_id'] != self.app.data['manifest']['id']
+ or claims['challenge'] != challenge
+ or claims['token_hash'] != hashlib.sha256(token.encode()).hexdigest()
+ or type(claims['iat']) is not int
+ or type(claims['exp']) is not int
+ or not 0 < claims['exp'] - claims['iat'] <= 60
+ ):
+ raise ValueError('Invalid app identity proof')
+ if saved and installation_id != self.app.data.get('installation_id'):
+ raise ValueError('App installation identity changed')
+ return token, installation_id
+
+ async def request(self, packet):
+ task = asyncio.current_task()
+ self.requests.add(task)
+ try:
+ if not isinstance(packet, dict) or len(JSON.dumps(packet).encode()) > MAX_PACKET - 128:
+ raise HTTPException(413, 'App request exceeds 1 MiB')
+ operation = packet.get('operation')
+ arguments = packet.get('arguments', {})
+ if not self.verified or (not self.ready and operation != 'ping'):
+ raise HTTPException(503 if self.verified else 403, 'App connection is not ready')
+ if operation == 'ping' and arguments == {}:
+ return {'status': 200, 'data': {'received': True}}
+ async with asyncio.timeout(10):
+ async with get_async_db() as db:
+ app = await get_app_by_id(self.app.id, db=db, require_active=True)
+ if connection_config(app) != connection_config(self.app):
+ raise HTTPException(403, 'App credentials changed')
+ value = await execute_request(app, operation, arguments, db)
+ result = {'status': 200, 'data': value}
+ if len(JSON.dumps(result).encode()) > MAX_PACKET - 128:
+ raise HTTPException(413, 'App response exceeds 1 MiB')
+ return result
+ except HTTPException as exc:
+ return {'status': exc.status_code, 'error': exc.detail}
+ except ValidationError:
+ return {'status': 400, 'error': 'Invalid app operation arguments'}
+ except TimeoutError:
+ return {'status': 504, 'error': 'App operation timed out; outcome may be unknown'}
+ except Exception:
+ log.exception('App operation failed for %s', self.app.id)
+ return {'status': 500, 'error': 'App operation failed'}
+ finally:
+ self.requests.discard(task)
+
+ async def close(self):
+ self.ready = False
+ self.connected.clear()
+ if self.sio:
+ await self.sio.disconnect()
+ if self.session:
+ await self.session.close()
+
+
+async def execute_request(app, operation, arguments, db):
+ from open_webui.utils import apps as runtime
+ from open_webui.utils.apps import create_app_message, update_app_message
+
+ if not isinstance(arguments, dict):
+ raise HTTPException(400, 'Invalid app operation arguments')
+ if operation in ('get_me', 'list_channels') and not arguments:
+ if operation == 'get_me':
+ return get_app_response(app)
+ return await runtime.get_app_channels(app=app, db=db)
+ if operation in ('authorize', 'verify_event'):
+ form = AppTokenForm.model_validate(arguments)
+ handler = runtime.authorize_app_execution if operation == 'authorize' else runtime.verify_app_event
+ return await handler(form_data=form, app=app, db=db)
+ if operation == 'list_messages':
+ form = AppHistoryRequest.model_validate(arguments)
+ return await runtime.get_app_channel_messages(**form.model_dump(), app=app, db=db)
+ if operation in ('create_message', 'update_message'):
+ arguments = dict(arguments)
+ identifier = arguments.pop('channel_id' if operation == 'create_message' else 'message_id', None)
+ if not isinstance(identifier, str) or not 1 <= len(identifier) <= 100:
+ raise HTTPException(400, 'Invalid resource ID')
+ etag = arguments.pop('etag', None) if operation == 'update_message' else None
+ form = AppChannelMessage.model_validate(arguments)
+ if operation == 'create_message':
+ return await create_app_message(db, app, identifier, form)
+ if not isinstance(etag, str) or not re.fullmatch(r'[0-9]+', etag):
+ raise HTTPException(400, 'If-Match is required')
+ return await update_app_message(db, app, identifier, form, f'"{etag}"')
+ raise HTTPException(400, 'Unknown app operation')
+
+
+class AppConnections:
+ def __init__(self):
+ self.peers = {}
+ self.tasks = {}
+ self.locks = {}
+ self.worker_id = str(uuid.uuid4())
+ self.redis = None
+ self.loop = None
+ self.closing = False
+
+ async def start(self, redis=None):
+ self.redis = redis
+ self.closing = False
+ self.loop = asyncio.create_task(self.reconcile())
+
+ async def maintain(self, app):
+ delay = 1
+ while not self.closing:
+ peer = AppConnection(app)
+ self.peers[app.id] = peer
+ try:
+ async with asyncio.timeout(20):
+ await peer.open()
+ delay = 1
+ await peer.sio.wait()
+ except asyncio.CancelledError:
+ raise
+ except Exception as exc:
+ log.warning('App %s socket unavailable: %s', app.id, type(exc).__name__)
+ finally:
+ await peer.close()
+ await asyncio.sleep(random.uniform(delay / 2, delay))
+ delay = min(delay * 2, 30)
+
+ async def reconcile(self):
+ while not self.closing:
+ try:
+ async with get_async_db() as db:
+ apps = (
+ list((await db.scalars(select(App).where(App.is_active.is_(True)))).all())
+ if await Config.get('apps.enable')
+ else []
+ )
+ snapshots = [await Apps.get_app_by_id(app.id, db=db) for app in apps]
+ active = {app.id: app for app in snapshots if app.data.get('connection_token')}
+ for app_id in set(self.tasks) | set(active):
+ await self.refresh(app_id)
+ if self.redis:
+ for app_id, peer in list(self.peers.items()):
+ key = f'{REDIS_KEY_PREFIX}:apps:connections:{app_id}'
+ if peer.ready:
+ await self.redis.zadd(key, {self.worker_id: time.time() + 15})
+ await self.redis.expire(key, 30)
+ else:
+ await self.redis.zrem(key, self.worker_id)
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ log.warning('App connection reconciliation failed')
+ await asyncio.sleep(5)
+
+ async def refresh(self, app_id):
+ async with self.locks.setdefault(app_id, asyncio.Lock()):
+ # Refresh after taking the lock: an admin may have just connected this app.
+ app = await Apps.get_app_by_id(app_id)
+ enabled = await Config.get('apps.enable')
+ peer = self.peers.get(app_id)
+ if not enabled or not app or not app.is_active or not app.data.get('connection_token'):
+ await self.remove(app_id)
+ else:
+ if peer and connection_config(app) != connection_config(peer.app):
+ await self.remove(app_id)
+ if app_id not in self.tasks:
+ self.tasks[app_id] = asyncio.create_task(self.maintain(app))
+
+ async def remove(self, app_id):
+ task = self.tasks.pop(app_id, None)
+ if task:
+ task.cancel()
+ await asyncio.gather(task, return_exceptions=True)
+ peer = self.peers.pop(app_id, None)
+ if peer:
+ await peer.close()
+ if self.redis:
+ try:
+ await self.redis.zrem(f'{REDIS_KEY_PREFIX}:apps:connections:{app_id}', self.worker_id)
+ except Exception:
+ pass # Liveness expires after a crashed worker or Redis outage.
+
+ async def connect(self, app):
+ async with self.locks.setdefault(app.id, asyncio.Lock()):
+ await self.remove(app.id)
+ peer = AppConnection(app)
+ try:
+ async with asyncio.timeout(20):
+ await peer.open(enroll=True)
+ except HTTPException:
+ await peer.close()
+ raise
+ except Exception as exc:
+ await peer.close()
+ log.warning('App %s connection failed: %s', app.id, type(exc).__name__)
+ raise HTTPException(
+ 502, 'App connection failed. Check the API key, pinned public key and Socket.IO endpoint.'
+ ) from None
+ self.peers[app.id] = peer
+
+ async def resume():
+ try:
+ await peer.sio.wait()
+ finally:
+ await peer.close()
+ await self.maintain(peer.app)
+
+ self.tasks[app.id] = asyncio.create_task(resume())
+ return peer.app
+
+ async def send(self, app, payload, timeout):
+ if self.closing:
+ raise HTTPException(503, 'App connections are shutting down')
+ async with self.locks.setdefault(app.id, asyncio.Lock()):
+ peer = self.peers.get(app.id)
+ if peer is None or connection_config(peer.app) != connection_config(app):
+ await self.remove(app.id)
+ if not app.data.get('connection_token'):
+ raise HTTPException(409, 'Reconnect this app to upgrade its transport')
+ self.tasks[app.id] = asyncio.create_task(self.maintain(app))
+ await asyncio.sleep(0)
+ peer = self.peers[app.id]
+ if not peer.ready:
+ raise HTTPException(503, 'App is disconnected; event was not sent')
+ try:
+ result = await peer.sio.call('app:event', payload, namespace=NAMESPACE, timeout=timeout)
+ except Exception:
+ raise HTTPException(504, 'App connection lost or timed out; execution outcome may be unknown') from None
+ if not isinstance(result, dict) or len(JSON.dumps(result).encode()) > 262144 + 128:
+ raise HTTPException(502, 'App returned an invalid or oversized response')
+ status = result.get('status')
+ if type(status) is not int:
+ raise HTTPException(502, 'App returned an invalid status')
+ return status, result.get('data')
+
+ async def status(self, app):
+ if not app.is_active:
+ return 'disconnected'
+ if self.redis:
+ try:
+ if await self.redis.zcount(f'{REDIS_KEY_PREFIX}:apps:connections:{app.id}', time.time(), '+inf'):
+ return 'connected'
+ except Exception:
+ return 'disconnected'
+ peer = self.peers.get(app.id)
+ return 'connected' if peer and peer.ready else 'connecting' if peer else 'disconnected'
+
+ async def close(self):
+ self.closing = True
+ if self.loop:
+ self.loop.cancel()
+ await asyncio.gather(self.loop, return_exceptions=True)
+ for peer in self.peers.values():
+ peer.ready = False
+ pending = set().union(*(peer.requests for peer in self.peers.values())) if self.peers else set()
+ if pending:
+ _, unfinished = await asyncio.wait(pending, timeout=30)
+ for task in unfinished:
+ task.cancel()
+ await asyncio.gather(*unfinished, return_exceptions=True)
+ for app_id in list(self.peers):
+ await self.remove(app_id)
+
+
+connections = AppConnections()
diff --git a/backend/open_webui/utils/apps.py b/backend/open_webui/utils/apps.py
index 4d4cad3cda..09ca6c5eb5 100644
--- a/backend/open_webui/utils/apps.py
+++ b/backend/open_webui/utils/apps.py
@@ -14,20 +14,20 @@ from typing import Any
from urllib.parse import urlsplit
import aiohttp
-import jwt
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
-from fastapi import Depends, Header, HTTPException, status
+from fastapi import HTTPException, status
from open_webui.env import WEBUI_SECRET_KEY
-from open_webui.internal.db import get_async_db, get_async_session
+from open_webui.internal.db import get_async_db
from open_webui.models.access_grants import AccessGrants
-from open_webui.models.apps import AppChannelMessage, AppChannelReply, AppModel, Apps
-from open_webui.models.channels import ChannelModel, Channels
+from open_webui.models.apps import AppChannelMessage, AppChannelReply, AppModel, Apps, AppTokenForm
+from open_webui.models.channels import Channel, ChannelMember, ChannelModel, Channels
from open_webui.models.config import Config
-from open_webui.models.messages import MessageForm, Messages
+from open_webui.models.messages import Message, MessageForm, Messages
from open_webui.models.users import UserModel, Users
from open_webui.tasks import create_task
+from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
log = logging.getLogger(__name__)
@@ -50,6 +50,8 @@ def get_app_url_origin(url: str) -> str:
def validate_app_url(url: str) -> bool:
"""Validate the endpoint and return whether its private origin is explicitly allowed."""
origin = get_app_url_origin(url)
+ if urlsplit(url).query:
+ raise ValueError('App URL must not contain a query')
private_origins = {
get_app_url_origin(x.strip()) for x in os.getenv('APP_ALLOWED_PRIVATE_ORIGINS', '').split(',') if x.strip()
}
@@ -133,8 +135,13 @@ async def send_app_request(
# Only local, fixed diagnostics may be shown; never echo arbitrary app responses.
details = {
'Invalid service URL': 'Invalid app URL. Use an HTTP(S) URL without credentials or a fragment.',
- 'App endpoints require HTTPS': 'App endpoints require HTTPS. For local HTTP, configure APP_ALLOWED_PRIVATE_ORIGINS.',
- 'App endpoint resolves to a prohibited address': 'App resolves to a private or reserved address. For local development, configure APP_ALLOWED_PRIVATE_ORIGINS.',
+ 'App endpoints require HTTPS': (
+ 'App endpoints require HTTPS. For local HTTP, configure APP_ALLOWED_PRIVATE_ORIGINS.'
+ ),
+ 'App endpoint resolves to a prohibited address': (
+ 'App resolves to a private or reserved address. '
+ 'For local development, configure APP_ALLOWED_PRIVATE_ORIGINS.'
+ ),
'App response is too large': 'App response exceeds the allowed size.',
}
raise HTTPException(502, details.get(str(exc), 'App returned an invalid response. Expected JSON.')) from None
@@ -203,39 +210,14 @@ def decode_app_context_token(app: AppModel, purpose: str, token: str) -> dict:
raise HTTPException(403, 'Invalid or expired app context') from None
-def verify_app_token(app: AppModel, token: str, token_type: str = 'webui-app+jwt') -> dict:
- header = jwt.get_unverified_header(token)
- if header.get('typ') != token_type or any(k in header for k in ('jku', 'jwk', 'x5u', 'crit')):
- raise ValueError('Invalid app token type or key source')
- claims = jwt.decode(
- token,
- app.data['manifest']['public_key'],
- algorithms=['EdDSA'],
- audience=app.data['webui_url'] + '/api/v1/apps/runtime',
- options={'require': ['sub', 'aud', 'iat', 'exp'], 'strict_aud': True},
- )
- if (
- claims['sub'] != app.id
- or type(claims['iat']) is not int
- or type(claims['exp']) is not int
- or not 0 < claims['exp'] - claims['iat'] <= 60
- ):
- raise ValueError('Invalid app identity or lifetime')
- return claims
-
-
async def send_app_event(app: AppModel, event_data: dict, *, timeout: float = 3) -> tuple[int, Any]:
- body = _serialize_json({**event_data, 'app_id': app.id, 'webui_url': app.data['webui_url']})
- if len(body) > 1048576:
+ from open_webui.socket.apps import connections
+
+ payload = {**event_data, 'app_id': app.id, 'installation_id': app.data.get('installation_id')}
+ if len(_serialize_json(payload)) > 1048576 - 128:
log.warning('App %s event exceeds the 1 MiB delivery limit', app.id)
raise ValueError('App event exceeds 1 MiB')
- api_key = get_app_encryption_cipher().decrypt(app.api_key.encode()).decode()
- return await send_app_request(
- app.data['url'],
- body=body,
- timeout=timeout,
- headers={'Content-Type': 'application/json', 'Authorization': f'Bearer {api_key}'},
- )
+ return await connections.send(app, payload, timeout)
def get_app_response(app: AppModel) -> dict:
@@ -245,7 +227,7 @@ def get_app_response(app: AppModel) -> dict:
'name': app.name,
'is_active': app.is_active,
'configured': bool(app.api_key),
- 'data': app.data,
+ 'data': {k: v for k, v in app.data.items() if k not in ('connection_token', 'webui_url')},
'created_at': app.created_at,
'updated_at': app.updated_at,
}
@@ -262,28 +244,6 @@ async def get_app_by_id(app_id: str, db: AsyncSession, require_active: bool = Fa
return app
-async def get_verified_app(
- authorization: str = Header(default=''), db: AsyncSession = Depends(get_async_session)
-) -> AppModel:
- if not await Config.get('apps.enable'):
- raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Apps are disabled')
- try:
- if not authorization.startswith('Bearer '):
- raise ValueError()
- token = authorization.removeprefix('Bearer ')
- if len(token) > 4096:
- raise ValueError()
- claims = jwt.decode(token, options={'verify_signature': False})
- app_id = claims.get('sub')
- if not isinstance(app_id, str) or len(app_id) > 100:
- raise ValueError()
- app = await get_app_by_id(app_id, db=db, require_active=True)
- verify_app_token(app, token)
- return app
- except (ValueError, KeyError, jwt.PyJWTError, HTTPException):
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Invalid app credential') from None
-
-
async def has_app_access(app: AppModel, user: UserModel, db: AsyncSession) -> bool:
return user.role == 'admin' or await AccessGrants.has_access(user.id, 'app', app.id, 'read', db=db)
@@ -370,7 +330,11 @@ def get_app_message_data(message):
async def publish_app_message_event(message, kind):
# Model output is streamed into this row; only publish its completed update.
- if kind == 'channel.message.updated' and (message.meta or {}).get('model_id') and not (message.meta or {}).get('done'):
+ if (
+ kind == 'channel.message.updated'
+ and (message.meta or {}).get('model_id')
+ and not (message.meta or {}).get('done')
+ ):
return
value = deepcopy(get_app_message_data(message))
if kind == 'channel.message.deleted':
@@ -391,7 +355,7 @@ async def send_app_message_event(message, kind, created_at):
app = await Apps.get_app_by_id(app_id, db=db)
if not app or 'channels:messages:read' not in app.data.get('granted_scopes', []):
continue
- if not app.data.get('webui_url'):
+ if not app.data.get('installation_id'):
continue
try:
await check_app_channel_membership(app, message['channel_id'], db=db)
@@ -402,7 +366,7 @@ async def send_app_message_event(message, kind, created_at):
'type': kind,
'message': message,
'app_id': app.id,
- 'webui_url': app.data['webui_url'],
+ 'installation_id': app.data['installation_id'],
'channel': {'id': message['channel_id']},
}
envelope['event_token'] = create_app_context_token(app, 'event', envelope, expires_in=300)
@@ -565,3 +529,88 @@ async def generate_app_channel_reply(
message = AppChannelMessage(**{k: getattr(reply, k) for k in AppChannelMessage.model_fields})
value['message'] = await update_app_message(db, app, source['id'], message, f'"{source["updated_at"]}"')
return value
+
+
+async def get_app_channels(app: AppModel, db: AsyncSession):
+ if not await Config.get('channels.enable'):
+ raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='Channels are disabled')
+ await get_app_by_id(app.id, require_active=True, db=db)
+ require_scope(app, 'channels:read')
+ channels = (
+ await db.scalars(
+ select(Channel)
+ .join(ChannelMember, Channel.id == ChannelMember.channel_id)
+ .where(
+ ChannelMember.app_id == app.id,
+ ChannelMember.is_active.is_(True),
+ Channel.deleted_at.is_(None),
+ Channel.archived_at.is_(None),
+ )
+ )
+ ).all()
+ result = []
+ for channel in channels:
+ channel_data = await Channels.get_channel_by_id(channel.id, db=db)
+ is_private = (
+ channel_data.is_private is True
+ if channel_data.type == 'group'
+ else not any(grant.principal_id == '*' for grant in channel_data.access_grants)
+ )
+ result.append({'id': channel.id, 'name': channel.name, 'type': channel.type, 'is_private': is_private})
+ return result
+
+
+async def authorize_app_execution(form_data: AppTokenForm, app: AppModel, db: AsyncSession):
+ value = decode_app_context_token(app, 'execution', form_data.token)
+ user = await Users.get_user_by_id(value['actor_id'], db=db)
+ if not user:
+ raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='Actor is unavailable')
+ await check_app_invocation_access(app, value['channel_id'], user, db=db)
+ require_scope(app, value['scope'])
+ return {
+ 'authorized': True,
+ 'operation_id': value['operation_id'],
+ 'actor': {'id': user.id, 'is_admin': user.role == 'admin'},
+ 'channel_id': value['channel_id'],
+ 'type': value.get('type'),
+ 'data': value.get('data'),
+ }
+
+
+async def verify_app_event(form_data: AppTokenForm, app: AppModel, db: AsyncSession):
+ value = decode_app_context_token(app, 'event', form_data.token)
+ await check_app_channel_membership(app, value['channel']['id'], db=db)
+ require_scope(app, 'channels:messages:read')
+ return {k: v for k, v in value.items() if k not in ('purpose', 'exp', 'nonce')}
+
+
+async def get_app_channel_messages(
+ app: AppModel,
+ db: AsyncSession,
+ channel_id: str,
+ before: str | None = None,
+ parent_id: str | None = None,
+ limit: int = 50,
+):
+ await check_app_channel_membership(app, channel_id, db=db)
+ require_scope(app, 'channels:messages:read')
+ query = select(Message).where(Message.channel_id == channel_id, Message.parent_id == parent_id)
+ if parent_id:
+ parent = await db.get(Message, parent_id)
+ if not parent or parent.channel_id != channel_id:
+ raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Thread not found')
+ if before:
+ cursor = await db.get(Message, before)
+ if not cursor or cursor.channel_id != channel_id or cursor.parent_id != parent_id:
+ raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Invalid message cursor')
+ query = query.where(
+ or_(
+ Message.created_at < cursor.created_at,
+ and_(Message.created_at == cursor.created_at, Message.id < cursor.id),
+ )
+ )
+ rows = (await db.scalars(query.order_by(Message.created_at.desc(), Message.id.desc()).limit(limit + 1))).all()
+ return {
+ 'messages': [get_app_message_data(row) for row in rows[:limit]],
+ 'next_cursor': rows[limit - 1].id if len(rows) > limit else None,
+ }
diff --git a/src/lib/apis/apps/index.ts b/src/lib/apis/apps/index.ts
index e86e9724a9..7fce6faf2a 100644
--- a/src/lib/apis/apps/index.ts
+++ b/src/lib/apis/apps/index.ts
@@ -1,6 +1,7 @@
import { WEBUI_API_BASE_URL } from '$lib/constants';
export type App = {
+ connection_status: 'connecting' | 'connected' | 'disconnected';
id: string;
slug: string;
name: string;
diff --git a/src/lib/components/admin/Settings/Apps.svelte b/src/lib/components/admin/Settings/Apps.svelte
index bf51ae47bd..92832695ed 100644
--- a/src/lib/components/admin/Settings/Apps.svelte
+++ b/src/lib/components/admin/Settings/Apps.svelte
@@ -16,12 +16,14 @@
let selected: App | null = null;
let show = false;
let busy = '';
- async function load() {
+ async function load(silent = false) {
try {
apps = await getApps(localStorage.token);
} catch (error) {
- toast.error(String(error));
- apps = [];
+ if (!silent) {
+ toast.error(String(error));
+ apps = [];
+ }
}
}
function configure(app: App | null) {
@@ -43,7 +45,11 @@
busy = '';
}
}
- onMount(load);
+ onMount(() => {
+ load();
+ const interval = setInterval(() => load(true), 5000);
+ return () => clearInterval(interval);
+ });
@@ -86,6 +92,15 @@
{app.slug}
+ {#if app.is_active}
+ {app.connection_status === 'connected'
+ ? $i18n.t('Connected')
+ : app.connection_status === 'connecting'
+ ? $i18n.t('Connecting')
+ : $i18n.t('Disconnected')}
+ {/if}