refac

backend/open_webui/models/auths.py

@@ -7,7 +7,6 @@ from open_webui.models.users import UserModel, Users
 from open_webui.env import SRC_LOG_LEVELS
 from pydantic import BaseModel
 from sqlalchemy import Boolean, Column, String, Text
-from open_webui.utils.auth import verify_password
 
 log = logging.getLogger(__name__)
 log.setLevel(SRC_LOG_LEVELS["MODELS"])
@@ -122,7 +121,9 @@ class AuthsTable:
             else:
                 return None
 
-    def authenticate_user(self, email: str, password: str) -> Optional[UserModel]:
+    def authenticate_user(
+        self, email: str, verify_password: callable
+    ) -> Optional[UserModel]:
         log.info(f"authenticate_user: {email}")
 
         user = Users.get_user_by_email(email)
@@ -133,7 +134,7 @@ class AuthsTable:
             with get_db() as db:
                 auth = db.query(Auth).filter_by(id=user.id, active=True).first()
                 if auth:
-                    if verify_password(password, auth.password):
+                    if verify_password(auth.password):
                         return user
                     else:
                         return None

backend/open_webui/routers/auths.py

@@ -45,6 +45,7 @@ from pydantic import BaseModel
 
 from open_webui.utils.misc import parse_duration, validate_email_format
 from open_webui.utils.auth import (
+    verify_password,
     decode_token,
     invalidate_token,
     create_api_key,
@@ -175,7 +176,9 @@ async def update_password(
     if WEBUI_AUTH_TRUSTED_EMAIL_HEADER:
         raise HTTPException(400, detail=ERROR_MESSAGES.ACTION_PROHIBITED)
     if session_user:
-        user = Auths.authenticate_user(session_user.email, form_data.password)
+        user = Auths.authenticate_user(
+            session_user.email, lambda pw: verify_password(form_data.password, pw)
+        )
 
         if user:
             hashed = get_password_hash(form_data.new_password)
@@ -514,7 +517,9 @@ async def signin(request: Request, response: Response, form_data: SigninForm):
         admin_password = "admin"
 
         if Users.get_user_by_email(admin_email.lower()):
-            user = Auths.authenticate_user(admin_email.lower(), admin_password)
+            user = Auths.authenticate_user(
+                admin_email.lower(), lambda pw: verify_password(admin_password, pw)
+            )
         else:
             if Users.has_users():
                 raise HTTPException(400, detail=ERROR_MESSAGES.EXISTING_USERS)
@@ -525,7 +530,9 @@ async def signin(request: Request, response: Response, form_data: SigninForm):
                 SignupForm(email=admin_email, password=admin_password, name="User"),
             )
 
-            user = Auths.authenticate_user(admin_email.lower(), admin_password)
+            user = Auths.authenticate_user(
+                admin_email.lower(), lambda pw: verify_password(admin_password, pw)
+            )
     else:
         password_bytes = form_data.password.encode("utf-8")
         if len(password_bytes) > 72:
@@ -536,7 +543,9 @@ async def signin(request: Request, response: Response, form_data: SigninForm):
             # decode safely — ignore incomplete UTF-8 sequences
             form_data.password = password_bytes.decode("utf-8", errors="ignore")
 
-        user = Auths.authenticate_user(form_data.email.lower(), form_data.password)
+        user = Auths.authenticate_user(
+            form_data.email.lower(), lambda pw: verify_password(form_data.password, pw)
+        )
 
     if user: