scriptoriumadmin/open-webui
1
0
Fork 0
mirror of https://github.com/open-webui/open-webui.git synced 2026-09-13 23:14:11 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

update code with format

Browse source
This commit is contained in:
MiXaiLL76 2026-03-31 00:14:41 +03:00
parent 9c078fa5da
commit 11cebdf904
3 changed files with 146 additions and 125 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
backend/open_webui/config.py
View file

@ -4011,7 +4011,7 @@ LDAP_VALIDATE_CERT = PersistentConfig(
LDAP_CIPHERS = PersistentConfig('LDAP_CIPHERS', 'ldap.server.ciphers', os.environ.get('LDAP_CIPHERS', 'ALL'))
LDAP_USE_AD_SID = PersistentConfig(
"LDAP_USE_AD_SID", "ldap.server.use_ad_sid", os.environ.get("LDAP_USE_AD_SID", "False").lower() == "true"
'LDAP_USE_AD_SID', 'ldap.server.use_ad_sid', os.environ.get('LDAP_USE_AD_SID', 'False').lower() == 'true'
)
# For LDAP Group Management

95
backend/open_webui/models/auths.py
View file

@ -1,13 +1,13 @@
import logging
import uuid
from typing import Optional
from sqlalchemy.orm import Session
from open_webui.internal.db import Base, JSONField, get_db, get_db_context
from open_webui.internal.db import Base, get_db_context
from open_webui.models.users import User, UserModel, UserProfileImageResponse, Users
from open_webui.utils.validate import validate_profile_image_url
from pydantic import BaseModel, field_validator
from sqlalchemy import Boolean, Column, String, Text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
log = logging.getLogger(__name__)
@ -43,7 +43,7 @@ class Token(BaseModel):
class ApiKey(BaseModel):
api_key: Optional[str] = None
api_key: str | None = None
class SigninResponse(Token, UserProfileImageResponse):
@ -73,18 +73,18 @@ class SignupForm(BaseModel):
name: str
email: str
password: str
profile_image_url: Optional[str] = '/user.png'
profile_image_url: str | None = '/user.png'
@field_validator('profile_image_url')
@classmethod
def check_profile_image_url(cls, v: Optional[str]) -> Optional[str]:
def check_profile_image_url(cls, v: str | None) -> str | None:
if v is not None:
return validate_profile_image_url(v)
return v
class AddUserForm(SignupForm):
role: Optional[str] = 'pending'
role: str | None = 'pending'
class AuthsTable:
@ -95,33 +95,68 @@ class AuthsTable:
name: str,
profile_image_url: str = '/user.png',
role: str = 'pending',
oauth: Optional[dict] = None,
db: Optional[Session] = None,
id: Optional[str] = None,
) -> Optional[UserModel]:
oauth: dict | None = None,
db: Session | None = None,
id: str | None = None,
) -> UserModel | None:
with get_db_context(db) as db:
log.info('insert_new_auth')
if id is None:
id = str(uuid.uuid4())
auth = AuthModel(**{'id': id, 'email': email, 'password': password, 'active': True})
result = Auth(**auth.model_dump())
db.add(result)
try:
auth = AuthModel(**{'id': id, 'email': email, 'password': password, 'active': True})
result = Auth(**auth.model_dump())
db.add(result)
user = Users.insert_new_user(id, name, email, profile_image_url, role, oauth=oauth, db=db)
user = Users.insert_new_user(id, name, email, profile_image_url, role, oauth=oauth, db=db)
db.commit()
db.refresh(result)
db.commit()
db.refresh(result)
if result and user:
return user
else:
return None
if result and user:
return user
else:
return None
except IntegrityError as e:
db.rollback()
# Handle case where auth ID already exists (e.g., user deleted from UI
# but auth record remained, common with stable AD SID-based IDs)
if 'UNIQUE constraint failed' in str(e) or 'duplicate key' in str(e).lower():
log.info(f'Auth ID {id} already exists, reactivating and updating existing records')
try:
existing_auth = db.query(Auth).filter_by(id=id).first()
if existing_auth:
existing_auth.email = email
existing_auth.password = password
existing_auth.active = True
else:
new_auth = Auth(
**AuthModel(id=id, email=email, password=password, active=True).model_dump()
)
db.add(new_auth)
db.commit()
def authenticate_user(
self, email: str, verify_password: callable, db: Optional[Session] = None
) -> Optional[UserModel]:
user = Users.get_user_by_id(id, db=db)
if not user:
user = Users.insert_new_user(id, name, email, profile_image_url, role, oauth=oauth, db=db)
else:
Users.update_user_by_id(
id,
{'email': email, 'name': name, 'profile_image_url': profile_image_url},
db=db,
)
return user
except Exception as recovery_err:
log.error(f'Failed to recover from duplicate auth ID: {str(recovery_err)}')
return None
else:
log.error(f'Failed to insert auth: {str(e)}')
return None
def authenticate_user(self, email: str, verify_password: callable, db: Session | None = None) -> UserModel | None:
log.info(f'authenticate_user: {email}')
user = Users.get_user_by_email(email, db=db)
@ -141,8 +176,8 @@ class AuthsTable:
except Exception:
return None
def authenticate_user_by_api_key(self, api_key: str, db: Optional[Session] = None) -> Optional[UserModel]:
log.info(f'authenticate_user_by_api_key')
def authenticate_user_by_api_key(self, api_key: str, db: Session | None = None) -> UserModel | None:
log.info('authenticate_user_by_api_key')
# if no api_key, return None
if not api_key:
return None
@ -153,7 +188,7 @@ class AuthsTable:
except Exception:
return False
def authenticate_user_by_email(self, email: str, db: Optional[Session] = None) -> Optional[UserModel]:
def authenticate_user_by_email(self, email: str, db: Session | None = None) -> UserModel | None:
log.info(f'authenticate_user_by_email: {email}')
try:
with get_db_context(db) as db:
@ -171,7 +206,7 @@ class AuthsTable:
except Exception:
return None
def update_user_password_by_id(self, id: str, new_password: str, db: Optional[Session] = None) -> bool:
def update_user_password_by_id(self, id: str, new_password: str, db: Session | None = None) -> bool:
try:
with get_db_context(db) as db:
result = db.query(Auth).filter_by(id=id).update({'password': new_password})
@ -180,7 +215,7 @@ class AuthsTable:
except Exception:
return False
def update_email_by_id(self, id: str, email: str, db: Optional[Session] = None) -> bool:
def update_email_by_id(self, id: str, email: str, db: Session | None = None) -> bool:
try:
with get_db_context(db) as db:
result = db.query(Auth).filter_by(id=id).update({'email': email})
@ -192,7 +227,7 @@ class AuthsTable:
except Exception:
return False
def delete_auth_by_id(self, id: str, db: Optional[Session] = None) -> bool:
def delete_auth_by_id(self, id: str, db: Session | None = None) -> bool:
try:
with get_db_context(db) as db:
# Delete User

