Merge pull request #19 from ZhangJinHaHaHa/main

注册漏洞修复，增强用户认证系统安全性

Merge pull request #19 from ZhangJinHaHaHa/main
注册漏洞修复，增强用户认证系统安全性
戒酒的李白 · GitHub
Commit a14f5d6538a51d1e266197f944b247a914a3e8f5 a14f5d65 2 parents f85298c0 673eb024
Showing 1 changed file with 106 additions and 47 deletions
views/user/user.py
--- a/views/user/user.py
View file @a14f5d6
+++ b/views/user/user.py
View file @a14f5d6
 import time
 import hashlib
-from flask import Blueprint, redirect, render_template, request, Flask, session
-
+from flask import Blueprint, redirect, render_template, request, Flask, session, current_app
+from datetime import datetime, timedelta
+import re
 from utils.query import query
 from utils.errorResponse import errorResponse
 from utils.logger import app_logger as logging
+from functools import wraps
+import secrets
 ub = Blueprint('user',
                __name__,
                url_prefix='/user',
                template_folder='templates')
+def login_required(f):
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+        if 'username' not in session:
+            return redirect('/user/login')
+        return f(*args, **kwargs)
+    return decorated_function
+
 # 密码加密函数
-def hash_password(password: str, salt: str = 'XiaoXueQi2024') -> str:
+def hash_password(password: str, salt: str = None) -> tuple:
     """
     使用 SHA256 对密码进行加盐哈希
     :param password: 用户输入的密码
-    :param salt: 加盐值，默认值为 'XiaoXueQi2024'
-    :return: 哈希后的密码
+    :param salt: 可选的盐值
+    :return: (哈希后的密码, 盐值)
+    """
+    if not salt:
+        salt = secrets.token_hex(16)
+    hash_obj = hashlib.sha256()
+    hash_obj.update(salt.encode('utf-8'))
+    hash_obj.update(password.encode('utf-8'))
+    return hash_obj.hexdigest(), salt
+
+def validate_password(password: str) -> bool:
+    """
+    验证密码强度
     """
-    hash_with_salt = hashlib.sha256(salt.encode('utf-8'))
-    hash_with_salt.update(password.encode('utf-8'))
-    return hash_with_salt.hexdigest()
-  
+    if len(password) < 8:
+        return False
+    if not re.search(r"[A-Z]", password):
+        return False
+    if not re.search(r"[a-z]", password):
+        return False
+    if not re.search(r"\d", password):
+        return False
+    if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", password):
+        return False
+    return True
+
 @ub.route('/login', methods=['GET', 'POST'])
 def login():
     """
     处理用户登录请求
-    :return: 登录页面或重定向到主页
     """
     if request.method == 'GET':
-        return render_template('login_and_register.html')  # 显示登录页面
+        return render_template('login_and_register.html')
     try:
         username = request.form.get('username')
@@ -40,59 +69,89 @@ def login():
             logging.warning("登录失败：用户名或密码为空")
             return render_template('login_and_register.html', msg='用户名和密码不能为空')
-        # 查询用户
-        sql = "SELECT * FROM user WHERE username = %s AND password = %s"
-        result = query(sql, [username, password], "select")
+        # 查询用户和盐值
+        sql = "SELECT password, salt FROM user WHERE username = %s"
+        result = query(sql, [username], "select")
         if result:
-            session['username'] = username
-            logging.info(f"用户 {username} 登录成功")
-            return redirect('/page/home')
-        else:
-            logging.warning(f"用户 {username} 登录失败：用户名或密码错误")
-            return render_template('login_and_register.html', msg='用户名或密码错误')
+            stored_password = result[0]['password']
+            salt = result[0]['salt']
+            
+            # 验证密码
+            hashed_input, _ = hash_password(password, salt)
+            
+            if hashed_input == stored_password:
+                session.clear()
+                session['username'] = username
+                session['login_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+                session['csrf_token'] = secrets.token_hex(32)
+                session.permanent = True
+                current_app.permanent_session_lifetime = timedelta(hours=2)
+                
+                logging.info(f"用户 {username} 登录成功")
+                return redirect('/page/home')
+        
+        # 使用相同的响应防止用户枚举
+        logging.warning(f"登录失败：用户名或密码错误")
+        return render_template('login_and_register.html', msg='用户名或密码错误')
     except Exception as e:
         logging.error(f"登录过程发生错误: {e}")
         return render_template('login_and_register.html', msg='登录失败，请稍后重试')
-
 @ub.route('/register', methods=['GET', 'POST'])
 def register():
     if request.method == 'GET':
         return render_template('login_and_register.html')
-    else:
+    
+    try:
+        username = request.form.get('username')
+        password = request.form.get('password')
-        def filter_fn(user):
-            return request.form['username'] in user
+        if not username or not password:
+            return errorResponse('用户名和密码不能为空')
-        users = query('select * from user', [], 'select')
-        filter_list = list(filter(filter_fn, users))
-        if len(filter_list):
-            return errorResponse('该用户名已被注册')
-        else:
-            time_tuple = time.localtime(time.time())
-            hash_with_salt = hashlib.sha256('XiaoXueQi2024'.encode('utf-8'))
-            hash_with_salt.update(request.form['password'].encode('utf-8'))
-            query(
-                '''
-                insert into user(username,password,createTime) values(%s,%s,%s)
-            ''', [
-                    request.form['username'],
-                    hash_with_salt.hexdigest(),
-                    str(time_tuple[0]) + '-' + str(time_tuple[1]) + '-' +
-                    str(time_tuple[2])
-                ])
+        # 验证用户名格式
+        if not re.match(r'^[a-zA-Z0-9_]{4,20}$', username):
+            return errorResponse('用户名只能包含字母、数字和下划线，长度4-20位')
-        return redirect('/user/login')
+        # 验证密码强度
+        if not validate_password(password):
+            return errorResponse('密码必须包含大小写字母、数字和特殊字符，且长度至少8位')
+        # 使用事务处理竞态条件
+        try:
+            # 检查用户名是否存在
+            check_sql = "SELECT COUNT(*) as count FROM user WHERE username = %s"
+            result = query(check_sql, [username], "select")
+            
+            if result[0]['count'] > 0:
+                return errorResponse('该用户名已被注册')
-@ub.route('/logOut')
-def logOut():
-    session.clear()
-    return redirect('/user/login')
+            # 生成密码哈希和盐值
+            hashed_password, salt = hash_password(password)
+            
+            # 插入新用户
+            insert_sql = '''
+                INSERT INTO user(username, password, salt, createTime) 
+                VALUES(%s, %s, %s, %s)
+            '''
+            current_time = datetime.now().strftime('%Y-%m-%d')
+            query(insert_sql, [username, hashed_password, salt, current_time])
+            
+            logging.info(f"新用户注册成功: {username}")
+            return redirect('/user/login')
+            
+        except Exception as e:
+            logging.error(f"注册过程发生错误: {e}")
+            return errorResponse('注册失败，请稍后重试')
+
+    except Exception as e:
+        logging.error(f"注册过程发生错误: {e}")
+        return errorResponse('注册失败，请稍后重试')
-@ub.route('/user/logout')
+@ub.route('/logout')
+@login_required
 def logout():
     """用户登出"""
     try: