feat(ldap): 完善 LDAP 认证功能和安全性

- 添加 LDAP 配置类型定义，移除 any 类型
- 首次配置 LDAP 时强制要求设置绑定密码
- 根据认证类型区分登录标识验证（本地需邮箱，LDAP 允许用户名）
- 添加 LDAP 过滤器转义函数防止注入攻击
- 增加 LDAP 连接超时设置
- 添加账户来源冲突检查，防止 LDAP 覆盖本地账户
- 添加用户名冲突自动重命名机制

src/_version.py

@@ -28,7 +28,7 @@ version_tuple: VERSION_TUPLE
 commit_id: COMMIT_ID
 __commit_id__: COMMIT_ID
 
-__version__ = version = '0.1.1.dev0+g393d4d13f.d20251213'
-__version_tuple__ = version_tuple = (0, 1, 1, 'dev0', 'g393d4d13f.d20251213')
+__version__ = version = '0.2.1.dev3+g612992fa1.d20260104'
+__version_tuple__ = version_tuple = (0, 2, 1, 'dev3', 'g612992fa1.d20260104')
 
 __commit_id__ = commit_id = None

src/api/admin/ldap.py

@@ -132,8 +132,12 @@ class AdminUpdateLDAPConfigAdapter(AdminApiAdapter):
             raise InvalidRequestException("请求数据验证失败")
 
         config = db.query(LDAPConfig).first()
+        is_new_config = config is None
 
-        if not config:
+        if is_new_config:
+            # 首次创建配置时必须提供密码
+            if not config_update.bind_password:
+                raise InvalidRequestException("首次配置 LDAP 时必须设置绑定密码")
             config = LDAPConfig()
             db.add(config)
 

src/models/api.py

@@ -6,7 +6,7 @@ import re
 from datetime import datetime
 from typing import Any, Dict, List, Literal, Optional
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from ..core.enums import UserRole
 
@@ -15,19 +15,10 @@ from ..core.enums import UserRole
 class LoginRequest(BaseModel):
     """登录请求"""
 
-    email: str = Field(..., min_length=3, max_length=255, description="邮箱地址")
+    email: str = Field(..., min_length=1, max_length=255, description="邮箱/用户名")
     password: str = Field(..., min_length=1, max_length=128, description="密码")
     auth_type: Literal["local", "ldap"] = Field(default="local", description="认证类型")
 
-    @classmethod
-    @field_validator("email")
-    def validate_email(cls, v):
-        """验证邮箱格式"""
-        email_pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
-        if not re.match(email_pattern, v):
-            raise ValueError("邮箱格式无效")
-        return v.lower()
-
     @classmethod
     @field_validator("password")
     def validate_password(cls, v):
@@ -37,6 +28,23 @@ class LoginRequest(BaseModel):
             raise ValueError("密码不能为空")
         return v
 
+    @model_validator(mode="after")
+    def validate_login(self):
+        """根据认证类型校验并规范化登录标识"""
+        identifier = self.email.strip()
+
+        if self.auth_type == "local":
+            email_pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
+            if not re.match(email_pattern, identifier):
+                raise ValueError("邮箱格式无效")
+            self.email = identifier.lower()
+        else:
+            if not identifier:
+                raise ValueError("用户名/邮箱不能为空")
+            self.email = identifier
+
+        return self
+
 
 class LoginResponse(BaseModel):
     """登录响应"""

src/services/auth/ldap.py

@@ -7,6 +7,32 @@ from sqlalchemy.orm import Session
 from src.core.logger import logger
 from src.models.database import LDAPConfig
 
+# LDAP 连接超时时间（秒）
+LDAP_CONNECT_TIMEOUT = 10
+
+
+def escape_ldap_filter(value: str) -> str:
+    """
+    转义 LDAP 过滤器中的特殊字符，防止 LDAP 注入攻击
+
+    Args:
+        value: 需要转义的字符串
+
+    Returns:
+        转义后的安全字符串
+    """
+    # LDAP 过滤器特殊字符: \ * ( ) NUL
+    escape_chars = {
+        "\\": r"\5c",
+        "*": r"\2a",
+        "(": r"\28",
+        ")": r"\29",
+        "\x00": r"\00",
+    }
+    for char, escaped in escape_chars.items():
+        value = value.replace(char, escaped)
+    return value
+
 
 class LDAPService:
     """LDAP 认证服务"""
@@ -57,7 +83,12 @@ class LDAPService:
         try:
             # 创建服务器连接
             use_ssl = config.server_url.startswith("ldaps://")
-            server = Server(config.server_url, use_ssl=use_ssl, get_info=ldap3.ALL)
+            server = Server(
+                config.server_url,
+                use_ssl=use_ssl,
+                get_info=ldap3.ALL,
+                connect_timeout=LDAP_CONNECT_TIMEOUT,
+            )
 
             # 使用管理员账号连接
             bind_password = config.get_bind_password()
@@ -70,8 +101,9 @@ class LDAPService:
                 logger.error(f"LDAP 管理员绑定失败: {admin_conn.result}")
                 return None
 
-            # 搜索用户
-            search_filter = config.user_search_filter.replace("{username}", username)
+            # 搜索用户（转义用户名防止 LDAP 注入）
+            safe_username = escape_ldap_filter(username)
+            search_filter = config.user_search_filter.replace("{username}", safe_username)
             admin_conn.search(
                 search_base=config.base_dn,
                 search_filter=search_filter,
@@ -140,7 +172,12 @@ class LDAPService:
 
         try:
             use_ssl = config.server_url.startswith("ldaps://")
-            server = Server(config.server_url, use_ssl=use_ssl, get_info=ldap3.ALL)
+            server = Server(
+                config.server_url,
+                use_ssl=use_ssl,
+                get_info=ldap3.ALL,
+                connect_timeout=LDAP_CONNECT_TIMEOUT,
+            )
             bind_password = config.get_bind_password()
             conn = Connection(server, user=config.bind_dn, password=bind_password)
 

src/services/auth/service.py

@@ -117,6 +117,12 @@ class AuthService:
 
             # 获取或创建本地用户
             user = await AuthService._get_or_create_ldap_user(db, ldap_user)
+            if not user:
+                # 已有本地账号但来源不匹配等情况
+                return None
+            if not user.is_active:
+                logger.warning(f"登录失败 - 用户已禁用: {email}")
+                return None
             return user
 
         # 本地认证
@@ -154,7 +160,7 @@ class AuthService:
         return user
 
     @staticmethod
-    async def _get_or_create_ldap_user(db: Session, ldap_user: dict) -> User:
+    async def _get_or_create_ldap_user(db: Session, ldap_user: dict) -> Optional[User]:
         """获取或创建 LDAP 用户
 
         Args:
@@ -166,17 +172,33 @@ class AuthService:
         if user:
             # 更新 auth_source（如果是首次 LDAP 登录）
             if user.auth_source != AuthSource.LDAP:
-                user.auth_source = AuthSource.LDAP
+                # 避免覆盖已有本地账户（不同来源时拒绝登录）
+                logger.warning(
+                    f"LDAP 登录拒绝 - 账户来源不匹配(现有:{user.auth_source}, 请求:LDAP): {ldap_user['email']}"
+                )
+                return None
             user.last_login_at = datetime.now(timezone.utc)
             db.commit()
             await UserCacheService.invalidate_user_cache(user.id, user.email)
             logger.info(f"LDAP 用户登录成功: {ldap_user['email']} (ID: {user.id})")
             return user
 
+        # 检查 username 是否已被占用
+        username = ldap_user["username"]
+        existing_user_with_username = db.query(User).filter(User.username == username).first()
+        if existing_user_with_username:
+            # 如果 username 已存在，添加后缀使其唯一
+            base_username = username
+            counter = 1
+            while db.query(User).filter(User.username == username).first():
+                username = f"{base_username}_ldap_{counter}"
+                counter += 1
+            logger.info(f"LDAP 用户名冲突，使用新用户名: {base_username} -> {username}")
+
         # 创建新用户
         user = User(
             email=ldap_user["email"],
-            username=ldap_user["username"],
+            username=username,
             password_hash="",  # LDAP 用户无本地密码
             auth_source=AuthSource.LDAP,
             role=UserRole.USER,