174
backend/open_webui/routers/auths.py
View file

@ -1,91 +1,79 @@
import asyncio
import re
import uuid
import time
import datetime
import logging
from aiohttp import ClientSession
import re
import time
import urllib
import uuid
from ssl import CERT_NONE, CERT_REQUIRED, PROTOCOL_TLS
from aiohttp import ClientSession
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import JSONResponse, Response
from ldap3 import NONE, Connection, Server, Tls
from ldap3.utils.conv import escape_filter_chars
from open_webui.config import (
ENABLE_PASSWORD_AUTH,
OAUTH_MERGE_ACCOUNTS_BY_EMAIL,
OAUTH_PROVIDERS,
OPENID_END_SESSION_ENDPOINT,
OPENID_PROVIDER_URL,
)
from open_webui.constants import ERROR_MESSAGES, WEBHOOK_MESSAGES
from open_webui.env import (
ENABLE_INITIAL_ADMIN_SIGNUP,
ENABLE_OAUTH_TOKEN_EXCHANGE,
WEBUI_AUTH,
WEBUI_AUTH_COOKIE_SAME_SITE,
WEBUI_AUTH_COOKIE_SECURE,
WEBUI_AUTH_SIGNOUT_REDIRECT_URL,
WEBUI_AUTH_TRUSTED_EMAIL_HEADER,
WEBUI_AUTH_TRUSTED_GROUPS_HEADER,
WEBUI_AUTH_TRUSTED_NAME_HEADER,
WEBUI_AUTH_TRUSTED_ROLE_HEADER,
)
from open_webui.internal.db import get_session
from open_webui.models.auths import (
AddUserForm,
ApiKey,
Auths,
Token,
LdapForm,
SigninForm,
SigninResponse,
SignupForm,
Token,
UpdatePasswordForm,
)
from open_webui.models.users import (
UserModel,
UserProfileImageResponse,
Users,
UpdateProfileForm,
UserStatus,
)
from open_webui.models.groups import Groups
from open_webui.models.oauth_sessions import OAuthSessions
from open_webui.constants import ERROR_MESSAGES, WEBHOOK_MESSAGES
from open_webui.env import (
WEBUI_AUTH,
WEBUI_AUTH_TRUSTED_EMAIL_HEADER,
WEBUI_AUTH_TRUSTED_NAME_HEADER,
WEBUI_AUTH_TRUSTED_GROUPS_HEADER,
WEBUI_AUTH_TRUSTED_ROLE_HEADER,
WEBUI_AUTH_COOKIE_SAME_SITE,
WEBUI_AUTH_COOKIE_SECURE,
WEBUI_AUTH_SIGNOUT_REDIRECT_URL,
ENABLE_INITIAL_ADMIN_SIGNUP,
ENABLE_OAUTH_TOKEN_EXCHANGE,
AIOHTTP_CLIENT_SESSION_SSL,
from open_webui.models.users import (
UpdateProfileForm,
UserModel,
UserProfileImageResponse,
Users,
UserStatus,
)
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import RedirectResponse, Response, JSONResponse
from open_webui.config import (
OPENID_PROVIDER_URL,
OPENID_END_SESSION_ENDPOINT,
ENABLE_OAUTH_SIGNUP,
ENABLE_LDAP,
ENABLE_PASSWORD_AUTH,
OAUTH_PROVIDERS,
OAUTH_MERGE_ACCOUNTS_BY_EMAIL,
)
from pydantic import BaseModel
from open_webui.utils.misc import parse_duration, validate_email_format
from open_webui.utils.access_control import get_permissions, has_permission
from open_webui.utils.auth import (
validate_password,
verify_password,
decode_token,
invalidate_token,
create_api_key,
create_token,
decode_token,
get_admin_user,
get_verified_user,
get_current_user,
get_password_hash,
get_http_authorization_cred,
get_password_hash,
get_verified_user,
invalidate_token,
validate_password,
verify_password,
)
from open_webui.internal.db import get_session
from sqlalchemy.orm import Session
from open_webui.utils.webhook import post_webhook
from open_webui.utils.access_control import get_permissions, has_permission
from open_webui.utils.groups import apply_default_group_assignment
from open_webui.utils.redis import get_redis_client
from open_webui.utils.misc import parse_duration, validate_email_format
from open_webui.utils.rate_limit import RateLimiter
from typing import Optional, List
from ssl import CERT_NONE, CERT_REQUIRED, PROTOCOL_TLS
from ldap3 import Server, Connection, NONE, Tls
from ldap3.utils.conv import escape_filter_chars
from open_webui.utils.redis import get_redis_client
from open_webui.utils.webhook import post_webhook
from pydantic import BaseModel
from sqlalchemy.orm import Session
router = APIRouter()
@ -119,7 +107,7 @@ def create_session_response(request: Request, user, db, response: Response = Non
)
if set_cookie and response:
datetime_expires_at = datetime.datetime.fromtimestamp(expires_at, datetime.timezone.utc) if expires_at else None
datetime_expires_at = datetime.datetime.fromtimestamp(expires_at, datetime.UTC) if expires_at else None
max_age = int(expires_delta.total_seconds()) if expires_delta else None
response.set_cookie(
key='token',
@ -152,14 +140,14 @@ def create_session_response(request: Request, user, db, response: Response = Non
class SessionUserResponse(Token, UserProfileImageResponse):
expires_at: Optional[int] = None
permissions: Optional[dict] = None
expires_at: int | None = None
permissions: dict | None = None
class SessionUserInfoResponse(SessionUserResponse, UserStatus):
bio: Optional[str] = None
gender: Optional[str] = None
date_of_birth: Optional[datetime.date] = None
bio: str | None = None
gender: str | None = None
date_of_birth: datetime.date | None = None
@router.get('/', response_model=SessionUserInfoResponse)
@ -190,7 +178,7 @@ async def get_session_user(
response.set_cookie(
key='token',
value=token,
expires=(datetime.datetime.fromtimestamp(expires_at, datetime.timezone.utc) if expires_at else None),
expires=(datetime.datetime.fromtimestamp(expires_at, datetime.UTC) if expires_at else None),
httponly=True, # Ensures the cookie is not accessible via JavaScript
samesite=WEBUI_AUTH_COOKIE_SAME_SITE,
secure=WEBUI_AUTH_COOKIE_SECURE,
@ -336,21 +324,18 @@ def convert_ad_sid_to_string(sid_bytes):
sub_authorities = []
for i in range(sub_auth_count):
offset = 8 + (i * 4)
sub_auth = int.from_bytes(
sid_bytes[offset:offset + 4],
byteorder='little'
)
sub_auth = int.from_bytes(sid_bytes[offset : offset + 4], byteorder='little')
sub_authorities.append(str(sub_auth))
# Construct SID string
sid_string = f"S-{revision}-{identifier_authority}"
sid_string = f'S-{revision}-{identifier_authority}'
if sub_authorities:
sid_string += "-" + "-".join(sub_authorities)
sid_string += '-' + '-'.join(sub_authorities)
return sid_string
except Exception as e:
log.error(f"Failed to convert AD SID to string: {str(e)}")
log.error(f'Failed to convert AD SID to string: {str(e)}')
return None
@ -470,19 +455,19 @@ async def ldap_auth(
# Extract and convert AD SID if enabled
user_id = None
if LDAP_USE_AD_SID and "objectSid" in entry:
if LDAP_USE_AD_SID and 'objectSid' in entry:
try:
sid_bytes = entry["objectSid"].value
sid_bytes = entry['objectSid'].value
if sid_bytes:
user_id = convert_ad_sid_to_string(sid_bytes)
if user_id:
log.debug(f"Successfully extracted AD SID for user {username_list}: {user_id}")
log.debug(f'Successfully extracted AD SID for user {username_list}: {user_id}')
else:
log.warning(f"Failed to convert AD SID for user {username_list}, will use UUID fallback")
log.warning(f'Failed to convert AD SID for user {username_list}, will use UUID fallback')
except Exception as e:
log.warning(f"Error extracting AD SID for user {username_list}: {str(e)}, will use UUID fallback")
log.warning(f'Error extracting AD SID for user {username_list}: {str(e)}, will use UUID fallback')
elif LDAP_USE_AD_SID:
log.debug(f"LDAP_USE_AD_SID enabled but objectSid not found for user {username_list}")
log.debug(f'LDAP_USE_AD_SID enabled but objectSid not found for user {username_list}')
user_groups = []
if ENABLE_LDAP_GROUP_MANAGEMENT and LDAP_ATTRIBUTE_FOR_GROUPS in entry:
@ -554,7 +539,7 @@ async def ldap_auth(
role = 'admin' if not Users.has_users(db=db) else request.app.state.config.DEFAULT_USER_ROLE
if user_id:
log.debug(f"Creating LDAP user with AD SID as ID: {user_id}")
log.debug(f'Creating LDAP user with AD SID as ID: {user_id}')
user = Auths.insert_new_auth(
email=email,
@ -566,6 +551,7 @@ async def ldap_auth(
)
if not user:
log.error(f'Failed to create or recover LDAP user: email={email}, id={user_id}')
raise HTTPException(500, detail=ERROR_MESSAGES.CREATE_USER_ERROR)
apply_default_group_assignment(
@ -631,7 +617,7 @@ async def signin(
name = request.headers.get(WEBUI_AUTH_TRUSTED_NAME_HEADER, email)
try:
name = urllib.parse.unquote(name, encoding='utf-8')
except Exception as e:
except Exception:
pass
if not Users.get_user_by_email(email.lower(), db=db):
@ -1038,7 +1024,7 @@ async def get_admin_config(request: Request, user=Depends(get_admin_user)):
class AdminConfig(BaseModel):
SHOW_ADMIN_DETAILS: bool
ADMIN_EMAIL: Optional[str] = None
ADMIN_EMAIL: str | None = None
WEBUI_URL: str
ENABLE_SIGNUP: bool
ENABLE_API_KEYS: bool
@ -1050,15 +1036,15 @@ class AdminConfig(BaseModel):
ENABLE_COMMUNITY_SHARING: bool
ENABLE_MESSAGE_RATING: bool
ENABLE_FOLDERS: bool
FOLDER_MAX_FILE_COUNT: Optional[int | str] = None
FOLDER_MAX_FILE_COUNT: int | str | None = None
ENABLE_CHANNELS: bool
ENABLE_MEMORIES: bool
ENABLE_NOTES: bool
ENABLE_USER_WEBHOOKS: bool
ENABLE_USER_STATUS: bool
PENDING_USER_OVERLAY_TITLE: Optional[str] = None
PENDING_USER_OVERLAY_CONTENT: Optional[str] = None
RESPONSE_WATERMARK: Optional[str] = None
PENDING_USER_OVERLAY_TITLE: str | None = None
PENDING_USER_OVERLAY_CONTENT: str | None = None
RESPONSE_WATERMARK: str | None = None
@router.post('/admin/config')
@ -1131,7 +1117,7 @@ async def update_admin_config(request: Request, form_data: AdminConfig, user=Dep
class LdapServerConfig(BaseModel):
label: str
host: str
port: Optional[int] = None
port: int | None = None
attribute_for_mail: str = 'mail'
attribute_for_username: str = 'uid'
app_dn: str
@ -1139,9 +1125,9 @@ class LdapServerConfig(BaseModel):
search_base: str
search_filters: str = ''
use_tls: bool = True
certificate_path: Optional[str] = None
certificate_path: str | None = None
validate_cert: bool = True
ciphers: Optional[str] = 'ALL'
ciphers: str | None = 'ALL'
@router.get('/admin/config/ldap/server', response_model=LdapServerConfig)
@ -1214,7 +1200,7 @@ async def get_ldap_config(request: Request, user=Depends(get_admin_user)):
class LdapConfigForm(BaseModel):
enable_ldap: Optional[bool] = None
enable_ldap: bool | None = None
@router.post('/admin/config/ldap')
@ -1337,7 +1323,7 @@ async def token_exchange(
# Get sub claim
sub = user_data.get(request.app.state.config.OAUTH_SUB_CLAIM or OAUTH_PROVIDERS[provider].get('sub_claim', 'sub'))
if not sub:
log.warning(f'Token exchange failed: sub claim missing from user data')
log.warning('Token exchange failed: sub claim missing from user data')
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Token missing required 'sub' claim",
@ -1345,7 +1331,7 @@ async def token_exchange(
email = user_data.get(email_claim, '')
if not email:
log.warning(f'Token exchange failed: email claim missing from user data')
log.warning('Token exchange failed: email claim missing from user data')
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail='Token missing required email claim',