diff --git a/.env.example b/.env.example index 39cec7649d..801c63a22a 100644 --- a/.env.example +++ b/.env.example @@ -19,6 +19,9 @@ OPENAI_API_KEY='' # CORS_ALLOW_ORIGIN='http://localhost:5173;http://localhost:8080' CORS_ALLOW_ORIGIN='*' +# Renewable sessions expire after this period without a refresh. +# SESSION_EXPIRES_IN='90d' + # MFA defaults; saved settings take precedence once persistent config is seeded. ENABLE_MFA=false MFA_ALLOW_OAUTH_BYPASS=false diff --git a/.github/workflows/actions.lock b/.github/workflows/actions.lock new file mode 100644 index 0000000000..d320c1fcfd --- /dev/null +++ b/.github/workflows/actions.lock @@ -0,0 +1,112 @@ +# This file is machine-generated by `gh actions-lock`. +# Do not edit by hand; run `gh actions-lock` to update. +# Docs: https://gh.io/actions-lockfile +version: 'v0.0.3' +workflows: + '.github/workflows/backend.yaml': + - 'actions/checkout@v5.1.0' + - 'actions/setup-python@v6.3.0' + '.github/workflows/docker.yaml': + - 'actions/checkout@v5.1.0' + - 'actions/create-github-app-token@v2.2.2' + - 'actions/download-artifact@v5.0.0' + - 'actions/github-script@v8.0.0' + - 'actions/upload-artifact@v4.6.2' + - 'docker/build-push-action@v5.4.0' + - 'docker/login-action@v3.7.0' + - 'docker/metadata-action@v5.10.0' + - 'docker/setup-buildx-action@v3.12.0' + - 'docker/setup-qemu-action@v3.7.0' + '.github/workflows/frontend.yaml': + - 'actions/checkout@v5.1.0' + - 'actions/setup-node@v5.0.0' + '.github/workflows/issue-label.yaml': + - 'actions/github-script@v7.1.0' + '.github/workflows/regression.yaml': [] + '.github/workflows/release-pypi.yml': + - 'actions/checkout@v5.1.0' + - 'actions/setup-node@v5.0.0' + - 'actions/setup-python@v6.3.0' + - 'pypa/gh-action-pypi-publish@release/v1' + '.github/workflows/release.yml': + - 'actions/checkout@v5.1.0' + - 'actions/github-script@v8.0.0' + - 'actions/upload-artifact@v4.6.2' +dependencies: + 'actions/checkout@v5.1.0': + ref: 'v5.1.0' + commit: 'sha1-fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09' + owner_id: 44036562 + repo_id: 197814629 + 'actions/create-github-app-token@v2.2.2': + ref: 'v2.2.2' + commit: 'sha1-fee1f7d63c2ff003460e3d139729b119787bc349' + owner_id: 44036562 + repo_id: 642580244 + 'actions/download-artifact@v5.0.0': + ref: 'v5.0.0' + commit: 'sha1-634f93cb2916e3fdff6788551b99b062d0335ce0' + owner_id: 44036562 + repo_id: 192626254 + 'actions/github-script@v7.1.0': + ref: 'v7.1.0' + commit: 'sha1-f28e40c7f34bde8b3046d885e986cb6290c5673b' + owner_id: 44036562 + repo_id: 205262760 + 'actions/github-script@v8.0.0': + ref: 'v8.0.0' + commit: 'sha1-ed597411d8f924073f98dfc5c65a23a2325f34cd' + owner_id: 44036562 + repo_id: 205262760 + 'actions/setup-node@v5.0.0': + ref: 'v5.0.0' + commit: 'sha1-a0853c24544627f65ddf259abe73b1d18a591444' + owner_id: 44036562 + repo_id: 189476904 + 'actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405': + ref: 'v6.2.0' + commit: 'sha1-a309ff8b426b58ec0e2a45f0f869d46889d02405' + owner_id: 44036562 + repo_id: 192625525 + 'actions/setup-python@v6.3.0': + ref: 'v6.3.0' + commit: 'sha1-ece7cb06caefa5fff74198d8649806c4678c61a1' + owner_id: 44036562 + repo_id: 192625525 + 'actions/upload-artifact@v4.6.2': + ref: 'v4.6.2' + commit: 'sha1-ea165f8d65b6e75b540449e92b4886f43607fa02' + owner_id: 44036562 + repo_id: 192625955 + 'docker/build-push-action@v5.4.0': + ref: 'v5.4.0' + commit: 'sha1-ca052bb54ab0790a636c9b5f226502c73d547a25' + owner_id: 5429470 + repo_id: 241092383 + 'docker/login-action@v3.7.0': + ref: 'v3.7.0' + commit: 'sha1-c94ce9fb468520275223c153574b00df6fe4bcc9' + owner_id: 5429470 + repo_id: 287743349 + 'docker/metadata-action@v5.10.0': + ref: 'v5.10.0' + commit: 'sha1-c299e40c65443455700f0fdfc63efafe5b349051' + owner_id: 5429470 + repo_id: 306769011 + 'docker/setup-buildx-action@v3.12.0': + ref: 'v3.12.0' + commit: 'sha1-8d2750c68a42422c14e847fe6c8ac0403b4cbd6f' + owner_id: 5429470 + repo_id: 288485773 + 'docker/setup-qemu-action@v3.7.0': + ref: 'v3.7.0' + commit: 'sha1-c7c53464625b32c7a7e944ae62b3e17d2b600130' + owner_id: 5429470 + repo_id: 288487894 + 'pypa/gh-action-pypi-publish@release/v1': + ref: 'release/v1' + commit: 'sha1-7245906fb61fa33a13554f9d5fb856fe61758086' + owner_id: 647025 + repo_id: 178055147 + uses: + - 'actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405' diff --git a/.github/workflows/backend.yaml b/.github/workflows/backend.yaml index 56dcf5a396..ea2966f96f 100644 --- a/.github/workflows/backend.yaml +++ b/.github/workflows/backend.yaml @@ -27,9 +27,9 @@ jobs: matrix: python-version: ['3.11', '3.12'] steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v5.1.0 - - uses: actions/setup-python@v6 + - uses: actions/setup-python@v6.3.0 with: python-version: ${{ matrix.python-version }} diff --git a/.github/workflows/docker.yaml b/.github/workflows/docker.yaml index 1802e8f336..40e9761972 100644 --- a/.github/workflows/docker.yaml +++ b/.github/workflows/docker.yaml @@ -67,13 +67,13 @@ jobs: run: rm -rf /opt/hostedtoolcache - name: Checkout repository - uses: actions/checkout@v5 + uses: actions/checkout@v5.1.0 - name: Set up QEMU - uses: docker/setup-qemu-action@v3 + uses: docker/setup-qemu-action@v3.7.0 - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 + uses: docker/setup-buildx-action@v3.12.0 - name: Prepare CI Dockerfile run: | @@ -87,7 +87,7 @@ jobs: ' Dockerfile > "${RUNNER_TEMP}/Dockerfile" - name: Log in to the Container registry - uses: docker/login-action@v3 + uses: docker/login-action@v3.7.0 with: registry: ${{ env.REGISTRY }} username: ${{ github.actor }} @@ -95,7 +95,7 @@ jobs: - name: Extract metadata for Docker images id: meta - uses: docker/metadata-action@v5 + uses: docker/metadata-action@v5.10.0 with: images: ${{ env.FULL_IMAGE_NAME }} tags: | @@ -111,7 +111,7 @@ jobs: - name: Extract metadata for Docker cache id: cache-meta - uses: docker/metadata-action@v5 + uses: docker/metadata-action@v5.10.0 with: images: ${{ env.FULL_IMAGE_NAME }} tags: | @@ -122,7 +122,7 @@ jobs: latest=false - name: Build Docker image - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v5.4.0 id: build with: context: . @@ -146,7 +146,7 @@ jobs: touch "/tmp/digests/${digest#sha256:}" - name: Upload digest - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v4.6.2 with: name: digests-${{ matrix.variant.name }}-${{ env.PLATFORM_PAIR }} path: /tmp/digests/* @@ -183,7 +183,7 @@ jobs: - name: Download digests id: download - uses: actions/download-artifact@v5 + uses: actions/download-artifact@v5.0.0 with: pattern: digests-${{ matrix.variant.name }}-* path: /tmp/digests @@ -204,11 +204,11 @@ jobs: - name: Set up Docker Buildx if: steps.check.outputs.skip != 'true' - uses: docker/setup-buildx-action@v3 + uses: docker/setup-buildx-action@v3.12.0 - name: Log in to the Container registry if: steps.check.outputs.skip != 'true' - uses: docker/login-action@v3 + uses: docker/login-action@v3.7.0 with: registry: ${{ env.REGISTRY }} username: ${{ github.actor }} @@ -217,7 +217,7 @@ jobs: - name: Extract metadata for Docker images if: steps.check.outputs.skip != 'true' id: meta - uses: docker/metadata-action@v5 + uses: docker/metadata-action@v5.10.0 with: images: ${{ env.FULL_IMAGE_NAME }} tags: | @@ -251,7 +251,7 @@ jobs: steps: - name: Create Helm charts app token id: helm-app-token - uses: actions/create-github-app-token@v2 + uses: actions/create-github-app-token@v2.2.2 with: app-id: ${{ secrets.HELM_CHARTS_APP_ID }} private-key: ${{ secrets.HELM_CHARTS_APP_PRIVATE_KEY }} @@ -259,7 +259,7 @@ jobs: repositories: helm-charts - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 + uses: docker/setup-buildx-action@v3.12.0 - name: Verify published Open WebUI image id: image @@ -279,7 +279,7 @@ jobs: echo "tag=${image_tag}" >> "${GITHUB_OUTPUT}" - name: Dispatch Helm chart automation - uses: actions/github-script@v8 + uses: actions/github-script@v8.0.0 with: github-token: ${{ steps.helm-app-token.outputs.token }} script: | @@ -334,17 +334,17 @@ jobs: echo "FULL_IMAGE_NAME=${REGISTRY}/${GITHUB_REPOSITORY,,}" >> ${GITHUB_ENV} - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 + uses: docker/setup-buildx-action@v3.12.0 - name: Log in to the Container registry - uses: docker/login-action@v3 + uses: docker/login-action@v3.7.0 with: registry: ${{ env.REGISTRY }} username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - name: Log in to Docker Hub - uses: docker/login-action@v3 + uses: docker/login-action@v3.7.0 with: username: ${{ secrets.DOCKERHUB_USERNAME }} password: ${{ secrets.DOCKERHUB_TOKEN }} diff --git a/.github/workflows/frontend.yaml b/.github/workflows/frontend.yaml index a977c5bf4e..fd37691174 100644 --- a/.github/workflows/frontend.yaml +++ b/.github/workflows/frontend.yaml @@ -23,9 +23,9 @@ jobs: runs-on: ubuntu-latest timeout-minutes: 15 steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v5.1.0 - - uses: actions/setup-node@v5 + - uses: actions/setup-node@v5.0.0 with: node-version: '22' @@ -52,9 +52,9 @@ jobs: runs-on: ubuntu-latest timeout-minutes: 10 steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v5.1.0 - - uses: actions/setup-node@v5 + - uses: actions/setup-node@v5.0.0 with: node-version: '22' diff --git a/.github/workflows/issue-label.yaml b/.github/workflows/issue-label.yaml index 1a138e2f11..cb9372fdc4 100644 --- a/.github/workflows/issue-label.yaml +++ b/.github/workflows/issue-label.yaml @@ -12,7 +12,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Add "bug" label to unlabeled bug reports - uses: actions/github-script@v7 + uses: actions/github-script@v7.1.0 with: script: | const issue = context.payload.issue; @@ -72,7 +72,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Add "enhancement" label to unlabeled feature requests - uses: actions/github-script@v7 + uses: actions/github-script@v7.1.0 with: script: | const issue = context.payload.issue; diff --git a/.github/workflows/release-pypi.yml b/.github/workflows/release-pypi.yml index f5b353dbcc..a9194d2301 100644 --- a/.github/workflows/release-pypi.yml +++ b/.github/workflows/release-pypi.yml @@ -16,15 +16,15 @@ jobs: id-token: write steps: - name: Checkout repository - uses: actions/checkout@v5 + uses: actions/checkout@v5.1.0 with: fetch-depth: 0 - name: Install Git run: sudo apt-get update && sudo apt-get install -y git - - uses: actions/setup-node@v5 + - uses: actions/setup-node@v5.0.0 with: node-version: 22 - - uses: actions/setup-python@v6 + - uses: actions/setup-python@v6.3.0 with: python-version: 3.12 - name: Build diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index bcc6b6dcff..7d6b7fae2f 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -21,7 +21,7 @@ jobs: contents: write actions: write steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v5.1.0 - name: Abort if package.json unchanged run: | @@ -65,7 +65,7 @@ jobs: fi - name: Archive source - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v4.6.2 with: name: release-archive path: | @@ -75,7 +75,7 @@ jobs: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - name: Trigger Docker build - uses: actions/github-script@v8 + uses: actions/github-script@v8.0.0 with: script: | github.rest.actions.createWorkflowDispatch({ diff --git a/README.md b/README.md index 1f2735fd8c..82d20f3a2b 100644 --- a/README.md +++ b/README.md @@ -43,9 +43,11 @@ #### Python -Requires Python 3.12. - ```bash +# With uv installed (macOS/Linux) +DATA_DIR=~/.open-webui uvx --python 3.12 open-webui@latest serve + +# Or install and run with pip (Python 3.12) pip install open-webui open-webui serve ``` diff --git a/backend/open_webui/config.py b/backend/open_webui/config.py index 3e98856ec4..89476c4934 100644 --- a/backend/open_webui/config.py +++ b/backend/open_webui/config.py @@ -2549,6 +2549,7 @@ ENABLE_API_KEYS_ENDPOINT_RESTRICTIONS = ( API_KEYS_ALLOWED_ENDPOINTS = os.getenv('API_KEYS_ALLOWED_ENDPOINTS', os.getenv('API_KEY_ALLOWED_ENDPOINTS', '')) JWT_EXPIRES_IN = os.getenv('JWT_EXPIRES_IN', '4w') +SESSION_EXPIRES_IN = os.getenv('SESSION_EXPIRES_IN', '90d') if JWT_EXPIRES_IN == '-1': log.warning( @@ -3292,6 +3293,7 @@ DEFAULT_CONFIG = { 'auth.api_key.endpoint_restrictions': ENABLE_API_KEYS_ENDPOINT_RESTRICTIONS, 'auth.api_key.allowed_endpoints': API_KEYS_ALLOWED_ENDPOINTS, 'auth.jwt_expiry': JWT_EXPIRES_IN, + 'auth.session_expiry': SESSION_EXPIRES_IN, 'auth.mfa.enable': ENABLE_MFA, 'auth.mfa.allow_oauth_bypass': MFA_ALLOW_OAUTH_BYPASS, 'auth.mfa.allow_trusted_header_bypass': MFA_ALLOW_TRUSTED_HEADER_BYPASS, diff --git a/backend/open_webui/main.py b/backend/open_webui/main.py index 25ff758870..64850bd824 100644 --- a/backend/open_webui/main.py +++ b/backend/open_webui/main.py @@ -827,6 +827,9 @@ app.include_router(retrieval.router, prefix='/api/v1/retrieval', tags=['retrieva app.include_router(configs.router, prefix='/api/v1/configs', tags=['configs']) +from open_webui.env import UVICORN_WORKERS, REDIS_URL +from open_webui.routers import tether +app.include_router(tether.router, prefix='/api/v1/tether', tags=['tether']) app.include_router(auths.router, prefix='/api/v1/auths', tags=['auths']) app.include_router(mfa.router, prefix='/api/v1/auths/mfa', tags=['auths']) app.include_router(users.router, prefix='/api/v1/users', tags=['users']) @@ -1506,7 +1509,7 @@ async def chat_completion( asyncio.create_task(run_initial_title_generation()) else: - chat = await Chats.get_accessible_chat_by_id(chat_id, user, permission='write') + chat = await Chats.get_accessible_chat_by_id(chat_id, user, permission='write', chat=existing_chat) if not chat: raise HTTPException(status_code=404, detail=ERROR_MESSAGES.NOT_FOUND) @@ -2387,6 +2390,8 @@ async def get_app_config(request: Request): }, 'features': { 'slim': USE_SLIM, + 'auth_sessions': True, + 'tether': ENABLE_WEBSOCKET_SUPPORT and (UVICORN_WORKERS == 1 or bool(REDIS_URL)), # Public: required by login/signup page pre-auth 'auth': WEBUI_AUTH, 'auth_trusted_header': bool(WEBUI_AUTH_TRUSTED_EMAIL_HEADER), diff --git a/backend/open_webui/migrations/versions/8bc7e5a902d1_add_auth_session.py b/backend/open_webui/migrations/versions/8bc7e5a902d1_add_auth_session.py new file mode 100644 index 0000000000..05fe02ab2a --- /dev/null +++ b/backend/open_webui/migrations/versions/8bc7e5a902d1_add_auth_session.py @@ -0,0 +1,26 @@ +"""Add renewable authentication sessions.""" + +from alembic import op +import sqlalchemy as sa + +revision = '8bc7e5a902d1' +down_revision = 'e42c8f710a36' +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + 'auth_session', + sa.Column('id', sa.Text(), primary_key=True), + sa.Column('user_id', sa.Text(), nullable=True), + sa.Column('refresh_token_id', sa.Text(), nullable=True), + sa.Column('data', sa.JSON(), nullable=False), + *[sa.Column(key, sa.BigInteger(), nullable=False) for key in ('created_at', 'updated_at', 'expires_at')], + ) + op.create_index('idx_auth_session_user', 'auth_session', ['user_id']) + op.create_index('idx_auth_session_expires', 'auth_session', ['expires_at']) + + +def downgrade(): + op.drop_table('auth_session') diff --git a/backend/open_webui/models/auth_sessions.py b/backend/open_webui/models/auth_sessions.py new file mode 100644 index 0000000000..7a3ab0160d --- /dev/null +++ b/backend/open_webui/models/auth_sessions.py @@ -0,0 +1,98 @@ +"""Renewable human sessions. Pending browser logins become the same session row.""" + +import time +from uuid import uuid4 + +from open_webui.internal.db import Base, get_async_db +from pydantic import BaseModel, ConfigDict +from sqlalchemy import JSON, BigInteger, Column, Index, Text, delete, select, update + + +class AuthSession(Base): + __tablename__ = 'auth_session' + + id = Column(Text, primary_key=True) + user_id = Column(Text, nullable=True) + refresh_token_id = Column(Text, nullable=True) + data = Column(JSON, nullable=False) + created_at = Column(BigInteger, nullable=False) + updated_at = Column(BigInteger, nullable=False) + expires_at = Column(BigInteger, nullable=False) + + __table_args__ = ( + Index('idx_auth_session_user', 'user_id'), + Index('idx_auth_session_expires', 'expires_at'), + ) + + +class AuthSessionModel(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: str + user_id: str | None + refresh_token_id: str | None = None + data: dict + created_at: int + updated_at: int + expires_at: int + + +class AuthSessionsTable: + async def get(self, session_id: str) -> AuthSessionModel | None: + async with get_async_db() as db: + row = await db.get(AuthSession, session_id) + return AuthSessionModel.model_validate(row) if row else None + + async def create(self, **values) -> AuthSessionModel: + now = int(time.time()) + row = AuthSession(id=str(uuid4()), created_at=now, updated_at=now, **values) + async with get_async_db() as db: + await db.execute(delete(AuthSession).where(AuthSession.expires_at <= now)) + db.add(row) + await db.flush() + result = AuthSessionModel.model_validate(row) + await db.commit() + return result + + async def change(self, session_id: str, conditions: list, **values) -> bool: + async with get_async_db() as db: + result = await db.execute( + update(AuthSession) + .where(AuthSession.id == session_id, *conditions) + .values(updated_at=int(time.time()), **values) + .execution_options(synchronize_session=False) + ) + await db.commit() + return result.rowcount == 1 + + async def revoke(self, session_id: str, user_id: str) -> bool: + async with get_async_db() as db: + result = await db.execute( + delete(AuthSession).where(AuthSession.id == session_id, AuthSession.user_id == user_id) + ) + await db.commit() + return result.rowcount == 1 + + async def list(self, user_id: str) -> list[dict]: + async with get_async_db() as db: + rows = ( + await db.execute( + select(AuthSession) + .where( + AuthSession.user_id == user_id, + AuthSession.refresh_token_id.is_not(None), + AuthSession.expires_at > int(time.time()), + ) + .order_by(AuthSession.created_at.desc()) + ) + ).scalars() + return [ + { + **{key: getattr(row, key) for key in ('id', 'created_at', 'updated_at', 'expires_at')}, + 'name': row.data['name'], + } + for row in rows + ] + + +AuthSessions = AuthSessionsTable() diff --git a/backend/open_webui/models/channels.py b/backend/open_webui/models/channels.py index 1f0d15f47b..179ed83fd2 100644 --- a/backend/open_webui/models/channels.py +++ b/backend/open_webui/models/channels.py @@ -613,6 +613,33 @@ class ChannelTable: memberships = result.scalars().all() return [ChannelMemberModel.model_validate(membership) for membership in memberships] + async def get_members_by_user_id_and_channel_ids( + self, user_id: str, channel_ids: list[str], db: Optional[AsyncSession] = None + ) -> dict[str, ChannelMemberModel]: + async with get_async_db_context(db) as db: + result = await db.execute( + select(ChannelMember).filter( + ChannelMember.channel_id.in_(channel_ids), + ChannelMember.user_id == user_id, + ) + ) + return { + membership.channel_id: ChannelMemberModel.model_validate(membership) + for membership in result.scalars().all() + } + + async def get_members_by_channel_ids( + self, channel_ids: list[str], db: Optional[AsyncSession] = None + ) -> list[ChannelMemberModel]: + async with get_async_db_context(db) as db: + result = await db.execute( + select(ChannelMember).filter( + ChannelMember.channel_id.in_(channel_ids), ChannelMember.user_id.is_not(None) + ) + ) + memberships = result.scalars().all() + return [ChannelMemberModel.model_validate(membership) for membership in memberships] + async def pin_channel( self, channel_id: str, diff --git a/backend/open_webui/models/chats.py b/backend/open_webui/models/chats.py index 57f44331e3..ec855eb272 100644 --- a/backend/open_webui/models/chats.py +++ b/backend/open_webui/models/chats.py @@ -1308,7 +1308,7 @@ class ChatTable: async def insert_chat_turn(self, chat_id, user, user_message, message_ids): from fastapi import HTTPException - if not await self.get_accessible_chat_by_id(chat_id, user, permission='write'): + if not await self.has_chat_access(chat_id, user, permission='write'): raise HTTPException(403, 'Chat continuation is not allowed.') async with self._chat_transaction(chat_id) as (db, chat): if not chat: @@ -1997,15 +1997,34 @@ class ChatTable: if user.role not in {'user', 'admin'}: return None chat = chat if chat is not None else await self.get_chat_by_id(id, db=db) + if chat and await self.has_chat_access(id, user, db=db, permission=permission, chat=chat): + return ChatModel.model_validate(chat) + return None + + async def has_chat_access( + self, + id: str, + user, + db: AsyncSession | None = None, + *, + permission: Literal['read', 'write'] = 'read', + chat: Chat | ChatModel | None = None, + ) -> bool: + if user.role not in {'user', 'admin'}: + return False + if chat is None: + async with get_async_db_context(db) as session: + result = await session.execute(select(Chat.user_id, Chat.meta, Chat.folder_id).filter_by(id=id)) + chat = result.first() if not chat: - return None + return False if ( chat.user_id == user.id or user.role == 'admin' and permission == 'read' and (ENABLE_ADMIN_CHAT_ACCESS or is_internal_chat(chat.meta)) ): - return ChatModel.model_validate(chat) + return True from open_webui.models.shared_chats import SharedChats shared = await SharedChats.get_by_chat_id(id, db=db) @@ -2016,7 +2035,7 @@ class ChatTable: user_id=user.id, resource_type='shared_chat', resource_id=id, permission='read', db=db ) ): - return ChatModel.model_validate(chat) + return True if chat.folder_id: from open_webui.utils.access_control.folders import has_folder_access @@ -2026,8 +2045,8 @@ class ChatTable: and (permission == 'read' or (folder.data or {}).get('share_mode') == 'continue') and await has_folder_access(user.id, folder, 'read', db) ): - return ChatModel.model_validate(chat) - return None + return True + return False async def filter_task_ids_by_user_id(self, chat, user_id: str, task_ids: list[str]) -> list[str]: if not task_ids: diff --git a/backend/open_webui/models/messages.py b/backend/open_webui/models/messages.py index 0ba78c2eb2..a3f6c45349 100644 --- a/backend/open_webui/models/messages.py +++ b/backend/open_webui/models/messages.py @@ -412,14 +412,11 @@ class MessageTable: ) all_messages = result.scalars().all() + reply_to_messages = await self._get_reply_to_messages(all_messages, db) + messages = [] for message in all_messages: - reply_to_message = ( - await self.get_message_by_id(message.reply_to_id, include_thread_replies=False, db=db) - if message.reply_to_id - else None - ) - + reply_to_message = reply_to_messages.get(message.reply_to_id) user_info = await self._resolve_user_info(message, db) messages.append( @@ -427,12 +424,33 @@ class MessageTable: { **MessageModel.model_validate(message).model_dump(), 'user': user_info, - 'reply_to_message': (reply_to_message.model_dump() if reply_to_message else None), + 'reply_to_message': reply_to_message, } ) ) return messages + async def _get_reply_to_messages(self, messages: list[Message], db: AsyncSession) -> dict[str, dict]: + """Fetch every reply target of the given messages in a fixed number of queries, keyed by message id.""" + reply_to_ids = {message.reply_to_id for message in messages if message.reply_to_id} + if not reply_to_ids: + return {} + + result = await db.execute(select(Message).filter(Message.id.in_(reply_to_ids))) + targets = {message.id: message for message in result.scalars().all()} + + user_infos = {target_id: await self._resolve_user_info(target, db) for target_id, target in targets.items()} + author_ids = [target.user_id for target_id, target in targets.items() if user_infos[target_id] is None] + authors = {user.id: user for user in await Users.get_users_by_user_ids(author_ids, db=db)} + + reply_to_messages = {} + for target_id, target in targets.items(): + user_info = user_infos[target_id] + if user_info is None and target.user_id in authors: + user_info = authors[target.user_id].model_dump() + reply_to_messages[target_id] = {**MessageModel.model_validate(target).model_dump(), 'user': user_info} + return reply_to_messages + async def get_messages_by_parent_id( self, channel_id: str, @@ -513,15 +531,36 @@ class MessageTable: ) return messages - async def get_last_message_by_channel_id( - self, channel_id: str, db: Optional[AsyncSession] = None - ) -> Optional[MessageModel]: + async def get_last_message_times_by_channel_ids( + self, channel_ids: list[str], db: Optional[AsyncSession] = None + ) -> dict[str, int]: async with get_async_db_context(db) as db: result = await db.execute( - select(Message).filter_by(channel_id=channel_id).order_by(Message.created_at.desc()).limit(1) + select(Message.channel_id, func.max(Message.created_at)) + .filter(Message.channel_id.in_(channel_ids)) + .group_by(Message.channel_id) ) - message = result.scalars().first() - return MessageModel.model_validate(message) if message else None + return dict(result.all()) + + async def get_unread_message_counts_by_user_id( + self, user_id: str, channel_ids: list[str], db: Optional[AsyncSession] = None + ) -> dict[str, int]: + async with get_async_db_context(db) as db: + result = await db.execute( + select(Message.channel_id, func.count(Message.id)) + .join( + ChannelMember, + and_(ChannelMember.channel_id == Message.channel_id, ChannelMember.user_id == user_id), + ) + .filter( + Message.channel_id.in_(channel_ids), + Message.parent_id == None, # only count top-level messages + Message.created_at > func.coalesce(ChannelMember.last_read_at, 0), + or_(Message.user_id != user_id, Message.user_id.is_(None)), + ) + .group_by(Message.channel_id) + ) + return dict(result.all()) async def get_pinned_messages_by_channel_id( self, diff --git a/backend/open_webui/models/models.py b/backend/open_webui/models/models.py index 3ceae31be0..4849f4050a 100755 --- a/backend/open_webui/models/models.py +++ b/backend/open_webui/models/models.py @@ -14,6 +14,7 @@ from open_webui.models.access_grants import AccessGrant, AccessGrantModel, Acces from open_webui.models.groups import Groups from open_webui.models.model_history import ModelHistories, ModelHistory, model_snapshot from open_webui.models.users import User, UserModel, UserResponse, Users +from open_webui.utils.chat_variables import is_nested_too_deep from open_webui.utils.misc import json_text_variants from open_webui.utils.validate import validate_image_url from pydantic import BaseModel, ConfigDict, Field, JsonValue, ValidationInfo, field_validator, model_validator @@ -279,6 +280,8 @@ class ModelsTable: async def _write_model(self, session, form, user_id, current=None, production_version_id=None): """Write configuration, history, and grants in the caller's transaction.""" data = form.model_dump(exclude={'access_grants', 'commit_message'}) + if is_nested_too_deep(data['meta']) or is_nested_too_deep(data['params']): + raise HTTPException(400, 'Model meta and params are nested too deeply') data['meta'].pop('chat_variables_schema', None) snapshot = model_snapshot(data) if current is None: diff --git a/backend/open_webui/retrieval/external.py b/backend/open_webui/retrieval/external.py index ca95153cce..df94dfd055 100644 --- a/backend/open_webui/retrieval/external.py +++ b/backend/open_webui/retrieval/external.py @@ -2,6 +2,7 @@ import asyncio import logging import re import time +from itertools import zip_longest from typing import Any, Optional from open_webui.models.config import Config @@ -325,42 +326,26 @@ async def retrieve_external_knowledge_for_connection( chunks = [] provider = (connection.get('provider') or '').lower() - for query in queries: - if provider == 'qdrant': - chunks.extend( - await _retrieve_qdrant( - connection, - auth_config, - knowledge, - query, - count, - getattr(request.app.state, 'EMBEDDING_FUNCTION', None), - ) - ) - elif provider == 'milvus': - chunks.extend( - await _retrieve_milvus( - connection, - auth_config, - knowledge, - query, - count, - getattr(request.app.state, 'EMBEDDING_FUNCTION', None), - ) - ) - elif provider == 'pgvector': - chunks.extend( - await _retrieve_pgvector( - connection, - auth_config, - knowledge, - query, - count, - getattr(request.app.state, 'EMBEDDING_FUNCTION', None), - ) - ) - else: - raise RuntimeError(f'Unsupported external knowledge provider: {connection.get("provider")}') + retrieve = { + 'qdrant': _retrieve_qdrant, + 'milvus': _retrieve_milvus, + 'pgvector': _retrieve_pgvector, + }.get(provider) + if not retrieve: + raise RuntimeError(f'Unsupported external knowledge provider: {connection.get("provider")}') + + embedding_function = getattr(request.app.state, 'EMBEDDING_FUNCTION', None) + query_chunks = await asyncio.gather( + *(retrieve(connection, auth_config, knowledge, query, count, embedding_function) for query in queries) + ) + + # Score direction differs per provider and metric, so merge by rank + seen_contents = set() + for chunks_at_rank in zip_longest(*query_chunks): + for chunk in chunks_at_rank: + if chunk and chunk['content'] not in seen_contents: + seen_contents.add(chunk['content']) + chunks.append(chunk) chunks = chunks[:count] log.info( diff --git a/backend/open_webui/retrieval/loaders/mineru.py b/backend/open_webui/retrieval/loaders/mineru.py index 7a0c3af9eb..6677fa2ed7 100644 --- a/backend/open_webui/retrieval/loaders/mineru.py +++ b/backend/open_webui/retrieval/loaders/mineru.py @@ -69,6 +69,8 @@ class MinerULoader: try: if self.api_mode == 'cloud': return self._load_cloud_api() + elif self._is_tasks_api(): + return self._load_tasks_api() elif self._is_v1_api(): return self._load_v1_api() else: @@ -192,6 +194,13 @@ class MinerULoader: except requests.RequestException: return False + def _is_tasks_api(self) -> bool: + """Detect MinerU 4.1, whose Tasks API replaced the upload and parse job routes of 4.0.""" + try: + return 'task_output_formats' in requests.get(f'{self.api_url}/v1/health', timeout=10).json()['features'] + except (requests.RequestException, ValueError, KeyError): + return False + def _load_v1_api(self) -> List[Document]: """ Load document using the MinerU 4.x V1 API (asynchronous). @@ -269,16 +278,75 @@ class MinerULoader: return [Document(page_content=markdown_content, metadata=metadata)] - def _request_v1_api(self, method: str, path: str, **kwargs) -> requests.Response: + def _load_tasks_api(self) -> List[Document]: + """Load document using the MinerU 4.1 Tasks API (asynchronous).""" + log.info('Using MinerU Tasks API at %s', self.api_url) + + filename = os.path.basename(self.file_path) + + file_entry = {'file_name': filename, 'bytes': os.path.getsize(self.file_path)} + if self.page_ranges and filename.lower().endswith('.pdf'): + file_entry['page_range'] = self.page_ranges + + task = self._request_v1_api( + 'POST', + '/v1/tasks', + json={ + 'files': [file_entry], + 'tier': self.tier, + 'ocr_mode': 'ocr' if self.enable_ocr else 'auto', + }, + ).json() + + upload = task['files'][0]['upload'] + with open(self.file_path, 'rb') as f: + self._request_v1_api('PUT', upload['url'], headers=upload['headers'], data=f) + + log.info('Polling MinerU task: %s', task['task_id']) + deadline = time.monotonic() + self.timeout if self.timeout else None + while True: + task = self._request_v1_api('GET', f'/v1/tasks/{task["task_id"]}').json() + if task['status'] not in ['waiting-file', 'pending', 'running']: + break + if deadline and time.monotonic() > deadline: + raise HTTPException( + status.HTTP_504_GATEWAY_TIMEOUT, + detail=f'MinerU processing timed out after {self.timeout} seconds', + ) + time.sleep(2) + + file_result = task['files'][0] + if task['status'] != 'completed': + error = file_result.get('error') or task.get('error') or {} + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail=f'MinerU processing {task["status"]}: {error.get("message", "Unknown error")}', + ) + + markdown_content = self._download_and_extract_zip(file_result['result_url'], filename) + + log.info('Successfully parsed document with MinerU Tasks API: %s', filename) + + metadata = { + 'source': filename, + 'api_mode': 'local', + 'tier': file_result['parse']['tier'], + 'version': file_result['parse'].get('parser_version', 'unknown'), + } + + return [Document(page_content=markdown_content, metadata=metadata)] + + def _request_v1_api(self, method: str, path: str, headers: Optional[dict] = None, **kwargs) -> requests.Response: """ Send one request to the MinerU V1 API and raise an HTTPException on failure. """ + url = path if path.startswith('http') else f'{self.api_url}{path}' # A 4.x server started with --api-key rejects every V1 request without it - headers = {'Authorization': f'Bearer {self.api_key}'} if self.api_key else None + # The key stays off storage URLs returned by the server + if self.api_key and url.startswith(f'{self.api_url}/'): + headers = {**(headers or {}), 'Authorization': f'Bearer {self.api_key}'} try: - response = requests.request( - method, f'{self.api_url}{path}', headers=headers, timeout=self.timeout, **kwargs - ) + response = requests.request(method, url, headers=headers, timeout=self.timeout, **kwargs) response.raise_for_status() except requests.Timeout: raise HTTPException( @@ -546,7 +614,12 @@ class MinerULoader: log.info('Downloading results from: %s', zip_url) try: - response = requests.get(zip_url, timeout=60) + headers = ( + {'Authorization': f'Bearer {self.api_key}'} + if self.api_key and zip_url.startswith(f'{self.api_url}/') + else None + ) + response = requests.get(zip_url, headers=headers, timeout=60) response.raise_for_status() except requests.HTTPError as e: raise HTTPException( diff --git a/backend/open_webui/retrieval/models/external.py b/backend/open_webui/retrieval/models/external.py index a66a2c50bf..1e6263784d 100644 --- a/backend/open_webui/retrieval/models/external.py +++ b/backend/open_webui/retrieval/models/external.py @@ -36,7 +36,6 @@ class ExternalReranker(BaseReranker): try: log.info('ExternalReranker:predict:model %s', self.model) - log.info('ExternalReranker:predict:query %s', query) headers = { 'Content-Type': 'application/json', @@ -65,5 +64,5 @@ class ExternalReranker(BaseReranker): return None except Exception as e: - log.exception(f'Error in external reranking: {e}') + log.error('Error in external reranking: %s', type(e).__name__) return None diff --git a/backend/open_webui/retrieval/web/exa.py b/backend/open_webui/retrieval/web/exa.py index 803428a342..3b2061c356 100644 --- a/backend/open_webui/retrieval/web/exa.py +++ b/backend/open_webui/retrieval/web/exa.py @@ -24,7 +24,7 @@ def search_exa( filter_list (list[str] | None): List of domains to filter results by max_content_length (int | None): Maximum characters per result; None leaves text unlimited. """ - log.info('Searching with Exa for query: %s', query) + log.info('Searching with Exa') headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'} diff --git a/backend/open_webui/retrieval/web/ollama.py b/backend/open_webui/retrieval/web/ollama.py index 8535edf172..4900f07a6e 100644 --- a/backend/open_webui/retrieval/web/ollama.py +++ b/backend/open_webui/retrieval/web/ollama.py @@ -23,7 +23,7 @@ def search_ollama_cloud( count (int): Number of results to return filter_list (Optional[list[str]]): List of domains to filter results by """ - log.info('Searching with Ollama for query: %s', query) + log.info('Searching with Ollama') headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'} payload = {'query': query, 'max_results': count} diff --git a/backend/open_webui/routers/auths.py b/backend/open_webui/routers/auths.py index 318d2dc786..0565e57e9b 100644 --- a/backend/open_webui/routers/auths.py +++ b/backend/open_webui/routers/auths.py @@ -10,7 +10,7 @@ import uuid from ssl import CERT_NONE, CERT_REQUIRED, PROTOCOL_TLS import jwt -from aiohttp import BasicAuth, ClientSession +from aiohttp import ClientSession, encode_basic_auth from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.responses import JSONResponse, Response from ldap3 import NONE, Connection, Server, Tls @@ -87,7 +87,10 @@ from pydantic import BaseModel, Field, StrictStr, field_validator from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from open_webui.routers import sessions + router = APIRouter(route_class=MfaRoute) +router.include_router(sessions.router, prefix='/sessions') log = logging.getLogger(__name__) @@ -918,6 +921,9 @@ async def signout(request: Request, response: Response, db: AsyncSession = Depen data = decode_token(token) if data and data.get('id'): actor = await Users.get_user_by_id(data['id'], db=db) + if data and data.get('typ', 'session') == 'session' and data.get('sid') and data.get('id'): + from open_webui.models.auth_sessions import AuthSessions + await AuthSessions.revoke(data['sid'], data['id']) await invalidate_token(request, token) await publish_event( request, @@ -1571,7 +1577,9 @@ async def get_token_client_id(client, token: str) -> str | None: async with session.post( introspection_endpoint, data={'token': token, 'token_type_hint': 'access_token'}, - auth=BasicAuth(client.client_id, client.client_secret or ''), + headers={ + 'Authorization': encode_basic_auth(client.client_id, client.client_secret or '', encoding='latin1') + }, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as r: if r.status != 200: diff --git a/backend/open_webui/routers/channels.py b/backend/open_webui/routers/channels.py index 37e11b9d1b..52395b2987 100644 --- a/backend/open_webui/routers/channels.py +++ b/backend/open_webui/routers/channels.py @@ -193,40 +193,46 @@ async def get_channels( await check_channels_access(request, user) channels = await Channels.get_channels_by_user_id(user.id, db=db) + channel_ids = [channel.id for channel in channels] + dm_channel_ids = [channel.id for channel in channels if channel.type == 'dm'] + + channel_members = await Channels.get_members_by_user_id_and_channel_ids(user.id, channel_ids, db=db) + last_message_times = await Messages.get_last_message_times_by_channel_ids(channel_ids, db=db) + unread_counts = await Messages.get_unread_message_counts_by_user_id(user.id, channel_ids, db=db) + + dm_member_user_ids = {channel_id: set() for channel_id in dm_channel_ids} + for member in await Channels.get_members_by_channel_ids(dm_channel_ids, db=db): + dm_member_user_ids[member.channel_id].add(member.user_id) + all_dm_users = await Users.get_users_by_user_ids( + list({user_id for user_ids in dm_member_user_ids.values() for user_id in user_ids}), db=db + ) + channel_list = [] for channel in channels: - last_message = await Messages.get_last_message_by_channel_id(channel.id, db=db) - last_message_at = last_message.created_at if last_message else None - - channel_member = await Channels.get_member_by_channel_and_user_id(channel.id, user.id, db=db) - unread_count = ( - await Messages.get_unread_message_count(channel.id, user.id, channel_member.last_read_at, db=db) - if channel_member - else 0 - ) + channel_member = channel_members.get(channel.id) user_ids = None users = None if channel.type == 'dm': - member_user_ids = [member.user_id for member in await Channels.get_members_by_channel_id(channel.id, db=db)] users = [ UserIdNameStatusResponse( **{ - **u.model_dump(), - 'is_active': Users.is_active(u), + **dm_user.model_dump(), + 'is_active': Users.is_active(dm_user), } ) - for u in await Users.get_users_by_user_ids(member_user_ids, db=db) + for dm_user in all_dm_users + if dm_user.id in dm_member_user_ids[channel.id] ] - user_ids = [u.id for u in users] + user_ids = [dm_user.id for dm_user in users] channel_list.append( ChannelListItemResponse( **channel.model_dump(), user_ids=user_ids, users=users, - last_message_at=last_message_at, - unread_count=unread_count, + last_message_at=last_message_times.get(channel.id), + unread_count=unread_counts.get(channel.id, 0) if channel_member else 0, is_channel_muted=channel_member.is_channel_muted if channel_member else False, ) ) @@ -1109,6 +1115,17 @@ async def send_notification(request, channel, message, active_user_ids, db=None) member = members.get(u.id) if member and member.is_channel_muted and u.id not in mentioned_user_ids: continue + if member and u.id != message.user_id: + from open_webui.utils.tether import notify + from urllib.parse import quote + + try: + await notify( + u.id, f'channel:{channel.id}:{message.id}', f'{channel.name} / Open WebUI', + message.content, '/channels/' + quote(channel.id, safe=''), + ) + except Exception: + log.exception('Tether channel notification failed') if (u.id not in active_user_ids) and member: if enable_user_webhooks and u.settings: await publish_event( diff --git a/backend/open_webui/routers/chats.py b/backend/open_webui/routers/chats.py index b283f4f3d9..25f08ca67c 100644 --- a/backend/open_webui/routers/chats.py +++ b/backend/open_webui/routers/chats.py @@ -1215,7 +1215,7 @@ async def get_shared_chat_by_id( live = ( shared.chat.get('share_mode') == 'continue' and user is not None - and await Chats.get_accessible_chat_by_id(shared.chat_id, user, db=db, permission='write') is not None + and await Chats.has_chat_access(shared.chat_id, user, db=db, permission='write') ) chat = ( await Chats.get_chat_by_id(shared.chat_id, db=db) @@ -1239,9 +1239,7 @@ async def get_shared_chat_by_id( if chat: data = await shared_chat_response(chat, user, db=db) data['chat']['share_mode'] = ( - 'continue' - if await Chats.get_accessible_chat_by_id(chat.id, user, db=db, permission='write', chat=chat) - else None + 'continue' if await Chats.has_chat_access(chat.id, user, db=db, permission='write', chat=chat) else None ) return data @@ -1347,9 +1345,7 @@ async def get_chat_by_id( if chat: data = await shared_chat_response(chat, user, db=db) data['chat']['share_mode'] = ( - 'continue' - if await Chats.get_accessible_chat_by_id(id, user, db=db, permission='write', chat=chat) - else None + 'continue' if await Chats.has_chat_access(id, user, db=db, permission='write', chat=chat) else None ) data = overlay_response_streams( data, diff --git a/backend/open_webui/routers/configs.py b/backend/open_webui/routers/configs.py index b9f0cec932..1a0608ea51 100644 --- a/backend/open_webui/routers/configs.py +++ b/backend/open_webui/routers/configs.py @@ -385,7 +385,7 @@ async def verify_terminal_server_connection( raise HTTPException(status_code=400, detail='Terminal server URL is required') headers = {} - if form_data.auth_type == 'bearer' and form_data.key: + if form_data.auth_type in ('bearer', 'session') and form_data.key: headers.update(bearer_auth_header(form_data.key)) try: @@ -458,7 +458,7 @@ async def put_terminal_server_policy( raise HTTPException(status_code=400, detail='Terminal server URL is required') headers = {'Content-Type': 'application/json'} - if form_data.auth_type == 'bearer' and form_data.key: + if form_data.auth_type in ('bearer', 'session') and form_data.key: headers.update(bearer_auth_header(form_data.key)) try: @@ -498,7 +498,7 @@ async def put_terminal_server_lifecycle( raise HTTPException(status_code=400, detail='Terminal server URL is required') headers = {'Content-Type': 'application/json'} - if form_data.auth_type == 'bearer' and form_data.key: + if form_data.auth_type in ('bearer', 'session') and form_data.key: headers.update(bearer_auth_header(form_data.key)) try: @@ -540,7 +540,7 @@ async def refresh_terminal_server_terminals( raise HTTPException(status_code=400, detail='Terminal server URL is required') headers = {'Content-Type': 'application/json'} - if form_data.auth_type == 'bearer' and form_data.key: + if form_data.auth_type in ('bearer', 'session') and form_data.key: headers.update(bearer_auth_header(form_data.key)) body = { diff --git a/backend/open_webui/routers/images.py b/backend/open_webui/routers/images.py index fae2410649..ab9c07a92d 100644 --- a/backend/open_webui/routers/images.py +++ b/backend/open_webui/routers/images.py @@ -203,7 +203,7 @@ async def set_image_model(request: Request, model: str): async def get_image_model(request): image_config = await get_image_config() if image_config.IMAGE_GENERATION_ENGINE == 'openai': - return image_config.IMAGE_GENERATION_MODEL if image_config.IMAGE_GENERATION_MODEL else 'dall-e-2' + return image_config.IMAGE_GENERATION_MODEL if image_config.IMAGE_GENERATION_MODEL else 'gpt-image-2' elif image_config.IMAGE_GENERATION_ENGINE == 'gemini': return image_config.IMAGE_GENERATION_MODEL if image_config.IMAGE_GENERATION_MODEL else 'imagen-3.0-generate-002' elif image_config.IMAGE_GENERATION_ENGINE == 'comfyui': @@ -366,10 +366,9 @@ async def get_models(request: Request, user=Depends(get_verified_user)): try: if image_config.IMAGE_GENERATION_ENGINE == 'openai': return [ - {'id': 'dall-e-2', 'name': 'DALL·E 2'}, - {'id': 'dall-e-3', 'name': 'DALL·E 3'}, {'id': 'gpt-image-1', 'name': 'GPT-IMAGE 1'}, {'id': 'gpt-image-1.5', 'name': 'GPT-IMAGE 1.5'}, + {'id': 'gpt-image-2', 'name': 'GPT-IMAGE 2'}, ] elif image_config.IMAGE_GENERATION_ENGINE == 'gemini': return [ diff --git a/backend/open_webui/routers/openai.py b/backend/open_webui/routers/openai.py index 6e1229ed7b..f8323faa8c 100644 --- a/backend/open_webui/routers/openai.py +++ b/backend/open_webui/routers/openai.py @@ -1559,8 +1559,14 @@ async def generate_chat_completion( if not is_responses and 'messages' in payload: for message in payload['messages']: if message.get('role') == 'tool' and isinstance(message.get('content'), list): - message['content'] = ''.join( - part.get('text', '') for part in message['content'] if part.get('type') in ('input_text', 'text') + parts = message['content'] + text = ''.join(part.get('text', '') for part in parts if part.get('type') in ('input_text', 'text')) + cache_control = next( + (part['cache_control'] for part in reversed(parts) if part.get('cache_control')), None + ) + # Prompt-cache markers only work on a content block, so keep the text as one + message['content'] = ( + [{'type': 'text', 'text': text, 'cache_control': cache_control}] if cache_control and text else text ) is_streaming_request = bool(payload.get('stream', False)) diff --git a/backend/open_webui/routers/sessions.py b/backend/open_webui/routers/sessions.py new file mode 100644 index 0000000000..8313c06c5d --- /dev/null +++ b/backend/open_webui/routers/sessions.py @@ -0,0 +1,314 @@ +"""Persistent sessions for native clients, using existing human authentication.""" + +import base64 +import hashlib +import hmac +import time +from datetime import timedelta +from urllib.parse import urlencode, urlsplit, urlunsplit +from uuid import UUID, uuid4 + +import jwt +from fastapi import APIRouter, Depends, HTTPException, Request +from open_webui.models.auth_sessions import AuthSession, AuthSessions +from open_webui.models.config import Config +from open_webui.models.users import Users +from open_webui.routers.mfa import MfaRoute +from open_webui.utils.auth import ( + ALGORITHM, + SESSION_SECRET, + create_token, + decode_token, + get_human_user, + is_valid_token, +) +from open_webui.utils.misc import parse_duration +from open_webui.utils.rate_limit import RateLimiter +from pydantic import BaseModel, ConfigDict, Field, field_validator + +router = APIRouter(route_class=MfaRoute) +limiter = RateLimiter(limit=30, window=60) +PROOF_KEYS = ('auth_time', 'auth_method', 'session_stamp', 'mfa_verified') + + +class SessionForm(BaseModel): + model_config = ConfigDict(extra='forbid', str_strip_whitespace=True) + name: str = Field(min_length=1, max_length=100) + + +class AuthorizationRequest(SessionForm): + redirect_uri: str = Field(max_length=2048) + state: str = Field(min_length=32, max_length=256, pattern=r'^[A-Za-z0-9_-]+$') + code_challenge: str = Field(min_length=43, max_length=43, pattern=r'^[A-Za-z0-9_-]+$') + + @field_validator('redirect_uri') + @classmethod + def loopback(cls, value): + parsed = urlsplit(value) + try: + port = parsed.port + except ValueError: + raise ValueError('Invalid callback port') from None + if ( + parsed.scheme != 'http' + or parsed.hostname != '127.0.0.1' + or not port + or parsed.username + or parsed.password + or parsed.query + or parsed.fragment + ): + raise ValueError('A loopback callback is required') + return value + + +class ExchangeForm(BaseModel): + session_id: UUID + code: str = Field(max_length=8192) + code_verifier: str = Field(min_length=43, max_length=128, pattern=r'^[A-Za-z0-9._~-]+$') + redirect_uri: str = Field(max_length=2048) + + +class RefreshForm(BaseModel): + refresh_token: str = Field(max_length=8192) + request_id: UUID | None = None + + +def digest(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def challenge(verifier: str) -> str: + return base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b'=').decode() + + +async def throttle(request): + ip = request.client.host if request.client else 'unknown' + if await limiter.is_limited(getattr(request.app.state, 'redis', None), 'sessions:' + digest(ip)): + raise HTTPException(429, 'Too many login attempts. Try again shortly.') + + +async def idle_expiry(): + duration = parse_duration(await Config.get('auth.session_expiry')) + if duration is None or duration.total_seconds() < 1: + raise ValueError('SESSION_EXPIRES_IN must be at least 1 second') + return int(time.time()) + int(duration.total_seconds()) + + +async def verify_proof(claims, request): + # Codes and refresh tokens are never passed to ordinary API authentication. + proof = {key: claims.get(key) for key in PROOF_KEYS} + user_id = claims.get('id') + if not user_id or not isinstance(proof['auth_time'], int) or not proof['auth_method']: + raise HTTPException(401, 'Sign in again.') + if not await is_valid_token( + {'id': user_id, 'typ': 'session', 'iat': claims.get('iat'), **proof}, getattr(request.app.state, 'redis', None) + ): + raise HTTPException(401, 'Sign in again.') + user = await Users.get_user_by_id(user_id) + if not user or user.role not in ('user', 'admin'): + raise HTTPException(403, 'Account approval is required.') + return proof + + +def credentials(row, proof, refresh_issued_at=None): + now = int(time.time()) + access_expiry = min(now + 900, row.expires_at) + access = create_token( + {'id': row.user_id, 'sid': row.id, 'typ': 'session', **proof}, + timedelta(seconds=access_expiry - now), + ) + refresh = jwt.encode( + { + 'id': row.user_id, + 'sid': row.id, + 'typ': 'refresh', + **proof, + 'jti': row.refresh_token_id, + 'iat': now if refresh_issued_at is None else refresh_issued_at, + 'exp': row.expires_at, + }, + SESSION_SECRET, + algorithm=ALGORITHM, + ) + return { + 'session_id': row.id, + 'token': access, + 'token_type': 'Bearer', + 'expires_at': access_expiry, + 'refresh_token': refresh, + 'refresh_expires_at': row.expires_at, + } + + +@router.post('') +async def create_session(form: SessionForm, request: Request, user=Depends(get_human_user)): + await throttle(request) + proof = await verify_proof(request.state.claims, request) + row = await AuthSessions.create( + user_id=user.id, + data={'name': form.name}, + expires_at=await idle_expiry(), + refresh_token_id=str(uuid4()), + ) + return credentials(row, proof) + + +@router.post('/requests') +async def authorization_request(form: AuthorizationRequest, request: Request): + await throttle(request) + row = await AuthSessions.create( + data={ + 'name': form.name, + 'authorization': { + 'redirect_uri': form.redirect_uri, + 'state': form.state, + 'code_challenge': form.code_challenge, + }, + }, + expires_at=int(time.time()) + 600, + ) + return {'session_id': row.id} + + +@router.post('/exchange') +async def exchange(form: ExchangeForm, request: Request): + await throttle(request) + claims = decode_token(form.code) + row = await AuthSessions.get(str(form.session_id)) + authorization = row.data.get('authorization', {}) if row else {} + if ( + not claims + or claims.get('typ') != 'session_code' + or claims.get('sid') != str(form.session_id) + or not row + or row.refresh_token_id is not None + or row.expires_at <= time.time() + or claims.get('id') != row.user_id + or not authorization.get('code_hash') + or not hmac.compare_digest(authorization['code_hash'], digest(form.code)) + or authorization.get('redirect_uri') != form.redirect_uri + or not hmac.compare_digest(authorization.get('code_challenge', ''), challenge(form.code_verifier)) + ): + raise HTTPException(401, 'Invalid or expired login code.') + proof = await verify_proof(claims, request) + values = dict( + refresh_token_id=str(uuid4()), + expires_at=await idle_expiry(), + data={key: value for key, value in row.data.items() if key != 'authorization'}, + ) + if not await AuthSessions.change( + row.id, + [ + AuthSession.refresh_token_id.is_(None), + AuthSession.user_id == row.user_id, + AuthSession.data['authorization']['code_hash'].as_string() == authorization['code_hash'], + AuthSession.expires_at > int(time.time()), + ], + **values, + ): + raise HTTPException(401, 'Login code has already been consumed.') + return credentials(row.model_copy(update=values), proof) + + +@router.post('/refresh') +async def refresh(form: RefreshForm, request: Request): + await throttle(request) + claims = decode_token(form.refresh_token) + if not claims or claims.get('typ') != 'refresh' or not claims.get('jti') or not claims.get('sid'): + raise HTTPException(401, 'Invalid refresh token.') + proof = await verify_proof(claims, request) + session_id, token_id = claims['sid'], claims['jti'] + row = await AuthSessions.get(session_id) + if not row or row.user_id != claims.get('id') or row.expires_at <= time.time(): + raise HTTPException(401, 'Session expired or revoked.') + if row.refresh_token_id == token_id: + issued_at = int(time.time()) + data = {key: value for key, value in row.data.items() if key != 'refresh_retry'} + if form.request_id: + data['refresh_retry'] = { + 'token_hash': digest(form.refresh_token), + 'request_hash': digest(str(form.request_id)), + 'issued_at': issued_at, + } + values = dict(refresh_token_id=str(uuid4()), expires_at=await idle_expiry(), data=data) + if await AuthSessions.change( + row.id, + [ + AuthSession.user_id == claims['id'], + AuthSession.refresh_token_id == token_id, + AuthSession.expires_at > int(time.time()), + ], + **values, + ): + return credentials(row.model_copy(update=values), proof, issued_at) + # Another worker may have completed the same request while we were rotating. + row = await AuthSessions.get(session_id) + + retry = row.data.get('refresh_retry', {}) if row else {} + if ( + row + and row.user_id == claims['id'] + and row.refresh_token_id is not None + and row.expires_at > time.time() + and form.request_id + and hmac.compare_digest(retry.get('token_hash', ''), digest(form.refresh_token)) + and hmac.compare_digest(retry.get('request_hash', ''), digest(str(form.request_id))) + ): + # Only the last rotation is recoverable; token verification still enforces its original expiry. + return credentials(row, proof, retry['issued_at']) + raise HTTPException(401, 'Invalid refresh token. Sign in again.') + + +@router.get('') +async def list_sessions(user=Depends(get_human_user)): + return await AuthSessions.list(user.id) + + +@router.get('/{session_id}/request') +async def request_info(session_id: UUID, user=Depends(get_human_user)): + row = await AuthSessions.get(str(session_id)) + if not row or row.refresh_token_id is not None or row.expires_at <= time.time(): + raise HTTPException(404, 'Login request expired.') + return {'name': row.data['name'], 'expires_at': row.expires_at} + + +@router.post('/{session_id}/authorize') +async def authorize(session_id: UUID, request: Request, user=Depends(get_human_user)): + await throttle(request) + proof = await verify_proof(request.state.claims, request) + row = await AuthSessions.get(str(session_id)) + authorization = row.data.get('authorization', {}) if row else {} + if ( + not row + or row.refresh_token_id is not None + or row.user_id is not None + or row.expires_at <= time.time() + or not authorization + or authorization.get('code_hash') + ): + raise HTTPException(409, 'Login request expired or already authorized.') + code = create_token({'typ': 'session_code', 'id': user.id, 'sid': row.id, **proof}, timedelta(seconds=60)) + if not await AuthSessions.change( + row.id, + [ + AuthSession.refresh_token_id.is_(None), + AuthSession.user_id.is_(None), + AuthSession.data['authorization']['code_hash'].as_string().is_(None), + AuthSession.expires_at > int(time.time()), + ], + user_id=user.id, + data={**row.data, 'authorization': {**authorization, 'code_hash': digest(code)}}, + ): + raise HTTPException(409, 'Login request changed.') + callback = urlsplit(authorization['redirect_uri']) + return { + 'redirect_uri': urlunsplit(callback._replace(query=urlencode({'code': code, 'state': authorization['state']}))) + } + + +@router.delete('/{session_id}') +async def revoke(session_id: UUID, user=Depends(get_human_user)): + if not await AuthSessions.revoke(str(session_id), user.id): + raise HTTPException(404, 'Session not found.') + return {'status': True} diff --git a/backend/open_webui/routers/tasks.py b/backend/open_webui/routers/tasks.py index 75943a8764..145feb8bb5 100644 --- a/backend/open_webui/routers/tasks.py +++ b/backend/open_webui/routers/tasks.py @@ -419,7 +419,7 @@ async def generate_queries(request: Request, form_data: dict, user=Depends(get_v ) if getattr(request.state, 'cached_queries', None): - log.info('Reusing cached queries: %s', request.state.cached_queries) + log.info('Reusing cached queries') return request.state.cached_queries if getattr(request.state, 'direct', False) and hasattr(request.state, 'model'): diff --git a/backend/open_webui/routers/terminals.py b/backend/open_webui/routers/terminals.py index 494bc7c284..e3be8cee0a 100644 --- a/backend/open_webui/routers/terminals.py +++ b/backend/open_webui/routers/terminals.py @@ -20,11 +20,11 @@ from fastapi.responses import JSONResponse, StreamingResponse from open_webui.config import TERMINAL_PROXY_HEADERS from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL, ENABLE_TOOL_SERVERS from open_webui.events import EVENTS, publish_event -from open_webui.models.config import Config from open_webui.models.groups import Groups from open_webui.utils.access_control import has_connection_access from open_webui.utils.auth import get_verified_user, get_verified_user_by_token from open_webui.utils.browser import browser_access, browser_ticket +from open_webui.utils.tether import RelayInterrupted, get_terminal_connections, get_terminal_client from open_webui.utils.headers import bearer_auth_header, normalize_bearer_token from open_webui.utils.json_codec import JSONCodec from open_webui.utils.terminals import ( @@ -195,7 +195,7 @@ async def watch_access(server_id, sid, ticket): while True: await asyncio.sleep(10) current_user, _ = await browser_access(server_id, sid, ticket) - latest = await Config.get('terminal_server.connections', []) or [] + latest = await get_terminal_connections(current_user) current = next((c for c in latest if c.get('id') == server_id), None) if ( not current @@ -211,7 +211,7 @@ async def watch_access(server_id, sid, ticket): async def browser_ws(ws: WebSocket, server_id: str, sid: str, ticket: str, kind: str): try: user, data = await browser_access(server_id, sid, ticket) - connections = await Config.get('terminal_server.connections', []) or [] + connections = await get_terminal_connections(user) connection = next((c for c in connections if c.get('id') == server_id), None) if ( kind not in ('ws', 'stream', 'vnc') @@ -239,7 +239,7 @@ async def browser_ws(ws: WebSocket, server_id: str, sid: str, ticket: str, kind: headers['Authorization'] = data['relay']['authorization'] tasks = [] try: - async with aiohttp.ClientSession(cookies=data['relay'].get('cookies', {}), trust_env=True) as client: + async with get_terminal_client(base, user.id, cookies=data['relay'].get('cookies', {}), trust_env=True) as client: async with client.ws_connect( url, headers=headers, @@ -270,13 +270,13 @@ async def list_terminal_servers(request: Request, user=Depends(get_verified_user if not ENABLE_TOOL_SERVERS: return [] - connections = await Config.get('terminal_server.connections', []) or [] + connections = await get_terminal_connections(user) user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id, include_inherited=True)} return [ { 'id': connection.get('id', ''), - 'url': connection.get('url', ''), + 'url': '' if connection.get('owner_id') else connection.get('url', ''), 'name': connection.get('name', ''), 'contexts': terminal_contexts(connection), 'config': {'chat_uploads': terminal_chat_uploads(connection)}, @@ -302,7 +302,7 @@ async def proxy_terminal( if not ENABLE_TOOL_SERVERS: return JSONResponse({'error': 'Tool servers are disabled'}, status_code=403) - connections = await Config.get('terminal_server.connections', []) or [] + connections = await get_terminal_connections(user) connection = next((c for c in connections if c.get('id') == server_id), None) if connection is None: @@ -392,13 +392,13 @@ async def proxy_terminal( if content_type: headers['Content-Type'] = content_type - session = aiohttp.ClientSession( + session = get_terminal_client(base_url, user.id, timeout=aiohttp.ClientTimeout(total=300, connect=10), trust_env=True, ) try: - body = await request.body() + body = request.stream() upstream_response = await session.request( method=request.method, @@ -476,6 +476,9 @@ async def proxy_terminal( except ClientDisconnect: await session.close() return Response(status_code=499) + except RelayInterrupted as error: + await session.close() + return JSONResponse(error.result(), status_code=502) except (aiohttp.ClientConnectionError, TimeoutError) as error: await session.close() log.error('Terminal proxy error: %s', str(error) or type(error).__name__) @@ -517,6 +520,7 @@ async def _resolve_authenticated_connection(ws: WebSocket, server_id: str): if result is None: return None user, connection = result + ws.scope['terminal_resume'] = {key: payload[key] for key in ('resume', 'cursor') if key in payload} chat_id = payload.get('chat_id', '') return user, connection, chat_id if isinstance(chat_id, str) else '', token @@ -537,11 +541,14 @@ async def _resolve_terminal_access(ws: WebSocket, server_id: str, token: str): return None # Resolve terminal server - connections = await Config.get('terminal_server.connections', []) or [] + connections = await get_terminal_connections(user) connection = next((c for c in connections if c.get('id') == server_id), None) if connection is None: - await ws.close(code=4004, reason='Terminal server not found') + if server_id.startswith('tether:'): + await ws.close(code=1013, reason='Tether is offline. Reconnecting.') + else: + await ws.close(code=4004, reason='Terminal server not found') return None if not connection.get('enabled', True): @@ -645,7 +652,7 @@ async def _proxy_terminal_websocket( request = Request({**ws.scope, 'type': 'http', 'method': 'GET'}) opened = False - session = aiohttp.ClientSession() + session = get_terminal_client(base_url, user.id) try: async with session.ws_connect( upstream_url, @@ -658,11 +665,11 @@ async def _proxy_terminal_websocket( auth_type = connection.get('auth_type', 'bearer') if auth_type == 'bearer': key = normalize_bearer_token(connection.get('key', '')) - await upstream.send_str(_json.dumps({'type': 'auth', 'token': key})) + await upstream.send_str(_json.dumps({'type': 'auth', 'token': key, **ws.scope.get('terminal_resume', {})})) elif auth_type == 'session' and is_terminal_orchestrator(connection): await upstream.send_str(_json.dumps({'type': 'auth', 'token': token})) else: - await upstream.send_str(_json.dumps({'type': 'auth', 'token': ''})) + await upstream.send_str(_json.dumps({'type': 'auth', 'token': '', **ws.scope.get('terminal_resume', {})})) if session_id is not None: await publish_event( @@ -687,7 +694,8 @@ async def _proxy_terminal_websocket( elif 'text' in msg and msg['text']: await upstream.send_str(msg['text']) except Exception: - pass + with suppress(Exception): + await ws.close(code=1011, reason='Terminal connection interrupted. Reconnecting.') async def _upstream_to_client(): """Forward upstream → client.""" @@ -702,11 +710,11 @@ async def _proxy_terminal_websocket( aiohttp.WSMsgType.ERROR, ): break - if session_id is None: - code = upstream.close_code or 1011 - await ws.close(code=1011 if code in (1005, 1006, 1015) else code) + code = upstream.close_code or 1011 + await ws.close(code=1011 if code in (1005, 1006, 1015) else code) except Exception: - pass + with suppress(Exception): + await ws.close(code=1011, reason='Terminal connection interrupted. Reconnecting.') async def _watch_access(): try: @@ -717,6 +725,8 @@ async def _proxy_terminal_websocket( return except Exception: log.exception('Terminal access recheck failed') + with suppress(Exception): + await ws.close(code=1011, reason='Terminal access check unavailable. Reconnecting.') # End the proxy as soon as any task finishes (e.g. a # graceful upstream CLOSE) and cancel the rest, which would @@ -734,13 +744,10 @@ async def _proxy_terminal_websocket( await asyncio.gather(*tasks, return_exceptions=True) except Exception as e: log.exception('Terminal WebSocket proxy error: %s', e) - if session_id is None: - status = getattr(e, 'status', None) - code = {401: 4001, 403: 4003, 404: 4004}.get(status, 1011) - try: - await ws.close(code=code) - except Exception: - pass + status = getattr(e, 'status', getattr(e, 'status_code', None)) + code = {401: 4001, 403: 4003, 404: 4004}.get(status, 1011) + with suppress(Exception): + await ws.close(code=code) finally: await session.close() if opened: diff --git a/backend/open_webui/routers/tether.py b/backend/open_webui/routers/tether.py new file mode 100644 index 0000000000..214a3a8f21 --- /dev/null +++ b/backend/open_webui/routers/tether.py @@ -0,0 +1,175 @@ +"""Authenticated outbound connections from Tether.""" + +import asyncio +import json +from contextlib import suppress +from types import SimpleNamespace +from uuid import UUID, uuid4 + +from fastapi import APIRouter, WebSocket +from open_webui.models.auth_sessions import AuthSessions +from open_webui.utils.auth import decode_token, get_verified_user_by_token +from open_webui.utils.tether import hub, encode, decode, get_services + +router = APIRouter() + + +@router.websocket('/connect') +async def connect(ws: WebSocket): + await ws.accept() + connection = None + watcher = None + try: + hello = json.loads(await asyncio.wait_for(ws.receive_text(), 10)) + if hello.get('type') != 'hello' or hello.get('version') not in (1, 2): + raise ValueError('Unsupported Tether handshake') + token = hello.get('token', '') + user = await get_verified_user_by_token(token, getattr(ws.app.state, 'redis', None)) + claims = decode_token(token) + if not user or not claims or not claims.get('sid') or claims.get('typ') != 'session': + raise ValueError('A renewable human session is required') + row = await AuthSessions.get(claims['sid']) + if not row or row.user_id != user.id or row.refresh_token_id is None: + raise ValueError('Invalid session') + connection_id = str(UUID(hello['connection_id'])) + name = hello.get('name') + if not isinstance(name, str) or not 1 <= len(name.strip()) <= 100: + raise ValueError('A connection name of 1–100 characters is required') + await hub.start() + presence = { + 'connection': str(uuid4()), + 'worker': hub.worker, + 'session': row.id, + 'user': user.id, + 'connection_id': connection_id, + 'name': name.strip(), + 'notifications': hello.get('notifications') is True, + 'services': get_services(hello.get('services', ['open_terminal'] if hello['version'] == 1 else None)), + } + connection = SimpleNamespace(ws=ws, presence=presence, routes={}, lock=asyncio.Lock()) + + async def send(header, data): + async with connection.lock: + await asyncio.wait_for(ws.send_bytes(encode(header, data)), 10) + + connection.send = send + # A reconnect replaces the previous connection, including on another worker. + for old in list(hub.connections.values()): + if old.presence['user'] == user.id and old.presence['connection_id'] == connection_id: + await old.ws.close(code=1000) + hub.connections[presence['connection']] = connection + key = hub.key(user.id, connection_id) + record = json.dumps(presence) + presence_lock = asyncio.Lock() + if hub.redis: + await hub.redis.set(key, record, ex=60) + await ws.send_json({'type': 'ready', 'notifications': True}) + + async def watch(): + while True: + await asyncio.sleep(15) + current = await get_verified_user_by_token(token, getattr(ws.app.state, 'redis', None)) + if not current or current.id != user.id: + await ws.close(code=4001) + return + async with presence_lock: + if hub.redis: + refreshed = await hub.redis.eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], 60) else return 0 end", + 1, + key, + record, + ) + if not refreshed: + await ws.close(code=1000) + return + + async def guarded_watch(): + try: + await watch() + except Exception: + with suppress(Exception): + await ws.close(code=4001) + + watcher = asyncio.create_task(guarded_watch()) + while True: + message = await ws.receive() + if message['type'] == 'websocket.disconnect': + break + if message.get('bytes') is not None: + header, data = decode(message['bytes']) + await hub.returned(connection, header, data) + elif message.get('text') is not None: + update = json.loads(message['text']) + if update.get('type') == 'services': + services = get_services(update.get('services')) + name = update.get('name', presence['name']) + if not isinstance(name, str) or not 1 <= len(name.strip()) <= 100: + raise ValueError('A device name of 1–100 characters is required') + async with presence_lock: + updated_record = json.dumps({**presence, 'services': services, 'name': name.strip()}) + if hub.redis: + replaced = await hub.redis.eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then redis.call('set', KEYS[1], ARGV[2], 'EX', 60); return 1 else return 0 end", + 1, key, record, updated_record, + ) + if not replaced: + raise ValueError('Tether connection was replaced') + presence['services'] = services + presence['name'] = name.strip() + record = updated_record + continue + updated = decode_token(update.get('token', '')) + current = await get_verified_user_by_token( + update.get('token', ''), getattr(ws.app.state, 'redis', None) + ) + if ( + update.get('type') != 'auth' + or not current + or current.id != user.id + or not updated + or updated.get('sid') != row.id + or updated.get('typ') != 'session' + ): + raise ValueError('Cannot change relay identity') + token = update['token'] + except Exception: + with suppress(Exception): + await ws.close(code=4001) + finally: + if watcher: + watcher.cancel() + await asyncio.gather(watcher, return_exceptions=True) + if connection: + hub.connections.pop(connection.presence['connection'], None) + if hub.redis: + with suppress(Exception): + await hub.redis.eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", + 1, + key, + record, + ) + for stream_id in list(connection.routes): + stream = hub.streams.get(stream_id) + if stream: + stream.fail('Tether disconnected; operation completion may be unknown') + else: + worker = connection.routes[stream_id] + if hub.redis: + # An explicit failure is terminal even if a preceding frame was lost. + with suppress(Exception): + await hub.redis.publish( + hub.channel(worker), + encode( + { + 'type': 'error', + 'stream': stream_id, + 'sequence': 0, + 'route': 'stream', + 'connection': connection.presence['connection'], + 'message': 'Tether disconnected', + }, + b'', + ), + ) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index ea9ea1de9b..542bff41a9 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -658,6 +658,11 @@ async def user_join(sid, data): if not user: return + previous = await sio.get_session(sid) + if previous.get('user', {}).get('id') not in (None, user.id): + await sio.disconnect(sid) + return + socket_user = { **user.model_dump( exclude=[ @@ -711,7 +716,7 @@ async def refresh_chat_access(chat_id=None): key = (cid, session['id']) if key not in access: user = await Users.get_user_by_id(session['id']) - access[key] = bool(user and await Chats.get_accessible_chat_by_id(cid, user)) + access[key] = bool(user and await Chats.has_chat_access(cid, user)) if not access[key]: await sio.leave_room(sid, f'chat:{cid}') chat_ids.add(cid) @@ -891,7 +896,7 @@ async def chat_events(sid, data): if sid not in (get_room_sid_map(sio.manager, '/', room) or {}): return False sender = await Users.get_user_by_id(user['id']) - if not sender or not await Chats.get_accessible_chat_by_id(chat_id, sender, permission='write'): + if not sender or not await Chats.has_chat_access(chat_id, sender, permission='write'): return False await sio.emit( 'events', @@ -916,7 +921,7 @@ async def chat_events(sid, data): await sio.leave_room(sid, f'chat:{chat_id}') else: reader = await Users.get_user_by_id(user['id']) - if not reader or not await Chats.get_accessible_chat_by_id(chat_id, reader): + if not reader or not await Chats.has_chat_access(chat_id, reader): return False await sio.enter_room(sid, f'chat:{chat_id}') session = SESSION_POOL.get(sid) @@ -1040,6 +1045,9 @@ async def ydoc_document_join(sid, data): ): log.error(f'User {user.get("id")} does not have access to note {note_id}') return + else: + # Only notes have access checks, so no other document gets a room. + return user_id = user['id'] user_name = user.get('name', '') diff --git a/backend/open_webui/socket/utils.py b/backend/open_webui/socket/utils.py index 1d350d3696..6f332a59a0 100644 --- a/backend/open_webui/socket/utils.py +++ b/backend/open_webui/socket/utils.py @@ -214,12 +214,14 @@ class CachedRedisDict(RedisDict): super().__init__(name, redis_url, redis_sentinels, redis_cluster, cache_set_signature=True) self._cache: dict = {} self._cached_signature: str | None = None + self._cached_values: list | None = None def _refresh_cache(self) -> dict: stored_signature = self.redis.get(self._signature_name) if stored_signature is None or stored_signature != self._cached_signature: self._cache = self.redis.hgetall(self.name) self._cached_signature = stored_signature + self._cached_values = None return self._cache def __getitem__(self, key): @@ -238,7 +240,11 @@ class CachedRedisDict(RedisDict): return list(self._refresh_cache().keys()) def values(self): - return [JSONCodec.loads(v) for v in self._refresh_cache().values()] + cache = self._refresh_cache() + # Decoded once per signature; callers share these dicts read-only. + if self._cached_values is None: + self._cached_values = [JSONCodec.loads(v) for v in cache.values()] + return list(self._cached_values) def items(self): return [(k, JSONCodec.loads(v)) for k, v in self._refresh_cache().items()] diff --git a/backend/open_webui/tools/builtin.py b/backend/open_webui/tools/builtin.py index 1704b9d4d1..b439ff5093 100644 --- a/backend/open_webui/tools/builtin.py +++ b/backend/open_webui/tools/builtin.py @@ -2840,6 +2840,21 @@ async def grep_knowledge_files( return JSONCodec.dumps({'error': str(e)}) +def _numbered_lines_within_cap(all_lines: list[str], start: int, end: int) -> tuple[str, int]: + """Number lines start..end, stopping at the last whole line that fits VIEW_FILE_MAX_CHARS.""" + numbered_lines = [] + used_chars = 0 + for line_index in range(start, end): + numbered_line = f'{line_index + 1}: {all_lines[line_index]}' + if used_chars + len(numbered_line) > VIEW_FILE_MAX_CHARS: + if numbered_lines: + return '\n'.join(numbered_lines), line_index + return numbered_line[:VIEW_FILE_MAX_CHARS], line_index + 1 + numbered_lines.append(numbered_line) + used_chars += len(numbered_line) + 1 + return '\n'.join(numbered_lines), end + + async def view_file( file_id: str, offset: int = 0, @@ -2904,11 +2919,10 @@ async def view_file( if start_line is not None or end_line is not None: all_lines = content.split('\n') total_lines = len(all_lines) - s = max(1, int(start_line or 1)) - 1 # 1-indexed to 0-indexed - e = min(total_lines, int(end_line) if end_line else s + 100) - selected = all_lines[s:e] - sliced = '\n'.join(f'{s + i + 1}: {line}' for i, line in enumerate(selected)) - is_truncated = e < total_lines + start_index = max(1, int(start_line or 1)) - 1 # 1-indexed to 0-indexed + end_index = min(total_lines, int(end_line) if end_line else start_index + 100) + sliced, end_index = _numbered_lines_within_cap(all_lines, start_index, end_index) + is_truncated = end_index < total_lines result = { 'id': file.id, 'filename': file.filename, @@ -2916,11 +2930,11 @@ async def view_file( 'updated_at': file.updated_at, 'created_at': file.created_at, 'total_lines': total_lines, - 'showing_lines': f'{s + 1}-{e}', + 'showing_lines': f'{start_index + 1}-{end_index}', } if is_truncated: result['truncated'] = True - result['next_start_line'] = e + 1 + result['next_start_line'] = end_index + 1 return JSONCodec.dumps(result, ensure_ascii=False) sliced = content[offset : offset + max_chars] @@ -3045,11 +3059,10 @@ async def view_knowledge_file( if start_line is not None or end_line is not None: all_lines = content.split('\n') total_lines = len(all_lines) - s = max(1, int(start_line or 1)) - 1 - e = min(total_lines, int(end_line) if end_line else s + 100) - selected = all_lines[s:e] - sliced = '\n'.join(f'{s + i + 1}: {line}' for i, line in enumerate(selected)) - is_truncated = e < total_lines + start_index = max(1, int(start_line or 1)) - 1 + end_index = min(total_lines, int(end_line) if end_line else start_index + 100) + sliced, end_index = _numbered_lines_within_cap(all_lines, start_index, end_index) + is_truncated = end_index < total_lines result = { 'id': file.id, 'filename': file.filename, @@ -3057,14 +3070,14 @@ async def view_knowledge_file( 'updated_at': file.updated_at, 'created_at': file.created_at, 'total_lines': total_lines, - 'showing_lines': f'{s + 1}-{e}', + 'showing_lines': f'{start_index + 1}-{end_index}', } if knowledge_info: result['knowledge_id'] = knowledge_info['id'] result['knowledge_name'] = knowledge_info['name'] if is_truncated: result['truncated'] = True - result['next_start_line'] = e + 1 + result['next_start_line'] = end_index + 1 return JSONCodec.dumps(result, ensure_ascii=False) sliced = content[offset : offset + max_chars] @@ -4221,12 +4234,12 @@ async def create_automation( The automation will use the current chat model. The rrule parameter must be a valid iCalendar RRULE string. Common examples: - - Every day at 9am: "DTSTART:20250101T090000\\nRRULE:FREQ=DAILY" - - Every Monday at 8am: "DTSTART:20250106T080000\\nRRULE:FREQ=WEEKLY;BYDAY=MO" + - Every day at 9am: "DTSTART:20250101T090000\nRRULE:FREQ=DAILY" + - Every Monday at 8am: "DTSTART:20250106T080000\nRRULE:FREQ=WEEKLY;BYDAY=MO" - Every hour: "RRULE:FREQ=HOURLY;INTERVAL=1" - Every 30 minutes: "RRULE:FREQ=MINUTELY;INTERVAL=30" - - Once at a specific time: "DTSTART:20250415T140000\\nRRULE:FREQ=DAILY;COUNT=1" - - First day of every month: "DTSTART:20250101T090000\\nRRULE:FREQ=MONTHLY;BYMONTHDAY=1" + - Once at a specific time: "DTSTART:20250415T140000\nRRULE:FREQ=DAILY;COUNT=1" + - First day of every month: "DTSTART:20250101T090000\nRRULE:FREQ=MONTHLY;BYMONTHDAY=1" The DTSTART time should reflect the desired execution time. Use COUNT=1 for one-time automations. diff --git a/backend/open_webui/utils/access_control/files.py b/backend/open_webui/utils/access_control/files.py index 4052bd0798..03d12c2bb9 100644 --- a/backend/open_webui/utils/access_control/files.py +++ b/backend/open_webui/utils/access_control/files.py @@ -115,7 +115,7 @@ async def has_access_to_file( if accessible_ids: return True for chat_id in shared_chat_ids: - if await Chats.get_accessible_chat_by_id(chat_id, user, db=db): + if await Chats.has_chat_access(chat_id, user, db=db): return True # Note attachment JSON is user-controlled, so only the file owner's notes can grant access. diff --git a/backend/open_webui/utils/audit.py b/backend/open_webui/utils/audit.py index 7f492822e8..edd0cbffe4 100644 --- a/backend/open_webui/utils/audit.py +++ b/backend/open_webui/utils/audit.py @@ -141,11 +141,11 @@ def log_model_usage(request: Request, user: UserModel, model_id: str, usage: Opt logger.error('Failed to log model usage: {}', e) -def redact_passwords(body: str) -> str: - if 'password' not in body.lower(): - return body +def redact_secrets(body: str) -> str: + # Secrets cut off by the size limit have no closing quote. + quoted = r'"(?:[^"\\]|\\.)*\\?(?:"|$)' return re.sub( - r'"(\w*password)"\s*:\s*"(?:[^"\\]|\\.)*"', + rf'"(\w*(?:password|secret|token|key|authorization)s?)"\s*:\s*(?:{quoted}|\[(?:{quoted}|[^\]"])*(?:\]|$))', r'"\1": "********"', body, flags=re.IGNORECASE, @@ -330,8 +330,8 @@ class AuditLoggingMiddleware: response_body = context.response_body.decode('utf-8', errors='replace') # Redact sensitive information - request_body = redact_passwords(request_body) - response_body = redact_passwords(response_body) + request_body = redact_secrets(request_body) + response_body = redact_secrets(response_body) entry = AuditLogEntry( id=request.state.audit_request_id, diff --git a/backend/open_webui/utils/auth.py b/backend/open_webui/utils/auth.py index 1363601508..881ab06db9 100644 --- a/backend/open_webui/utils/auth.py +++ b/backend/open_webui/utils/auth.py @@ -279,6 +279,18 @@ async def is_valid_token(decoded, redis=None) -> bool: token_type = decoded.get('typ', 'session') if token_type not in {'session', 'automation', 'subagent'}: return False + if decoded.get('sid'): + from open_webui.models.auth_sessions import AuthSessions + import time + + session = await AuthSessions.get(decoded['sid']) + if ( + not session + or session.user_id != decoded.get('id') + or session.refresh_token_id is None + or session.expires_at <= int(time.time()) + ): + return False if token_type == 'session': config = await get_mfa_config() if is_mfa_required(decoded.get('auth_method', ''), config): diff --git a/backend/open_webui/utils/automations.py b/backend/open_webui/utils/automations.py index 3250b38614..c4dda6a7bc 100644 --- a/backend/open_webui/utils/automations.py +++ b/backend/open_webui/utils/automations.py @@ -46,6 +46,7 @@ from open_webui.utils.recurrence import ( validate_rrule, ) from open_webui.utils.task import prompt_template +from open_webui.utils.tether import get_terminal_connections, get_terminal_client from open_webui.utils.terminals import get_terminal_server_url from starlette.datastructures import Headers @@ -206,10 +207,7 @@ async def _set_terminal_cwd(app, server_id: str, user, cwd: str, chat_id: str) - import aiohttp from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL - connections = getattr(getattr(app, 'state', None), 'config', None) - if connections is None: - return - connections = getattr(connections, 'TERMINAL_SERVER_CONNECTIONS', None) or [] + connections = await get_terminal_connections(user) connection = next((c for c in connections if c.get('id') == server_id), None) if connection is None: log.warning(f'Terminal server {server_id} not found for CWD set') @@ -230,7 +228,7 @@ async def _set_terminal_cwd(app, server_id: str, user, cwd: str, chat_id: str) - headers['Authorization'] = f'Bearer {connection.get("key", "")}' try: - async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10)) as session: + async with get_terminal_client(base_url, user.id, timeout=aiohttp.ClientTimeout(total=10)) as session: async with session.post( target_url, json={'path': cwd}, diff --git a/backend/open_webui/utils/chat_variables.py b/backend/open_webui/utils/chat_variables.py index bc9413efcc..97fe05c4c8 100644 --- a/backend/open_webui/utils/chat_variables.py +++ b/backend/open_webui/utils/chat_variables.py @@ -63,10 +63,11 @@ def is_nested_too_deep(value: Any) -> bool: stack = [(value, 1)] while stack: item, depth = stack.pop() - if isinstance(item, list): + if isinstance(item, (list, dict)): if depth > MAX_JSON_VALUE_DEPTH: return True - stack.extend((child, depth + 1) for child in item) + children = item.values() if isinstance(item, dict) else item + stack.extend((child, depth + 1) for child in children) return False diff --git a/backend/open_webui/utils/comments.py b/backend/open_webui/utils/comments.py index 924f853982..f8eb10d436 100644 --- a/backend/open_webui/utils/comments.py +++ b/backend/open_webui/utils/comments.py @@ -63,7 +63,7 @@ async def comment_resource(resource_type, resource_id, user, db, *, lock=False, if resource_type == 'chat': from open_webui.models.chats import Chats - allowed = await Chats.get_accessible_chat_by_id(resource_id, user, db=db, chat=resource) is not None + allowed = await Chats.has_chat_access(resource_id, user, db=db, chat=resource) else: if resource_type == 'note': if not await Config.get('notes.enable') or ( diff --git a/backend/open_webui/utils/context_compaction.py b/backend/open_webui/utils/context_compaction.py index da2c03c6e4..861a0d4f1c 100644 --- a/backend/open_webui/utils/context_compaction.py +++ b/backend/open_webui/utils/context_compaction.py @@ -458,9 +458,6 @@ async def _generate_summary( if not isinstance(task_model_params, dict): task_model_params = {} task_model_params = {key: value for key, value in task_model_params.items() if value is not None and value != ''} - task_model_params = task_model_params or { - 'max_tokens': models[task_model_id].get('info', {}).get('params', {}).get('max_tokens', 1000) - } metadata = { **(request.state.metadata if hasattr(request.state, 'metadata') else {}), diff --git a/backend/open_webui/utils/middleware.py b/backend/open_webui/utils/middleware.py index e2284b1118..b789c48a67 100644 --- a/backend/open_webui/utils/middleware.py +++ b/backend/open_webui/utils/middleware.py @@ -152,6 +152,7 @@ from open_webui.utils.task import ( rag_template, tools_function_calling_generation_template, ) +from open_webui.utils.tether import get_terminal_connections from open_webui.utils.terminals import filter_user_terminal_tools, upload_files_to_terminal from open_webui.utils.tool_approval import complete_tool_call, pending_tool_calls from open_webui.utils.tool_search import apply_tool_search, resolve_tool_call @@ -2778,7 +2779,7 @@ async def process_chat_payload(request, form_data, user, metadata, model): # so system terminals work even when no other tools are selected) terminal_capability = (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get('terminal', True) terminal_connection_ids = { - connection.get('id') for connection in await Config.get('terminal_server.connections', []) or [] + connection.get('id') for connection in await get_terminal_connections(user) } if terminal_id and terminal_capability and terminal_id in terminal_connection_ids: try: diff --git a/backend/open_webui/utils/misc.py b/backend/open_webui/utils/misc.py index 043d987105..83c13da09f 100644 --- a/backend/open_webui/utils/misc.py +++ b/backend/open_webui/utils/misc.py @@ -927,26 +927,38 @@ def _strip_null_bytes_deep(obj): return obj +def _has_null_bytes_or_surrogates(obj) -> bool: + """True if any string, key or value, has a null byte or surrogate.""" + if isinstance(obj, str): + if '\x00' in obj: + return True + # O(1) in CPython, so ASCII strings such as base64 images skip the surrogate scan. + if obj.isascii(): + return False + try: + obj.encode('utf-8') + return False + except UnicodeEncodeError: + return True + elif isinstance(obj, dict): + return any(_has_null_bytes_or_surrogates(k) or _has_null_bytes_or_surrogates(v) for k, v in obj.items()) + elif isinstance(obj, list): + return any(_has_null_bytes_or_surrogates(v) for v in obj) + return False + + def sanitize_data_for_db(obj): """Recursively sanitize all strings in a data structure for database storage. - Performs a fast pre-check: serializes the structure once and scans for + Performs a fast pre-check: scans every string once for null bytes or invalid UTF-8 surrogates. If none are found, the original object is returned immediately, skipping the expensive recursive walk. """ if isinstance(obj, str): return sanitize_text_for_db(obj) - # Fast path: check for null bytes and surrogate code points in the serialized form. - # json.dumps is implemented in C and much faster than a Python-level - # recursive walk over every leaf string. - try: - serialized = JSONCodec.dumps(obj, ensure_ascii=False) - if '\\u0000' not in serialized: - serialized.encode('utf-8') - return obj - except (TypeError, ValueError, UnicodeEncodeError): - pass + if not _has_null_bytes_or_surrogates(obj): + return obj return _strip_null_bytes_deep(obj) diff --git a/backend/open_webui/utils/notifications.py b/backend/open_webui/utils/notifications.py index 7d824730de..36417e8fcd 100644 --- a/backend/open_webui/utils/notifications.py +++ b/backend/open_webui/utils/notifications.py @@ -4,7 +4,7 @@ import logging import re import time from typing import Any -from urllib.parse import urlparse +from urllib.parse import urlparse, quote from open_webui.events import EVENT_DEFINITIONS_BY_NAME, NOTIFICATION_EVENTS from open_webui.models.config import Config @@ -387,7 +387,26 @@ async def notify_target( async def dispatch_notification_event(app: Any, event: Any) -> None: - if event.event not in VALID_EVENTS or not await Config.get('ui.enable_user_webhooks'): + if event.event not in VALID_EVENTS: + return + + # Tether delivery is independent of webhooks and of whether a WebUI page is loaded. + if event.event in {CHAT_FINISHED_EVENT, CHAT_FAILED_EVENT, CALENDAR_ALERT_EVENT}: + from open_webui.utils.tether import notify + + data = event.data or {} + chat_id = data.get('chat_id') or (event.subject or {}).get('id') + if event.event == CALENDAR_ALERT_EVENT or (chat_id and not str(chat_id).startswith('local:')): + path = '/calendar' if event.event == CALENDAR_ALERT_EVENT else '/c/' + quote(str(chat_id), safe='') + title = str(data.get('title') or event.message or 'Open WebUI') + body = str(data.get('message') or '') + body = re.sub(r']*>.*?', '', body, flags=re.DOTALL | re.IGNORECASE) + try: + await notify(data.get('user_id'), event.id, f'{title} / Open WebUI', body, path) + except Exception: + log.exception('Tether notification delivery failed for event %s', event.id) + + if not await Config.get('ui.enable_user_webhooks'): return from open_webui.events import event_user_ids diff --git a/backend/open_webui/utils/oauth.py b/backend/open_webui/utils/oauth.py index 74abf72196..a6ff088f1d 100644 --- a/backend/open_webui/utils/oauth.py +++ b/backend/open_webui/utils/oauth.py @@ -1100,7 +1100,7 @@ class OAuthClientManager: # Get the OAuth session session = await OAuthSessions.get_session_by_provider_and_user_id(client_id, user_id) if not session: - log.warning(f'No OAuth session found for user {user_id}, client_id {client_id}') + log.debug('No OAuth session found for user %s, client_id %s', user_id, client_id) return None if ( @@ -1191,10 +1191,12 @@ class OAuthClientManager: if should_send_oauth_resource(client_info): refresh_data['resource'] = client_info.resource - client_auth = None + refresh_headers = {'Content-Type': 'application/x-www-form-urlencoded'} if hasattr(client, 'client_secret') and client.client_secret: if client.client_kwargs.get('token_endpoint_auth_method') == 'client_secret_basic': - client_auth = aiohttp.BasicAuth(client.client_id, client.client_secret) + refresh_headers['Authorization'] = aiohttp.encode_basic_auth( + client.client_id, client.client_secret, encoding='latin1' + ) # Some providers reject a body client_id next to Basic as a second credential (#19415) del refresh_data['client_id'] else: @@ -1213,8 +1215,7 @@ class OAuthClientManager: async with session_http.post( token_endpoint, data=refresh_data, - auth=client_auth, - headers={'Content-Type': 'application/x-www-form-urlencoded'}, + headers=refresh_headers, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as r: if r.status == 200: @@ -1559,11 +1560,13 @@ class OAuthManager: 'refresh_token': token_data['refresh_token'], 'client_id': client.client_id, } - client_auth = None + refresh_headers = {'Content-Type': 'application/x-www-form-urlencoded'} # Add client_secret if available (some providers require it) if hasattr(client, 'client_secret') and client.client_secret: if client.client_kwargs.get('token_endpoint_auth_method') == 'client_secret_basic': - client_auth = aiohttp.BasicAuth(client.client_id, client.client_secret) + refresh_headers['Authorization'] = aiohttp.encode_basic_auth( + client.client_id, client.client_secret, encoding='latin1' + ) if not OAUTH_ACCESS_TOKEN_REQUEST_INCLUDE_CLIENT_ID: del refresh_data['client_id'] else: @@ -1582,8 +1585,7 @@ class OAuthManager: async with session_http.post( token_endpoint, data=refresh_data, - auth=client_auth, - headers={'Content-Type': 'application/x-www-form-urlencoded'}, + headers=refresh_headers, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as r: if r.status == 200: diff --git a/backend/open_webui/utils/terminals.py b/backend/open_webui/utils/terminals.py index ec9fff01de..63de09cddc 100644 --- a/backend/open_webui/utils/terminals.py +++ b/backend/open_webui/utils/terminals.py @@ -140,15 +140,15 @@ async def get_terminal_json( return None import aiohttp + from open_webui.utils.tether import get_terminal_connections, get_terminal_client from open_webui.env import AIOHTTP_CLIENT_SESSION_TOOL_SERVER_SSL - from open_webui.models.config import Config from open_webui.models.groups import Groups from open_webui.models.users import UserModel from open_webui.utils.access_control import has_connection_access from open_webui.utils.tools import build_tool_server_headers user_model = user if isinstance(user, UserModel) else UserModel(**user) - connections = await Config.get('terminal_server.connections', []) or [] + connections = await get_terminal_connections(user_model) connection = next((item for item in connections if item.get('id') == terminal_id), None) if connection: @@ -191,14 +191,14 @@ async def get_terminal_json( body = aiohttp.FormData() body.add_field( 'file', - await asyncio.to_thread(Path(local_path).read_bytes), + open(local_path, 'rb'), filename=file['name'], content_type=(source.meta or {}).get('content_type') or 'application/octet-stream', ) headers = {key: value for key, value in headers.items() if key.lower() != 'content-type'} timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_TERMINAL) try: - async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: + async with get_terminal_client(get_terminal_server_url(connection), user_model.id, timeout=timeout, trust_env=True) as session: async with session.request( 'POST' if file else 'GET', f'{get_terminal_server_url(connection)}{path}', diff --git a/backend/open_webui/utils/tether.py b/backend/open_webui/utils/tether.py new file mode 100644 index 0000000000..7085360382 --- /dev/null +++ b/backend/open_webui/utils/tether.py @@ -0,0 +1,582 @@ +"""Ephemeral reverse connections and streaming service transport.""" + +import asyncio +import json +import re +import struct +import time +from contextlib import suppress +from types import SimpleNamespace +from urllib.parse import urlsplit, urlencode +from uuid import uuid4 + +import aiohttp +from multidict import CIMultiDict +from fastapi import HTTPException +from open_webui.env import REDIS_KEY_PREFIX, UVICORN_WORKERS +from open_webui.models.auth_sessions import AuthSessions +from open_webui.models.config import Config + +CHUNK = 65536 +WINDOW = 262144 +MAX_MESSAGE = 16 * 1024 * 1024 + + +class RelayInterrupted(ConnectionError): + def __init__(self, reason='Connection lost'): + super().__init__( + f'Tether connection interrupted: {reason}. ' + 'The operation may have completed. Check its result before trying again; do not automatically repeat it.' + ) + + def result(self): + return {'error': str(self), 'code': 'tether_interrupted', 'outcome': 'unknown', 'retryable': False} + + +def get_services(value): + if ( + not isinstance(value, list) + or len(value) > 16 + or any(not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9_]{0,31}', name) for name in value) + ): + raise ValueError('Invalid Tether services') + return list(dict.fromkeys(value)) + + +def encode(header, data=b''): + raw = json.dumps(header, separators=(',', ':')).encode() + if len(raw) > 4096 or len(data) > CHUNK: + raise ValueError('Relay packet too large') + return struct.pack('!I', len(raw)) + raw + data + + +def decode(raw): + if len(raw) < 4 or len(raw) > CHUNK + 4100: + raise ValueError('Invalid relay packet') + length = struct.unpack('!I', raw[:4])[0] + if length > 4096 or length > len(raw) - 4: + raise ValueError('Invalid relay header') + header = json.loads(raw[4 : 4 + length]) + if ( + not isinstance(header, dict) + or not isinstance(header.get('stream'), str) + or len(header['stream']) != 36 + or type(header.get('sequence')) is not int + or header['sequence'] < 0 + ): + raise ValueError('Invalid relay header') + return header, raw[4 + length :] + + +class Stream: + def __init__(self, owner, presence): + self.id = str(uuid4()) + self.owner, self.presence = owner, presence + self.sequence = self.incoming = 0 + self.credit = self.receive_credit = WINDOW + self.wake = asyncio.Event() + self.queue = asyncio.Queue(maxsize=32) + self.closed = False + self.lock = asyncio.Lock() + self.error = None + + async def send(self, operation, data=b'', **fields): + while len(data) > self.credit and not self.closed: + self.wake.clear() + try: + await asyncio.wait_for(self.wake.wait(), 300) + except TimeoutError as error: + self.fail('Send timed out') + raise self.error from error + if self.closed: + raise self.error or RelayInterrupted('Stream closed') + self.credit -= len(data) + async with self.lock: + header = dict(fields, type=operation, stream=self.id, sequence=self.sequence) + self.sequence += 1 + try: + await self.owner.send(self.presence, header, data) + except (OSError, TimeoutError) as error: + self.fail(str(error) or 'Send failed') + raise self.error from error + + def feed(self, header, data): + if self.closed: + return + if header['sequence'] != self.incoming: + self.fail('Missing relay packet') + return + self.incoming += 1 + if header['type'] == 'window': + count = header.get('bytes') + if type(count) is not int or count <= 0 or self.credit + count > WINDOW: + self.fail('Invalid relay window') + else: + self.credit += count + self.wake.set() + return + if len(data) > self.receive_credit: + self.fail('Receive window exceeded') + return + self.receive_credit -= len(data) + try: + self.queue.put_nowait((header, data)) + except asyncio.QueueFull: + self.fail('Relay consumer too slow') + + def fail(self, message): + self.error = RelayInterrupted(message) + self.closed = True + self.wake.set() + while not self.queue.empty(): + self.queue.get_nowait() + self.queue.put_nowait(({'type': 'error', 'message': message}, b'')) + + async def receive(self): + if self.error: + raise self.error + try: + header, data = await asyncio.wait_for(self.queue.get(), 300) + except TimeoutError as error: + self.fail('Response timed out') + raise self.error from error + if header['type'] == 'error': + self.fail(header.get('message', 'Response interrupted')) + raise self.error + return header, data + + async def consumed(self, count): + if count and not self.closed: + self.receive_credit += count + await self.send('window', bytes=count) + + async def close(self): + if not self.closed: + with suppress(Exception): + await self.send('cancel') + self.closed = True + self.wake.set() + self.owner.streams.pop(self.id, None) + + +class Hub: + def __init__(self): + self.worker = str(uuid4()) + self.connections = {} + self.streams = {} + self.redis = None + self.listener = None + self.lock = asyncio.Lock() + + def channel(self, worker): + return f'{REDIS_KEY_PREFIX}:tether:worker:{worker}' + + def key(self, user, connection_id): + return f'{REDIS_KEY_PREFIX}:tether:presence:{user}:{connection_id}' + + async def start(self): + async with self.lock: + if self.listener and not self.listener.done(): + return + from open_webui.env import REDIS_URL, REDIS_SENTINEL_HOSTS, REDIS_SENTINEL_PORT, REDIS_CLUSTER + from open_webui.utils.redis import get_redis_connection, get_sentinels_from_env + + sentinels = get_sentinels_from_env(REDIS_SENTINEL_HOSTS, REDIS_SENTINEL_PORT) + if not REDIS_URL and not sentinels: + if UVICORN_WORKERS > 1: + raise HTTPException(503, 'Tether requires Redis with multiple WebUI workers.') + return + self.redis = get_redis_connection( + REDIS_URL, + redis_sentinels=sentinels, + redis_cluster=REDIS_CLUSTER, + async_mode=True, + decode_responses=False, + ) + pubsub = self.redis.pubsub() + await pubsub.subscribe(self.channel(self.worker)) + self.listener = asyncio.create_task(self.listen(pubsub)) + + async def listen(self, pubsub): + try: + async for message in pubsub.listen(): + if message['type'] != 'message': + continue + header, data = decode(message['data']) + if header.pop('route', None) == 'connection': + connection = self.connections.get(header.pop('connection', '')) + if not connection: + continue + if header['type'] == 'notification': + await connection.send(header, data) + continue + reply = header.pop('reply', '') + if header['type'] in ('http-open', 'ws-open'): + if len(connection.routes) >= 16: + continue + connection.routes[header['stream']] = reply + elif connection.routes.get(header['stream']) != reply: + continue + await connection.send(header, data) + if header['type'] == 'cancel': + connection.routes.pop(header['stream'], None) + else: + stream = self.streams.get(header['stream']) + if stream and header.pop('connection', None) == stream.presence['connection']: + stream.feed(header, data) + except Exception: + for stream in list(self.streams.values()): + stream.fail('Tether coordination disconnected') + for connection in list(self.connections.values()): + with suppress(Exception): + await connection.ws.close(code=1011) + finally: + await pubsub.aclose() + + async def presence(self, user_id, connection_id): + await self.start() + if self.redis: + raw = await self.redis.get(self.key(user_id, connection_id)) + return json.loads(raw) if raw else None + return next( + ( + c.presence + for c in self.connections.values() + if c.presence['user'] == user_id and c.presence['connection_id'] == connection_id + ), + None, + ) + + async def presences(self, user_id): + await self.start() + if self.redis: + records = [] + async for key in self.redis.scan_iter(match=self.key(user_id, '*'), count=100): + raw = await self.redis.get(key) + if raw: + records.append(json.loads(raw)) + else: + records = [connection.presence for connection in self.connections.values()] + # SCAN may return a key more than once while connections are changing. + return list({record['connection_id']: record for record in records if record['user'] == user_id}.values()) + + async def open(self, user_id, connection_id, service): + presence = await self.presence(user_id, connection_id) + if not presence or presence['user'] != user_id or presence['connection_id'] != connection_id: + raise HTTPException(503, 'Tether is offline.') + row = await AuthSessions.get(presence['session']) + if not row or row.user_id != user_id or row.refresh_token_id is None or row.expires_at <= time.time(): + raise HTTPException(403, 'Tether session is no longer available.') + if service not in presence.get('services', []): + raise HTTPException(503, 'Tether service is unavailable.') + if sum(s.presence['connection'] == presence['connection'] for s in self.streams.values()) >= 16: + raise HTTPException(429, 'Tether has too many active operations.') + stream = Stream(self, presence) + self.streams[stream.id] = stream + return stream + + async def send(self, presence, header, data): + connection = self.connections.get(presence['connection']) + if presence['worker'] == self.worker: + if not connection: + raise ConnectionError('Tether disconnected') + if header['type'] in ('http-open', 'ws-open'): + if len(connection.routes) >= 16: + raise HTTPException(429, 'Tether has too many active operations.') + connection.routes[header['stream']] = self.worker + await connection.send(header, data) + if header['type'] == 'cancel': + connection.routes.pop(header['stream'], None) + else: + packet = encode( + dict(header, route='connection', connection=presence['connection'], reply=self.worker), data + ) + if not await self.redis.publish(self.channel(presence['worker']), packet): + raise ConnectionError('Tether worker is unavailable') + + async def returned(self, connection, header, data): + worker = connection.routes.get(header['stream']) + if worker == self.worker: + stream = self.streams.get(header['stream']) + if stream: + stream.feed(header, data) + elif worker: + await self.redis.publish( + self.channel(worker), + encode(dict(header, route='stream', connection=connection.presence['connection']), data), + ) + if header['type'] in ('end', 'error'): + connection.routes.pop(header['stream'], None) + + +hub = Hub() + + +async def notify(user_id, notification_id, title, body, path): + """Deliver only to authenticated background clients that opted into notifications.""" + from open_webui.models.users import Users + + if not user_id: + return + presences = [p for p in await hub.presences(user_id) if p.get('notifications')] + if not presences: + return + user = await Users.get_user_by_id(user_id) + if not user or user.role == 'pending' or not user.settings or not (user.settings.ui or {}).get('notificationEnabled', False): + return + payload = json.dumps({ + 'id': str(notification_id)[:200], 'user_id': user_id, + 'title': str(title)[:256], 'body': str(body)[:4096], 'path': path, + }).encode() + for presence in presences: + row = await AuthSessions.get(presence['session']) + if not row or row.user_id != user_id or row.refresh_token_id is None or row.expires_at <= time.time(): + continue + with suppress(Exception): + await asyncio.wait_for(hub.send(presence, { + 'type': 'notification', 'stream': str(uuid4()), 'sequence': 0, + }, payload), 10) + + +async def get_terminal_connections(user): + """Resolve personal targets alongside existing configured servers.""" + user_id = user['id'] if isinstance(user, dict) else user.id + connections = list(await Config.get('terminal_server.connections', []) or []) + for presence in await hub.presences(user_id): + if 'open_terminal' not in presence.get('services', []): + continue + row = await AuthSessions.get(presence['session']) + if not row or row.user_id != user_id or row.refresh_token_id is None or row.expires_at <= time.time(): + continue + connection_id = presence['connection_id'] + connections.append( + { + 'id': 'tether:' + connection_id, + 'name': presence['name'], + 'url': 'tether://' + connection_id, + 'auth_type': 'none', + 'enabled': True, + 'owner_id': user_id, + 'config': { + 'access_grants': [{'principal_type': 'user', 'principal_id': user_id, 'permission': 'read'}] + }, + } + ) + return connections + + +class Operation: + def __init__(self, coroutine): + self.coroutine, self.value = coroutine, None + + def __await__(self): + return self.coroutine.__await__() + + async def __aenter__(self): + self.value = await self.coroutine + return self.value + + async def __aexit__(self, *_): + if self.value: + await self.value.close() + + +class RelayResponse: + def __init__(self, stream, header, upload): + self.stream, self.upload = stream, upload + self.status = header['status'] + self.headers = CIMultiDict({k: str(v) for k, v in header.get('headers', {}).items()}) + self.content_type = self.headers.get('Content-Type', '').split(';')[0] + self.content = self + self.cached = None + + async def iter_any(self): + while True: + header, data = await self.stream.receive() + if header['type'] == 'end': + return + if header['type'] != 'data': + raise RelayInterrupted('Unexpected response') + yield data + await self.stream.consumed(len(data)) + + def iter_chunked(self, size): + return self.iter_any() + + async def read(self): + if self.cached is None: + parts, size = [], 0 + async for data in self.iter_any(): + size += len(data) + if size > MAX_MESSAGE: + await self.close() + raise HTTPException(413, 'Response too large to buffer; use streaming.') + parts.append(data) + self.cached = b''.join(parts) + return self.cached + + async def text(self): + return (await self.read()).decode('utf-8', errors='replace') + + async def json(self): + return json.loads(await self.read()) + + async def close(self): + self.upload.cancel() + await asyncio.gather(self.upload, return_exceptions=True) + await self.stream.close() + + release = close + + +class RelaySocket: + close_code = 1000 + + def __init__(self, stream): + self.stream = stream + + async def send_bytes(self, value): + await self.send(value, 'binary') + + async def send_str(self, value): + await self.send(value.encode(), 'text') + + async def send(self, value, kind): + if len(value) > MAX_MESSAGE: + raise ValueError('WebSocket message too large') + for offset in range(0, max(1, len(value)), CHUNK): + await self.stream.send( + 'data', value[offset : offset + CHUNK], kind=kind, final=offset + CHUNK >= len(value) + ) + + def __aiter__(self): + return self + + async def __anext__(self): + parts, size = [], 0 + while True: + header, data = await self.stream.receive() + if header['type'] == 'end': + self.close_code = header.get('code', 1000) + raise StopAsyncIteration + if header['type'] != 'data': + raise ConnectionError('Invalid WebSocket relay message') + size += len(data) + if size > MAX_MESSAGE: + raise ValueError('WebSocket message too large') + parts.append(data) + await self.stream.consumed(len(data)) + if header.get('final'): + value = b''.join(parts) + binary = header.get('kind') != 'text' + return SimpleNamespace( + type=aiohttp.WSMsgType.BINARY if binary else aiohttp.WSMsgType.TEXT, + data=value if binary else value.decode(), + ) + + async def close(self, **kwargs): + await self.stream.close() + + +class RelayClient: + def __init__(self, user_id, service): + if not user_id: + raise HTTPException(403, 'A user is required for Tether.') + self.user_id = user_id + self.service = get_services([service])[0] + self.operations = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, *_): + await self.close() + + async def close(self): + await asyncio.gather(*(operation.close() for operation in self.operations), return_exceptions=True) + + def request(self, method, url, **kwargs): + return Operation(self._request(method, url, **kwargs)) + + def ws_connect(self, url, **kwargs): + return Operation(self._request(None, url, **kwargs)) + + def __getattr__(self, name): + if name in ('get', 'post', 'put', 'patch', 'delete', 'head', 'options'): + return lambda url, **kwargs: self.request(name.upper(), url, **kwargs) + raise AttributeError(name) + + async def _request(self, method, url, **kwargs): + parsed = urlsplit(str(url)) + if parsed.scheme != 'tether' or not parsed.hostname: + raise ValueError('Invalid Tether URL') + headers = dict(kwargs.get('headers') or {}) + headers['X-User-Id'] = self.user_id + query = parsed.query + if kwargs.get('params'): + query += ('&' if query else '') + urlencode(kwargs['params'], doseq=True) + stream = await hub.open(self.user_id, parsed.hostname, self.service) + body = kwargs.get('data') + if 'json' in kwargs: + body = json.dumps(kwargs['json']).encode() + headers['Content-Type'] = 'application/json' + if isinstance(body, aiohttp.FormData): + body = body() + headers['Content-Type'] = body.headers['Content-Type'] + upload = None + try: + await stream.send( + 'http-open' if method else 'ws-open', + service=self.service, + method=method, + path=parsed.path or '/', + query=query, + headers=headers, + ) + if method: + + async def write(data): + for offset in range(0, len(data), CHUNK): + await stream.send('data', data[offset : offset + CHUNK]) + + async def upload_body(): + try: + if isinstance(body, str): + await write(body.encode()) + elif isinstance(body, (bytes, bytearray)): + await write(body) + elif isinstance(body, aiohttp.payload.Payload): + await body.write(SimpleNamespace(write=write)) + elif body is not None: + async for data in body: + await write(data) + await stream.send('end') + except asyncio.CancelledError: + raise + except Exception: + stream.fail('Upload interrupted; completion may be unknown') + + upload = asyncio.create_task(upload_body()) + header, _ = await asyncio.wait_for(stream.receive(), 30) + if method and header['type'] == 'http-response': + result = RelayResponse(stream, header, upload) + elif not method and header['type'] == 'ws-ready': + result = RelaySocket(stream) + else: + raise ConnectionError('Unexpected relay handshake') + self.operations.append(result) + return result + except BaseException as error: + if upload: + upload.cancel() + await asyncio.gather(upload, return_exceptions=True) + await stream.close() + if isinstance(error, (OSError, TimeoutError)) and not isinstance(error, RelayInterrupted): + raise RelayInterrupted(str(error) or 'Response timed out') from error + raise + + +def get_terminal_client(url, user_id=None, **kwargs): + if str(url).startswith('tether://'): + return RelayClient(user_id, service='open_terminal') + return aiohttp.ClientSession(**kwargs) diff --git a/backend/open_webui/utils/tools.py b/backend/open_webui/utils/tools.py index 97ef2c8030..09c1a7bbf1 100644 --- a/backend/open_webui/utils/tools.py +++ b/backend/open_webui/utils/tools.py @@ -125,6 +125,7 @@ from open_webui.utils.plugin import ( load_tool_module_by_id, set_tool_module_in_cache, ) +from open_webui.utils.tether import RelayInterrupted, get_terminal_connections, get_terminal_client from open_webui.utils.terminals import ( TERMINAL_CONTEXT_HEADER, get_terminal_server_url, @@ -1370,7 +1371,7 @@ async def get_terminal_cwd( """Fetch the current working directory from a terminal server.""" try: cwd_url = f'{base_url.rstrip("/")}/files/cwd' - async with aiohttp.ClientSession( + async with get_terminal_client(base_url, headers.get('X-User-Id'), timeout=aiohttp.ClientTimeout(total=5), trust_env=True, ) as session: @@ -1398,7 +1399,7 @@ async def get_terminal_system_prompt( """ base = base_url.rstrip('/') try: - async with aiohttp.ClientSession( + async with get_terminal_client(base_url, headers.get('X-User-Id'), timeout=aiohttp.ClientTimeout(total=3), trust_env=True, ) as session: @@ -1442,11 +1443,16 @@ async def set_terminal_servers(request: Request): base_url = get_terminal_server_url(connection) + # Spec discovery has no user, so Session connections use the orchestrator key + auth_type = connection.get('auth_type', 'bearer') + if auth_type == 'session': + auth_type = 'bearer' + server_configs.append( { 'url': base_url, 'key': normalize_bearer_token(connection.get('key', '')), - 'auth_type': connection.get('auth_type', 'bearer'), + 'auth_type': auth_type, 'path': connection.get('path', '/openapi.json'), 'spec_type': 'url', # get_tool_servers_data reads config.enable to filter active servers @@ -1532,7 +1538,7 @@ async def get_terminal_tools( if not ENABLE_TOOL_SERVERS: return {} - connections = await Config.get('terminal_server.connections', []) or [] + connections = await get_terminal_connections(user) connection = next( (terminal_connection for terminal_connection in connections if terminal_connection.get('id') == terminal_id), None, @@ -1547,8 +1553,14 @@ async def get_terminal_tools( raise RuntimeError(f'Access denied to terminal {terminal_id}') # Find the cached spec data for this terminal - terminal_servers = await get_terminal_servers(request) - server_data = next((server for server in terminal_servers if server.get('id') == terminal_id), None) + if connection.get('owner_id') == user.id: + url = connection['url'] + openapi = await get_tool_server_data(url + '/openapi.json', {'X-User-Id': user.id}) + server_data = {'id': terminal_id, 'url': url, 'openapi': openapi, + 'specs': convert_openapi_to_tool_payload(openapi)} + else: + terminal_servers = await get_terminal_servers(request) + server_data = next((server for server in terminal_servers if server.get('id') == terminal_id), None) if server_data is None: raise RuntimeError(f"Terminal server '{terminal_id}' is unavailable") @@ -1658,7 +1670,7 @@ async def get_tool_server_data(url: str, headers: dict | None) -> dict[str, Any] error = None try: timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_TOOL_SERVER_DATA) - async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: + async with get_terminal_client(url, _headers.get('X-User-Id'), timeout=timeout, trust_env=True) as session: async with session.get(url, headers=_headers, ssl=AIOHTTP_CLIENT_SESSION_TOOL_SERVER_SSL) as response: if response.status != 200: error_body = await response.json() @@ -1899,7 +1911,7 @@ async def execute_tool_server( else: body_params = params - async with aiohttp.ClientSession( + async with get_terminal_client(url, (headers or {}).get('X-User-Id'), trust_env=True, timeout=aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_TOOL_SERVER) ) as session: request_method = getattr(session, http_method.lower()) @@ -1956,6 +1968,8 @@ async def execute_tool_server( response_headers = response.headers return (response_data, response_headers) + except RelayInterrupted as err: + return (err.result(), None) except Exception as err: error = str(err) log.warning(f'API Request Error: {error}') diff --git a/package-lock.json b/package-lock.json index 45cb579aa3..8d69ee7184 100644 --- a/package-lock.json +++ b/package-lock.json @@ -105,7 +105,7 @@ "svelte-sonner": "^0.3.19", "three": "0.186.1", "tippy.js": "^6.3.7", - "turndown": "^7.2.0", + "turndown": "^7.2.5", "turndown-plugin-gfm": "^1.0.2", "undici": "^7.28.0", "uuid": "^11.1.1", @@ -14715,12 +14715,16 @@ } }, "node_modules/turndown": { - "version": "7.2.2", - "resolved": "https://registry.npmjs.org/turndown/-/turndown-7.2.2.tgz", - "integrity": "sha512-1F7db8BiExOKxjSMU2b7if62D/XOyQyZbPKq/nUwopfgnHlqXHqQ0lvfUTeUIr1lZJzOPFn43dODyMSIfvWRKQ==", + "version": "7.2.5", + "resolved": "https://registry.npmjs.org/turndown/-/turndown-7.2.5.tgz", + "integrity": "sha512-EKZvCCFHh8sWttpAMs10U6niZFJ2cpz8LkCPlylxoG4JcHF50J7EHnnBW3AY1qXtUMu7kDjCp7dnggB1sIbqEQ==", "license": "MIT", "dependencies": { "@mixmark-io/domino": "^2.2.0" + }, + "engines": { + "node": ">=18", + "npm": ">=9" } }, "node_modules/turndown-plugin-gfm": { diff --git a/package.json b/package.json index cd6a19c3dc..4aa9d6a7db 100644 --- a/package.json +++ b/package.json @@ -157,7 +157,7 @@ "svelte-sonner": "^0.3.19", "three": "0.186.1", "tippy.js": "^6.3.7", - "turndown": "^7.2.0", + "turndown": "^7.2.5", "turndown-plugin-gfm": "^1.0.2", "undici": "^7.28.0", "uuid": "^11.1.1", diff --git a/src/app.d.ts b/src/app.d.ts index d23728b866..fcf04f2002 100644 --- a/src/app.d.ts +++ b/src/app.d.ts @@ -1,6 +1,27 @@ // See https://kit.svelte.dev/docs/types#app // for information about these interfaces declare global { + interface Window { + tether?: { + version: 1; + isFocused?(): Promise; + focus?(): Promise; + onEvent?( + callback: ( + event: + | import('$lib/stores').TetherEvent + | { type: 'theme:update'; data: { theme: 'system' | 'light' | 'dark' } } + ) => Promise + ): () => void; + signedIn(token: string): Promise<{ token: string; expiresAt: number | null }>; + refresh(force?: boolean): Promise<{ token: string; expiresAt: number | null } | null>; + signedOut(): Promise; + onToken( + callback: (access: { token: string; expiresAt: number | null } | null) => void + ): () => void; + }; + } + class AudioWorkletProcessor { readonly port: MessagePort; } diff --git a/src/lib/apis/auths/index.ts b/src/lib/apis/auths/index.ts index 57f72fb27d..47264965b6 100644 --- a/src/lib/apis/auths/index.ts +++ b/src/lib/apis/auths/index.ts @@ -380,12 +380,16 @@ export const userSignUp = async ( }; export const userSignOut = async () => { + const token = localStorage.getItem('token'); + // Stop local access even if the hosted server cannot be reached. + await window.tether?.signedOut(); let error = null; const res = await fetch(`${WEBUI_API_BASE_URL}/auths/signout`, { method: 'POST', headers: { - 'Content-Type': 'application/json' + 'Content-Type': 'application/json', + ...(token ? { Authorization: `Bearer ${token}` } : {}) }, credentials: 'include' }) diff --git a/src/lib/apis/terminal/socket.ts b/src/lib/apis/terminal/socket.ts index 9b0a7ae9cf..ea8396fe42 100644 --- a/src/lib/apis/terminal/socket.ts +++ b/src/lib/apis/terminal/socket.ts @@ -39,7 +39,8 @@ export function openTerminalSocket( connection: TerminalConnection, chatId: string | null, path: string, - onOpen?: (socket: WebSocket) => void + onOpen?: (socket: WebSocket) => void, + resume?: { resume: true; cursor: number } ) { const url = new URL(`${connection.baseUrl.replace(/\/$/, '')}${path}`, location.href); url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; @@ -47,7 +48,12 @@ export function openTerminalSocket( socket.binaryType = 'arraybuffer'; socket.onopen = () => { socket.send( - JSON.stringify({ type: 'auth', token: connection.key.trim(), chat_id: chatId ?? '' }) + JSON.stringify({ + ...resume, + type: 'auth', + token: connection.key.trim(), + chat_id: chatId ?? '' + }) ); onOpen?.(socket); }; diff --git a/src/lib/components/AddTerminalServerModal.svelte b/src/lib/components/AddTerminalServerModal.svelte index 7c3f6d1486..a578e1e99c 100644 --- a/src/lib/components/AddTerminalServerModal.svelte +++ b/src/lib/components/AddTerminalServerModal.svelte @@ -1061,20 +1061,23 @@
- {#if auth_type === 'bearer'} - + {#if auth_type === 'bearer' || auth_type === 'session'} +
+ + {#if auth_type === 'session'} +
+ {$i18n.t('Forwards system user session credentials to authenticate')} +
+ {/if} +
{:else if auth_type === 'none'}
{$i18n.t('No authentication')}
- {:else if auth_type === 'session'} -
- {$i18n.t('Forwards system user session credentials to authenticate')} -
{:else if auth_type === 'system_oauth'}
{$i18n.t('Forwards system user OAuth access token to authenticate')} diff --git a/src/lib/components/admin/Settings/Audio.svelte b/src/lib/components/admin/Settings/Audio.svelte index 6ac3ae1a6b..43c632564a 100644 --- a/src/lib/components/admin/Settings/Audio.svelte +++ b/src/lib/components/admin/Settings/Audio.svelte @@ -224,6 +224,7 @@ }; const sttModelUpdateHandler = async () => { + if (STT_WHISPER_MODEL_LOADING) return; STT_WHISPER_MODEL_LOADING = true; await updateConfigHandler(); STT_WHISPER_MODEL_LOADING = false; @@ -652,12 +653,12 @@ bind:value={STT_WHISPER_MODEL} />
-
+
{#if selectedModelIds.length === 1 && $models.find((m) => m.id === selectedModelIds[0])?.has_user_valves && ($_user?.role === 'admin' || ($_user?.permissions?.chat?.valves ?? true))}
@@ -2645,7 +2681,7 @@ > - + {(selectedToolIds ?? []).length} @@ -2668,7 +2704,7 @@ > - + {(selectedSkillIds ?? []).length} @@ -2759,9 +2795,6 @@ : 'bg-transparent text-gray-600 dark:text-gray-300 hover:bg-gray-50 dark:hover:bg-gray-800 '}" > - {/if} @@ -2779,9 +2812,6 @@ : 'bg-transparent text-gray-600 dark:text-gray-300 hover:bg-gray-50 dark:hover:bg-gray-800 '}" > - {/if} @@ -2804,10 +2834,6 @@ : 'focus:outline-hidden rounded-full'}" > - - {/if} diff --git a/src/lib/components/chat/MessageInput/TerminalMenu.svelte b/src/lib/components/chat/MessageInput/TerminalMenu.svelte index 8ef4c9dda0..6bdb194c4c 100644 --- a/src/lib/components/chat/MessageInput/TerminalMenu.svelte +++ b/src/lib/components/chat/MessageInput/TerminalMenu.svelte @@ -135,18 +135,18 @@
{#if disabled} - +