jiazhizhong/LiteOps
1
0
Fork 0
mirror of https://github.com/opsre/LiteOps.git synced 2026-03-16 19:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

✨ feat: 新增LDAP认证

Browse Source
This commit is contained in:
hukdoesn
2025-07-23 13:44:47 +08:00
parent 5c89db4337
commit 82b7614c27
14 changed files with 1495 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
backend/apps/models.py
View File

@@ -10,6 +10,8 @@ class User(models.Model):
name = models.CharField(max_length=50, null=True, verbose_name='姓名')
password = models.CharField(max_length=128, null=True, verbose_name='密码')
email = models.EmailField(max_length=100, unique=True, null=True, verbose_name='邮箱')
user_type = models.CharField(max_length=20, default='system', null=True, verbose_name='用户类型')
ldap_dn = models.CharField(max_length=255, null=True, blank=True, verbose_name='LDAP DN')
status = models.SmallIntegerField(null=True, verbose_name='状态')
login_time = models.DateTimeField(null=True, verbose_name='最后登录时间')
create_time = models.DateTimeField(auto_now_add=True, null=True, verbose_name='创建时间')
@@ -388,4 +390,30 @@ class LoginAttempt(models.Model):
unique_together = ['user', 'ip_address'] # 确保用户和IP组合唯一
def __str__(self):
return f"{self.user.username if self.user else 'Unknown'} - {self.ip_address}"
return f"{self.user.username if self.user else 'Unknown'} - {self.ip_address}"
class LDAPConfig(models.Model):
"""
LDAP配置表 - 存储LDAP服务器配置信息
"""
id = models.AutoField(primary_key=True)
enabled = models.BooleanField(default=False, verbose_name='启用LDAP认证')
server_host = models.CharField(max_length=255, null=True, verbose_name='LDAP服务器地址')
server_port = models.IntegerField(default=389, verbose_name='LDAP服务器端口')
use_ssl = models.BooleanField(default=False, verbose_name='使用SSL/TLS')
base_dn = models.CharField(max_length=255, null=True, verbose_name='Base DN')
bind_dn = models.CharField(max_length=255, null=True, blank=True, verbose_name='绑定DN', help_text='管理员DN，用于搜索用户')
bind_password = models.CharField(max_length=255, null=True, blank=True, verbose_name='绑定密码')
user_search_filter = models.CharField(max_length=255, default='(uid={username})', verbose_name='用户搜索过滤器')
user_attr_map = models.JSONField(default=dict, verbose_name='用户属性映射', help_text='LDAP属性到系统属性的映射')
timeout = models.IntegerField(default=10, verbose_name='连接超时时间(秒)')
update_time = models.DateTimeField(auto_now=True, null=True, verbose_name='更新时间')
class Meta:
db_table = 'ldap_config'
verbose_name = 'LDAP配置'
verbose_name_plural = verbose_name
def __str__(self):
return f"LDAP配置 - {self.server_host}:{self.server_port}"

44
backend/apps/utils/crypto.py Normal file
View File

@@ -0,0 +1,44 @@
import base64
from cryptography.fernet import Fernet
from django.conf import settings
import hashlib
class CryptoUtils:
"""加密解密工具类"""
@staticmethod
def _get_key():
"""获取加密密钥"""
key_material = settings.SECRET_KEY.encode('utf-8')
digest = hashlib.sha256(key_material).digest()
key = base64.urlsafe_b64encode(digest)
return key
@staticmethod
def encrypt_password(password):
"""加密密码"""
if not password:
return password
try:
key = CryptoUtils._get_key()
fernet = Fernet(key)
encrypted = fernet.encrypt(password.encode('utf-8'))
return base64.urlsafe_b64encode(encrypted).decode('utf-8')
except Exception:
return password
@staticmethod
def decrypt_password(encrypted_password):
"""解密密码"""
if not encrypted_password:
return encrypted_password
try:
key = CryptoUtils._get_key()
fernet = Fernet(key)
encrypted_data = base64.urlsafe_b64decode(encrypted_password.encode('utf-8'))
decrypted = fernet.decrypt(encrypted_data)
return decrypted.decode('utf-8')
except Exception:
return encrypted_password

490
backend/apps/views/ldap.py Normal file
View File

@@ -0,0 +1,490 @@
import json
import uuid
import hashlib
import logging
import ldap3
from django.http import JsonResponse
from django.views import View
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt
from django.db import transaction
from ..models import User, LDAPConfig
from ..utils.auth import jwt_auth_required
from ..utils.crypto import CryptoUtils
logger = logging.getLogger('apps')
def generate_id():
"""生成唯一ID"""
return hashlib.sha256(str(uuid.uuid4()).encode()).hexdigest()[:32]
class LDAPError(Exception):
"""LDAP相关异常"""
pass
class APIResponse:
"""统一的API响应处理"""
@staticmethod
def success(message, data=None):
return JsonResponse({'code': 200, 'message': message, 'data': data})
@staticmethod
def error(code, message):
return JsonResponse({'code': code, 'message': message})
@method_decorator(csrf_exempt, name='dispatch')
class LDAPConfigView(View):
@method_decorator(jwt_auth_required)
def get(self, request):
"""获取LDAP配置"""
try:
config = self._get_or_create_config()
return APIResponse.success('获取LDAP配置成功', self._serialize_config(config))
except Exception as e:
logger.error(f'获取LDAP配置失败: {str(e)}', exc_info=True)
return APIResponse.error(500, f'服务器错误: {str(e)}')
@method_decorator(jwt_auth_required)
def put(self, request):
"""更新LDAP配置"""
try:
with transaction.atomic():
data = json.loads(request.body)
config = self._get_or_create_config()
self._update_config(config, data)
config.save()
return APIResponse.success('更新LDAP配置成功')
except Exception as e:
logger.error(f'更新LDAP配置失败: {str(e)}', exc_info=True)
return APIResponse.error(500, f'服务器错误: {str(e)}')
def _get_or_create_config(self):
"""获取或创建LDAP配置"""
config, created = LDAPConfig.objects.get_or_create(
defaults={
'enabled': False,
'server_host': '',
'server_port': 389,
'use_ssl': False,
'base_dn': '',
'bind_dn': '',
'bind_password': '',
'user_search_filter': '(cn={username})',
'user_attr_map': {'username': 'cn', 'name': 'uid', 'email': 'mail'},
'timeout': 10
}
)
return config
def _serialize_config(self, config):
"""序列化配置，隐藏敏感信息"""
return {
'enabled': config.enabled,
'server_host': config.server_host,
'server_port': config.server_port,
'use_ssl': config.use_ssl,
'base_dn': config.base_dn,
'bind_dn': config.bind_dn,
'bind_password': '******' if config.bind_password else '', # 隐藏密码或显示为空
'user_search_filter': config.user_search_filter,
'user_attr_map': config.user_attr_map,
'timeout': config.timeout,
'update_time': config.update_time.strftime('%Y-%m-%d %H:%M:%S') if config.update_time else None
}
def _update_config(self, config, data):
"""更新配置字段"""
update_fields = [
'enabled', 'server_host', 'server_port', 'use_ssl', 'base_dn',
'bind_dn', 'user_search_filter', 'user_attr_map', 'timeout'
]
# 检查是否禁用LDAP认证
if 'enabled' in data and not data['enabled']:
# 如果禁用LDAP，清除所有配置数据
config.enabled = False
config.server_host = ''
config.server_port = 389
config.use_ssl = False
config.base_dn = ''
config.bind_dn = ''
config.bind_password = ''
config.user_search_filter = '(cn={username})'
config.user_attr_map = {'username': 'cn', 'name': 'uid', 'email': 'mail'}
config.timeout = 10
else:
# 启用状态下正常更新字段
for field in update_fields:
if field in data:
setattr(config, field, data[field])
if 'bind_password' in data:
if data['bind_password']:
config.bind_password = CryptoUtils.encrypt_password(data['bind_password'])
@method_decorator(csrf_exempt, name='dispatch')
class LDAPTestView(View):
@method_decorator(jwt_auth_required)
def post(self, request):
"""测试LDAP管理员账户连接"""
try:
config = self._get_config()
if not all([config.bind_dn, config.bind_password]):
return APIResponse.error(400, '绑定DN和绑定密码是必需的')
result = self._test_admin_connection(config)
return APIResponse.success('LDAP服务器连接测试成功', result)
except LDAPError as e:
return APIResponse.error(400, str(e))
except Exception as e:
logger.error(f'LDAP测试失败: {str(e)}', exc_info=True)
return APIResponse.error(500, f'服务器错误: {str(e)}')
def _get_config(self):
"""获取LDAP配置"""
try:
config = LDAPConfig.objects.get()
except LDAPConfig.DoesNotExist:
raise LDAPError('LDAP配置不存在，请先配置LDAP服务器')
if not all([config.server_host, config.base_dn, config.bind_dn]):
raise LDAPError('LDAP配置不完整，请检查服务器地址、Base DN和绑定DN')
return config
def _test_admin_connection(self, config):
"""测试LDAP管理员账户连接"""
try:
server_uri = f"{'ldaps://' if config.use_ssl else 'ldap://'}{config.server_host}:{config.server_port}"
server = ldap3.Server(server_uri, use_ssl=config.use_ssl, connect_timeout=config.timeout)
logger.info(f"LDAP测试 - 服务器: {server_uri}")
logger.info(f"LDAP测试 - 绑定DN: {config.bind_dn}")
logger.info(f"LDAP测试 - Base DN: {config.base_dn}")
# 解密绑定密码
bind_password = CryptoUtils.decrypt_password(config.bind_password)
# 使用管理员账户连接
connection = ldap3.Connection(
server,
user=config.bind_dn,
password=bind_password,
auto_bind=True
)
# 测试基本搜索功能
connection.search(
search_base=config.base_dn,
search_filter='(objectClass=*)',
search_scope=ldap3.BASE,
attributes=['*']
)
connection.unbind()
return {
'server': server_uri,
'bind_dn': config.bind_dn,
'base_dn': config.base_dn,
'connection_status': '连接成功'
}
except ldap3.core.exceptions.LDAPBindError as e:
logger.error(f"LDAP绑定错误: {str(e)}")
if 'invalidCredentials' in str(e):
raise LDAPError('管理员账户认证失败，请检查绑定DN和密码')
elif 'invalidDNSyntax' in str(e):
raise LDAPError('绑定DN格式错误，请检查DN语法是否正确')
else:
raise LDAPError(f'绑定失败: {str(e)}')
except ldap3.core.exceptions.LDAPSocketOpenError as e:
logger.error(f"LDAP连接错误: {str(e)}")
raise LDAPError('LDAP服务器连接失败，请检查服务器地址和端口')
except ldap3.core.exceptions.LDAPSocketReceiveError as e:
logger.error(f"LDAP超时错误: {str(e)}")
raise LDAPError('LDAP服务器连接超时，请检查网络连接')
except ldap3.core.exceptions.LDAPException as e:
logger.error(f"LDAP通用错误: {str(e)}")
raise LDAPError(f'LDAP错误: {str(e)}')
except Exception as e:
logger.error(f"LDAP测试未知错误: {str(e)}", exc_info=True)
raise LDAPError(f'连接失败: {str(e)}')
@method_decorator(csrf_exempt, name='dispatch')
class LDAPSyncView(View):
"""LDAP用户同步视图"""
@method_decorator(jwt_auth_required)
def post(self, request):
"""同步LDAP用户到系统"""
try:
data = json.loads(request.body)
action = data.get('action')
config = self._get_config()
if action == 'search':
# 搜索LDAP用户
search_filter = data.get('search_filter', '')
users = self._search_ldap_users(config, search_filter)
return APIResponse.success('搜索LDAP用户成功', {'users': users})
elif action == 'sync':
# 同步用户到系统
users_data = data.get('users', [])
if not users_data:
return APIResponse.error(400, '请选择要同步的用户')
synced_users = self._sync_users_to_system(config, users_data)
return APIResponse.success(
f'成功同步{len(synced_users)}个用户',
{'synced_users': synced_users}
)
else:
return APIResponse.error(400, '无效的操作类型')
except LDAPError as e:
return APIResponse.error(400, str(e))
except Exception as e:
logger.error(f'LDAP用户同步失败: {str(e)}', exc_info=True)
return APIResponse.error(500, f'服务器错误: {str(e)}')
def _get_config(self):
"""获取LDAP配置"""
try:
config = LDAPConfig.objects.get()
if not config.enabled:
raise LDAPError('LDAP未启用')
except LDAPConfig.DoesNotExist:
raise LDAPError('LDAP配置不存在')
if not all([config.server_host, config.base_dn, config.bind_dn, config.bind_password]):
raise LDAPError('LDAP配置不完整')
return config
def _search_ldap_users(self, config, search_filter=''):
"""搜索LDAP用户"""
try:
server_uri = f"{'ldaps://' if config.use_ssl else 'ldap://'}{config.server_host}:{config.server_port}"
server = ldap3.Server(server_uri, use_ssl=config.use_ssl, connect_timeout=config.timeout)
# 解密绑定密码
bind_password = CryptoUtils.decrypt_password(config.bind_password)
# 使用管理员账户连接
connection = ldap3.Connection(
server,
user=config.bind_dn,
password=bind_password,
auto_bind=True
)
# 构建搜索过滤器
if search_filter:
final_filter = f"(&(objectClass=person)({search_filter}))"
else:
final_filter = "(objectClass=person)"
# 搜索用户
connection.search(
search_base=config.base_dn,
search_filter=final_filter,
attributes=list(config.user_attr_map.values())
)
users = []
for entry in connection.entries:
user_info = self._extract_user_info(entry, config.user_attr_map)
if user_info.get('username'): # 确保有用户名
users.append(user_info)
connection.unbind()
return users
except ldap3.core.exceptions.LDAPException as e:
logger.error(f"LDAP搜索用户错误: {str(e)}")
raise LDAPError(f'搜索用户失败: {str(e)}')
except Exception as e:
logger.error(f"搜索LDAP用户未知错误: {str(e)}", exc_info=True)
raise LDAPError(f'搜索失败: {str(e)}')
def _extract_user_info(self, entry, attr_map):
"""提取用户信息"""
user_info = {'dn': entry.entry_dn}
for local_attr, ldap_attr in attr_map.items():
if hasattr(entry, ldap_attr):
attr_value = getattr(entry, ldap_attr)
if attr_value:
user_info[local_attr] = str(attr_value.value) if hasattr(attr_value, 'value') else str(attr_value)
return user_info
def _sync_users_to_system(self, config, users_data):
"""同步用户到系统"""
synced_users = []
for user_data in users_data:
try:
with transaction.atomic():
username = user_data.get('username')
if not username:
continue
# 检查用户是否已存在
try:
user = User.objects.get(username=username)
if user.user_type != 'ldap':
logger.warning(f'用户{username}已存在但不是LDAP用户，跳过')
continue
# 更新用户信息
user.name = user_data.get('name', username)
user.email = user_data.get('email')
user.ldap_dn = user_data.get('dn')
user.save()
synced_users.append({
'username': username,
'action': 'updated'
})
except User.DoesNotExist:
# 创建新用户
user = User.objects.create(
user_id=generate_id(),
username=username,
name=user_data.get('name', username),
email=user_data.get('email'),
user_type='ldap',
ldap_dn=user_data.get('dn'),
status=1
)
synced_users.append({
'username': username,
'action': 'created'
})
except Exception as e:
logger.error(f'同步用户{user_data.get("username")}失败: {str(e)}')
continue
return synced_users
@method_decorator(csrf_exempt, name='dispatch')
class LDAPStatusView(View):
def get(self, request):
"""检查LDAP是否启用（无需认证）"""
try:
try:
config = LDAPConfig.objects.get()
enabled = config.enabled
except LDAPConfig.DoesNotExist:
enabled = False
return APIResponse.success('获取LDAP状态成功', {'enabled': enabled})
except Exception as e:
logger.error(f'获取LDAP状态失败: {str(e)}', exc_info=True)
return APIResponse.error(500, f'服务器错误: {str(e)}')
class LDAPAuthenticator:
"""LDAP认证器"""
@staticmethod
def authenticate(username, password):
"""LDAP用户认证 - 只认证已存在的用户"""
try:
config = LDAPAuthenticator._validate_config()
# 先检查用户是否存在于本地数据库
user = LDAPAuthenticator._get_existing_user(username)
# 进行LDAP认证
LDAPAuthenticator._ldap_authenticate(config, username, password)
return True, user
except LDAPError as e:
return False, str(e)
except Exception as e:
logger.error(f'LDAP认证失败: {str(e)}', exc_info=True)
return False, f'认证失败: {str(e)}'
@staticmethod
def _validate_config():
"""验证LDAP配置"""
try:
config = LDAPConfig.objects.get()
if not config.enabled:
raise LDAPError('未启用LDAP认证')
except LDAPConfig.DoesNotExist:
raise LDAPError('LDAP配置不存在')
if not all([config.server_host, config.base_dn, config.bind_dn, config.bind_password]):
raise LDAPError('LDAP配置不完整，缺少必要的服务器信息或管理员账户')
return config
@staticmethod
def _ldap_authenticate(config, username, password):
"""执行LDAP认证 - 只使用绑定DN模式"""
try:
server_uri = f"{'ldaps://' if config.use_ssl else 'ldap://'}{config.server_host}:{config.server_port}"
server = ldap3.Server(server_uri, use_ssl=config.use_ssl, connect_timeout=config.timeout)
# 解密绑定密码
bind_password = CryptoUtils.decrypt_password(config.bind_password)
# 先用管理员账户连接并搜索用户
admin_conn = ldap3.Connection(server, user=config.bind_dn, password=bind_password, auto_bind=True)
# 搜索用户 - 支持两种过滤器格式
search_filter = config.user_search_filter
if '{username}' in search_filter:
search_filter = search_filter.format(username=username)
elif '%(user)s' in search_filter:
search_filter = search_filter.replace('%(user)s', username)
else:
# 如果都不匹配，尝试直接替换
search_filter = search_filter.format(username=username)
admin_conn.search(
search_base=config.base_dn,
search_filter=search_filter,
attributes=list(config.user_attr_map.values())
)
if not admin_conn.entries:
admin_conn.unbind()
raise LDAPError('用户不存在于LDAP服务器')
# 获取搜索到的用户DN
found_user_dn = admin_conn.entries[0].entry_dn
admin_conn.unbind()
# DN和用户密码进行认证
user_conn = ldap3.Connection(server, user=found_user_dn, password=password, auto_bind=True)
user_conn.unbind()
except ldap3.core.exceptions.LDAPBindError as e:
if 'invalidCredentials' in str(e):
raise LDAPError('用户名或密码错误')
else:
raise LDAPError('LDAP认证失败')
except ldap3.core.exceptions.LDAPSocketOpenError:
raise LDAPError('LDAP服务器连接失败')
except ldap3.core.exceptions.LDAPSocketReceiveError:
raise LDAPError('LDAP服务器连接超时')
except ldap3.core.exceptions.LDAPException as e:
raise LDAPError(f'LDAP错误: {str(e)}')
@staticmethod
def _get_existing_user(username):
"""获取已存在的LDAP用户"""
try:
return User.objects.get(username=username, user_type='ldap')
except User.DoesNotExist:
raise LDAPError(f'用户{username}不存在，请联系管理员同步LDAP用户')

322
backend/apps/views/login.py
View File

@@ -9,6 +9,7 @@ from django.views.decorators.http import require_http_methods
from django.conf import settings
from ..models import User, UserToken, LoginLog
from ..utils.security import SecurityValidator
from .ldap import LDAPAuthenticator
def generate_token_id():
"""生成token_id"""
@@ -18,6 +19,135 @@ def generate_log_id():
"""生成log_id"""
return hashlib.sha256(str(uuid.uuid4()).encode()).hexdigest()[:32]
class LoginHandler:
"""登录处理器"""
def __init__(self, username, password, ip_address, user_agent):
self.username = username
self.password = password
self.ip_address = ip_address
self.user_agent = user_agent
self.log_id = generate_log_id()
def authenticate_system_user(self):
"""系统用户认证"""
try:
user = User.objects.get(username=self.username)
except User.DoesNotExist:
self._log_failed_login(fail_reason='用户不存在')
return JsonResponse({'code': 404, 'message': '用户不存在'})
# 检查账户锁定
is_not_locked, lockout_message = SecurityValidator.check_account_lockout(user, self.ip_address)
if not is_not_locked:
self._log_failed_login(user=user, fail_reason=lockout_message)
return JsonResponse({'code': 423, 'message': lockout_message})
# 验证密码
password_hash = hashlib.sha256(self.password.encode()).hexdigest()
if user.password != password_hash:
return self._handle_password_error(user)
# 检查用户状态
if user.status == 0:
self._log_failed_login(user=user, fail_reason='账号已被锁定')
return JsonResponse({'code': 423, 'message': '账号已被锁定，请联系管理员解锁'})
# 登录成功
SecurityValidator.record_successful_login(user, self.ip_address)
return self._create_login_response(user)
def authenticate_ldap_user(self):
"""LDAP用户认证"""
success, result = LDAPAuthenticator.authenticate(self.username, self.password)
if not success:
self._log_failed_login(fail_reason=f'LDAP认证失败: {result}')
return JsonResponse({'code': 401, 'message': result})
user = result
if user.status == 0:
self._log_failed_login(user=user, fail_reason='账号已被锁定')
return JsonResponse({'code': 423, 'message': '账号已被锁定，请联系管理员解锁'})
return self._create_login_response(user)
def _handle_password_error(self, user):
"""处理密码错误"""
failed_attempts, max_attempts = SecurityValidator.record_failed_login(user, self.ip_address)
self._log_failed_login(user=user, fail_reason='密码错误')
remaining_attempts = max_attempts - failed_attempts
if remaining_attempts > 0:
message = f'密码错误，还可尝试{remaining_attempts}次'
else:
config = SecurityValidator.get_security_config()
message = f'密码错误次数过多，账户已被锁定{config.lockout_duration}分钟'
return JsonResponse({'code': 401, 'message': message})
def _create_login_response(self, user):
"""创建登录成功响应"""
config = SecurityValidator.get_security_config()
# 生成token
token_id = generate_token_id()
token_payload = {
'user_id': user.user_id,
'username': user.username,
'exp': datetime.utcnow() + timedelta(minutes=config.session_timeout)
}
token = jwt.encode(token_payload, settings.SECRET_KEY, algorithm='HS256')
# 更新token记录
UserToken.objects.update_or_create(
user=user,
defaults={'token_id': token_id, 'token': token}
)
# 更新登录时间
user.login_time = datetime.now()
user.save()
# 记录成功登录日志
self._log_successful_login(user)
return JsonResponse({
'code': 200,
'message': '登录成功',
'data': {
'token_id': token_id,
'token': token,
'user': {
'user_id': user.user_id,
'username': user.username,
'name': user.name,
'email': user.email
}
}
})
def _log_failed_login(self, user=None, fail_reason=''):
"""记录失败登录日志"""
LoginLog.objects.create(
log_id=self.log_id,
user=user,
ip_address=self.ip_address,
user_agent=self.user_agent,
status='failed',
fail_reason=fail_reason
)
def _log_successful_login(self, user):
"""记录成功登录日志"""
LoginLog.objects.create(
log_id=self.log_id,
user=user,
ip_address=self.ip_address,
user_agent=self.user_agent,
status='success'
)
@csrf_exempt
@require_http_methods(["POST"])
def login(request):
@@ -25,160 +155,26 @@ def login(request):
data = json.loads(request.body)
username = data.get('username')
password = data.get('password')
auth_type = data.get('auth_type', 'system')
# 获取客户端IP和用户代理
if not username or not password:
return JsonResponse({'code': 400, 'message': '用户名和密码不能为空'})
# 获取客户端信息
ip_address = request.META.get('HTTP_X_FORWARDED_FOR', request.META.get('REMOTE_ADDR', ''))
user_agent = request.META.get('HTTP_USER_AGENT', '')
if not username or not password:
return JsonResponse({
'code': 400,
'message': '用户名和密码不能为空'
})
# 密码加密
password_hash = hashlib.sha256(password.encode()).hexdigest()
try:
user = User.objects.get(username=username)
# 记录登录日志
log_id = generate_log_id()
# 检查账户是否被锁定
is_not_locked, lockout_message = SecurityValidator.check_account_lockout(user, ip_address)
if not is_not_locked:
# 账户被锁定，记录失败登录
LoginLog.objects.create(
log_id=log_id,
user=user,
ip_address=ip_address,
user_agent=user_agent,
status='failed',
fail_reason=lockout_message
)
return JsonResponse({
'code': 423, # 423 Locked
'message': lockout_message
})
if user.password != password_hash:
# 密码错误，记录失败尝试
failed_attempts, max_attempts = SecurityValidator.record_failed_login(user, ip_address)
# 记录失败登录日志
LoginLog.objects.create(
log_id=log_id,
user=user,
ip_address=ip_address,
user_agent=user_agent,
status='failed',
fail_reason='密码错误'
)
# 提供剩余尝试次数信息
remaining_attempts = max_attempts - failed_attempts
if remaining_attempts > 0:
message = f'密码错误，还可尝试{remaining_attempts}次'
else:
config = SecurityValidator.get_security_config()
message = f'密码错误次数过多，账户已被锁定{config.lockout_duration}分钟'
return JsonResponse({
'code': 401,
'message': message
})
if user.status == 0:
# 账号锁定，记录失败登录
LoginLog.objects.create(
log_id=log_id,
user=user,
ip_address=ip_address,
user_agent=user_agent,
status='failed',
fail_reason='账号已被锁定'
)
return JsonResponse({
'code': 423, # 423 Locked
'message': '账号已被锁定，请联系管理员解锁'
})
# 登录成功，重置失败尝试次数
SecurityValidator.record_successful_login(user, ip_address)
# 检查会话超时设置
config = SecurityValidator.get_security_config()
# 生成token_id和JWT token
token_id = generate_token_id()
token_payload = {
'user_id': user.user_id,
'username': user.username,
'exp': datetime.utcnow() + timedelta(minutes=config.session_timeout) # 使用配置的会话超时时间
}
token = jwt.encode(token_payload, settings.SECRET_KEY, algorithm='HS256')
# 更新或创建token记录
UserToken.objects.update_or_create(
user=user,
defaults={
'token_id': token_id,
'token': token
}
)
# 更新登录时间
user.login_time = datetime.now()
user.save()
# 记录成功登录日志
LoginLog.objects.create(
log_id=log_id,
user=user,
ip_address=ip_address,
user_agent=user_agent,
status='success'
)
return JsonResponse({
'code': 200,
'message': '登录成功',
'data': {
'token_id': token_id,
'token': token,
'user': {
'user_id': user.user_id,
'username': user.username,
'name': user.name,
'email': user.email
}
}
})
except User.DoesNotExist:
# 用户不存在，记录失败登录
log_id = generate_log_id()
LoginLog.objects.create(
log_id=log_id,
ip_address=ip_address,
user_agent=user_agent,
status='failed',
fail_reason='用户不存在'
)
return JsonResponse({
'code': 404,
'message': '用户不存在'
})
# 创建登录处理器
handler = LoginHandler(username, password, ip_address, user_agent)
# 根据认证类型进行认证
if auth_type == 'ldap':
return handler.authenticate_ldap_user()
else:
return handler.authenticate_system_user()
except Exception as e:
return JsonResponse({
'code': 500,
'message': f'服务器错误: {str(e)}'
})
return JsonResponse({'code': 500, 'message': f'服务器错误: {str(e)}'})
@csrf_exempt
@require_http_methods(["POST"])
@@ -186,36 +182,18 @@ def logout(request):
try:
token = request.headers.get('Authorization')
if not token:
return JsonResponse({
'code': 400,
'message': '未提供Token'
})
return JsonResponse({'code': 400, 'message': '未提供Token'})
try:
# 解析token获取user_id
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=['HS256'])
user_id = payload.get('user_id')
# 删除该用户的token记录
UserToken.objects.filter(user_id=user_id).delete()
return JsonResponse({'code': 200, 'message': '退出成功'})
return JsonResponse({
'code': 200,
'message': '退出成功'
})
except jwt.ExpiredSignatureError:
return JsonResponse({
'code': 401,
'message': 'Token已过期'
})
return JsonResponse({'code': 401, 'message': 'Token已过期'})
except jwt.InvalidTokenError:
return JsonResponse({
'code': 401,
'message': '无效的Token'
})
return JsonResponse({'code': 401, 'message': '无效的Token'})
except Exception as e:
return JsonResponse({
'code': 500,
'message': f'服务器错误: {str(e)}'
})
return JsonResponse({'code': 500, 'message': f'服务器错误: {str(e)}'})

4
backend/apps/views/user.py
View File

@@ -57,6 +57,7 @@ class UserView(View):
'email': user.email,
'status': user.status,
'roles': roles,
'user_type': user.user_type,
'login_time': user.login_time.strftime('%Y-%m-%d %H:%M:%S') if user.login_time else None,
'create_time': user.create_time.strftime('%Y-%m-%d %H:%M:%S'),
})
@@ -188,7 +189,6 @@ class UserView(View):
})
user.email = email
if status is not None:
# 如果是解锁操作（从0改为1），使用安全验证器的解锁方法
if user.status == 0 and status == 1:
success, message = SecurityValidator.unlock_user_account(user)
if not success:
@@ -196,7 +196,6 @@ class UserView(View):
'code': 400,
'message': message
})
# 如果是锁定操作（从1改为0），使用安全验证器的锁定方法
elif user.status == 1 and status == 0:
success, message = SecurityValidator.lock_user_account(user)
if not success:
@@ -315,6 +314,7 @@ class UserProfileView(View):
'email': user.email,
'status': user.status,
'roles': roles,
'user_type': user.user_type,
'login_time': user.login_time.strftime('%Y-%m-%d %H:%M:%S') if user.login_time else None,
'create_time': user.create_time.strftime('%Y-%m-%d %H:%M:%S') if user.create_time else None,
'update_time': user.update_time.strftime('%Y-%m-%d %H:%M:%S') if user.update_time else None,