#【Redis连接池开启SSL加密】【Redis.ConnectionPool】【小白指南】【生产环境可用】python连接Redis数据库代码（连接池技术高效运行的加密通讯版）
由于上一篇文章教了如何使用SSL加密连接的方式来连接Redis数据库，实现了有效的加密连接通讯，这一篇文章我们来根据生产环境来应用Redis数据库加密通讯的基础来实现用户登录验证的状态保存和判断，避免登录界面被暴力破解和暴力攻击。使用Redis数据库连接池（Redis.ConnectionPool）可以有效地控制web服务器的Redis连接的资源消耗，用Redis数据库服务器记录登录次数和时间和IP地址，这样就可以有效地阻止非法暴力访问登录页面。在实践中有非常重要的安全作用。
下面我们开始分几步来实现吧！

【第一步：创建加密的Redis连接池：redis_pool = redis.ConnectionPool】

根据官方文档Redis—SSL配置和connection_to_the_Redis，单个连接的初始化方式和连接池是不一样的。

单个Redis对象进行SSL连接的代码实例可以看我的上一篇文章Redis数据库开启SSL加密连接。
那么连接池的官方给出的实例是：

import redis

redis_pool = redis.ConnectionPool(
    host="localhost",
    port=6666,
    connection_class=redis.SSLConnection,
    ssl_cert_reqs="none",
)

r = redis.StrictRedis(connection_pool=redis_pool)
r.ping()

很明显，这里有一个redis.SSLConnection的引用。
那么我新建一个 app/utils/redis_pool.py 文件：

import redis
from config import Config

# 创建Redis连接池

redis_pool = redis.ConnectionPool(
    host=Config.REDIS_HOST_IP,
    port=Config.REDIS_PORT,
    password=Config.REDIS_PASSWORD,
    connection_class=redis.SSLConnection,
    ssl_certfile=Config.REDIS_SSL_CERTFILE,
    ssl_keyfile=Config.REDIS_SSL_KEYFILE,
    ssl_ca_certs=Config.REDIS_SSL_CA_CERTS,
    ssl_cert_reqs='none', #这个可以设置为required，当你的证书和服务器IP或者域名匹配的时候。
    #decode_responses=True, # flask session can't not use this parm
    max_connections=100  # 连接池最大连接数，可根据需求调整
)

# 从连接池获取Redis客户端实例
def get_redis_client():
    """从连接池获取Redis客户端客户端"""
    return redis.StrictRedis(connection_pool=redis_pool)

【第二步：创建配套的配置信息（如果测试可以直接填字符串）】

还有config.py文件作为配置文件：

# config.py
import os
# 获取当前配置文件所在目录的绝对路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))

class Config:
    # 配置类，用于存储应用程序的各种配置参数
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'your-secret-key-here'      # 密钥从环境变量获取

    # Redis服务器配置
    REDIS_HOST_IP = os.environ.get('REDIS_HOST_IP') # Redis服务器IP地址
    REDIS_PORT = os.environ.get('REDIS_PORT')       # Redis服务器端口
    REDIS_PASSWORD = os.environ.get('REDIS_PWD')    # 从环境变量获取Redis密码
    REDIS_SSL_CERTFILE = os.path.join(BASE_DIR, 'certs', 'redis.crt')   # SSL证书文件路径
    REDIS_SSL_KEYFILE = os.path.join(BASE_DIR, 'certs', 'redis.key')     # SSL密钥文件路径
    REDIS_SSL_CA_CERTS = os.path.join(BASE_DIR, 'certs', 'redis.crt')   # SSL证书文件路径
    print("证书路径是否存在：", os.path.exists(REDIS_SSL_KEYFILE))

【第三步：启用 Flask session，并指定使用Redis_pool来获取Redis数据库连接】

然后修改app下面的初始化文件 app/init.py： 获取Redis对象的要点 redis.StrictRedis(connection_pool=redis_pool)

# app/__init__.py
from flask import Flask
from flask_session import Session  # Flask会话管理扩展
import redis  # Redis客户端库，用于连接和操作Redis服务器
from config import Config 
from .utils.redis_pool import redis_pool  # 导入Redis连接池配置
# ====================
# 应用工厂函数
# ====================
def create_app():
    """
    Flask应用工厂函数,用于创建和配置应用实例
    
    Returns:
        Flask应用实例
    """
    app = Flask(
        __name__,
        template_folder='../templates',  # 模板文件夹路径（相对于当前文件）
        static_folder='../static',       # 静态文件文件夹路径（相对于当前文件）
        static_url_path='/static'        # 静态文件的URL路径（默认值）
    )
    # 配置应用的密钥（在生产环境中应使用更强的随机密钥）
    # 这个密钥用于加密会话和其他安全操作
    app.secret_key = 'your-secret-key-here'  # 在生产环境中替换为随机密钥
    # ====================
    # Redis会话配置 -- flask session
    # ====================
    # 设置会话类型为Redis，使用Redis作为会话存储后端
    app.config['SESSION_TYPE'] = 'redis'
    
    # 配置Redis连接池，使用之前定义的redis_pool
    app.config['SESSION_REDIS'] = redis.StrictRedis(connection_pool=redis_pool)
    
    # 配置会话的过期时间（单位：秒）
    # 这里设置为1800秒（0.5小时），即用户在0.5小时内无操作后会话自动过期
    app.config['PERMANENT_SESSION_LIFETIME'] = 1800
    # 初始化Flask-Session扩展，将Redis配置应用到应用中
    Session(app)
    # ====================
    # 路由注册
    # ====================
    # 从routes模块导入蓝图列表
    # 蓝图是Flask的一种模块化方式，用于组织相关路由
    from .routes import blueprints
    # 注册所有蓝图到应用中
    # 这样应用就能处理所有蓝图定义的路由
    for blueprint in blueprints:
        app.register_blueprint(blueprint)
    # 返回配置好的应用实例
    return app

【第四步：利用Redis连接池高速高效的办持久化缓存，来对登录界面进行保护】

保护策略：
总的登录检测逻辑：
# 1，让访问端的IP地址一个小时内只有50次尝试次数，如果超过了50次那么就要屏蔽这个IP地址的访问60分钟
# 2，让相同用户名的登录尝试次数在3分钟内不能超过10次，如果超过了10次那么就要屏蔽这个用户名的登录3分钟
第一种情况当然是针对恶意暴力登录尝试的用户。第二种情况就针对一般的登录尝试情形。
根据策略编写Login.py的代码就很爽了：

# app/routes/login.py
from flask import Blueprint, render_template, request, redirect, url_for, session, flash
from datetime import datetime, timedelta
from ..utils.redis_pool import get_redis_client  # 导入连接池
from config import Config
import json  # 新增

login_bp = Blueprint('login', __name__)

# 配置rate limiting参数
RATE_LIMIT_ATTEMPTS = 50
RATE_LIMIT_TIME_WINDOW = 3600  # 1小时（秒）
BLOCK_TIME = 3600  # 1小时（秒））

USER_LIMIT_ATTEMPTS = 10
USER_LIMIT_TIME_WINDOW = 180  # 3分钟（秒）
USER_BLOCK_TIME = 180  # 3分钟（秒）

