账号或密码多次输入错误，进行账号封禁

目录

账号封禁的规划

 账号封禁的流程图​编辑

实际场景中如何应用

 使用mysql作为计数器

使用redis作为计数器

---

账号封禁的规划

 账号封禁的流程图

实际场景中如何应用

 使用mysql作为计数器

• 本人的具体实现是基于tornado框架，django，flask其接口方法也差不多
• tornado是一个采用异步非阻塞的方式来写的接口
• 其中登录成功后传的的token，请看博主主页的详细介绍哦
• 使用mysql作计数操作会比redis麻烦许多，并且需要单独定义一张表来存放每次输错的时间，来判断是否超过规定时间，并且计数，但是redis就不会涉及到时间减法问题，redis本身就可以设置过期删除，而mysql需要写手动删除（代码删除）
• mysql是普遍运用的，redis对性能有一定要求，存储在磁盘上，会有一定的消耗

  # 计数器表(这是在tornado框架中定义的表)
  class CheckNumModel(peewee.Model):
      
      id = peewee.IntegerField(primary_key = True,unique=True,constraints=[peewee.SQL('AUTO_INCREMENT')])
  
      #入库时间
      create_time = peewee.DateTimeField(default=datetime.datetime.now(),help_text='入库时间')
      
      email = peewee.CharField(null=False,unique=True)
  
      num = peewee.IntegerField(null=False,default=0) 
  
      class Meta:
  
          # 声明表名
          db_table = 'check_num'

# 登录页面
class UserLogin(BaseHandler):
    async def get(self):
         # 接收参数
        email = self.get_argument('email', None)
        password = self.get_argument('password', None)
        print('获取到的邮箱和密码：',email,password)

        #判断是否获取到数据
        if not all([email,password]):
            self.finish({'msg':'邮箱或验证码为空','errcode':0})  

        try:

            # 获取计数器（mysql版本）
            try:
                #从数据库表中查询
                num_mysql =await self.application.objects.get(CheckNumModel.select().where(CheckNumModel.email==email))
            except Exception as e:
                num_mysql = None
            # 拿到计数器
            if  num_mysql:
                # 获取当前时间
                now_time =  int(time.time())
                # 获取首次输错密码的时间点
                num_time =  num_mysql.create_time.timestamp()
                print(now_time,num_time)
                # 减法运算
                mytime = now_time - num_time
                print(mytime)
                if  mytime  <=  30:
                    # 判断计数器
                    if  num_mysql.num>5:
                        self.finish({'msg':'您已经超过错误次数','errcode':3})

                # 超过30秒从数据库删除
                else:
                    # 清除计数器
                    await self.application.objects.delete(num_mysql)


            # 查找用户是否存在
            user = await self.application.objects.get(UserModel.select().where((UserModel.email==email )  & (UserModel.password ==  make_password(password)) ))
                
            # 判断账号是否激活
            if  user.state == 0:
                self.finish({'msg':'账号尚未激活','errcode':400})
            else:
                # 生成jwt
                myjwt=MyJwt()
                self.finish({"msg":"邮箱已激活 登陆成功", 'email':user.email,"errcode": 1,'token':myjwt.encode({'id':user.id})})


        except  Exception  as e:
            # print('登录失败的原因:',e)
            self.finish({'msg':'用户名或密码错误','errcode':2}            

            # mysql逻辑
            if num_mysql:
                # 错误次数累加
                num_mysql.num += 1
                # 修改数据库
                await self.application.objects.update(num_mysql)

            # 第一次输错,插入计数器
            else:
                await   self.application.objects.create(CheckNumModel,email=email,num=1)

使用redis作为计数器

class UserLogin(BaseHandler):
    async def get(self):
         # 接收前端传递的参数
        email = self.get_argument('email', None)
        password = self.get_argument('password', None)
        print('获取到的邮箱和密码：',email,password)

        #判断是否获取到数据
        if not all([email,password]):
            self.finish({'msg':'邮箱或验证码为空','errcode':0})  

        try:
            
            # 获取计数器(redis)
            num=self.application.redis.get('num_'+email)
             # 如果计数器存在
            if num:
                if  int(num) >= 5:
                    #当然自己在测试接口时，时间就可以设的短一些
                    self.finish({'msg':'您已经超过错误次数，请30分钟后尝试','errcode':3})

            
            # 查找用户是否存在
            user = await self.application.objects.get(UserModel.select().where((UserModel.email==email )  & (UserModel.password ==  make_password(password)) ))
                
            # 判断账号是否激活
            if  user.state == 0:
                self.finish({'msg':'账号尚未激活','errcode':400})
            else:
                # 生成jwt
                myjwt=MyJwt()
                self.finish({"msg":"邮箱已激活 登陆成功", 'email':user.email,"errcode": 1,'token':myjwt.encode({'id':user.id})})


        except  Exception  as e:
            # print('登录失败的原因:',e)
            self.finish({'msg':'用户名或密码错误','errcode':2})

            # redis版本
            # 如果计数器存在
            if  num:
                # 对错误次数进行累加操作
                self.application.redis.incr('num_'+email)
            # 计数器不存在，则说明是用户第一次输错
            else:
                # 30s以内，输错做累加
                self.application.redis.setex('num_'+email,30,1)