fix(auth): migrate local credentials to salted Argon2id (#1432)

backend/alembic/versions/075_local_password_hashes.py

@@ -0,0 +1,64 @@
+"""Protect dormant local credentials with salted Argon2id.
+
+Irreversible data migration: never downgrade credentials back to MD5. Stop old
+writers before upgrading. Each CAS update commits independently so interruption
+is resumable; historical initialization scripts remain unchanged.
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+from common.core.passwords import is_legacy_password, wrap_legacy_password
+
+revision = "d91543ca7b21"
+down_revision = "ed42b7f169a3"
+branch_labels = None
+depends_on = None
+BATCH_SIZE = 100
+
+
+def upgrade():
+    # No external services, configuration, CRUD or plaintext passwords are used.
+    # Release row locks after each update rather than holding them during hashing.
+    with op.get_context().autocommit_block():
+        bind = op.get_bind()
+        schema = None
+        if bind.dialect.name == "postgresql":
+            schema = bind.scalar(sa.text("SELECT current_schema()"))
+            if not schema or not sa.inspect(bind).has_table("sys_user", schema=schema):
+                raise RuntimeError(
+                    "Local credentials must exist in the current migration schema"
+                )
+        users = sa.table(
+            "sys_user",
+            sa.column("id", sa.BigInteger),
+            sa.column("password", sa.String(255)),
+            schema=schema,
+        )
+        last_id = None
+        while True:
+            query = (
+                sa.select(users.c.id, users.c.password)
+                .order_by(users.c.id)
+                .limit(BATCH_SIZE)
+            )
+            if last_id is not None:
+                query = query.where(users.c.id > last_id)
+            rows = bind.execute(query).all()
+            if not rows:
+                break
+            for uid, stored in rows:
+                if is_legacy_password(stored):
+                    replacement = wrap_legacy_password(stored)
+                    bind.execute(
+                        users.update()
+                        .where(users.c.id == uid, users.c.password == stored)
+                        .values(password=replacement)
+                    )
+            last_id = rows[-1][0]
+
+
+def downgrade():
+    # Schema is unchanged. Keeping strong credentials is intentional; old
+    # applications are incompatible and require a coordinated restore/reset.
+    pass

backend/apps/mcp/mcp.py

@@ -5,6 +5,7 @@
 from typing import Optional
 
 import jwt
+from starlette.concurrency import run_in_threadpool
 from fastapi import HTTPException, status, APIRouter
 # from fastapi.security import OAuth2PasswordBearer
 from jwt.exceptions import InvalidTokenError
@@ -16,7 +17,7 @@
     ChatFinishStep, McpDs, ChatToken, WsMcp, Chat
 from apps.datasource.crud.datasource import get_datasource_list
 from apps.system.crud.aimodel_manage import get_ai_model_list_by_workspace
-from apps.system.crud.user import authenticate, user_ws_options
+from apps.system.crud.user import authenticate, upgrade_password, user_ws_options
 from apps.system.crud.user import get_db_user
 from apps.system.models.system_model import UserWsModel
 from apps.system.models.user import UserModel
@@ -40,12 +41,13 @@
 
 @router.post("/access_token", operation_id="access_token")
 async def access_token(session: SessionDep, chat: ChatToken):
-    user: BaseUserDTO = authenticate(session=session, account=chat.username, password=chat.password)
-    if not user:
+    user: BaseUserDTO = await run_in_threadpool(authenticate, session=session, account=chat.username, password=chat.password)
+    if not user or user.status != 1 or user.origin not in (None, 0):
         raise HTTPException(status_code=400, detail="Incorrect account or password")
 
     if not user.oid or user.oid == 0:
         raise HTTPException(status_code=400, detail="No associated workspace, Please contact the administrator")
+    await run_in_threadpool(upgrade_password, session=session, user=user, password=chat.password)
     access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
     user_dict = user.to_dict()
     t = Token(access_token=create_access_token(
@@ -101,12 +103,13 @@ async def mcp_start(session: SessionDep, trans: Trans, chat: ChatStart):
         res_token = chat.token
         user = get_user(session, chat.token)
     else:
-        user = authenticate(session=session, account=chat.username, password=chat.password)
-        if not user:
+        user = await run_in_threadpool(authenticate, session=session, account=chat.username, password=chat.password)
+        if not user or user.status != 1 or user.origin not in (None, 0):
             raise HTTPException(status_code=400, detail="Incorrect account or password")
 
         if not user.oid or user.oid == 0:
             raise HTTPException(status_code=400, detail="No associated workspace, Please contact the administrator")
+        await run_in_threadpool(upgrade_password, session=session, user=user, password=chat.password)
         access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
         user_dict = user.to_dict()
         t = Token(access_token=create_access_token(

backend/apps/system/api/login.py

@@ -1,12 +1,13 @@
 from typing import Annotated
+from starlette.concurrency import run_in_threadpool
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.security import OAuth2PasswordRequestForm
 from apps.system.schemas.logout_schema import LogoutSchema
 from apps.system.schemas.system_schema import BaseUserDTO, LoginPwdEditor
 from common.core.deps import SessionDep, Trans
 from common.utils.crypto import sqlbot_decrypt
-from ..crud.user import authenticate, check_pwd_format, clean_user_cache, get_db_user
-from common.core.security import create_access_token, default_pwd, md5pwd, verify_md5pwd
+from ..crud.user import authenticate, upgrade_password, check_pwd_format, clean_user_cache, get_db_user
+from common.core.security import create_access_token, default_pwd, get_password_hash, verify_password
 from datetime import timedelta
 from common.core.config import settings
 from common.core.schemas import Token
@@ -32,7 +33,7 @@ async def local_login(
 ) -> Token:
     origin_account = await sqlbot_decrypt(form_data.username)
     origin_pwd = await sqlbot_decrypt(form_data.password)
-    user: BaseUserDTO = authenticate(session=session, account=origin_account, password=origin_pwd)
+    user: BaseUserDTO = await run_in_threadpool(authenticate, session=session, account=origin_account, password=origin_pwd)
     if not user:
         raise HTTPException(status_code=400, detail=trans('i18n_login.account_pwd_error'))
     if not user.oid or user.oid == 0:
@@ -44,9 +45,10 @@ async def local_login(
     if (
         user.id != 1
         and await initial_pwd_disabled(session)
-        and verify_md5pwd(default_pwd(), user.password)
+        and await run_in_threadpool(verify_password, default_pwd(), user.password)
     ):
         return Token(access_token='', need_change_pwd=True)
+    await run_in_threadpool(upgrade_password, session=session, user=user, password=origin_pwd)
     access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
     user_dict = user.to_dict()
     return Token(
@@ -65,7 +67,7 @@ async def login_change_pwd(session: SessionDep, trans: Trans, editor: LoginPwdEd
     origin_account = await sqlbot_decrypt(editor.account)
     origin_pwd = await sqlbot_decrypt(editor.pwd)
     new_pwd = await sqlbot_decrypt(editor.new_pwd)
-    user: BaseUserDTO = authenticate(session=session, account=origin_account, password=origin_pwd)
+    user: BaseUserDTO = await run_in_threadpool(authenticate, session=session, account=origin_account, password=origin_pwd)
     if not user:
         raise HTTPException(status_code=400, detail=trans('i18n_login.account_pwd_error'))
     if user.origin is not None and user.origin != 0:
@@ -75,10 +77,10 @@ async def login_change_pwd(session: SessionDep, trans: Trans, editor: LoginPwdEd
     if await initial_pwd_disabled(session) and new_pwd == default_pwd():
         raise HTTPException(status_code=400, detail=trans('i18n_login.new_pwd_is_initial'))
     db_user = get_db_user(session=session, user_id=user.id)
-    db_user.password = md5pwd(new_pwd)
+    db_user.password = await run_in_threadpool(get_password_hash, new_pwd)
     session.add(db_user)
     await clean_user_cache(user.id)
-    return db_user
+    return {"id": db_user.id}
 
 @router.post("/logout")
 async def logout(session: SessionDep, request: Request, dto: LogoutSchema):

backend/apps/system/api/user.py

@@ -1,5 +1,6 @@
 from collections import defaultdict
 from typing import Optional
+from starlette.concurrency import run_in_threadpool
 from fastapi import APIRouter, File, HTTPException, Path, Query, UploadFile
 from sqlmodel import SQLModel, case, or_, select, delete as sqlmodel_delete
 from apps.system.crud.user import check_account_exists, check_email_exists, check_email_format, check_pwd_format, get_db_user, single_delete, user_ws_options
@@ -15,7 +16,7 @@
 from common.core.deps import CurrentUser, SessionDep, Trans
 from common.core.pagination import Paginator
 from common.core.schemas import PaginatedResponse, PaginationParams
-from common.core.security import default_md5_pwd, md5pwd, verify_md5pwd
+from common.core.security import default_password_hash, get_password_hash, verify_password
 from common.core.sqlbot_cache import clear_cache
 from common.core.config import settings
 from apps.swagger.i18n import PLACEHOLDER_PREFIX
@@ -208,7 +209,7 @@ async def pager(
         user_id = item['id']
         merged[user_id].append(ws_oid)
         if user_id not in extra_attrs:
-            extra_attrs[user_id] = {k: v for k, v in item.items() if k != "ws_oid"}
+            extra_attrs[user_id] = {k: v for k, v in item.items() if k not in {"ws_oid", "password"}}
 
     # 组合结果
     result = [
@@ -280,6 +281,7 @@ async def create(session: SessionDep, creator: UserCreator, trans: Trans):
         raise Exception(trans('i18n_format_invalid', key = f"{trans('i18n_user.email')} [{creator.email}]"))
     #data = creator.model_dump(exclude_unset=True)
     data = creator.model_dump()
+    data["password"] = await run_in_threadpool(default_password_hash)
     user_model = UserModel.model_validate(data)
     #user_model.create_time = get_timestamp()
     user_model.language = "zh-CN"
@@ -297,7 +299,7 @@ async def create(session: SessionDep, creator: UserCreator, trans: Trans):
         session.add_all(db_model_list)
         user_model.oid = creator.oid_list[0]   
     session.add(user_model)
-    return user_model
+    return user_model.model_dump(exclude={"password"})
 
 
 @router.put("", summary=f"{PLACEHOLDER_PREFIX}user_update_api", description=f"{PLACEHOLDER_PREFIX}user_update_api")
@@ -386,7 +388,7 @@ async def pwdReset(session: SessionDep, current_user: CurrentUser, trans: Trans,
     if not current_user.isAdmin:
         raise Exception(trans('i18n_permission.no_permission', url = " patch[/user/pwd/id],", msg = trans('i18n_permission.only_admin')))
     db_user: UserModel = get_db_user(session=session, user_id=id)
-    db_user.password = default_md5_pwd()
+    db_user.password = await run_in_threadpool(default_password_hash)
     session.add(db_user)
 
 @router.put("/pwd", summary=f"{PLACEHOLDER_PREFIX}update_pwd", description=f"{PLACEHOLDER_PREFIX}update_pwd")
@@ -397,13 +399,13 @@ async def pwdUpdate(session: SessionDep, current_user: CurrentUser, trans: Trans
     if not check_pwd_format(new_pwd):
         raise Exception(trans('i18n_format_invalid', key = trans('i18n_user.password')))
     db_user: UserModel = get_db_user(session=session, user_id=current_user.id)
-    if not verify_md5pwd(editor.pwd, db_user.password):
+    if not await run_in_threadpool(verify_password, editor.pwd, db_user.password):
         raise Exception(trans('i18n_error', key = trans('i18n_user.password')))
     if await initial_pwd_disabled(session) and new_pwd == settings.DEFAULT_PWD:
         raise HTTPException(status_code=400, detail=trans('i18n_login.new_pwd_is_initial'))
-    db_user.password = md5pwd(new_pwd)
+    db_user.password = await run_in_threadpool(get_password_hash, new_pwd)
     session.add(db_user)
-    return db_user
+    return {"id": db_user.id}
 
 
 @router.patch("/status", summary=f"{PLACEHOLDER_PREFIX}update_status", description=f"{PLACEHOLDER_PREFIX}update_status")

backend/apps/system/crud/user.py

@@ -1,12 +1,14 @@
 from typing import Optional
 
+from sqlalchemy import update
 from sqlmodel import Session, func, select, delete as sqlmodel_delete
 
 from apps.system.models.system_model import UserWsModel, WorkspaceModel
 from apps.system.schemas.auth import CacheName, CacheNamespace
 from apps.system.schemas.system_schema import EMAIL_REGEX, PWD_REGEX, BaseUserDTO, UserInfoDTO, UserWs
+from common.core.passwords import hash_password, needs_password_upgrade
 from common.core.deps import SessionDep
-from common.core.security import verify_md5pwd
+from common.core.security import verify_password
 from common.core.sqlbot_cache import cache, clear_cache
 from common.utils.locale import I18n, I18nHelper
 from common.utils.utils import SQLBotLogUtil
@@ -45,7 +47,7 @@ def authenticate(*, session: Session, account: str, password: str) -> BaseUserDT
     db_user = get_user_by_account(session=session, account=account)
     if not db_user:
         return None
-    if not verify_md5pwd(password, db_user.password):
+    if not verify_password(password, db_user.password):
         return None
     return db_user
 
@@ -105,3 +107,19 @@ def check_email_format(email: str) -> bool:
 
 def check_pwd_format(pwd: str) -> bool:
     return bool(PWD_REGEX.fullmatch(pwd))
+
+
+def upgrade_password(*, session: Session, user: BaseUserDTO, password: str) -> None:
+    """CAS protects a concurrent reset; the caller owns the transaction.
+
+    Invoke only after all login eligibility checks have passed.
+    """
+    if user.status != 1 or user.origin not in (None, 0) or not user.oid:
+        return
+    if not needs_password_upgrade(user.password):
+        return
+    replacement = hash_password(password)
+    session.execute(
+        update(UserModel).where(UserModel.id == user.id, UserModel.password == user.password)
+        .values(password=replacement).execution_options(synchronize_session=False)
+    )

backend/apps/system/crud/user_excel.py

@@ -1,3 +1,4 @@
+from common.core.security import default_password_hash
 
 
 import asyncio
@@ -7,6 +8,7 @@
 import uuid
 import atexit
 import threading
+from starlette.concurrency import run_in_threadpool
 from fastapi import HTTPException
 from fastapi.responses import StreamingResponse, FileResponse
 import os
@@ -129,6 +131,8 @@ async def batchUpload(session: SessionDep, trans, file) -> UploadResultDTO:
         error_file_id = generate_error_file(error_list, head_list)
     result = UploadResultDTO(successCount=len(success_list), errorCount=len(error_list), dataKey=error_file_id)
     if success_list:
+        for row in success_list:
+            row["password"] = await run_in_threadpool(default_password_hash)
         user_po_list = [UserModel.model_validate(row) for row in success_list]
         session.add_all(user_po_list)
         session.commit()

backend/apps/system/models/user.py

@@ -5,15 +5,15 @@
 from sqlmodel import SQLModel, Field
 
 from common.core.models import SnowflakeBase
-from common.core.security import default_md5_pwd
+from common.core.security import default_password_hash
 from common.utils.time import get_timestamp
 
 
 class BaseUserPO(SQLModel):
     account: str = Field(max_length=255, unique=True)
     oid: int = Field(nullable=False, sa_type=BigInteger(), default=0)
     name: str = Field(max_length=255, unique=True)
-    password: str = Field(default_factory=default_md5_pwd, max_length=255)
+    password: str = Field(default_factory=default_password_hash, max_length=255)
     email: str = Field(max_length=255)
     status: int = Field(default=0, nullable=False)
     origin: int = Field(nullable=False, default=0)