@login_bp.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        current_time = datetime.utcnow()
        # 初始化 Redis 连接
        redis_client = get_redis_client()
        # ==================
        # 总的登录检测逻辑：
        # 1，让访问端的IP地址一个小时内只有50次尝试次数，如果超过了50次那么就要屏蔽这个IP地址的访问60分钟
        # 2，让相同用户名的登录尝试次数在3分钟内不能超过10次，如果超过了10次那么就要屏蔽这个用户名的登录3分钟
        # ==================
        client_ip = request.remote_addr
        ip_redis_key = f"login_attempts:{client_ip}"
        ip_attempt_data = redis_client.get(ip_redis_key)
        
        if ip_attempt_data:
            ip_attempt_data = json.loads(ip_attempt_data.decode('utf-8'))
            ip_last_attempt = datetime.fromisoformat(ip_attempt_data['last_attempt'])
            ip_attempt_count = ip_attempt_data['count']

            if (current_time - ip_last_attempt).total_seconds() < RATE_LIMIT_TIME_WINDOW:
                if ip_attempt_count >= RATE_LIMIT_ATTEMPTS:
                    flash('登录失败! IP 地址已被锁定 60 分钟，请稍后再试。')
                    return render_template('login.html')
                ip_attempt_count += 1
            else:
                ip_attempt_count = 1
        else:
            ip_attempt_count = 1

        redis_client.set(
            ip_redis_key,
            json.dumps({
                'count': ip_attempt_count,
                'last_attempt': current_time.isoformat()
            }),
            ex=BLOCK_TIME  # 1小时后自动删除该键（避免永久存储）
        )
        # 用户名登录检测逻辑：
        # 从 Redis 获取尝试次数数据（键格式：login_attempts:{username}）
        redis_key = f"login_attempts:{username}"
        attempt_data = redis_client.get(redis_key)
        
        if attempt_data:
            # 解析 JSON 数据
            attempt_data = json.loads(attempt_data.decode('utf-8')) #需要转码
            last_attempt = datetime.fromisoformat(attempt_data['last_attempt'])
            attempt_count = attempt_data['count']
            
            # 检查是否在锁定时间内（3分钟）
            if (current_time - last_attempt).total_seconds() < USER_LIMIT_TIME_WINDOW:
                if attempt_count >= USER_LIMIT_ATTEMPTS:
                    flash('登录失败！账户已被锁定 3 分钟，请稍后再试。')
                    return render_template('login.html')
                # 未锁定但次数未满，更新计数
                attempt_count += 1
            else:
                # 锁定时间已过，重置计数
                attempt_count = 1
        else:
            # 首次尝试
            attempt_count = 1

        # 更新 Redis 中的尝试数据
        redis_client.set(
            redis_key,
            json.dumps({
                'count': attempt_count,
                'last_attempt': current_time.isoformat()
            }),
            ex=USER_BLOCK_TIME  # 3 分钟后自动删除该键（避免永久存储）
        )

        # ======================
        # 用户名和密码验证：
        #     由于已经防止了暴力破解，所以现在可以直接进行MSSQL数据库的用户名和密码比对。
        #     如果验证失败，则将尝试次数增加，并记录最后一次尝试的时间。
        # ======================
        
        # 第一步：根据用户名生成SQL查询语句，从数据库中获取这个用户的数据。
        # 第二步：比对用户输入的密码和数据库中的密码是否一致。（数据库中密码是明文保存的）
        import pyodbc  # 需要导入pyodbc库用于连接MSSQL
        from ..utils.sqlStr_pool import SqlStrPool_Login  # 导入SQL语句池
        try:
            # 建立数据库连接
            #print("连接字符串：", Config.MSSQL_CONSTR_MainFashionShop)
            conn = pyodbc.connect(Config.MSSQL_CONSTR_MainFashionShop)
            cursor = conn.cursor()
            
            # 执行参数化查询，防止SQL注入
            query = SqlStrPool_Login.user_info_query
            cursor.execute(query, (username,))
            result = cursor.fetchone()
            
            # 第二步：比对用户输入的密码和数据库中的密码是否一致（数据库中密码是明文保存的）
            if result and password == result[0]:
                # 登录成功，清除尝试记录
                redis_client.delete(redis_key)
                redis_client.delete(ip_redis_key)
                session['user'] = {
                    'username': username,
                    'phoneNum': result[1],
                    'permission': result[2],
                    'default_shopName': result[3],
                    'default_DBName': result[4],
                    'default_DBServer': result[5]
                }
                flash('登录成功！')
                # 更新最后登录时间和IP
                query = SqlStrPool_Login.update_query
                cursor.execute(query, (client_ip, username))
                conn.commit()
                return redirect(url_for('main.dashboard'))
            else:
                # 密码错误或用户不存在，检查是否达到锁定条件
                if attempt_count >= USER_LIMIT_ATTEMPTS:
                    flash('登录失败！账户已被锁定 3 分钟，请稍后再试。')
                else:
                    remaining = USER_LIMIT_ATTEMPTS - attempt_count
                    flash(f'无效的用户名或密码，还剩 {remaining} 次尝试机会。')
                return render_template('login.html')
                
        except pyodbc.Error as e:
            # 数据库连接或查询出错
            flash(f'数据库错误：{str(e)}')
            return render_template('login.html')
        finally:
            # 确保连接关闭
            if 'conn' in locals() and conn:
                conn.close()
    return render_template('login.html')

后面的SQL用户验证是我自己项目中使用的代码。你根据自己的情况进行调整就行了。

【结束语】

我们通过创建Redis连接池，并且用安全SSL加密为连接池的连接提供加密配置。让你的Redis数据库更加安全，通讯也防止了中间者攻击。
有了安全的基础，我们将session放入了Redis数据库作为Session的保存后端，有效地减轻了web服务器的负担，提高了并发量，Flask Session+Redis的组合非常适合管理Session，因为Redis是高速内存缓存型数据库，可以自动管理Session的时间，减轻了Web服务器资源消耗。
同时，我们还可以用Redis连接池提供防止恶意访问、暴力破解登录密码的可能。这样的方式比验证码或者图形检测类的验证方式更加安全，毕竟是在后端对访问者进行控制。当然，你可以结合验证码，图形校验等方式增强安全性。
好了，说了这么多，最喜欢的还是你的赞和支持，感谢读者！

希望能够得到你的一个免费赞哦~~
最好就关注我拉~哈哈
本文章原创，转载请注明出处。

—深深浅水湾