mirror of
https://github.com/open-webui/open-webui.git
synced 2026-10-14 04:08:46 +00:00
refac
This commit is contained in:
Timothy Jaeryang Baek
parent
53a8f131f3
commit
1786d6cdb0
8 changed files with 619 additions and 321 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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'}
|
||||
|
||||
|
||||
############################
|
||||
|
|
|
|||
446
backend/open_webui/socket/apps.py
Normal file
446
backend/open_webui/socket/apps.py
Normal file
|
|
@ -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()
|
||||
|
|
@ -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,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
</script>
|
||||
|
||||
<AppModal bind:show app={selected} onChange={load} />
|
||||
|
|
@ -86,6 +92,15 @@
|
|||
<span class="text-gray-400 dark:text-gray-600">{app.slug}</span></span
|
||||
>
|
||||
</div>
|
||||
{#if app.is_active}
|
||||
<span class="text-gray-400 dark:text-gray-500"
|
||||
>{app.connection_status === 'connected'
|
||||
? $i18n.t('Connected')
|
||||
: app.connection_status === 'connecting'
|
||||
? $i18n.t('Connecting')
|
||||
: $i18n.t('Disconnected')}</span
|
||||
>
|
||||
{/if}
|
||||
<div class="flex shrink-0 items-center gap-1">
|
||||
<Tooltip content={$i18n.t('Configure')}
|
||||
><button
|
||||
|
|
|
|||
|
|
@ -2138,7 +2138,8 @@
|
|||
text,
|
||||
history?.currentId
|
||||
? createMessagesList(history, history.currentId)
|
||||
: null
|
||||
: null,
|
||||
'general'
|
||||
).catch((error) => {
|
||||
console.log(error);
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue