import time import hashlib from flask import Blueprint, redirect, render_template, request, Flask, session, current_app, make_response 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 from flask_limiter import Limiter from flask_limiter.util import get_remote_address import redis import json import bleach from argon2 import PasswordHasher from argon2.exceptions import VerifyMismatchError import html # 创建Argon2密码哈希器 ph = PasswordHasher() # Redis连接 redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True) # 创建限流器 limiter = Limiter( key_func=get_remote_address, default_limits=["200 per day", "50 per hour"] ) ub = Blueprint('user', __name__, url_prefix='/user', template_folder='templates') def sanitize_input(text): """清理用户输入，防止XSS攻击""" if text is None: return None return bleach.clean(str(text), strip=True) def validate_csrf_token(): """验证CSRF令牌""" token = request.form.get('csrf_token') stored_token = session.get('csrf_token') if not token or not stored_token or token != stored_token: return False return True def get_client_info(): """获取客户端信息""" return { 'ip': request.remote_addr, 'user_agent': str(request.user_agent.string), 'platform': str(request.user_agent.platform), 'browser': str(request.user_agent.browser), } def is_suspicious_ip(ip): """检查IP是否可疑""" key = f"login_attempts:{ip}" attempts = redis_client.get(key) if attempts and int(attempts) >= 5: # 5次失败尝试 return True return False def record_failed_attempt(ip): """记录失败的登录尝试""" key = f"login_attempts:{ip}" pipe = redis_client.pipeline() pipe.incr(key) pipe.expire(key, 1800) # 30分钟后重置 pipe.execute() def clear_login_attempts(ip): """清除登录尝试记录""" redis_client.delete(f"login_attempts:{ip}") def set_secure_headers(response): """设置安全响应头""" response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['X-Frame-Options'] = 'SAMEORIGIN' response.headers['X-XSS-Protection'] = '1; mode=block' response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains' response.headers['Content-Security-Policy'] = "default-src 'self'" return response def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'username' not in session: return redirect('/user/login') # 验证会话完整性 if 'client_info' not in session or 'session_id' not in session: session.clear() return redirect('/user/login') # 验证客户端信息 current_client = get_client_info() stored_client = session['client_info'] if (current_client['ip'] != stored_client['ip'] or current_client['user_agent'] != stored_client['user_agent']): session.clear() return redirect('/user/login') # 验证会话ID stored_session_id = redis_client.get(f"session:{session['username']}") if not stored_session_id or stored_session_id != session['session_id']: session.clear() return redirect('/user/login') return f(*args, **kwargs) return decorated_function def hash_password(password: str) -> str: """ 使用Argon2id算法哈希密码 :param password: 用户输入的密码 :return: 哈希后的密码 """ return ph.hash(password) def verify_password(stored_hash: str, password: str) -> bool: """ 验证密码 :param stored_hash: 存储的密码哈希 :param password: 用户输入的密码 :return: 是否匹配 """ try: return ph.verify(stored_hash, password) except VerifyMismatchError: return False def validate_password(password: str) -> bool: """ 验证密码强度 """ if len(password) < 12: # 增加最小长度要求 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 # 检查常见密码模式 common_patterns = ['password', '123456', 'qwerty'] if any(pattern in password.lower() for pattern in common_patterns): return False return True @ub.route('/login', methods=['GET', 'POST']) @limiter.limit("5 per minute") def login(): """处理用户登录请求""" if request.method == 'GET': response = make_response(render_template('login_and_register.html')) return set_secure_headers(response) try: if request.method == 'POST' and not validate_csrf_token(): logging.warning("CSRF验证失败") return errorResponse('无效的请求') client_ip = request.remote_addr if is_suspicious_ip(client_ip): logging.warning(f"可疑IP尝试登录: {client_ip}") return errorResponse('由于多次失败尝试，请30分钟后再试') username = sanitize_input(request.form.get('username')) password = request.form.get('password') # 密码不需要sanitize if not username or not password: logging.warning("登录失败：用户名或密码为空") return errorResponse('用户名和密码不能为空') # 查询用户信息 sql = "SELECT password, status FROM user WHERE username = %s" result = query(sql, [username], "select") if result: stored_password = result[0]['password'] status = result[0]['status'] if status != 'active': logging.warning(f"已禁用的账户尝试登录: {username}") return errorResponse('账户已被禁用') if verify_password(stored_password, password): session.clear() session.regenerate() # 生成唯一会话ID session_id = secrets.token_hex(32) client_info = get_client_info() # 存储会话信息 session['username'] = username session['login_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') session['csrf_token'] = secrets.token_hex(32) session['client_info'] = client_info session['session_id'] = session_id session.permanent = True current_app.permanent_session_lifetime = timedelta(hours=2) # 在Redis中存储会话ID redis_client.setex( f"session:{username}", int(current_app.permanent_session_lifetime.total_seconds()), session_id ) clear_login_attempts(client_ip) # 记录登录历史 login_history_sql = ''' INSERT INTO login_history (username, login_time, ip_address, user_agent, success, attempt_count) VALUES (%s, %s, %s, %s, %s, %s) ''' query(login_history_sql, [ username, datetime.now(), client_info['ip'], client_info['user_agent'], True, redis_client.get(f"login_attempts:{client_ip}") or 0 ]) logging.info(f"用户 {username} 登录成功") response = make_response(redirect('/page/home')) return set_secure_headers(response) record_failed_attempt(client_ip) logging.warning(f"登录失败：用户名或密码错误") return errorResponse('用户名或密码错误') except Exception as e: logging.error(f"登录过程发生错误: {e}") return errorResponse('登录失败，请稍后重试') @ub.route('/register', methods=['GET', 'POST']) @limiter.limit("3 per hour") def register(): if request.method == 'GET': response = make_response(render_template('login_and_register.html')) return set_secure_headers(response) try: if request.method == 'POST' and not validate_csrf_token(): logging.warning("CSRF验证失败") return errorResponse('无效的请求') username = sanitize_input(request.form.get('username')) password = request.form.get('password') email = sanitize_input(request.form.get('email')) if not username or not password or not email: return errorResponse('用户名、密码和邮箱不能为空') # 验证用户名格式 if not re.match(r'^[a-zA-Z0-9_]{4,20}$', username): return errorResponse('用户名只能包含字母、数字和下划线，长度4-20位') # 验证邮箱格式 if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email): return errorResponse('邮箱格式不正确') # 验证密码强度 if not validate_password(password): return errorResponse('密码必须包含大小写字母、数字和特殊字符，且长度至少12位') try: # 检查用户名和邮箱是否存在 check_sql = """ SELECT (SELECT COUNT(*) FROM user WHERE LOWER(username) = LOWER(%s)) as username_count, (SELECT COUNT(*) FROM user WHERE LOWER(email) = LOWER(%s)) as email_count """ result = query(check_sql, [username.lower(), email.lower()], "select") if result[0]['username_count'] > 0: return errorResponse('该用户名已被注册') if result[0]['email_count'] > 0: return errorResponse('该邮箱已被注册') # 哈希密码 hashed_password = hash_password(password) # 插入新用户 insert_sql = ''' INSERT INTO user(username, password, email, status, createTime, last_password_change) VALUES(%s, %s, %s, %s, %s, %s) ''' current_time = datetime.now() query(insert_sql, [ username, hashed_password, email, 'active', current_time, current_time ]) # 记录注册信息 client_info = get_client_info() register_history_sql = ''' INSERT INTO register_history (username, register_time, ip_address, user_agent, email) VALUES (%s, %s, %s, %s, %s) ''' query(register_history_sql, [ username, current_time, client_info['ip'], client_info['user_agent'], email ]) logging.info(f"新用户注册成功: {username}") response = make_response(redirect('/user/login')) return set_secure_headers(response) except Exception as e: logging.error(f"注册过程发生错误: {e}") return errorResponse('注册失败，请稍后重试') except Exception as e: logging.error(f"注册过程发生错误: {e}") return errorResponse('注册失败，请稍后重试') @ub.route('/logout') @login_required def logout(): """用户登出""" try: username = session.get('username') client_info = session.get('client_info', {}) # 记录登出历史 logout_history_sql = ''' INSERT INTO logout_history (username, logout_time, ip_address, user_agent, session_id) VALUES (%s, %s, %s, %s, %s) ''' query(logout_history_sql, [ username, datetime.now(), client_info.get('ip'), client_info.get('user_agent'), session.get('session_id') ]) # 删除Redis中的会话 redis_client.delete(f"session:{username}") session.clear() logging.info(f"用户 {username} 成功登出") response = make_response(redirect('/user/login')) return set_secure_headers(response) except Exception as e: logging.error(f"登出过程发生错误: {e}") response = make_response(redirect('/user/login')) return set_secure_headers(response)