feat: 添加 OAuth 认证支持及相关改进

- 新增 OAuth 模块，支持 LinuxDo/GitHub/Google 等第三方登录
- 用户邮箱改为可选字段，支持无邮箱注册
- 新增模块配置验证状态 (config_validated/config_error)
- 系统设置界面改为分块独立保存
- 用户设置新增 OAuth 绑定管理和首次密码设置
- 登录界面支持 OAuth 按钮展示
- 邮箱验证设置移至邮件设置页面

src/api/admin/modules.py

@@ -27,6 +27,8 @@ class ModuleStatusResponse(BaseModel):
     available: bool
     enabled: bool
     active: bool
+    config_validated: bool
+    config_error: Optional[str]
     display_name: str
     description: str
     category: str
@@ -43,6 +45,8 @@ class ModuleStatusResponse(BaseModel):
             available=status.available,
             enabled=status.enabled,
             active=status.active,
+            config_validated=status.config_validated,
+            config_error=status.config_error,
             display_name=status.display_name,
             description=status.description,
             category=status.category.value,
@@ -180,6 +184,12 @@ class AdminSetModuleEnabledAdapter(AdminApiAdapter):
         except Exception:
             raise InvalidRequestException("请求体格式错误，需要 enabled 字段")
 
+        # 如果是启用模块，必须先通过配置验证
+        if req.enabled:
+            config_validated, config_error = registry.validate_config(self.module_name, context.db)
+            if not config_validated:
+                raise InvalidRequestException(f"模块配置未验证通过: {config_error}")
+
         # 设置启用状态
         registry.set_enabled(self.module_name, req.enabled, context.db)
 

src/api/admin/provider_query.py

@@ -3,7 +3,6 @@ Provider Query API 端点
 用于查询提供商的模型列表等信息
 """
 
-import asyncio
 from typing import Optional
 
 import httpx
@@ -11,14 +10,17 @@ from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlalchemy.orm import Session, joinedload
 
-from src.api.handlers.base.chat_adapter_base import get_adapter_class
-from src.api.handlers.base.cli_adapter_base import get_cli_adapter_class
 from src.config.constants import TimeoutDefaults
 from src.core.crypto import crypto_service
 from src.core.headers import get_extra_headers_from_endpoint
 from src.core.logger import logger
 from src.database.database import get_db
-from src.models.database import Provider, ProviderAPIKey, ProviderEndpoint, User
+from src.models.database import Provider, ProviderEndpoint, User
+from src.services.model.upstream_fetcher import (
+    _get_adapter_for_format,
+    build_all_format_configs,
+    fetch_models_from_endpoints,
+)
 from src.utils.auth_utils import get_current_user
 
 
@@ -50,21 +52,6 @@ class TestModelRequest(BaseModel):
 # ============ API Endpoints ============
 
 
-def _get_adapter_for_format(api_format: str):
-    """根据 API 格式获取对应的 Adapter 类"""
-    # 先检查 Chat Adapter 注册表
-    adapter_class = get_adapter_class(api_format)
-    if adapter_class:
-        return adapter_class
-
-    # 再检查 CLI Adapter 注册表
-    cli_adapter_class = get_cli_adapter_class(api_format)
-    if cli_adapter_class:
-        return cli_adapter_class
-
-    return None
-
-
 @router.post("/models")
 async def query_available_models(
     request: ModelsQueryRequest,
@@ -75,11 +62,7 @@ async def query_available_models(
     查询提供商可用模型
 
     优先从缓存获取（缓存由定时任务刷新），缓存未命中时实时调用上游 API。
-
-    遍历所有活跃端点，根据端点的 API 格式选择正确的 Adapter 进行请求：
-    - OPENAI/OPENAI_CLI: 使用 OpenAIChatAdapter.fetch_models
-    - CLAUDE/CLAUDE_CLI: 使用 ClaudeChatAdapter.fetch_models
-    - GEMINI/GEMINI_CLI: 使用 GeminiChatAdapter.fetch_models
+    从所有 API 格式尝试获取模型，然后聚合去重。
 
     Args:
         request: 查询请求
@@ -107,9 +90,6 @@ async def query_available_models(
         raise HTTPException(status_code=404, detail="Provider not found")
 
     # 如果指定了 api_key_id 且不是强制刷新，优先从缓存获取
-    # 注：不指定 api_key_id 时（Provider 级别查询）不使用缓存，因为：
-    # 1. Provider 级别查询会遍历多个 Key，结果不稳定
-    # 2. 缓存按 Key 粒度存储，与定时任务的刷新逻辑一致
     if request.api_key_id and not request.force_refresh:
         cached_models = await get_upstream_models_from_cache(
             request.provider_id, request.api_key_id
@@ -126,127 +106,42 @@ async def query_available_models(
 
     # 缓存未命中或强制刷新，实时获取
 
-    # 收集所有活跃端点的配置
-    endpoint_configs: list[dict] = []
-
     # 构建 api_format -> endpoint 映射
     format_to_endpoint: dict[str, ProviderEndpoint] = {}
     for endpoint in provider.endpoints:
         if endpoint.is_active:
             format_to_endpoint[endpoint.api_format] = endpoint
 
+    if not format_to_endpoint:
+        raise HTTPException(status_code=400, detail="No active endpoints found for this provider")
+
+    # 获取 API Key
     if request.api_key_id:
-        # 指定了特定的 API Key（从 provider.api_keys 查找）
+        # 指定了特定的 API Key
         api_key = next(
             (key for key in provider.api_keys if key.id == request.api_key_id),
             None
         )
-
         if not api_key:
             raise HTTPException(status_code=404, detail="API Key not found")
-
-        try:
-            api_key_value = crypto_service.decrypt(api_key.api_key)
-        except Exception as e:
-            logger.error(f"Failed to decrypt API key: {e}")
-            raise HTTPException(status_code=500, detail="Failed to decrypt API key")
-
-        # 根据 Key 的 api_formats 找对应的 Endpoint
-        key_formats = api_key.api_formats or []
-        for fmt in key_formats:
-            endpoint = format_to_endpoint.get(fmt)
-            if endpoint:
-                endpoint_configs.append({
-                    "api_key": api_key_value,
-                    "base_url": endpoint.base_url,
-                    "api_format": fmt,
-                    "extra_headers": get_extra_headers_from_endpoint(endpoint),
-                })
-
-        if not endpoint_configs:
-            raise HTTPException(
-                status_code=400,
-                detail="No matching endpoint found for this API Key's formats"
-            )
     else:
-        # 遍历所有活跃端点，为每个端点找一个支持该格式的 Key
-        for endpoint in provider.endpoints:
-            if not endpoint.is_active:
-                continue
-
-            # 找第一个支持该格式的可用 Key
-            for api_key in provider.api_keys:
-                if not api_key.is_active:
-                    continue
-                key_formats = api_key.api_formats or []
-                if endpoint.api_format not in key_formats:
-                    continue
-                try:
-                    api_key_value = crypto_service.decrypt(api_key.api_key)
-                except Exception as e:
-                    logger.error(f"Failed to decrypt API key: {e}")
-                    continue
-                endpoint_configs.append({
-                    "api_key": api_key_value,
-                    "base_url": endpoint.base_url,
-                    "api_format": endpoint.api_format,
-                    "extra_headers": get_extra_headers_from_endpoint(endpoint),
-                })
-                break  # 只取第一个可用的 Key
-
-        if not endpoint_configs:
+        # 使用第一个可用的 Key
+        api_key = next(
+            (key for key in provider.api_keys if key.is_active),
+            None
+        )
+        if not api_key:
             raise HTTPException(status_code=400, detail="No active API Key found for this provider")
 
-    # 并发请求所有端点的模型列表
-    all_models: list = []
-    errors: list[str] = []
+    try:
+        api_key_value = crypto_service.decrypt(api_key.api_key)
+    except Exception as e:
+        logger.error(f"Failed to decrypt API key: {e}")
+        raise HTTPException(status_code=500, detail="Failed to decrypt API key")
 
-    async def fetch_endpoint_models(
-        client: httpx.AsyncClient, config: dict
-    ) -> tuple[list, Optional[str]]:
-        base_url = config["base_url"]
-        if not base_url:
-            return [], None
-        base_url = base_url.rstrip("/")
-        api_format = config["api_format"]
-        api_key_value = config["api_key"]
-        extra_headers = config.get("extra_headers")
-
-        try:
-            # 获取对应的 Adapter 类并调用 fetch_models
-            adapter_class = _get_adapter_for_format(api_format)
-            if not adapter_class:
-                return [], f"Unknown API format: {api_format}"
-            models, error = await adapter_class.fetch_models(
-                client, base_url, api_key_value, extra_headers
-            )
-            # 确保所有模型都有 api_format 字段
-            for m in models:
-                if "api_format" not in m:
-                    m["api_format"] = api_format
-            return models, error
-        except Exception as e:
-            logger.error(f"Error fetching models from {api_format} endpoint: {e}")
-            return [], f"{api_format}: {str(e)}"
-
-    # 限制并发请求数量，避免触发上游速率限制
-    MAX_CONCURRENT_REQUESTS = 5
-    semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
-
-    async def fetch_with_semaphore(
-        client: httpx.AsyncClient, config: dict
-    ) -> tuple[list, Optional[str]]:
-        async with semaphore:
-            return await fetch_endpoint_models(client, config)
-
-    async with httpx.AsyncClient(timeout=30.0) as client:
-        results = await asyncio.gather(
-            *[fetch_with_semaphore(client, c) for c in endpoint_configs]
-        )
-        for models, error in results:
-            all_models.extend(models)
-            if error:
-                errors.append(error)
+    # 使用公共函数构建所有格式的端点配置并获取模型
+    endpoint_configs = build_all_format_configs(api_key_value, format_to_endpoint)  # type: ignore[arg-type]
+    all_models, errors, has_success = await fetch_models_from_endpoints(endpoint_configs)
 
     # 按 model id + api_format 去重（保留第一个）
     seen_keys: set[str] = set()

src/api/admin/system.py

@@ -1486,8 +1486,17 @@ class AdminImportUsersAdapter(AdminApiAdapter):
                     stats["users"]["skipped"] += 1
                     continue
 
+                # 导入必须有邮箱（email 是导入的主键）
+                import_email = user_data.get("email")
+                if not import_email:
+                    stats["errors"].append(
+                        f"跳过无邮箱用户: {user_data.get('username', '未知')}"
+                    )
+                    stats["users"]["skipped"] += 1
+                    continue
+
                 existing_user = (
-                    db.query(User).filter(User.email == user_data["email"]).first()
+                    db.query(User).filter(User.email == import_email).first()
                 )
 
                 if existing_user:
@@ -1496,7 +1505,7 @@ class AdminImportUsersAdapter(AdminApiAdapter):
                         stats["users"]["skipped"] += 1
                     elif merge_mode == "error":
                         raise InvalidRequestException(
-                            f"用户 '{user_data['email']}' 已存在"
+                            f"用户 '{import_email}' 已存在"
                         )
                     elif merge_mode == "overwrite":
                         # 更新现有用户
@@ -1527,8 +1536,9 @@ class AdminImportUsersAdapter(AdminApiAdapter):
 
                     new_user = User(
                         id=str(uuid.uuid4()),
-                        email=user_data["email"],
-                        username=user_data.get("username", user_data["email"].split("@")[0]),
+                        email=import_email,
+                        email_verified=user_data.get("email_verified", True),
+                        username=user_data.get("username") or import_email.split("@")[0],
                         password_hash=user_data.get("password_hash", ""),
                         role=role,
                         allowed_providers=user_data.get("allowed_providers"),

src/api/announcements/routes.py

@@ -266,8 +266,21 @@ class AnnouncementOptionalAuthAdapter(ApiAdapter):
             if not user_id:
                 return None
             user = (
-                context.db.query(User).filter(User.id == user_id, User.is_active.is_(True)).first()
+                context.db.query(User)
+                .filter(
+                    User.id == user_id,
+                    User.is_active.is_(True),
+                    User.is_deleted.is_(False),
+                )
+                .first()
             )
+
+            if not user:
+                return None
+
+            if not AuthService.token_identity_matches_user(payload, user):
+                return None
+
             return user
         except Exception:
             return None

src/api/auth/routes.py

@@ -299,13 +299,12 @@ class AuthLoginAdapter(AuthPublicAdapter):
         access_token = AuthService.create_access_token(
             data={
                 "user_id": user.id,
-                "email": user.email,
                 "role": user.role.value,
                 "created_at": user.created_at.isoformat() if user.created_at else None,
             }
         )
         refresh_token = AuthService.create_refresh_token(
-            data={"user_id": user.id, "email": user.email}
+            data={"user_id": user.id, "created_at": user.created_at.isoformat() if user.created_at else None}
         )
         response = LoginResponse(
             access_token=access_token,
@@ -345,19 +344,23 @@ class AuthRefreshAdapter(AuthPublicAdapter):
                 )
             if not user.is_active:
                 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="用户已禁用")
+            if user.is_deleted:
+                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="用户不存在或已禁用")
+
+            if not AuthService.token_identity_matches_user(token_payload, user):
+                raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的刷新令牌")
 
             new_access_token = AuthService.create_access_token(
                 data={
                     "user_id": user.id,
-                    "email": user.email,
                     "role": user.role.value,
                     "created_at": user.created_at.isoformat() if user.created_at else None,
                 }
             )
             new_refresh_token = AuthService.create_refresh_token(
-                data={"user_id": user.id, "email": user.email}
+                data={"user_id": user.id, "created_at": user.created_at.isoformat() if user.created_at else None}
             )
-            logger.info(f"令牌刷新成功: {user.email}")
+            logger.info(f"令牌刷新成功: user_id={user.id}")
             return RefreshTokenResponse(
                 access_token=new_access_token,
                 refresh_token=new_refresh_token,
@@ -378,10 +381,16 @@ class AuthRegistrationSettingsAdapter(AuthPublicAdapter):
 
         enable_registration = SystemConfigService.get_config(db, "enable_registration", default=False)
         require_verification = SystemConfigService.get_config(db, "require_email_verification", default=False)
+        email_configured = EmailSenderService.is_smtp_configured(db)
+
+        # 如果邮箱服务未配置，强制 require_email_verification 为 False
+        if not email_configured:
+            require_verification = False
 
         return RegistrationSettingsResponse(
             enable_registration=bool(enable_registration),
             require_email_verification=bool(require_verification),
+            email_configured=email_configured,
         ).model_dump()
 
 
@@ -430,61 +439,78 @@ class AuthRegisterAdapter(AuthPublicAdapter):
             AuditService.log_event(
                 db=db,
                 event_type=AuditEventType.UNAUTHORIZED_ACCESS,
-                description=f"Registration attempt rejected - registration disabled: {register_request.email}",
+                description=f"Registration attempt rejected - registration disabled: {register_request.username}",
                 ip_address=client_ip,
                 user_agent=user_agent,
-                metadata={"email": register_request.email, "reason": "registration_disabled"},
+                metadata={"username": register_request.username, "reason": "registration_disabled"},
             )
             db.commit()
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="系统暂不开放注册")
 
-        # 检查邮箱后缀是否允许
-        suffix_allowed, suffix_error = validate_email_suffix(db, register_request.email)
-        if not suffix_allowed:
-            logger.warning(f"注册失败：邮箱后缀不允许: {register_request.email}")
-            AuditService.log_event(
-                db=db,
-                event_type=AuditEventType.UNAUTHORIZED_ACCESS,
-                description=f"Registration attempt rejected - email suffix not allowed: {register_request.email}",
-                ip_address=client_ip,
-                user_agent=user_agent,
-                metadata={"email": register_request.email, "reason": "email_suffix_not_allowed"},
-            )
-            db.commit()
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail=suffix_error,
-            )
-
-        # 检查是否需要邮箱验证
+        email = register_request.email
+        email_configured = EmailSenderService.is_smtp_configured(db)
         require_verification = SystemConfigService.get_config(db, "require_email_verification", default=False)
 
+        # 如果邮箱服务未配置，强制不要求邮箱验证
+        if not email_configured:
+            require_verification = False
+
+        # 如果系统要求邮箱验证，则必须提供邮箱
         if require_verification:
+            if not email:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="系统要求邮箱验证，请填写邮箱",
+                )
             # 检查邮箱是否已验证
-            is_verified = await EmailVerificationService.is_email_verified(register_request.email)
+            is_verified = await EmailVerificationService.is_email_verified(email)
             if not is_verified:
-                logger.warning(f"注册失败：邮箱未验证: {register_request.email}")
+                logger.warning(f"注册失败：邮箱未验证: {email}")
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST,
                     detail="请先完成邮箱验证。请发送验证码并验证后再注册。",
                 )
 
+        # 如果提供了邮箱，进行后缀验证
+        if email:
+            suffix_allowed, suffix_error = validate_email_suffix(db, email)
+            if not suffix_allowed:
+                logger.warning(f"注册失败：邮箱后缀不允许: {email}")
+                AuditService.log_event(
+                    db=db,
+                    event_type=AuditEventType.UNAUTHORIZED_ACCESS,
+                    description=f"Registration attempt rejected - email suffix not allowed: {email}",
+                    ip_address=client_ip,
+                    user_agent=user_agent,
+                    metadata={"email": email, "reason": "email_suffix_not_allowed"},
+                )
+                db.commit()
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=suffix_error,
+                )
+
         try:
             # 读取系统配置的默认配额
             default_quota = SystemConfigService.get_config(db, "default_user_quota_usd", default=10.0)
 
+            # email_verified 逻辑：
+            # - 要求邮箱验证且已通过验证：True
+            # - 提供了邮箱但不要求验证：False（用户可后续自行验证）
+            # - 未提供邮箱：False
             user = UserService.create_user(
                 db=db,
-                email=register_request.email,
+                email=email,  # 可以为 None
                 username=register_request.username,
                 password=register_request.password,
                 role=UserRole.USER,
                 quota_usd=default_quota,
+                email_verified=bool(require_verification and email),
             )
             AuditService.log_event(
                 db=db,
                 event_type=AuditEventType.USER_CREATED,
-                description=f"User registered: {user.email}",
+                description=f"User registered: {user.username}" + (f" ({user.email})" if user.email else ""),
                 user_id=user.id,
                 ip_address=client_ip,
                 user_agent=user_agent,
@@ -494,9 +520,9 @@ class AuthRegisterAdapter(AuthPublicAdapter):
             db.commit()
 
             # 注册成功后清除验证状态（在 commit 后清理，即使清理失败也不影响注册结果）
-            if require_verification:
+            if require_verification and email:
                 try:
-                    await EmailVerificationService.clear_verification(register_request.email)
+                    await EmailVerificationService.clear_verification(email)
                 except Exception as e:
                     logger.warning(f"清理验证状态失败: {e}")
 
@@ -510,10 +536,10 @@ class AuthRegisterAdapter(AuthPublicAdapter):
             AuditService.log_event(
                 db=db,
                 event_type=AuditEventType.UNAUTHORIZED_ACCESS,
-                description=f"Registration failed: {register_request.email} - {exc}",
+                description=f"Registration failed: {register_request.username} - {exc}",
                 ip_address=client_ip,
                 user_agent=user_agent,
-                metadata={"email": register_request.email, "error": str(exc)},
+                metadata={"username": register_request.username, "error": str(exc)},
             )
             db.commit()
             raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))

src/api/base/pipeline.py

@@ -238,9 +238,12 @@ class ApiRequestPipeline:
 
         # 直接查询数据库，确保返回的是当前 Session 绑定的对象
         user = db.query(User).filter(User.id == user_id).first()
-        if not user or not user.is_active:
+        if not user or not user.is_active or user.is_deleted:
             raise HTTPException(status_code=403, detail="用户不存在或已禁用")
 
+        if not self.auth_service.token_identity_matches_user(payload, user):
+            raise HTTPException(status_code=403, detail="无效的管理员令牌")
+
         # 检查管理员权限
         if user.role != UserRole.ADMIN:
             logger.warning(f"非管理员尝试通过 JWT 访问管理端点: {user.email}")
@@ -291,9 +294,12 @@ class ApiRequestPipeline:
             raise HTTPException(status_code=401, detail="无效的用户令牌")
 
         user = db.query(User).filter(User.id == user_id).first()
-        if not user or not user.is_active:
+        if not user or not user.is_active or user.is_deleted:
             raise HTTPException(status_code=403, detail="用户不存在或已禁用")
 
+        if not self.auth_service.token_identity_matches_user(payload, user):
+            raise HTTPException(status_code=403, detail="无效的用户令牌")
+
         request.state.user_id = user.id
         return user, None
 

src/api/oauth/__init__.py

@@ -0,0 +1,15 @@
+"""OAuth API 路由聚合。"""
+
+from fastapi import APIRouter
+
+from src.api.oauth.admin import router as admin_router
+from src.api.oauth.public import router as public_router
+from src.api.oauth.user import router as user_router
+
+router = APIRouter()
+router.include_router(public_router)
+router.include_router(user_router)
+router.include_router(admin_router)
+
+__all__ = ["router"]
+

src/api/oauth/admin.py

@@ -0,0 +1,264 @@
+"""OAuth 管理端点（管理员）。"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, Request
+from pydantic import BaseModel, Field, ValidationError
+from sqlalchemy.orm import Session
+
+from src.api.base.admin_adapter import AdminApiAdapter
+from src.api.base.context import ApiRequestContext
+from src.api.base.pipeline import ApiRequestPipeline
+from src.core.exceptions import InvalidRequestException, translate_pydantic_error
+from src.database import get_db
+from src.models.database import OAuthProvider
+from src.services.auth.oauth.registry import get_oauth_provider_registry
+from src.services.auth.oauth.service import OAuthService
+
+router = APIRouter(prefix="/api/admin/oauth", tags=["Admin - OAuth"])
+pipeline = ApiRequestPipeline()
+
+
+class SupportedOAuthType(BaseModel):
+    provider_type: str
+    display_name: str
+    default_authorization_url: str
+    default_token_url: str
+    default_userinfo_url: str
+    default_scopes: List[str]
+
+
+class OAuthProviderUpsertRequest(BaseModel):
+    display_name: str = Field(..., min_length=1, max_length=100)
+    client_id: str = Field(..., min_length=1, max_length=255)
+    client_secret: Optional[str] = Field(None, max_length=2048)
+
+    authorization_url_override: Optional[str] = Field(None, max_length=500)
+    token_url_override: Optional[str] = Field(None, max_length=500)
+    userinfo_url_override: Optional[str] = Field(None, max_length=500)
+    scopes: Optional[List[str]] = None
+
+    redirect_uri: str = Field(..., min_length=1, max_length=500)
+    frontend_callback_url: str = Field(..., min_length=1, max_length=500)
+
+    attribute_mapping: Optional[Dict[str, Any]] = None
+    extra_config: Optional[Dict[str, Any]] = None
+
+    is_enabled: bool = False
+    force: bool = False
+
+
+class OAuthProviderAdminResponse(BaseModel):
+    provider_type: str
+    display_name: str
+    client_id: str
+    has_secret: bool
+    authorization_url_override: Optional[str] = None
+    token_url_override: Optional[str] = None
+    userinfo_url_override: Optional[str] = None
+    scopes: Optional[List[str]] = None
+    redirect_uri: str
+    frontend_callback_url: str
+    attribute_mapping: Optional[Dict[str, Any]] = None
+    extra_config: Optional[Dict[str, Any]] = None
+    is_enabled: bool
+
+
+class OAuthProviderTestResponse(BaseModel):
+    authorization_url_reachable: bool
+    token_url_reachable: bool
+    secret_status: str
+    details: str = ""
+
+
+class OAuthProviderTestRequest(BaseModel):
+    """测试请求，使用表单数据而非数据库配置"""
+
+    client_id: str = Field(..., min_length=1)
+    client_secret: Optional[str] = None
+    authorization_url_override: Optional[str] = None
+    token_url_override: Optional[str] = None
+    redirect_uri: str = Field(..., min_length=1)
+
+
+@router.get("/supported-types", response_model=List[SupportedOAuthType])
+async def get_supported_types(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = GetSupportedTypesAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.get("/providers", response_model=List[OAuthProviderAdminResponse])
+async def list_provider_configs(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = ListOAuthProviderConfigsAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.get("/providers/{provider_type}", response_model=OAuthProviderAdminResponse)
+async def get_provider_config(provider_type: str, request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = GetOAuthProviderConfigAdapter(provider_type=provider_type)
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.put("/providers/{provider_type}", response_model=OAuthProviderAdminResponse)
+async def upsert_provider_config(provider_type: str, request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = UpsertOAuthProviderConfigAdapter(provider_type=provider_type)
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.delete("/providers/{provider_type}")
+async def delete_provider_config(provider_type: str, request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = DeleteOAuthProviderConfigAdapter(provider_type=provider_type)
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.post("/providers/{provider_type}/test", response_model=OAuthProviderTestResponse)
+async def test_provider_config(provider_type: str, request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = TestOAuthProviderConfigAdapter(provider_type=provider_type)
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+class GetSupportedTypesAdapter(AdminApiAdapter):
+    async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
+        registry = get_oauth_provider_registry()
+        types = registry.get_supported_types()
+        return [
+            SupportedOAuthType(
+                provider_type=t.provider_type,
+                display_name=t.display_name,
+                default_authorization_url=t.default_authorization_url,
+                default_token_url=t.default_token_url,
+                default_userinfo_url=t.default_userinfo_url,
+                default_scopes=list(t.default_scopes),
+            ).model_dump()
+            for t in types
+        ]
+
+
+class ListOAuthProviderConfigsAdapter(AdminApiAdapter):
+    async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
+        rows = context.db.query(OAuthProvider).order_by(OAuthProvider.provider_type.asc()).all()
+        return [
+            OAuthProviderAdminResponse(
+                provider_type=str(row.provider_type or ""),
+                display_name=str(row.display_name or ""),
+                client_id=str(row.client_id or ""),
+                has_secret=bool(row.client_secret_encrypted),
+                authorization_url_override=row.authorization_url_override,
+                token_url_override=row.token_url_override,
+                userinfo_url_override=row.userinfo_url_override,
+                scopes=row.scopes,
+                redirect_uri=str(row.redirect_uri or ""),
+                frontend_callback_url=str(row.frontend_callback_url or ""),
+                attribute_mapping=row.attribute_mapping,
+                extra_config=row.extra_config,
+                is_enabled=bool(row.is_enabled),
+            ).model_dump()
+            for row in rows
+        ]
+
+
+class GetOAuthProviderConfigAdapter(AdminApiAdapter):
+    def __init__(self, provider_type: str):
+        self.provider_type = provider_type
+
+    async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
+        row = context.db.query(OAuthProvider).filter(OAuthProvider.provider_type == self.provider_type).first()
+        if not row:
+            raise InvalidRequestException("Provider 配置不存在")
+        return OAuthProviderAdminResponse(
+            provider_type=str(row.provider_type or ""),
+            display_name=str(row.display_name or ""),
+            client_id=str(row.client_id or ""),
+            has_secret=bool(row.client_secret_encrypted),
+            authorization_url_override=row.authorization_url_override,
+            token_url_override=row.token_url_override,
+            userinfo_url_override=row.userinfo_url_override,
+            scopes=row.scopes,
+            redirect_uri=str(row.redirect_uri or ""),
+            frontend_callback_url=str(row.frontend_callback_url or ""),
+            attribute_mapping=row.attribute_mapping,
+            extra_config=row.extra_config,
+            is_enabled=bool(row.is_enabled),
+        ).model_dump()
+
+
+class UpsertOAuthProviderConfigAdapter(AdminApiAdapter):
+    def __init__(self, provider_type: str):
+        self.provider_type = provider_type
+
+    async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
+        payload = context.ensure_json_body()
+        try:
+            req = OAuthProviderUpsertRequest.model_validate(payload)
+        except ValidationError as exc:
+            errors = exc.errors()
+            if errors:
+                raise InvalidRequestException(translate_pydantic_error(errors[0]))
+            raise InvalidRequestException("请求数据验证失败")
+
+        row = await OAuthService.upsert_provider_config(
+            db=context.db,
+            provider_type=self.provider_type,
+            data=req,
+        )
+
+        return OAuthProviderAdminResponse(
+            provider_type=str(row.provider_type or ""),
+            display_name=str(row.display_name or ""),
+            client_id=str(row.client_id or ""),
+            has_secret=bool(row.client_secret_encrypted),
+            authorization_url_override=row.authorization_url_override,
+            token_url_override=row.token_url_override,
+            userinfo_url_override=row.userinfo_url_override,
+            scopes=row.scopes,
+            redirect_uri=str(row.redirect_uri or ""),
+            frontend_callback_url=str(row.frontend_callback_url or ""),
+            attribute_mapping=row.attribute_mapping,
+            extra_config=row.extra_config,
+            is_enabled=bool(row.is_enabled),
+        ).model_dump()
+
+
+class DeleteOAuthProviderConfigAdapter(AdminApiAdapter):
+    def __init__(self, provider_type: str):
+        self.provider_type = provider_type
+
+    async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
+        await OAuthService.delete_provider_config(context.db, self.provider_type)
+        return {"message": "删除成功"}
+
+
+class TestOAuthProviderConfigAdapter(AdminApiAdapter):
+    def __init__(self, provider_type: str):
+        self.provider_type = provider_type
+
+    async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
+        payload = context.ensure_json_body()
+        try:
+            req = OAuthProviderTestRequest.model_validate(payload)
+        except ValidationError as exc:
+            errors = exc.errors()
+            if errors:
+                raise InvalidRequestException(translate_pydantic_error(errors[0]))
+            raise InvalidRequestException("请求数据验证失败")
+
+        # 如果没有提供 client_secret，尝试从数据库获取已保存的
+        client_secret = req.client_secret
+        if not client_secret:
+            existing = context.db.query(OAuthProvider).filter(
+                OAuthProvider.provider_type == self.provider_type
+            ).first()
+            if existing and existing.client_secret_encrypted:
+                client_secret = existing.get_client_secret()
+
+        result = await OAuthService.test_provider_config_with_data(
+            provider_type=self.provider_type,
+            client_id=req.client_id,
+            client_secret=client_secret,
+            authorization_url_override=req.authorization_url_override,
+            token_url_override=req.token_url_override,
+            redirect_uri=req.redirect_uri,
+        )
+        return OAuthProviderTestResponse(**result).model_dump()

src/api/oauth/public.py

@@ -0,0 +1,59 @@
+"""OAuth 公开端点（无需登录）。"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from fastapi import APIRouter, Depends, Query, status
+from sqlalchemy.orm import Session
+from starlette.responses import RedirectResponse
+
+from src.database import get_db
+from src.services.auth.oauth.service import OAuthService
+
+router = APIRouter(prefix="/api/oauth", tags=["OAuth"])
+
+
+@router.get("/providers")
+async def list_oauth_providers(db: Session = Depends(get_db)) -> dict[str, Any]:
+    """
+    获取可用 OAuth Providers 列表。
+
+    模块未启用时返回空列表（前端友好）。
+    """
+    providers = await OAuthService.list_public_providers(db)
+    return {"providers": providers}
+
+
+@router.get("/{provider_type}/authorize")
+async def oauth_authorize(provider_type: str, db: Session = Depends(get_db)) -> RedirectResponse:
+    """
+    发起 OAuth 登录（login flow）。
+    """
+    url = await OAuthService.build_login_authorize_url(db, provider_type)
+    return RedirectResponse(url=url, status_code=status.HTTP_302_FOUND)
+
+
+@router.get("/{provider_type}/callback")
+async def oauth_callback(
+    provider_type: str,
+    db: Session = Depends(get_db),
+    code: Optional[str] = Query(None),
+    state: Optional[str] = Query(None),
+    error: Optional[str] = Query(None),
+    error_description: Optional[str] = Query(None),
+) -> RedirectResponse:
+    """
+    OAuth 回调端点。
+
+    成功/失败都会重定向到前端回调页。
+    """
+    redirect_url = await OAuthService.handle_callback(
+        db=db,
+        provider_type=provider_type,
+        state=state or "",
+        code=code,
+        error=error,
+        error_description=error_description,
+    )
+    return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)

src/api/oauth/user.py

@@ -0,0 +1,84 @@
+"""OAuth 用户端点（需登录）。"""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+from fastapi import APIRouter, Depends, Request, status
+from sqlalchemy.orm import Session
+from starlette.responses import RedirectResponse
+
+from src.api.base.authenticated_adapter import AuthenticatedApiAdapter
+from src.api.base.context import ApiRequestContext
+from src.api.base.pipeline import ApiRequestPipeline
+from src.database import get_db
+from src.services.auth.oauth.service import OAuthService
+
+router = APIRouter(prefix="/api/user/oauth", tags=["User - OAuth"])
+pipeline = ApiRequestPipeline()
+
+
+@router.get("/bindable-providers")
+async def list_bindable_providers(request: Request, db: Session = Depends(get_db)) -> dict[str, Any]:
+    adapter = ListBindableProvidersAdapter()
+    result = await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+    return cast(dict[str, Any], result)
+
+
+@router.get("/links")
+async def list_my_oauth_links(request: Request, db: Session = Depends(get_db)) -> dict[str, Any]:
+    adapter = ListMyOAuthLinksAdapter()
+    result = await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+    return cast(dict[str, Any], result)
+
+
+@router.get("/{provider_type}/bind")
+async def bind_oauth_provider(
+    provider_type: str, request: Request, db: Session = Depends(get_db)
+) -> RedirectResponse:
+    adapter = BindOAuthProviderAdapter(provider_type=provider_type)
+    result = await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+    return cast(RedirectResponse, result)
+
+
+@router.delete("/{provider_type}")
+async def unbind_oauth_provider(
+    provider_type: str, request: Request, db: Session = Depends(get_db)
+) -> dict[str, Any]:
+    adapter = UnbindOAuthProviderAdapter(provider_type=provider_type)
+    result = await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+    return cast(dict[str, Any], result)
+
+
+class ListBindableProvidersAdapter(AuthenticatedApiAdapter):
+    async def handle(self, context: ApiRequestContext) -> dict[str, Any]:  # type: ignore[override]
+        assert context.user is not None
+        providers = await OAuthService.list_bindable_providers(context.db, context.user)
+        return {"providers": providers}
+
+
+class ListMyOAuthLinksAdapter(AuthenticatedApiAdapter):
+    async def handle(self, context: ApiRequestContext) -> dict[str, Any]:  # type: ignore[override]
+        assert context.user is not None
+        links = await OAuthService.list_user_links(context.db, context.user)
+        return {"links": links}
+
+
+class BindOAuthProviderAdapter(AuthenticatedApiAdapter):
+    def __init__(self, provider_type: str):
+        self.provider_type = provider_type
+
+    async def handle(self, context: ApiRequestContext) -> RedirectResponse:  # type: ignore[override]
+        assert context.user is not None
+        url = await OAuthService.build_bind_authorize_url(context.db, context.user, self.provider_type)
+        return RedirectResponse(url=url, status_code=status.HTTP_302_FOUND)
+
+
+class UnbindOAuthProviderAdapter(AuthenticatedApiAdapter):
+    def __init__(self, provider_type: str):
+        self.provider_type = provider_type
+
+    async def handle(self, context: ApiRequestContext) -> dict[str, Any]:  # type: ignore[override]
+        assert context.user is not None
+        await OAuthService.unbind_provider(context.db, context.user, self.provider_type)
+        return {"message": "解绑成功"}

src/api/user_me/routes.py

@@ -446,16 +446,31 @@ class ChangePasswordAdapter(AuthenticatedApiAdapter):
                 raise InvalidRequestException(translate_pydantic_error(errors[0]))
             raise InvalidRequestException("请求数据验证失败")
 
-        if not user.verify_password(request.old_password):
-            raise InvalidRequestException("旧密码错误")
+        # LDAP 用户不能修改密码
+        from src.core.enums import AuthSource
+        if user.auth_source == AuthSource.LDAP:
+            raise ForbiddenException("LDAP 用户不能在此修改密码")
+
+        # 判断用户是否已有密码
+        has_password = bool(user.password_hash)
+
+        if has_password:
+            # 已有密码：需要验证旧密码
+            if not request.old_password:
+                raise InvalidRequestException("请输入当前密码")
+            if not user.verify_password(request.old_password):
+                raise InvalidRequestException("旧密码错误")
+        # 无密码（如 OAuth 用户首次设置）：无需旧密码
+
         if len(request.new_password) < 6:
             raise InvalidRequestException("密码长度至少6位")
 
         user.set_password(request.new_password)
         user.updated_at = datetime.now(timezone.utc)
         db.commit()
-        logger.info(f"用户修改密码: {user.email}")
-        return {"message": "密码修改成功"}
+        action = "修改" if has_password else "设置"
+        logger.info(f"用户{action}密码: {user.email}")
+        return {"message": f"密码{action}成功"}
 
 
 class ListMyApiKeysAdapter(AuthenticatedApiAdapter):

src/core/enums.py

@@ -37,3 +37,4 @@ class AuthSource(str, Enum):
 
     LOCAL = "local"  # 本地认证
     LDAP = "ldap"  # LDAP 认证
+    OAUTH = "oauth"  # OAuth 认证（账号首创来源）

src/core/exceptions.py

@@ -81,6 +81,13 @@ FIELD_NAME_TRANSLATIONS = {
     "is_pinned": "置顶状态",
     "start_time": "开始时间",
     "end_time": "结束时间",
+    # OAuth 相关字段
+    "client_id": "Client ID",
+    "client_secret": "Client Secret",
+    "redirect_uri": "回调地址",
+    "frontend_callback_url": "前端回调地址",
+    "display_name": "显示名称",
+    "scopes": "授权范围",
 }
 
 
@@ -340,6 +347,18 @@ class NotFoundException(ProxyException):
         )
 
 
+class ConfirmationRequiredException(ProxyException):
+    """需要用户确认的操作"""
+
+    def __init__(self, message: str, affected_count: int, action: str = "disable"):
+        super().__init__(
+            status_code=status.HTTP_409_CONFLICT,
+            error_type="confirmation_required",
+            message=message,
+            details={"affected_count": affected_count, "action": action},
+        )
+
+
 class ForbiddenException(ProxyException):
     """权限不足"""
 

src/core/modules/base.py

@@ -6,10 +6,11 @@
 
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Optional
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Optional, Tuple
 
 if TYPE_CHECKING:
     from fastapi import APIRouter
+    from sqlalchemy.orm import Session
 
 
 class ModuleCategory(str, Enum):
@@ -84,6 +85,9 @@ class ModuleDefinition:
     # 自定义依赖检测（可选，用于检测 ldap3 等库是否安装）
     check_dependencies: Optional[Callable[[], bool]] = None
 
+    # 配置验证（可选，启用模块时调用，返回 (success, error_message)）
+    validate_config: Optional[Callable[["Session"], Tuple[bool, str]]] = None
+
 
 @dataclass
 class ModuleStatus:
@@ -97,6 +101,8 @@ class ModuleStatus:
     available: bool  # 部署级可用（环境变量 + 依赖库）
     enabled: bool  # 运行级启用（数据库配置）
     active: bool  # 最终激活状态 (available && enabled && dependencies_ok)
+    config_validated: bool  # 配置验证通过（只有验证通过才允许启用）
+    config_error: Optional[str]  # 配置验证失败的错误信息
 
     # 显示信息
     display_name: str

src/core/modules/registry.py

@@ -175,6 +175,34 @@ class ModuleRegistry:
 
         return True
 
+    # ========== 配置验证 ==========
+
+    def validate_config(self, name: str, db: "Session") -> tuple[bool, str]:
+        """
+        验证模块配置是否有效
+
+        Args:
+            name: 模块名称
+            db: 数据库会话
+
+        Returns:
+            (validated, error_message) - validated 为 True 表示配置有效
+        """
+        if name not in self._modules:
+            return False, "模块不存在"
+
+        module = self._modules[name]
+
+        # 没有配置验证函数的模块，默认配置有效
+        if not module.validate_config:
+            return True, ""
+
+        try:
+            return module.validate_config(db)
+        except Exception as e:
+            logger.warning(f"Module [{name}] config validation error: {e}")
+            return False, f"配置验证出错: {str(e)}"
+
     # ========== 状态查询 ==========
 
     def get_module_status(
@@ -195,11 +223,28 @@ class ModuleRegistry:
         meta = module.metadata
         available = self.is_available(name)
 
+        # 获取配置验证状态
+        config_validated = False
+        config_error: Optional[str] = None
+        if available:
+            config_validated, config_error = self.validate_config(name, db)
+            if config_validated:
+                config_error = None  # 验证通过时清空错误信息
+
+        # 获取启用状态
+        enabled = self.is_enabled(name, db) if available else False
+
+        # 注意：配置验证失败时不自动禁用模块
+        # 自动禁用会在查询方法中产生写操作副作用，违反幂等性原则
+        # 配置验证状态通过 config_validated/config_error 字段返回，由调用方决定如何处理
+
         return ModuleStatus(
             name=name,
             available=available,
-            enabled=self.is_enabled(name, db) if available else False,
+            enabled=enabled,
             active=self.is_active(name, db) if available else False,
+            config_validated=config_validated,
+            config_error=config_error,
             display_name=meta.display_name,
             description=meta.description,
             category=meta.category,

src/database/database.py

@@ -364,6 +364,7 @@ def init_admin_user(db: Session) -> None:
         # 创建管理员账户
         admin = User(
             email=config.admin_email,
+            email_verified=True,
             username=config.admin_username,
             role=UserRole.ADMIN,
             is_active=True,

src/models/api.py

@@ -55,7 +55,7 @@ class LoginResponse(BaseModel):
     token_type: str = "bearer"
     expires_in: int = 86400  # Token有效期（秒），默认24小时
     user_id: str
-    email: str
+    email: Optional[str] = None
     username: str
     role: str
 
@@ -78,14 +78,19 @@ class RefreshTokenResponse(BaseModel):
 class RegisterRequest(BaseModel):
     """注册请求"""
 
-    email: str = Field(..., min_length=3, max_length=255, description="邮箱地址")
+    email: Optional[str] = Field(None, max_length=255, description="邮箱地址（可选）")
     username: str = Field(..., min_length=2, max_length=50, description="用户名")
     password: str = Field(..., min_length=6, max_length=128, description="密码")
 
-    @classmethod
     @field_validator("email")
+    @classmethod
     def validate_email(cls, v):
-        """验证邮箱格式"""
+        """验证邮箱格式（如果提供）"""
+        if v is None:
+            return None
+        v = v.strip()
+        if not v:
+            return None
         email_pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
         if not re.match(email_pattern, v):
             raise ValueError("邮箱格式无效")
@@ -121,7 +126,7 @@ class RegisterResponse(BaseModel):
     """注册响应"""
 
     user_id: str
-    email: str
+    email: Optional[str] = None
     username: str
     message: str
 
@@ -223,6 +228,7 @@ class RegistrationSettingsResponse(BaseModel):
 
     enable_registration: bool
     require_email_verification: bool
+    email_configured: bool = Field(description="是否配置了邮箱服务")
 
 
 # ========== 用户管理 ==========
@@ -335,7 +341,7 @@ class UserResponse(BaseModel):
     """用户响应"""
 
     id: str
-    email: str
+    email: Optional[str] = None
     username: str
     role: UserRole
     allowed_providers: Optional[List[str]] = None  # 允许使用的提供商 ID 列表
@@ -699,7 +705,7 @@ class UpdatePreferencesRequest(BaseModel):
 class ChangePasswordRequest(BaseModel):
     """修改密码请求"""
 
-    old_password: str
+    old_password: Optional[str] = None  # 可选：首次设置密码时不需要
     new_password: str
 
 

src/models/database.py

@@ -42,9 +42,13 @@ class User(Base):
     __tablename__ = "users"
 
     id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()), index=True)
-    email = Column(String(255), unique=True, index=True, nullable=False)
+    # OAuth 用户可能没有邮箱；Postgres unique 允许多个 NULL
+    email = Column(String(255), unique=True, index=True, nullable=True)
+    # 注意：所有创建用户的入口必须显式写入 true/false，禁止依赖默认值
+    email_verified = Column(Boolean, nullable=False)
     username = Column(String(100), unique=True, index=True, nullable=False)
-    password_hash = Column(String(255), nullable=False)
+    # OAuth 用户可能没有本地密码（v1 仅做字段兼容）
+    password_hash = Column(String(255), nullable=True)
     role = Column(
         Enum(
             UserRole,
@@ -509,6 +513,95 @@ class LDAPConfig(Base):
         return crypto_service.decrypt(self.bind_password_encrypted)
 
 
+class OAuthProvider(Base):
+    """OAuth Provider 配置表（按 provider_type 唯一）"""
+
+    __tablename__ = "oauth_providers"
+
+    # 使用 provider_type 作为主键，便于通过 URL 参数直接定位配置
+    provider_type = Column(String(50), primary_key=True)
+    display_name = Column(String(100), nullable=False)
+
+    client_id = Column(String(255), nullable=False)
+    client_secret_encrypted = Column(Text, nullable=True)  # 允许 NULL 表示尚未配置/已清除
+
+    # 可选覆盖端点（需在业务层做白名单校验）
+    authorization_url_override = Column(String(500), nullable=True)
+    token_url_override = Column(String(500), nullable=True)
+    userinfo_url_override = Column(String(500), nullable=True)
+
+    # 可选覆盖 scopes（JSON 列表）
+    scopes = Column(JSON, nullable=True)
+
+    # 服务端控制 redirect_uri 与前端回调 URL
+    redirect_uri = Column(String(500), nullable=False)
+    frontend_callback_url = Column(String(500), nullable=False)
+
+    # Provider 特定配置/映射
+    attribute_mapping = Column(JSON, nullable=True)
+    extra_config = Column(JSON, nullable=True)
+
+    is_enabled = Column(Boolean, default=False, nullable=False)
+
+    created_at = Column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False
+    )
+    updated_at = Column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+        nullable=False,
+    )
+
+    def set_client_secret(self, secret: str) -> None:
+        """设置并加密 client_secret"""
+        from src.core.crypto import crypto_service
+
+        self.client_secret_encrypted = crypto_service.encrypt(secret)
+
+    def get_client_secret(self) -> str:
+        """获取解密后的 client_secret（未配置时返回空串）"""
+        from src.core.crypto import crypto_service
+
+        if not self.client_secret_encrypted:
+            return ""
+        return crypto_service.decrypt(self.client_secret_encrypted)
+
+
+class UserOAuthLink(Base):
+    """用户与 OAuth Provider 的绑定关系"""
+
+    __tablename__ = "user_oauth_links"
+
+    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
+    user_id = Column(
+        String(36),
+        ForeignKey("users.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    provider_type = Column(
+        String(50),
+        ForeignKey("oauth_providers.provider_type", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    provider_user_id = Column(String(255), nullable=False)
+    provider_username = Column(String(255), nullable=True)
+    provider_email = Column(String(255), nullable=True)
+    extra_data = Column(JSON, nullable=True)
+
+    linked_at = Column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False
+    )
+    last_login_at = Column(DateTime(timezone=True), nullable=True)
+
+    __table_args__ = (
+        UniqueConstraint("provider_type", "provider_user_id", name="uq_oauth_provider_user"),
+        UniqueConstraint("user_id", "provider_type", name="uq_user_oauth_provider"),
+    )
+
+
 class Provider(Base):
     """提供商配置表"""
 

src/modules/__init__.py

@@ -10,10 +10,12 @@ from src.core.modules.base import ModuleDefinition
 
 # 导入所有模块定义
 from src.modules.ldap import ldap_module
+from src.modules.oauth import oauth_module
 
 # 所有模块列表
 ALL_MODULES: List[ModuleDefinition] = [
     ldap_module,
+    oauth_module,
 ]
 
 __all__ = ["ALL_MODULES"]

src/modules/ldap/__init__.py

@@ -4,6 +4,8 @@ LDAP 认证模块
 提供 LDAP/Active Directory 用户认证支持
 """
 
+from typing import TYPE_CHECKING, Tuple
+
 from src.core.modules.base import (
     ModuleCategory,
     ModuleDefinition,
@@ -11,6 +13,9 @@ from src.core.modules.base import (
     ModuleMetadata,
 )
 
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
 
 def _get_router():
     """延迟导入路由（避免启动时加载重依赖）"""
@@ -26,6 +31,46 @@ async def _health_check() -> ModuleHealth:
     return ModuleHealth.UNKNOWN
 
 
+def _validate_config(db: "Session") -> Tuple[bool, str]:
+    """
+    验证 LDAP 配置是否可以启用模块
+
+    检查项：
+    1. 配置是否存在
+    2. 必填字段是否完整
+    3. 绑定密码是否可解密
+
+    注意：不在此处执行连接测试，因为 validate_config 会在每次查询模块状态时调用，
+    同步阻塞等待 LDAP 服务器响应会严重影响性能。连接测试应在专门的测试接口中进行。
+    """
+    from src.core.crypto import crypto_service
+    from src.models.database import LDAPConfig
+
+    config = db.query(LDAPConfig).first()
+    if not config:
+        return False, "请先配置 LDAP 连接信息"
+
+    # 检查必填字段
+    if not config.server_url:
+        return False, "请配置 LDAP 服务器地址"
+    if not config.bind_dn:
+        return False, "请配置绑定 DN"
+    if not config.base_dn:
+        return False, "请配置搜索基准 DN"
+    if not config.bind_password_encrypted:
+        return False, "请配置绑定密码"
+
+    # 尝试解密密码（仅验证可解密，不执行连接测试）
+    try:
+        bind_password = crypto_service.decrypt(config.bind_password_encrypted)
+        if not bind_password:
+            return False, "绑定密码为空，请重新设置"
+    except Exception:
+        return False, "绑定密码解密失败，请重新设置"
+
+    return True, ""
+
+
 # LDAP 模块定义
 ldap_module = ModuleDefinition(
     metadata=ModuleMetadata(
@@ -47,4 +92,5 @@ ldap_module = ModuleDefinition(
     ),
     router_factory=_get_router,
     health_check=_health_check,
+    validate_config=_validate_config,
 )

src/modules/oauth/__init__.py

@@ -0,0 +1,88 @@
+"""
+OAuth 认证模块
+
+提供可配置的 OAuth 登录/绑定能力。
+"""
+
+from typing import TYPE_CHECKING, Tuple
+
+from src.core.modules.base import (
+    ModuleCategory,
+    ModuleDefinition,
+    ModuleHealth,
+    ModuleMetadata,
+)
+
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+
+def _get_router():
+    """延迟导入路由（避免启动时加载重依赖/副作用）。"""
+    # 延迟 discover，避免 alembic/mypy 等场景导入时触发 entry_points 解析
+    from src.services.auth.oauth.registry import get_oauth_provider_registry
+
+    get_oauth_provider_registry().discover_providers()
+
+    from src.api.oauth import router
+
+    return router
+
+
+async def _health_check() -> ModuleHealth:
+    # v1：不做外部网络探测，避免启动时阻塞
+    return ModuleHealth.UNKNOWN
+
+
+def _validate_config(db: "Session") -> Tuple[bool, str]:
+    """
+    验证 OAuth 配置是否可以启用模块
+
+    检查项：
+    1. 至少有一个已启用的 Provider 配置
+    2. 已启用的 Provider 必须有 client_id 和 client_secret
+    """
+    from src.models.database import OAuthProvider
+
+    # 查找所有已启用的 Provider
+    enabled_providers = (
+        db.query(OAuthProvider)
+        .filter(OAuthProvider.is_enabled.is_(True))
+        .all()
+    )
+
+    if not enabled_providers:
+        return False, "请先配置并启用至少一个 OAuth Provider"
+
+    # 检查每个已启用的 Provider 配置完整性
+    for provider in enabled_providers:
+        if not provider.client_id:
+            return False, f"Provider [{provider.display_name}] 未配置 Client ID"
+        if not provider.client_secret_encrypted:
+            return False, f"Provider [{provider.display_name}] 未配置 Client Secret"
+        if not provider.redirect_uri:
+            return False, f"Provider [{provider.display_name}] 未配置回调地址"
+
+    return True, ""
+
+
+oauth_module = ModuleDefinition(
+    metadata=ModuleMetadata(
+        name="oauth",
+        display_name="OAuth 登录",
+        description="支持通过第三方 OAuth Provider 登录/绑定账号",
+        category=ModuleCategory.AUTH,
+        env_key="OAUTH_AVAILABLE",
+        default_available=True,
+        required_packages=["httpx", "redis"],
+        api_prefix="/api/oauth",
+        admin_route="/admin/oauth",
+        admin_menu_icon="Key",
+        admin_menu_group="system",
+        admin_menu_order=55,
+    ),
+    router_factory=_get_router,
+    health_check=_health_check,
+    validate_config=_validate_config,
+)
+

src/plugins/auth/jwt.py

@@ -74,6 +74,13 @@ class JwtAuthPlugin(AuthPlugin):
             if not user.is_active:
                 logger.warning(f"JWT认证失败 - 用户已禁用: {user.email}")
                 return None
+            if user.is_deleted:
+                logger.warning(f"JWT认证失败 - 用户已删除: {user.email}")
+                return None
+
+            if not AuthService.token_identity_matches_user(payload, user):
+                logger.warning("JWT认证失败 - Token身份校验失败")
+                return None
 
             # 创建认证上下文
             auth_context = AuthContext(

src/services/auth/oauth/__init__.py

@@ -0,0 +1,6 @@
+"""OAuth 认证相关服务。"""
+
+from .service import OAuthService
+
+__all__ = ["OAuthService"]
+

src/services/auth/oauth/base.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Optional
+from urllib.parse import urlencode, urlparse, urlunparse
+
+import httpx
+
+from src.services.auth.oauth.models import OAuthToken, OAuthUserInfo
+
+if TYPE_CHECKING:
+    from src.models.database import OAuthProvider
+
+
+class OAuthProviderBase(ABC):
+    """
+    OAuth Provider 基类（稳定扩展点）。
+
+    v1 收敛点：仅实现 OAuth2 授权码流程所需的最小接口。
+    """
+
+    provider_type: str
+    display_name: str
+
+    # 允许的 host 白名单（用于端点覆盖校验，支持子域名）
+    allowed_domains: tuple[str, ...] = ()
+
+    authorization_url: str
+    token_url: str
+    userinfo_url: str
+    default_scopes: tuple[str, ...] = ()
+
+    def get_effective_authorization_url(self, config: "OAuthProvider") -> str:
+        return config.authorization_url_override or self.authorization_url
+
+    def get_effective_token_url(self, config: "OAuthProvider") -> str:
+        return config.token_url_override or self.token_url
+
+    def get_effective_userinfo_url(self, config: "OAuthProvider") -> str:
+        return config.userinfo_url_override or self.userinfo_url
+
+    def get_effective_scopes(self, config: "OAuthProvider") -> str:
+        scopes = config.scopes or list(self.default_scopes)
+        return " ".join(scopes)
+
+    def get_authorization_url(self, config: "OAuthProvider", state: str) -> str:
+        """
+        构造 provider 授权 URL。
+
+        redirect_uri 必须由服务端控制，不从客户端传入。
+        """
+        base = self.get_effective_authorization_url(config)
+        # 避免覆盖原有 query（若 provider 默认 url 带 query，保留）
+        parsed = urlparse(base)
+        query: dict[str, str] = {}
+        if parsed.query:
+            # 保留已有 query 参数
+            for kv in parsed.query.split("&"):
+                if not kv:
+                    continue
+                if "=" in kv:
+                    k, v = kv.split("=", 1)
+                    query[k] = v
+                else:
+                    query[kv] = ""
+
+        client_id = config.client_id
+        redirect_uri = config.redirect_uri
+        if not client_id or not redirect_uri:
+            raise ValueError("OAuthProvider 配置不完整：client_id/redirect_uri 不能为空")
+
+        query.update(
+            {
+                "response_type": "code",
+                "client_id": client_id,
+                "redirect_uri": redirect_uri,
+                "scope": self.get_effective_scopes(config),
+                "state": state,
+            }
+        )
+
+        return urlunparse(parsed._replace(query=urlencode(query)))
+
+    @abstractmethod
+    async def exchange_code(self, config: "OAuthProvider", code: str) -> OAuthToken:
+        """使用授权码兑换 token。"""
+
+    @abstractmethod
+    async def get_user_info(self, config: "OAuthProvider", access_token: str) -> OAuthUserInfo:
+        """获取用户信息。"""
+
+    async def _http_post_form(
+        self,
+        url: str,
+        data: dict[str, str],
+        *,
+        timeout_seconds: float = 5.0,
+        headers: Optional[dict[str, str]] = None,
+    ) -> httpx.Response:
+        async with httpx.AsyncClient(timeout=httpx.Timeout(timeout_seconds)) as client:
+            return await client.post(url, data=data, headers=headers)
+
+    async def _http_get(
+        self,
+        url: str,
+        *,
+        timeout_seconds: float = 5.0,
+        headers: Optional[dict[str, str]] = None,
+    ) -> httpx.Response:
+        async with httpx.AsyncClient(timeout=httpx.Timeout(timeout_seconds)) as client:
+            return await client.get(url, headers=headers)

src/services/auth/oauth/models.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+
+@dataclass(frozen=True)
+class OAuthToken:
+    access_token: str
+    token_type: str = "bearer"
+    refresh_token: Optional[str] = None
+    expires_in: Optional[int] = None
+    id_token: Optional[str] = None
+    scope: Optional[str] = None
+    raw: Optional[dict[str, Any]] = None
+
+
+@dataclass(frozen=True)
+class OAuthUserInfo:
+    id: str
+    username: Optional[str] = None
+    email: Optional[str] = None
+    email_verified: Optional[bool] = None
+    raw: Optional[dict[str, Any]] = None
+
+
+class OAuthFlowError(Exception):
+    """用于 OAuth 流程的可控错误（会映射到 error_code）。"""
+
+    def __init__(self, error_code: str, detail: str = ""):
+        super().__init__(error_code)
+        self.error_code = error_code
+        self.detail = detail
+

src/services/auth/oauth/providers/__init__.py

@@ -0,0 +1,6 @@
+"""内置 OAuth providers（v1）。"""
+
+from .linuxdo import LinuxDoOAuthProvider
+
+__all__ = ["LinuxDoOAuthProvider"]
+

src/services/auth/oauth/providers/linuxdo.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from src.core.logger import logger
+from src.services.auth.oauth.base import OAuthProviderBase
+from src.services.auth.oauth.models import OAuthFlowError, OAuthToken, OAuthUserInfo
+
+if TYPE_CHECKING:
+    from src.models.database import OAuthProvider
+
+
+class LinuxDoOAuthProvider(OAuthProviderBase):
+    """
+    LinuxDo OAuth Provider。
+
+    基于论坛信任等级（trust_level 0-4）的 OAuth2 认证，
+    用于通过用户等级进行额度配给和频率限制。
+
+    参考：https://linux.do/t/topic/329408
+
+    返回的用户信息示例：
+    {
+        "id": 1,
+        "username": "neo",
+        "name": "Neo",
+        "active": true,
+        "trust_level": 4,
+        "email": "u1@linux.do",
+        "avatar_url": "https://linux.do/xxxx",
+        "silenced": false
+    }
+    """
+
+    provider_type = "linuxdo"
+    display_name = "Linux Do"
+
+    allowed_domains = ("linux.do", "connect.linux.do", "connect.linuxdo.org")
+
+    # 默认端点
+    authorization_url = "https://connect.linux.do/oauth2/authorize"
+    token_url = "https://connect.linux.do/oauth2/token"
+    userinfo_url = "https://connect.linux.do/api/user"
+
+    # LinuxDo 不需要 scope
+    default_scopes = ()
+
+    async def exchange_code(self, config: "OAuthProvider", code: str) -> OAuthToken:
+        url = self.get_effective_token_url(config)
+        client_secret = config.get_client_secret()
+        if not client_secret:
+            raise OAuthFlowError("provider_unavailable", "client_secret 未配置")
+
+        redirect_uri = config.redirect_uri
+        client_id = config.client_id
+        if not redirect_uri or not client_id:
+            raise OAuthFlowError("provider_unavailable", "redirect_uri/client_id 未配置")
+
+        resp = await self._http_post_form(
+            url,
+            data={
+                "grant_type": "authorization_code",
+                "code": code,
+                "redirect_uri": redirect_uri,
+                "client_id": client_id,
+                "client_secret": client_secret,
+            },
+        )
+
+        if resp.status_code >= 400:
+            logger.warning("LinuxDo token 兑换失败: status={}", resp.status_code)
+            raise OAuthFlowError("token_exchange_failed", f"status={resp.status_code}")
+
+        data = resp.json()
+        access_token = data.get("access_token")
+        if not access_token:
+            raise OAuthFlowError("token_exchange_failed", "missing access_token")
+
+        return OAuthToken(
+            access_token=str(access_token),
+            token_type=str(data.get("token_type") or "bearer"),
+            refresh_token=(str(data["refresh_token"]) if data.get("refresh_token") else None),
+            expires_in=(int(data["expires_in"]) if data.get("expires_in") is not None else None),
+            id_token=(str(data["id_token"]) if data.get("id_token") else None),
+            scope=(str(data["scope"]) if data.get("scope") else None),
+            raw=data,
+        )
+
+    async def get_user_info(self, config: "OAuthProvider", access_token: str) -> OAuthUserInfo:
+        url = self.get_effective_userinfo_url(config)
+        resp = await self._http_get(url, headers={"Authorization": f"Bearer {access_token}"})
+
+        if resp.status_code >= 400:
+            logger.warning("LinuxDo userinfo 获取失败: status={}", resp.status_code)
+            raise OAuthFlowError("userinfo_fetch_failed", f"status={resp.status_code}")
+
+        data: dict[str, Any] = resp.json()
+
+        # LinuxDo 返回的 id 是数字类型
+        provider_user_id = data.get("id")
+        if provider_user_id is None:
+            raise OAuthFlowError("userinfo_fetch_failed", "missing user id")
+
+        return OAuthUserInfo(
+            id=str(provider_user_id),
+            username=data.get("username"),
+            email=str(data["email"]).lower() if data.get("email") else None,
+            email_verified=None,  # LinuxDo 不返回此字段
+            raw=data,  # 包含 trust_level, active, silenced, avatar_url, name 等
+        )

src/services/auth/oauth/registry.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+from src.core.logger import logger
+from src.services.auth.oauth.base import OAuthProviderBase
+
+
+@dataclass(frozen=True)
+class SupportedOAuthType:
+    provider_type: str
+    display_name: str
+    # 默认端点（用于前端 placeholder 展示）
+    default_authorization_url: str
+    default_token_url: str
+    default_userinfo_url: str
+    default_scopes: tuple[str, ...]
+
+
+class OAuthProviderRegistry:
+    """Provider 注册表（支持延迟 discover）。"""
+
+    def __init__(self) -> None:
+        self._providers: Dict[str, OAuthProviderBase] = {}
+        self._discovered: bool = False
+
+    def discover_providers(self) -> None:
+        """发现并注册 providers（幂等）。"""
+        if self._discovered:
+            return
+        self._discovered = True
+
+        # 1) 内置 providers（v1：至少保证 linuxdo 可用）
+        try:
+            from src.services.auth.oauth.providers.linuxdo import LinuxDoOAuthProvider
+
+            self.register(LinuxDoOAuthProvider())
+        except Exception as exc:
+            logger.warning("OAuth 内置 provider 加载失败: {}", exc)
+
+        # 2) entry_points 插件（可选）
+        try:
+            from importlib.metadata import entry_points
+
+            eps = entry_points()
+            # Python 3.10+ 支持 select；旧接口返回 dict
+            if hasattr(eps, "select"):
+                candidates = list(eps.select(group="aether.oauth_providers"))  # type: ignore[attr-defined]
+            else:
+                candidates = list(eps.get("aether.oauth_providers", []))  # type: ignore[call-arg]
+
+            for ep in candidates:
+                try:
+                    loaded = ep.load()
+                    provider = loaded() if isinstance(loaded, type) else loaded
+                    if not isinstance(provider, OAuthProviderBase):
+                        logger.warning(
+                            "OAuth provider entry_point 无效: {} (type={})", ep.name, type(provider)
+                        )
+                        continue
+                    self.register(provider)
+                except Exception as e:
+                    logger.warning("OAuth provider entry_point 加载失败: {}: {}", ep.name, e)
+        except Exception as exc:
+            # entry_points 不可用不影响主流程
+            logger.debug("OAuth entry_points discover skipped: {}", exc)
+
+    def register(self, provider: OAuthProviderBase) -> None:
+        self._providers[provider.provider_type] = provider
+
+    def get_provider(self, provider_type: str) -> Optional[OAuthProviderBase]:
+        return self._providers.get(provider_type)
+
+    def get_supported_types(self) -> List[SupportedOAuthType]:
+        return [
+            SupportedOAuthType(
+                provider_type=p.provider_type,
+                display_name=p.display_name,
+                default_authorization_url=p.authorization_url,
+                default_token_url=p.token_url,
+                default_userinfo_url=p.userinfo_url,
+                default_scopes=p.default_scopes,
+            )
+            for p in sorted(self._providers.values(), key=lambda x: x.provider_type)
+        ]
+
+
+_registry: Optional[OAuthProviderRegistry] = None
+
+
+def get_oauth_provider_registry() -> OAuthProviderRegistry:
+    global _registry
+    if _registry is None:
+        _registry = OAuthProviderRegistry()
+    return _registry
+

src/services/auth/oauth/service.py

@@ -0,0 +1,945 @@
+from __future__ import annotations
+
+import re
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Optional
+from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
+
+import httpx
+from fastapi import HTTPException, status
+from sqlalchemy import func
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from src.clients.redis_client import get_redis_client
+from src.core.enums import AuthSource, UserRole
+from src.core.exceptions import ConfirmationRequiredException, InvalidRequestException
+from src.core.logger import logger
+from src.core.modules import get_module_registry
+from src.models.database import OAuthProvider, User, UserOAuthLink
+from src.services.auth.ldap import LDAPService
+from src.services.auth.oauth.models import OAuthFlowError, OAuthUserInfo
+from src.services.auth.oauth.registry import get_oauth_provider_registry
+from src.services.auth.oauth.state import consume_oauth_state, create_oauth_state
+from src.services.auth.service import AuthService
+from src.services.cache.user_cache import UserCacheService
+from src.services.system.config import SystemConfigService
+from src.services.auth.oauth.base import OAuthProviderBase
+
+
+class OAuthService:
+    """OAuth 核心业务服务（v1）。"""
+
+    @staticmethod
+    def _require_module_active(db: Session) -> None:
+        registry = get_module_registry()
+        if not registry.is_active("oauth", db):
+            raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="OAuth 模块未启用")
+
+    @staticmethod
+    def _get_provider_impl(provider_type: str) -> Optional[OAuthProviderBase]:
+        registry = get_oauth_provider_registry()
+        registry.discover_providers()
+        provider = registry.get_provider(provider_type)
+        return provider
+
+    @staticmethod
+    def _get_provider_config(db: Session, provider_type: str) -> OAuthProvider:
+        row = db.query(OAuthProvider).filter(OAuthProvider.provider_type == provider_type).first()
+        if not row:
+            raise InvalidRequestException("Provider 配置不存在")
+        return row
+
+    @staticmethod
+    def _get_enabled_provider_config(db: Session, provider_type: str) -> OAuthProvider:
+        row = OAuthService._get_provider_config(db, provider_type)
+        if not row.is_enabled:
+            raise OAuthFlowError("provider_disabled", "provider 未启用")
+        return row
+
+    @staticmethod
+    def _build_frontend_error_redirect(
+        frontend_callback_url: str, *, error_code: str, error_detail: str = ""
+    ) -> str:
+        parsed = urlparse(frontend_callback_url)
+        query = dict(parse_qsl(parsed.query, keep_blank_values=True))
+        query["error_code"] = error_code
+        if error_detail:
+            query["error_detail"] = (error_detail[:200]).strip()
+        return urlunparse(parsed._replace(query=urlencode(query), fragment=""))
+
+    @staticmethod
+    def _build_frontend_bind_success_redirect(frontend_callback_url: str, display_name: str) -> str:
+        parsed = urlparse(frontend_callback_url)
+        query = dict(parse_qsl(parsed.query, keep_blank_values=True))
+        query["oauth_bound"] = display_name
+        return urlunparse(parsed._replace(query=urlencode(query), fragment=""))
+
+    @staticmethod
+    def _build_frontend_login_success_redirect(
+        frontend_callback_url: str, *, access_token: str, refresh_token: str
+    ) -> str:
+        parsed = urlparse(frontend_callback_url)
+        fragment = urlencode(
+            {
+                "access_token": access_token,
+                "refresh_token": refresh_token,
+                "token_type": "bearer",
+                "expires_in": 86400,
+            }
+        )
+        # fragment 不会被发送回后端，适配当前 localStorage 登录态方案
+        return urlunparse(parsed._replace(fragment=fragment))
+
+    @staticmethod
+    async def list_public_providers(db: Session) -> list[dict[str, str]]:
+        registry = get_module_registry()
+        if not registry.is_active("oauth", db):
+            return []
+
+        supported = get_oauth_provider_registry()
+        supported.discover_providers()
+
+        rows = (
+            db.query(OAuthProvider)
+            .filter(OAuthProvider.is_enabled.is_(True))
+            .order_by(OAuthProvider.provider_type.asc())
+            .all()
+        )
+
+        result: list[dict[str, str]] = []
+        for row in rows:
+            provider_type_value = row.provider_type
+            if not provider_type_value:
+                continue
+            provider_type_str = str(provider_type_value)
+            if supported.get_provider(provider_type_str) is None:
+                continue
+            display_name = row.display_name or provider_type_str
+            result.append({"provider_type": provider_type_str, "display_name": str(display_name)})
+        return result
+
+    @staticmethod
+    async def build_login_authorize_url(db: Session, provider_type: str) -> str:
+        OAuthService._require_module_active(db)
+
+        provider = OAuthService._get_provider_impl(provider_type)
+        if not provider:
+            raise HTTPException(status_code=404, detail="不支持的 OAuth provider")
+
+        try:
+            config = OAuthService._get_enabled_provider_config(db, provider_type)
+        except OAuthFlowError as exc:
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=exc.error_code
+            )
+
+        redis = await get_redis_client(require_redis=True)
+        if redis is None:
+            raise HTTPException(status_code=503, detail="Redis 不可用")
+
+        state = await create_oauth_state(redis, provider_type=provider_type, action="login", user_id=None)
+        return provider.get_authorization_url(config, state)
+
+    @staticmethod
+    async def build_bind_authorize_url(db: Session, user: User, provider_type: str) -> str:
+        OAuthService._require_module_active(db)
+
+        if user.auth_source == AuthSource.LDAP:
+            raise HTTPException(status_code=403, detail="LDAP 用户不允许绑定 OAuth")
+
+        provider = OAuthService._get_provider_impl(provider_type)
+        if not provider:
+            raise HTTPException(status_code=404, detail="不支持的 OAuth provider")
+
+        try:
+            config = OAuthService._get_enabled_provider_config(db, provider_type)
+        except OAuthFlowError as exc:
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=exc.error_code
+            )
+
+        redis = await get_redis_client(require_redis=True)
+        if redis is None:
+            raise HTTPException(status_code=503, detail="Redis 不可用")
+
+        state = await create_oauth_state(redis, provider_type=provider_type, action="bind", user_id=user.id)
+        return provider.get_authorization_url(config, state)
+
+    @staticmethod
+    def _sanitize_username(raw: Optional[str]) -> str:
+        if not raw or not raw.strip():
+            return f"user_{uuid.uuid4().hex[:8]}"
+
+        cleaned = re.sub(r"[^a-zA-Z0-9_-]", "_", raw.strip())
+        cleaned = re.sub(r"_+", "_", cleaned).strip("_")
+
+        if cleaned and cleaned[0].isdigit():
+            cleaned = f"u_{cleaned}"
+
+        # 预留后缀空间，避免后续重试超长
+        max_len = 90
+        if len(cleaned) > max_len:
+            cleaned = cleaned[:max_len]
+
+        return cleaned or f"user_{uuid.uuid4().hex[:8]}"
+
+    @staticmethod
+    def _generate_unique_username(db: Session, base: str, max_retries: int = 3) -> str:
+        base = OAuthService._sanitize_username(base)
+        candidates = [base]
+        for i in range(max_retries - 1):
+            suffix_len = 4 if i == 0 else 8
+            candidates.append(f"{base}_{uuid.uuid4().hex[:suffix_len]}")
+
+        for cand in candidates:
+            exists = db.query(User).filter(User.username == cand).first()
+            if not exists:
+                return cand
+        raise ValueError("无法生成唯一用户名")
+
+    @staticmethod
+    def _validate_email_suffix(db: Session, email: str) -> bool:
+        mode = SystemConfigService.get_config(db, "email_suffix_mode", default="none")
+        if mode == "none":
+            return True
+
+        suffix_list = SystemConfigService.get_config(db, "email_suffix_list", default=[])
+        if isinstance(suffix_list, str):
+            suffix_list = [s.strip().lower() for s in suffix_list.split(",") if s.strip()]
+
+        if not suffix_list:
+            return True
+
+        if "@" not in email:
+            return False
+        email_suffix = email.split("@", 1)[1].lower()
+
+        if mode == "whitelist":
+            return email_suffix in suffix_list
+        if mode == "blacklist":
+            return email_suffix not in suffix_list
+        return True
+
+    @staticmethod
+    def _get_constraint_name(err: IntegrityError) -> Optional[str]:
+        orig = getattr(err, "orig", None)
+        diag = getattr(orig, "diag", None)
+        name = getattr(diag, "constraint_name", None)
+        return str(name) if name else None
+
+    @staticmethod
+    async def handle_callback(
+        *,
+        db: Session,
+        provider_type: str,
+        state: str,
+        code: Optional[str],
+        error: Optional[str],
+        error_description: Optional[str],
+    ) -> str:
+        OAuthService._require_module_active(db)
+
+        provider = OAuthService._get_provider_impl(provider_type)
+        if not provider:
+            raise HTTPException(status_code=404, detail="不支持的 OAuth provider")
+
+        config = OAuthService._get_provider_config(db, provider_type)
+        frontend_callback_url = config.frontend_callback_url
+        if not frontend_callback_url:
+            # 无法重定向到前端时，直接返回 500（配置错误）
+            raise HTTPException(status_code=500, detail="frontend_callback_url 未配置")
+        frontend_callback_url = str(frontend_callback_url)
+
+        # display_name 用于 bind 成功 toast；兜底为 provider_type
+        display_name = str(config.display_name or provider_type)
+
+        # provider 被禁用时仍引导回前端（给出明确提示）
+        if not config.is_enabled:
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code="provider_disabled"
+            )
+
+        # provider 侧 error
+        if error:
+            code_map = "authorization_denied" if error == "access_denied" else "provider_error"
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url,
+                error_code=code_map,
+                error_detail=error_description or error,
+            )
+
+        if not code or not state:
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code="invalid_callback"
+            )
+
+        # 一次性消费 state
+        try:
+            redis = await get_redis_client(require_redis=True)
+            if redis is None:
+                raise RuntimeError("redis unavailable")
+            state_data = await consume_oauth_state(redis, state)
+        except Exception as exc:
+            logger.warning("OAuth state 消费失败: {}", exc)
+            state_data = None
+
+        if not state_data:
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code="invalid_state"
+            )
+
+        if state_data.provider_type != provider_type:
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code="invalid_state"
+            )
+
+        if state_data.action not in ("login", "bind"):
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code="invalid_state"
+            )
+
+        if state_data.action == "bind" and not state_data.user_id:
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code="invalid_bind_state"
+            )
+
+        try:
+            token = await provider.exchange_code(config, code)
+            oauth_user = await provider.get_user_info(config, token.access_token)
+        except OAuthFlowError as exc:
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code=exc.error_code, error_detail=exc.detail
+            )
+        except Exception as exc:
+            logger.warning("OAuth callback 处理失败: {}", exc)
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code="provider_error"
+            )
+
+        if state_data.action == "bind":
+            try:
+                await OAuthService._handle_bind(db, user_id=state_data.user_id or "", config=config, oauth_user=oauth_user)
+            except OAuthFlowError as exc:
+                return OAuthService._build_frontend_error_redirect(
+                    frontend_callback_url, error_code=exc.error_code, error_detail=exc.detail
+                )
+
+            return OAuthService._build_frontend_bind_success_redirect(
+                frontend_callback_url, display_name
+            )
+
+        # login
+        try:
+            user = await OAuthService._handle_login(db, config=config, oauth_user=oauth_user)
+        except OAuthFlowError as exc:
+            return OAuthService._build_frontend_error_redirect(
+                frontend_callback_url, error_code=exc.error_code, error_detail=exc.detail
+            )
+
+        assert user.id is not None
+        assert user.role is not None
+
+        access_token = AuthService.create_access_token(
+            data={
+                "user_id": user.id,
+                "role": user.role.value,
+                "created_at": user.created_at.isoformat() if user.created_at else None,
+            }
+        )
+        refresh_token = AuthService.create_refresh_token(
+            data={"user_id": user.id, "created_at": user.created_at.isoformat() if user.created_at else None}
+        )
+
+        return OAuthService._build_frontend_login_success_redirect(
+            frontend_callback_url, access_token=access_token, refresh_token=refresh_token
+        )
+
+    @staticmethod
+    async def _handle_login(db: Session, *, config: OAuthProvider, oauth_user: OAuthUserInfo) -> User:
+        now = datetime.now(timezone.utc)
+
+        # 1) 已绑定账号：直接登录
+        existing_link = (
+            db.query(UserOAuthLink)
+            .filter(
+                UserOAuthLink.provider_type == config.provider_type,
+                UserOAuthLink.provider_user_id == oauth_user.id,
+            )
+            .first()
+        )
+        if existing_link:
+            linked_user = db.query(User).filter(User.id == existing_link.user_id).first()
+            if not linked_user or not linked_user.is_active or linked_user.is_deleted:
+                raise OAuthFlowError("account_disabled", "用户不存在或已禁用")
+
+            linked_user.last_login_at = now
+            existing_link.last_login_at = now
+            db.commit()
+            assert linked_user.id is not None
+            await UserCacheService.invalidate_user_cache(linked_user.id, linked_user.email)
+            return linked_user
+
+        # 2) 未绑定账号：可能需要新建用户（受注册开关控制）
+        enable_registration = SystemConfigService.get_config(db, "enable_registration", default=False)
+        if not enable_registration:
+            raise OAuthFlowError("registration_disabled")
+
+        email = oauth_user.email
+        if email:
+            if not OAuthService._validate_email_suffix(db, email):
+                raise OAuthFlowError("email_suffix_denied")
+
+            existing_user = db.query(User).filter(User.email == email).first()
+            # 已删除用户不阻塞新建（邮箱可复用）
+            if existing_user and not existing_user.is_deleted:
+                if existing_user.auth_source == AuthSource.LOCAL:
+                    raise OAuthFlowError("email_exists_local")
+                if existing_user.auth_source == AuthSource.LDAP:
+                    raise OAuthFlowError("email_is_ldap")
+                raise OAuthFlowError("email_is_oauth")
+
+        base_username = oauth_user.username or (email.split("@", 1)[0] if email else None) or f"user_{uuid.uuid4().hex[:8]}"
+        default_quota = SystemConfigService.get_config(db, "default_user_quota_usd", default=10.0)
+
+        # 生成唯一用户名 + 创建用户（简单重试）
+        user: Optional[User] = None
+        last_error: Optional[Exception] = None
+        for _ in range(3):
+            try:
+                username = OAuthService._generate_unique_username(db, base_username)
+                user = User(
+                    email=email,
+                    email_verified=bool(oauth_user.email_verified) if email else False,
+                    username=username,
+                    password_hash=None,
+                    auth_source=AuthSource.OAUTH,
+                    role=UserRole.USER,
+                    is_active=True,
+                    last_login_at=now,
+                    quota_usd=default_quota,
+                )
+                db.add(user)
+                db.commit()
+                db.refresh(user)
+                last_error = None
+                break
+            except IntegrityError as e:
+                db.rollback()
+                last_error = e
+            except Exception as e:
+                db.rollback()
+                last_error = e
+
+        if last_error is not None or user is None:
+            raise OAuthFlowError("provider_error", "user_create_failed")
+
+        # 创建绑定关系
+        assert user.id is not None
+        try:
+            link = UserOAuthLink(
+                user_id=user.id,
+                provider_type=config.provider_type,
+                provider_user_id=oauth_user.id,
+                provider_username=oauth_user.username,
+                provider_email=email,
+                extra_data=oauth_user.raw,
+                linked_at=now,
+                last_login_at=now,
+            )
+            db.add(link)
+            db.commit()
+        except IntegrityError as e:
+            db.rollback()
+            constraint = OAuthService._get_constraint_name(e)
+            if constraint == "uq_oauth_provider_user":
+                # 并发：该第三方账号已先被绑定，尝试读取并登录
+                existing_link = (
+                    db.query(UserOAuthLink)
+                    .filter(
+                        UserOAuthLink.provider_type == config.provider_type,
+                        UserOAuthLink.provider_user_id == oauth_user.id,
+                    )
+                    .first()
+                )
+                if existing_link:
+                    existing_user = db.query(User).filter(User.id == existing_link.user_id).first()
+                    if existing_user and existing_user.is_active and not existing_user.is_deleted:
+                        existing_user.last_login_at = now
+                        existing_link.last_login_at = now
+                        db.commit()
+                        assert existing_user.id is not None
+                        await UserCacheService.invalidate_user_cache(existing_user.id, existing_user.email)
+                        return existing_user
+                raise OAuthFlowError("oauth_already_bound")
+            raise OAuthFlowError("provider_error", "link_create_failed")
+
+        assert user.id is not None
+        await UserCacheService.invalidate_user_cache(user.id, user.email)
+        return user
+
+    @staticmethod
+    async def _handle_bind(
+        db: Session, *, user_id: str, config: OAuthProvider, oauth_user: OAuthUserInfo
+    ) -> UserOAuthLink:
+        now = datetime.now(timezone.utc)
+        user = db.query(User).filter(User.id == user_id).first()
+        if not user or not user.is_active or user.is_deleted:
+            raise OAuthFlowError("user_not_found")
+        if user.auth_source == AuthSource.LDAP:
+            raise OAuthFlowError("ldap_no_oauth")
+
+        link = UserOAuthLink(
+            user_id=user.id,
+            provider_type=config.provider_type,
+            provider_user_id=oauth_user.id,
+            provider_username=oauth_user.username,
+            provider_email=oauth_user.email,
+            extra_data=oauth_user.raw,
+            linked_at=now,
+        )
+
+        try:
+            db.add(link)
+            db.commit()
+            db.refresh(link)
+            return link
+        except IntegrityError as e:
+            db.rollback()
+            constraint = OAuthService._get_constraint_name(e)
+
+            if constraint == "uq_oauth_provider_user":
+                existing = (
+                    db.query(UserOAuthLink)
+                    .filter(
+                        UserOAuthLink.provider_type == config.provider_type,
+                        UserOAuthLink.provider_user_id == oauth_user.id,
+                    )
+                    .first()
+                )
+                if existing and existing.user_id == user.id:
+                    return existing
+                raise OAuthFlowError("oauth_already_bound")
+
+            if constraint == "uq_user_oauth_provider":
+                raise OAuthFlowError("already_bound_provider")
+
+            raise OAuthFlowError("provider_error", "bind_failed")
+
+    @staticmethod
+    async def list_bindable_providers(db: Session, user: User) -> list[dict[str, str]]:
+        OAuthService._require_module_active(db)
+
+        if user.auth_source == AuthSource.LDAP:
+            return []
+
+        supported = get_oauth_provider_registry()
+        supported.discover_providers()
+
+        enabled_rows = (
+            db.query(OAuthProvider)
+            .filter(OAuthProvider.is_enabled.is_(True))
+            .order_by(OAuthProvider.provider_type.asc())
+            .all()
+        )
+        linked_types = {
+            provider_type
+            for (provider_type,) in db.query(UserOAuthLink.provider_type)
+            .filter(UserOAuthLink.user_id == user.id)
+            .all()
+        }
+
+        result: list[dict[str, str]] = []
+        for row in enabled_rows:
+            provider_type_value = row.provider_type
+            if not provider_type_value:
+                continue
+            provider_type_str = str(provider_type_value)
+            if provider_type_str in linked_types:
+                continue
+            if supported.get_provider(provider_type_str) is None:
+                continue
+            display_name = row.display_name or provider_type_str
+            result.append({"provider_type": provider_type_str, "display_name": str(display_name)})
+        return result
+
+    @staticmethod
+    async def list_user_links(db: Session, user: User) -> list[dict[str, Any]]:
+        OAuthService._require_module_active(db)
+
+        rows = (
+            db.query(UserOAuthLink, OAuthProvider)
+            .join(OAuthProvider, UserOAuthLink.provider_type == OAuthProvider.provider_type)
+            .filter(UserOAuthLink.user_id == user.id)
+            .order_by(UserOAuthLink.linked_at.desc())
+            .all()
+        )
+
+        result: list[dict[str, Any]] = []
+        for link, provider in rows:
+            result.append(
+                {
+                    "provider_type": link.provider_type,
+                    "display_name": provider.display_name,
+                    "provider_username": link.provider_username,
+                    "provider_email": link.provider_email,
+                    "linked_at": link.linked_at.isoformat() if link.linked_at else None,
+                    "last_login_at": link.last_login_at.isoformat() if link.last_login_at else None,
+                    "provider_enabled": bool(provider.is_enabled),
+                }
+            )
+        return result
+
+    @staticmethod
+    def _check_provider_disable_safety(db: Session, provider_type: str) -> list[str]:
+        """
+        v1 简化版防锁号检查：
+        - 只检查活跃用户（is_active && !is_deleted）
+        - OAUTH 用户：禁用后必须仍有其它启用的 OAuth provider 绑定
+        - LOCAL 用户：ldap_exclusive=true 且非 admin 时，同上
+        """
+        ldap_exclusive = LDAPService.is_ldap_exclusive(db)
+
+        users = (
+            db.query(User.id, User.auth_source, User.role)
+            .join(UserOAuthLink, User.id == UserOAuthLink.user_id)
+            .filter(
+                User.is_active.is_(True),
+                User.is_deleted.is_(False),
+                UserOAuthLink.provider_type == provider_type,
+            )
+            .all()
+        )
+
+        affected: list[str] = []
+        for user_id, auth_source, role in users:
+            other_enabled_count = (
+                db.query(func.count(UserOAuthLink.id))
+                .join(OAuthProvider, UserOAuthLink.provider_type == OAuthProvider.provider_type)
+                .filter(
+                    UserOAuthLink.user_id == user_id,
+                    UserOAuthLink.provider_type != provider_type,
+                    OAuthProvider.is_enabled.is_(True),
+                )
+                .scalar()
+                or 0
+            )
+
+            locked = False
+            if auth_source == AuthSource.OAUTH:
+                if other_enabled_count == 0:
+                    locked = True
+            elif auth_source == AuthSource.LOCAL and ldap_exclusive:
+                is_admin = role == UserRole.ADMIN
+                if not is_admin and other_enabled_count == 0:
+                    locked = True
+
+            if locked:
+                affected.append(str(user_id))
+
+        return affected
+
+    @staticmethod
+    async def upsert_provider_config(db: Session, provider_type: str, data: Any) -> OAuthProvider:
+        provider = OAuthService._get_provider_impl(provider_type)
+        if not provider:
+            raise InvalidRequestException("不支持的 provider_type")
+
+        OAuthService._validate_provider_config(provider, data)
+
+        row = db.query(OAuthProvider).filter(OAuthProvider.provider_type == provider_type).first()
+        creating = row is None
+        if not row:
+            row = OAuthProvider(provider_type=provider_type)
+            db.add(row)
+
+        # 禁用前防锁号检查（仅在从 enabled -> disabled 时触发）
+        if row.is_enabled and data.is_enabled is False:
+            affected = OAuthService._check_provider_disable_safety(db, provider_type)
+            if affected and not getattr(data, "force", False):
+                raise ConfirmationRequiredException(
+                    message=f"禁用该 Provider 会导致 {len(affected)} 个用户无法登录",
+                    affected_count=len(affected),
+                    action="disable_oauth_provider",
+                )
+
+        row.display_name = data.display_name
+        row.client_id = data.client_id
+        row.authorization_url_override = data.authorization_url_override
+        row.token_url_override = data.token_url_override
+        row.userinfo_url_override = data.userinfo_url_override
+        row.scopes = data.scopes
+        row.redirect_uri = data.redirect_uri
+        row.frontend_callback_url = data.frontend_callback_url
+        row.attribute_mapping = data.attribute_mapping
+        row.extra_config = data.extra_config
+        row.is_enabled = data.is_enabled
+
+        # client_secret 处理逻辑：
+        # - None 或空字符串：保持不变
+        # - "__CLEAR__"：清空 secret
+        # - 其他值：设置新 secret
+        if data.client_secret is not None:
+            secret_value = data.client_secret.strip()
+            if secret_value == "__CLEAR__":
+                row.client_secret_encrypted = None
+            elif secret_value:
+                row.set_client_secret(secret_value)
+
+        try:
+            db.commit()
+            db.refresh(row)
+        except Exception:
+            db.rollback()
+            raise
+
+        if creating:
+            logger.info("OAuth provider 配置已创建: {}", provider_type)
+        else:
+            logger.info("OAuth provider 配置已更新: {}", provider_type)
+        return row
+
+    @staticmethod
+    async def delete_provider_config(db: Session, provider_type: str) -> None:
+        row = db.query(OAuthProvider).filter(OAuthProvider.provider_type == provider_type).first()
+        if not row:
+            raise InvalidRequestException("Provider 配置不存在")
+
+        if row.is_enabled:
+            affected = OAuthService._check_provider_disable_safety(db, provider_type)
+            if affected:
+                raise InvalidRequestException(
+                    f"删除该 Provider 会导致部分用户无法登录（数量: {len(affected)}），已阻止操作"
+                )
+
+        db.delete(row)
+        db.commit()
+
+    @staticmethod
+    def _validate_provider_config(provider: OAuthProviderBase, data: Any) -> None:
+        # frontend_callback_url 校验：必须绝对 URL，path 以 /auth/callback 结尾（允许 basePath）
+        OAuthService._validate_frontend_callback_url(data.frontend_callback_url)
+
+        # redirect_uri：允许本地 http，其余建议 https（v1：仅做基本校验）
+        OAuthService._validate_redirect_uri(data.redirect_uri)
+
+        # 覆盖端点：必须 https 且 hostname 命中 provider 白名单
+        for field_name in ("authorization_url_override", "token_url_override", "userinfo_url_override"):
+            value = getattr(data, field_name)
+            if value:
+                OAuthService._validate_url_override(provider, value)
+
+    @staticmethod
+    def _validate_frontend_callback_url(url: str) -> None:
+        parsed = urlparse(url)
+        if not parsed.scheme or not parsed.netloc:
+            raise InvalidRequestException("frontend_callback_url 必须是绝对 URL")
+
+        if parsed.scheme not in {"http", "https"}:
+            raise InvalidRequestException("frontend_callback_url scheme 必须是 http/https")
+
+        path = (parsed.path or "").rstrip("/")
+        if not path.endswith("/auth/callback"):
+            raise InvalidRequestException("frontend_callback_url 路径必须以 /auth/callback 结尾")
+
+    @staticmethod
+    def _validate_redirect_uri(url: str) -> None:
+        parsed = urlparse(url)
+        if not parsed.scheme or not parsed.netloc:
+            raise InvalidRequestException("redirect_uri 必须是绝对 URL")
+        if parsed.scheme not in {"http", "https"}:
+            raise InvalidRequestException("redirect_uri scheme 必须是 http/https")
+
+    @staticmethod
+    def _validate_url_override(provider: OAuthProviderBase, url: str) -> None:
+        parsed = urlparse(url)
+        if parsed.scheme != "https" or not parsed.netloc:
+            raise InvalidRequestException("端点覆盖必须是 https 绝对 URL")
+
+        host = (parsed.hostname or "").lower().rstrip(".")
+        allowed = False
+        for domain in provider.allowed_domains:
+            d = domain.lower().rstrip(".")
+            if host == d or host.endswith(f".{d}"):
+                allowed = True
+                break
+        if not allowed:
+            raise InvalidRequestException("端点覆盖不在允许的域名白名单中")
+
+    @staticmethod
+    async def test_provider_config(db: Session, provider_type: str) -> dict[str, Any]:
+        provider = OAuthService._get_provider_impl(provider_type)
+        if not provider:
+            return {
+                "authorization_url_reachable": False,
+                "token_url_reachable": False,
+                "secret_status": "unknown",
+                "details": "provider 未安装/不可用",
+            }
+
+        cfg = OAuthService._get_provider_config(db, provider_type)
+
+        auth_url = provider.get_effective_authorization_url(cfg)
+        token_url = provider.get_effective_token_url(cfg)
+
+        async def _reachable(url: str) -> bool:
+            try:
+                async with httpx.AsyncClient(timeout=httpx.Timeout(5.0), follow_redirects=False) as client:
+                    await client.get(url)
+                return True
+            except Exception:
+                return False
+
+        authorization_url_reachable = await _reachable(auth_url)
+        token_url_reachable = await _reachable(token_url)
+
+        secret_status = "unknown"
+        details = ""
+
+        if cfg.client_secret_encrypted:
+            # 使用无效 code 做一次 token 请求（仅做粗略判定）
+            try:
+                async with httpx.AsyncClient(timeout=httpx.Timeout(5.0)) as client:
+                    resp = await client.post(
+                        token_url,
+                        data={
+                            "grant_type": "authorization_code",
+                            "code": "invalid",
+                            "redirect_uri": cfg.redirect_uri,
+                            "client_id": cfg.client_id,
+                            "client_secret": cfg.get_client_secret(),
+                        },
+                    )
+                try:
+                    body = resp.json()
+                except Exception:
+                    body = {}
+
+                err = str(body.get("error") or "").lower()
+                if err in {"invalid_client", "unauthorized_client"}:
+                    secret_status = "invalid"
+                elif err in {"invalid_grant", "invalid_code"}:
+                    secret_status = "likely_valid"
+                else:
+                    secret_status = "unknown"
+                details = f"status={resp.status_code}"
+            except Exception as exc:
+                secret_status = "unknown"
+                details = str(exc)
+
+        return {
+            "authorization_url_reachable": bool(authorization_url_reachable),
+            "token_url_reachable": bool(token_url_reachable),
+            "secret_status": secret_status,
+            "details": details,
+        }
+
+    @staticmethod
+    async def test_provider_config_with_data(
+        provider_type: str,
+        client_id: str,
+        client_secret: Optional[str],
+        authorization_url_override: Optional[str],
+        token_url_override: Optional[str],
+        redirect_uri: str,
+    ) -> dict[str, Any]:
+        """使用传入的表单数据测试配置，而非从数据库读取"""
+        provider = OAuthService._get_provider_impl(provider_type)
+        if not provider:
+            return {
+                "authorization_url_reachable": False,
+                "token_url_reachable": False,
+                "secret_status": "unknown",
+                "details": "provider 未安装/不可用",
+            }
+
+        # 使用传入的 override URL 或 provider 默认值
+        auth_url = authorization_url_override or provider.authorization_url
+        token_url = token_url_override or provider.token_url
+
+        async def _reachable(url: str) -> bool:
+            try:
+                async with httpx.AsyncClient(timeout=httpx.Timeout(5.0), follow_redirects=False) as client:
+                    await client.get(url)
+                return True
+            except Exception:
+                return False
+
+        authorization_url_reachable = await _reachable(auth_url)
+        token_url_reachable = await _reachable(token_url)
+
+        secret_status = "unknown"
+        details = ""
+
+        if client_secret:
+            # 使用无效 code 做一次 token 请求（仅做粗略判定）
+            try:
+                async with httpx.AsyncClient(timeout=httpx.Timeout(5.0)) as client:
+                    resp = await client.post(
+                        token_url,
+                        data={
+                            "grant_type": "authorization_code",
+                            "code": "invalid",
+                            "redirect_uri": redirect_uri,
+                            "client_id": client_id,
+                            "client_secret": client_secret,
+                        },
+                    )
+                try:
+                    body = resp.json()
+                except Exception:
+                    body = {}
+
+                err = str(body.get("error") or "").lower()
+                if err in {"invalid_client", "unauthorized_client"}:
+                    secret_status = "invalid"
+                elif err in {"invalid_grant", "invalid_code"}:
+                    secret_status = "likely_valid"
+                else:
+                    secret_status = "unknown"
+                details = f"status={resp.status_code}"
+            except Exception as exc:
+                secret_status = "unknown"
+                details = str(exc)
+        else:
+            secret_status = "not_provided"
+
+        return {
+            "authorization_url_reachable": bool(authorization_url_reachable),
+            "token_url_reachable": bool(token_url_reachable),
+            "secret_status": secret_status,
+            "details": details,
+        }
+
+    @staticmethod
+    async def unbind_provider(db: Session, user: User, provider_type: str) -> None:
+        OAuthService._require_module_active(db)
+
+        if user.auth_source == AuthSource.LDAP:
+            raise HTTPException(status_code=403, detail="LDAP 用户不允许解绑 OAuth")
+
+        link = (
+            db.query(UserOAuthLink)
+            .filter(UserOAuthLink.user_id == user.id, UserOAuthLink.provider_type == provider_type)
+            .first()
+        )
+        if not link:
+            raise InvalidRequestException("未绑定该 Provider")
+
+        total_links = db.query(func.count(UserOAuthLink.id)).filter(UserOAuthLink.user_id == user.id).scalar() or 0
+
+        if user.auth_source == AuthSource.OAUTH and total_links <= 1:
+            raise InvalidRequestException("OAUTH 用户必须至少保留一个 OAuth 绑定")
+
+        # 本地用户无密码时，解绑最后一个 OAuth 会导致无法登录
+        if user.auth_source == AuthSource.LOCAL and not user.password_hash and total_links <= 1:
+            raise InvalidRequestException("请先设置密码后再解绑")
+
+        if LDAPService.is_ldap_exclusive(db) and user.auth_source == AuthSource.LOCAL and user.role != UserRole.ADMIN:
+            # ldap_exclusive=true 时，普通本地用户解绑最后一个 OAuth 会锁死（密码登录被禁用）
+            if total_links <= 1:
+                raise InvalidRequestException("当前处于 LDAP 专属模式，解绑后将无法登录")
+
+        db.delete(link)
+        db.commit()

src/services/auth/oauth/state.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import json
+import secrets
+import time
+from dataclasses import dataclass
+from typing import Any, Awaitable, Optional, cast
+
+from redis.asyncio import Redis
+
+
+OAUTH_STATE_TTL_SECONDS = 600
+OAUTH_STATE_KEY_PREFIX = "oauth_state:"
+
+
+CONSUME_STATE_SCRIPT = r"""
+local value = redis.call("GET", KEYS[1])
+if value then
+    redis.call("DEL", KEYS[1])
+end
+return value
+"""
+
+
+@dataclass(frozen=True)
+class OAuthStateData:
+    nonce: str
+    provider_type: str
+    action: str  # "login" | "bind"
+    user_id: Optional[str]
+    created_at: int
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "OAuthStateData":
+        return cls(
+            nonce=str(data.get("nonce") or ""),
+            provider_type=str(data.get("provider_type") or ""),
+            action=str(data.get("action") or ""),
+            user_id=data.get("user_id"),
+            created_at=int(data.get("created_at") or 0),
+        )
+
+
+def _state_key(nonce: str) -> str:
+    return f"{OAUTH_STATE_KEY_PREFIX}{nonce}"
+
+
+async def create_oauth_state(
+    redis: Redis, *, provider_type: str, action: str, user_id: Optional[str] = None
+) -> str:
+    nonce = secrets.token_urlsafe(24)
+    data = {
+        "nonce": nonce,
+        "provider_type": provider_type,
+        "action": action,
+        "user_id": user_id,
+        "created_at": int(time.time()),
+    }
+    await redis.setex(_state_key(nonce), OAUTH_STATE_TTL_SECONDS, json.dumps(data))
+    return nonce
+
+
+async def consume_oauth_state(redis: Redis, nonce: str) -> Optional[OAuthStateData]:
+    if not nonce:
+        return None
+
+    key = _state_key(nonce)
+    # redis-py 的类型标注在 sync/async 之间会出现 Union；这里明确按 async 处理。
+    raw = await cast(Awaitable[Optional[str]], redis.eval(CONSUME_STATE_SCRIPT, 1, key))
+    if not raw:
+        return None
+
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError:
+        return None
+
+    return OAuthStateData.from_dict(parsed)

