If the account number or password is entered incorrectly for many times, the account will be banned.

目录

Account ban plan

 Flowchart of account ban​编辑

How to apply it in real scenarios

 使用mysql作为计数器

使用redis作为计数器

---

Account ban plan

 Flowchart of account ban

How to apply it in real scenarios

 使用mysql作为计数器

• My specific implementation is based on tornado框架,django,flaskIts interface methods are similar
• tornadoIt is an interface written in an asynchronous non-blocking way
• Which is uploaded after successful logintoken,Please see the detailed introduction of the blogger's homepage
• 使用mysqlDoing a counting operation will compareredis麻烦许多,And a separate table needs to be defined to store the time of each wrong input,to determine whether the specified time has been exceeded,并且计数,但是redisThere will be no time subtraction problem involved,redisIt can be set to expire and delete itself,而mysqlManual deletion is required（代码删除）
• mysqlis commonly used,redis对性能有一定要求,存储在磁盘上,会有一定的消耗

  # 计数器表(这是在tornadoTables defined in the framework)
  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('The obtained email and password：',email,password)

        #判断是否获取到数据
        if not all([email,password]):
            self.finish({'msg':'Email or verification code is empty','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
            # Get the counter
            if  num_mysql:
                # 获取当前时间
                now_time =  int(time.time())
                # Get the point in time when the wrong password was entered for the first 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':'You have exceeded the number of errors','errcode':3})

                # 超过30seconds to delete from the database
                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)) ))
                
            # Determine whether the account is activated
            if  user.state == 0:
                self.finish({'msg':'Account has not been activated','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:
                # The number of errors accumulates
                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('The obtained email and password：',email,password)

        #判断是否获取到数据
        if not all([email,password]):
            self.finish({'msg':'Email or verification code is empty','errcode':0})  

        try:
            
            # 获取计数器(redis)
            num=self.application.redis.get('num_'+email)
             # if the counter exists
            if num:
                if  int(num) >= 5:
                    #Of course, when testing the interface yourself,The time can be set shorter
                    self.finish({'msg':'You have exceeded the number of errors,请30try in minutes','errcode':3})

            
            # 查找用户是否存在
            user = await self.application.objects.get(UserModel.select().where((UserModel.email==email )  & (UserModel.password ==  make_password(password)) ))
                
            # Determine whether the account is activated
            if  user.state == 0:
                self.finish({'msg':'Account has not been activated','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 the counter exists
            if  num:
                # Accumulate the number of errors
                self.application.redis.incr('num_'+email)
            # 计数器不存在,It means that the user made a mistake for the first time
            else:
                # 30s以内,Accumulate by mistake
                self.application.redis.setex('num_'+email,30,1)