src/services/auth/service.py

@@ -91,6 +91,43 @@ REFRESH_TOKEN_EXPIRATION_DAYS = 7
 class AuthService:
     """认证服务"""
 
+    @staticmethod
+    def token_identity_matches_user(payload: Dict[str, Any], user: User) -> bool:
+        """
+        校验 token 的身份字段是否与用户一致。
+
+        兼容策略：
+        - email：旧 token 可能包含；新 token 允许不包含（支持无邮箱用户）
+        - created_at：用于替代 email 作为"防止身份混淆"的校验字段；旧 token 可能没有
+
+        时区处理说明：
+        - 本项目所有 created_at 统一使用 UTC 时区存储（PostgreSQL TIMESTAMPTZ）
+        - 对于 naive datetime（无时区信息），假定为 UTC
+        - 若历史数据使用了非 UTC 本地时区的 naive datetime，可能导致校验失败
+        """
+        token_email = payload.get("email")
+        if token_email is not None and user.email is not None and user.email != token_email:
+            return False
+
+        token_created_at = payload.get("created_at")
+        if not token_created_at or not user.created_at:
+            return True
+
+        try:
+            token_created = datetime.fromisoformat(str(token_created_at).replace("Z", "+00:00"))
+        except ValueError:
+            return False
+
+        # 统一时区：若是 naive datetime，按 UTC 处理
+        # 注意：本项目约定所有时间戳使用 UTC，若旧数据不符合此约定可能导致校验失败
+        user_created = user.created_at
+        if user_created.tzinfo is None:
+            user_created = user_created.replace(tzinfo=timezone.utc)
+        if token_created.tzinfo is None:
+            token_created = token_created.replace(tzinfo=timezone.utc)
+
+        return abs((user_created - token_created).total_seconds()) <= 1
+
     @staticmethod
     def create_access_token(data: dict) -> str:
         """创建JWT访问令牌"""
@@ -194,6 +231,9 @@ class AuthService:
             if not user:
                 # 已有本地账号但来源不匹配等情况
                 return None
+            if user.is_deleted:
+                logger.warning(f"登录失败 - 用户已删除: {email}")
+                return None
             if not user.is_active:
                 logger.warning(f"登录失败 - 用户已禁用: {email}")
                 return None
@@ -211,6 +251,10 @@ class AuthService:
             logger.warning(f"登录失败 - 用户不存在: {email}")
             return None
 
+        if user.is_deleted:
+            logger.warning(f"登录失败 - 用户已删除: {email}")
+            return None
+
         # 检查 LDAP exclusive 模式：仅允许本地管理员登录（紧急恢复通道）
         if LDAPService.is_ldap_exclusive(db):
             if user.role != UserRole.ADMIN or user.auth_source != AuthSource.LOCAL:
@@ -275,6 +319,10 @@ class AuthService:
             user = db.query(User).filter(User.email == email).with_for_update().first()
 
         if user:
+            if user.is_deleted:
+                logger.warning(f"LDAP 登录失败 - 用户已删除: {email}")
+                return None
+
             if user.auth_source != AuthSource.LDAP:
                 # 避免覆盖已有本地账户（不同来源时拒绝登录）
                 logger.warning(
@@ -293,6 +341,7 @@ class AuthService:
                     logger.warning(f"LDAP 登录拒绝 - 新邮箱已被占用: {email}")
                     return None
                 user.email = email
+                user.email_verified = True
 
             # 同步 LDAP 标识（首次填充或 LDAP 侧发生变化）
             if ldap_dn and user.ldap_dn != ldap_dn:
@@ -325,8 +374,9 @@ class AuthService:
             # 创建新用户
             user = User(
                 email=email,
+                email_verified=True,
                 username=username,
-                password_hash="",  # LDAP 用户无本地密码
+                password_hash=None,  # LDAP 用户无本地密码
                 auth_source=AuthSource.LDAP,
                 ldap_dn=ldap_dn,
                 ldap_username=ldap_username,
@@ -416,6 +466,9 @@ class AuthService:
         if not user.is_active:
             logger.warning(f"API认证失败 - 用户已禁用: {user.email}")
             return None
+        if user.is_deleted:
+            logger.warning(f"API认证失败 - 用户已删除: {user.email}")
+            return None
 
         # 更新最后使用时间（使用节流策略，减少数据库写入）
         if _should_update_last_used(key_record.id):
@@ -637,6 +690,9 @@ class AuthService:
         if not user or not user.is_active:
             logger.warning("Management Token 认证失败 - 用户不存在或已禁用")
             return None
+        if user.is_deleted:
+            logger.warning("Management Token 认证失败 - 用户不存在或已禁用")
+            return None
 
         # 使用 SQL 原子操作更新使用统计
         from sqlalchemy import func

src/services/cache/user_cache.py

@@ -126,9 +126,11 @@ class UserCacheService:
         return {
             "id": user.id,
             "email": user.email,
+            "email_verified": user.email_verified,
             "username": user.username,
             "role": user.role.value if user.role else None,
             "is_active": user.is_active,
+            "auth_source": user.auth_source.value if user.auth_source else None,
             "quota_usd": float(user.quota_usd) if user.quota_usd is not None else None,
             "used_usd": float(user.used_usd),
             "created_at": user.created_at.isoformat() if user.created_at else None,
@@ -146,11 +148,13 @@ class UserCacheService:
         """
         from datetime import datetime
 
+        from src.core.enums import AuthSource
         from src.models.database import UserRole
 
         user = User(
             id=user_dict["id"],
-            email=user_dict["email"],
+            email=user_dict.get("email"),
+            email_verified=user_dict.get("email_verified", False),
             username=user_dict["username"],
             is_active=user_dict["is_active"],
             used_usd=user_dict["used_usd"],
@@ -160,6 +164,9 @@ class UserCacheService:
         if user_dict.get("role"):
             user.role = UserRole(user_dict["role"])
 
+        if user_dict.get("auth_source"):
+            user.auth_source = AuthSource(user_dict["auth_source"])
+
         if user_dict.get("quota_usd") is not None:
             user.quota_usd = user_dict["quota_usd"]
 

src/services/email/email_sender.py

@@ -98,6 +98,21 @@ class EmailSenderService:
 
         return True, None
 
+    @staticmethod
+    def is_smtp_configured(db: Session) -> bool:
+        """
+        检查 SMTP 是否已配置（用于前端显示判断）
+
+        Args:
+            db: 数据库会话
+
+        Returns:
+            是否已配置有效的 SMTP
+        """
+        config = EmailSenderService._get_smtp_config(db)
+        valid, _ = EmailSenderService._validate_smtp_config(config)
+        return valid
+
     @staticmethod
     async def send_verification_code(
         db: Session, to_email: str, code: str, expire_minutes: int = 30

src/services/model/fetch_scheduler.py

@@ -15,15 +15,17 @@ import os
 from datetime import datetime, timezone
 from typing import Optional
 
-import httpx
 from sqlalchemy.orm import Session, joinedload
 
 from src.core.cache_service import CacheService
 from src.core.crypto import crypto_service
-from src.core.headers import get_extra_headers_from_endpoint
 from src.core.logger import logger
 from src.database import create_session
-from src.models.database import Provider, ProviderAPIKey, ProviderEndpoint
+from src.models.database import Provider, ProviderAPIKey
+from src.services.model.upstream_fetcher import (
+    build_all_format_configs,
+    fetch_models_from_endpoints,
+)
 from src.services.system.scheduler import get_scheduler
 
 # 从环境变量读取间隔，默认 1440 分钟（1 天），限制在 60-10080 分钟之间
@@ -66,21 +68,6 @@ async def set_upstream_models_to_cache(
     logger.debug(f"上游模型已缓存: {cache_key}, 数量={len(models)}")
 
 
-def _get_adapter_for_format(api_format: str) -> Optional[type]:
-    """根据 API 格式获取对应的 Adapter 类"""
-    # 延迟导入避免循环依赖
-    from src.api.handlers.base.chat_adapter_base import get_adapter_class
-    from src.api.handlers.base.cli_adapter_base import get_cli_adapter_class
-
-    adapter_class = get_adapter_class(api_format)
-    if adapter_class:
-        return adapter_class
-    cli_adapter_class = get_cli_adapter_class(api_format)
-    if cli_adapter_class:
-        return cli_adapter_class
-    return None
-
-
 class ModelFetchScheduler:
     """模型自动获取调度器"""
 
@@ -277,7 +264,7 @@ class ModelFetchScheduler:
             return "error"
 
         # 构建 api_format -> endpoint 映射
-        format_to_endpoint: dict[str, ProviderEndpoint] = {}
+        format_to_endpoint: dict[str, object] = {}
         for endpoint in provider.endpoints:  # type: ignore[attr-defined]
             if endpoint.is_active:
                 format_to_endpoint[endpoint.api_format] = endpoint
@@ -288,29 +275,11 @@ class ModelFetchScheduler:
             key.last_models_fetch_at = now
             return "error"
 
-        # 收集端点配置
-        endpoint_configs: list[dict] = []
-        key_formats = key.api_formats or []
-        for fmt in key_formats:
-            endpoint = format_to_endpoint.get(fmt)
-            if endpoint:
-                endpoint_configs.append(
-                    {
-                        "api_key": api_key_value,
-                        "base_url": endpoint.base_url,
-                        "api_format": fmt,
-                        "extra_headers": get_extra_headers_from_endpoint(endpoint),
-                    }
-                )
-
-        if not endpoint_configs:
-            logger.warning(f"Provider {provider.name} 没有匹配 Key {key.id} 格式的端点配置")
-            key.last_models_fetch_error = "No matching endpoints for key formats"
-            key.last_models_fetch_at = now
-            return "error"
+        # 使用公共函数构建所有格式的端点配置
+        endpoint_configs = build_all_format_configs(api_key_value, format_to_endpoint)  # type: ignore[arg-type]
 
         # 并发获取模型
-        all_models, errors, has_success = await self._fetch_models_from_endpoints(endpoint_configs)
+        all_models, errors, has_success = await fetch_models_from_endpoints(endpoint_configs)
 
         # 记录获取结果
         error_msg = "; ".join(errors) if errors else None
@@ -401,62 +370,6 @@ class ModelFetchScheduler:
             logger.debug(f"Key {key.id} 模型列表无变化")
             return False
 
-    async def _fetch_models_from_endpoints(
-        self, endpoint_configs: list[dict]
-    ) -> tuple[list[dict], list[str], bool]:
-        """从多个端点并发获取模型，返回 (模型列表, 错误列表, 是否有成功)"""
-        all_models: list[dict] = []
-        errors: list[str] = []
-        has_success = False
-        semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
-
-        async def fetch_one(
-            client: httpx.AsyncClient, config: dict
-        ) -> tuple[list, Optional[str], bool]:
-            base_url = config["base_url"]
-            if not base_url:
-                return [], None, False
-            base_url = base_url.rstrip("/")
-            api_format = config["api_format"]
-            api_key_value = config["api_key"]
-            extra_headers = config.get("extra_headers")
-
-            try:
-                adapter_class = _get_adapter_for_format(api_format)
-                if not adapter_class:
-                    return [], f"Unknown API format: {api_format}", False
-
-                async with semaphore:
-                    models, error = await adapter_class.fetch_models(  # type: ignore[attr-defined]
-                        client, base_url, api_key_value, extra_headers
-                    )
-
-                for m in models:
-                    if "api_format" not in m:
-                        m["api_format"] = api_format
-
-                # 即使返回空列表，只要没有错误也算成功
-                success = error is None
-                return models, error, success
-            except httpx.TimeoutException:
-                logger.warning(f"获取 {api_format} 模型超时")
-                return [], f"{api_format}: timeout", False
-            except Exception as e:
-                # 只记录异常类型，避免泄露敏感信息
-                logger.exception(f"获取 {api_format} 模型出错")
-                return [], f"{api_format}: {type(e).__name__}", False
-
-        async with httpx.AsyncClient(timeout=30.0) as client:
-            results = await asyncio.gather(*[fetch_one(client, c) for c in endpoint_configs])
-            for models, error, success in results:
-                all_models.extend(models)
-                if error:
-                    errors.append(error)
-                if success:
-                    has_success = True
-
-        return all_models, errors, has_success
-
 
 # 单例模式
 _model_fetch_scheduler: Optional[ModelFetchScheduler] = None

src/services/model/upstream_fetcher.py

@@ -0,0 +1,160 @@
+"""
+上游模型获取公共模块
+
+提供从上游 API 获取模型列表的公共函数，供以下场景使用：
+- 定时任务自动获取（fetch_scheduler.py）
+- 管理后台手动查询（provider_query.py）
+"""
+
+import asyncio
+from typing import Dict, Optional
+
+import httpx
+
+from src.core.enums import APIFormat
+from src.core.headers import get_extra_headers_from_endpoint
+from src.core.logger import logger
+from src.models.database import ProviderEndpoint
+
+# 并发请求限制
+MAX_CONCURRENT_REQUESTS = 5
+
+
+def _get_adapter_for_format(api_format: str) -> Optional[type]:
+    """根据 API 格式获取对应的 Adapter 类"""
+    from src.api.handlers.base.chat_adapter_base import get_adapter_class
+    from src.api.handlers.base.cli_adapter_base import get_cli_adapter_class
+
+    adapter_class = get_adapter_class(api_format)
+    if adapter_class:
+        return adapter_class
+    cli_adapter_class = get_cli_adapter_class(api_format)
+    if cli_adapter_class:
+        return cli_adapter_class
+    return None
+
+
+def build_all_format_configs(
+    api_key_value: str,
+    format_to_endpoint: Dict[str, ProviderEndpoint],
+) -> list[dict]:
+    """
+    构建所有 API 格式的端点配置
+
+    从所有 APIFormat 枚举值构建配置，如果该格式有专门的端点配置则使用，
+    否则使用基础端点的 base_url 尝试。
+
+    Args:
+        api_key_value: 解密后的 API Key
+        format_to_endpoint: API 格式到端点的映射
+
+    Returns:
+        端点配置列表，每个配置包含 api_key, base_url, api_format, extra_headers
+    """
+    if not format_to_endpoint:
+        return []
+
+    # 获取任意一个端点的 base_url 作为基础（用于尝试所有格式）
+    # 优先使用 OPENAI 格式的端点，因为它最通用
+    base_endpoint = (
+        format_to_endpoint.get("OPENAI")
+        or format_to_endpoint.get("CLAUDE")
+        or format_to_endpoint.get("GEMINI")
+        or next(iter(format_to_endpoint.values()))
+    )
+    base_url = base_endpoint.base_url
+    extra_headers = get_extra_headers_from_endpoint(base_endpoint)
+
+    # 从所有 API 格式都尝试获取模型，然后聚合去重
+    endpoint_configs: list[dict] = []
+    for fmt in APIFormat:
+        fmt_value = fmt.value
+        # 如果该格式有专门的端点配置，使用其 base_url 和 headers
+        if fmt_value in format_to_endpoint:
+            ep = format_to_endpoint[fmt_value]
+            endpoint_configs.append(
+                {
+                    "api_key": api_key_value,
+                    "base_url": ep.base_url,
+                    "api_format": fmt_value,
+                    "extra_headers": get_extra_headers_from_endpoint(ep),
+                }
+            )
+        else:
+            # 没有专门配置，使用基础端点的 base_url 尝试
+            endpoint_configs.append(
+                {
+                    "api_key": api_key_value,
+                    "base_url": base_url,
+                    "api_format": fmt_value,
+                    "extra_headers": extra_headers,
+                }
+            )
+
+    return endpoint_configs
+
+
+async def fetch_models_from_endpoints(
+    endpoint_configs: list[dict],
+    timeout: float = 30.0,
+) -> tuple[list[dict], list[str], bool]:
+    """
+    从多个端点并发获取模型
+
+    Args:
+        endpoint_configs: 端点配置列表，每个配置包含 api_key, base_url, api_format, extra_headers
+        timeout: 请求超时时间（秒）
+
+    Returns:
+        (模型列表, 错误列表, 是否有成功)
+    """
+    all_models: list[dict] = []
+    errors: list[str] = []
+    has_success = False
+    semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
+
+    async def fetch_one(
+        client: httpx.AsyncClient, config: dict
+    ) -> tuple[list, Optional[str], bool]:
+        base_url = config["base_url"]
+        if not base_url:
+            return [], None, False
+        base_url = base_url.rstrip("/")
+        api_format = config["api_format"]
+        api_key_value = config["api_key"]
+        extra_headers = config.get("extra_headers")
+
+        try:
+            adapter_class = _get_adapter_for_format(api_format)
+            if not adapter_class:
+                return [], f"Unknown API format: {api_format}", False
+
+            async with semaphore:
+                models, error = await adapter_class.fetch_models(  # type: ignore[attr-defined]
+                    client, base_url, api_key_value, extra_headers
+                )
+
+            for m in models:
+                if "api_format" not in m:
+                    m["api_format"] = api_format
+
+            # 即使返回空列表，只要没有错误也算成功
+            success = error is None
+            return models, error, success
+        except httpx.TimeoutException:
+            logger.warning(f"获取 {api_format} 模型超时")
+            return [], f"{api_format}: timeout", False
+        except Exception:
+            logger.exception(f"获取 {api_format} 模型出错")
+            return [], f"{api_format}: error", False
+
+    async with httpx.AsyncClient(timeout=timeout) as client:
+        results = await asyncio.gather(*[fetch_one(client, c) for c in endpoint_configs])
+        for models, error, success in results:
+            all_models.extend(models)
+            if error:
+                errors.append(error)
+            if success:
+                has_success = True
+
+    return all_models, errors, has_success

src/services/user/preference.py

@@ -109,6 +109,8 @@ class PreferenceService:
             "is_active": user.is_active,
             "created_at": user.created_at,
             "last_login_at": user.last_login_at,
+            "auth_source": user.auth_source.value if user.auth_source else "local",
+            "has_password": bool(user.password_hash),
             "preferences": {
                 "avatar_url": preferences.avatar_url,
                 "bio": preferences.bio,

src/services/user/service.py

@@ -25,18 +25,23 @@ class UserService:
     @retry_on_database_error(max_retries=3)
     def create_user(
         db: Session,
-        email: str,
+        email: Optional[str],
         username: str,
         password: str,
         role: UserRole = UserRole.USER,
         quota_usd: Optional[float] = 10.0,
+        email_verified: bool = False,
     ) -> User:
-        """创建新用户，quota_usd 为 None 表示无限制"""
+        """创建新用户，quota_usd 为 None 表示无限制，email 为 None 表示无邮箱"""
 
-        # 验证邮箱格式
-        valid, error_msg = EmailValidator.validate(email)
-        if not valid:
-            raise ValueError(error_msg)
+        # 验证邮箱格式（仅当提供邮箱时）
+        if email is not None:
+            valid, error_msg = EmailValidator.validate(email)
+            if not valid:
+                raise ValueError(error_msg)
+            # 检查邮箱是否已存在
+            if db.query(User).filter(User.email == email).first():
+                raise ValueError(f"邮箱已存在: {email}")
 
         # 验证用户名格式
         valid, error_msg = UsernameValidator.validate(username)
@@ -48,16 +53,13 @@ class UserService:
         if not valid:
             raise ValueError(error_msg)
 
-        # 检查邮箱是否已存在
-        if db.query(User).filter(User.email == email).first():
-            raise ValueError(f"邮箱已存在: {email}")
-
         # 检查用户名是否已存在
         if db.query(User).filter(User.username == username).first():
             raise ValueError(f"用户名已存在: {username}")
 
         user = User(
             email=email,
+            email_verified=email_verified if email else False,
             username=username,
             role=role,
             quota_usd=quota_usd,
@@ -69,7 +71,8 @@ class UserService:
         db.commit()  # 立即提交事务,释放数据库锁
         db.refresh(user)
 
-        logger.info(f"创建新用户: {email} (ID: {user.id}, 角色: {role.value})")
+        log_identifier = email if email else username
+        logger.info(f"创建新用户: {log_identifier} (ID: {user.id}, 角色: {role.value})")
         return user
 
     @staticmethod

src/utils/auth_utils.py

@@ -59,16 +59,12 @@ async def get_current_user(
             raise ForbiddenException("无效的Token")
 
         user_id = payload.get("user_id")
-        token_email = payload.get("email")
-        token_created_at = payload.get("created_at")
 
         if not user_id:
-            logger.error(f"Token缺少user_id字段: payload={payload}")
+            logger.error("Token缺少user_id字段: payload={}", payload)
             raise ForbiddenException("无效的认证凭据")
 
-        if not token_email:
-            logger.error(f"Token缺少email字段: payload={payload}")
-            raise ForbiddenException("无效的认证凭据")
+        # 兼容旧 token：email 字段可能存在；新 token 不再包含 email（支持无邮箱用户）
 
         # 仅在DEBUG模式下记录详细信息
         token_fp = hashlib.sha256(token.encode()).hexdigest()[:12]
@@ -76,7 +72,7 @@ async def get_current_user(
 
         # 确保user_id是字符串格式（UUID）
         if not isinstance(user_id, str):
-            logger.error(f"Token中user_id格式错误: {type(user_id)} - {user_id}")
+            logger.error("Token中user_id格式错误: {} - {}", type(user_id), user_id)
             raise ForbiddenException("认证信息格式错误，请重新登录")
 
         # 使用新的数据库会话获取用户，避免会话状态问题
@@ -85,46 +81,35 @@ async def get_current_user(
 
             user = UserService.get_user(db, user_id)
         except Exception as db_error:
-            logger.error(f"数据库查询失败: user_id={user_id}, error={db_error}")
+            logger.error("数据库查询失败: user_id={}, error={}", user_id, db_error)
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail="数据库查询失败，请稍后重试",
             )
 
         if not user:
-            logger.error(f"用户不存在: user_id={user_id}")
+            logger.error("用户不存在: user_id={}", user_id)
             raise ForbiddenException("用户不存在或已禁用")
 
         if not user.is_active:
-            logger.error(f"用户已禁用: user_id={user_id}")
+            logger.error("用户已禁用: user_id={}", user_id)
             raise ForbiddenException("用户不存在或已禁用")
 
-        # 验证邮箱是否匹配（防止用户ID重用导致的身份混淆）
-        if user.email != token_email:
-            logger.error(f"Token邮箱不匹配: Token中的邮箱={token_email}, 数据库中的邮箱={user.email}")
+        if user.is_deleted:
+            logger.error("用户已删除: user_id={}", user_id)
+            raise ForbiddenException("用户不存在或已禁用")
+
+        if not AuthService.token_identity_matches_user(payload, user):
+            logger.error("Token身份校验失败: user_id={}, token_fp={}", user_id, token_fp)
             raise ForbiddenException("身份验证失败")
 
-        # 验证用户创建时间是否匹配（防止ID重用）
-        if token_created_at and user.created_at:
-            try:
-                from datetime import datetime
-
-                token_created = datetime.fromisoformat(token_created_at.replace("Z", "+00:00"))
-                # 允许1秒的时间差异（考虑到时间精度问题）
-                time_diff = abs((user.created_at - token_created).total_seconds())
-                if time_diff > 1:
-                    logger.error(f"Token创建时间不匹配: Token时间={token_created_at}, 用户创建时间={user.created_at}")
-                    raise ForbiddenException("身份验证失败")
-            except ValueError as e:
-                logger.warning(f"Token时间格式解析失败: {e}")
-
-        logger.debug(f"成功获取用户: user_id={user_id}, email={user.email}")
+        logger.debug("成功获取用户: user_id={}, email={}", user_id, user.email)
         return user
 
     except HTTPException:
         raise
     except Exception as e:
-        logger.error(f"认证失败，未预期的错误: {e}")
+        logger.error("认证失败，未预期的错误: {}", e)
         # 返回500而不是401，避免触发前端的退出逻辑
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="认证服务暂时不可用"
@@ -166,6 +151,12 @@ async def get_current_user_from_header(
         if not user.is_active:
             raise ForbiddenException("用户已被禁用")
 
+        if user.is_deleted:
+            raise ForbiddenException("用户不存在或已禁用")
+
+        if not AuthService.token_identity_matches_user(payload, user):
+            raise ForbiddenException("身份验证失败")
+
         return user
     except HTTPException:
         # 保持原始的HTTPException (包括401)