VUE Flask登录的初探-JWT的探索

上回简单实现了基于JWT的登录，并且留下了一些问题，jwt天生的弊端。本次用某些逻辑解决jwt的弊端

先列举jwt可能遇到的问题：

1.注销问题，当客户端注销登录后，token在有效期内依然有效，实际上从服务端无法让token失效
2.修改密码，当用户修改了密码，按常规需要让前次token失效。
3.续签问题，jwt虽然有超时机制，但没有实现自动续签。

为了解决上述三个问题，我考虑用如下手段：

1.注销问题，我将对每个用户，维护一个sessionID，作为此用户本次会话的有效ID，当注销用户后，服务端将此sessionID 删除。于是服务端就能主动控制token的有效性

2.修改密码问题，我将对用户ID做一个替代ID，此ID生成逻辑是由用户真实ID和用户密码加密产生，由此当用户修改密码后，替代ID将自动被更新，由此，客户端的TokenID，无法在服务端查询到，于是失效。

3.续签问题，解决此事，只能不停刷新和下发token。

以上方案都扩大服务器资源开销。

下面是代码。

前端代码，

1.对login.vue代码做了微调。添加了修改密码的按钮

2.对request.js，添加了response拦截器，token的保存到cookie逻辑，迁移到此处

<template>
  <div class='login'>
    <h1>{{ titleMsg }}</h1>
    <el-form ref="loginForm" :model="loginData" label-width="100px">
      <el-form-item label="用户名" prop="username" :rules="[{required: true, message: '用户名不能为空'}]">
        <el-input ref="username" v-model="loginData.username" autocomplete="off"></el-input>
      </el-form-item>
      <el-form-item label="密码" prop="password" :rules="[{required: true, message: '密码不能为空'}]">
        <el-input type="password" v-model="loginData.password" autocomplete="off"></el-input>
      </el-form-item>      
      <el-form-item>
        <el-button type="primary" @click="loginForm('loginForm')">提交</el-button>
        <el-button @click="resetForm('loginForm')">重置</el-button>
      </el-form-item>
      <el-form-item>
        <el-button @click="testForm()">测试</el-button>
        <el-button @click="logoutForm()">登出</el-button>
      </el-form-item>
      <el-form-item>
        <el-button @click="changePws()">修改密码为456</el-button>
      </el-form-item>
    </el-form>
  </div>
</template>

<script>
import qs from 'qs'
import service from '../utils/request'
export default {
  name: 'loginForm',
  data() {
    return {
      titleMsg: '欢迎来到旗帜世界',
      loginData: {
        username: '',
        password: ''
      }
    }
  },
  methods: {
    loginForm(formName) {
      this.$refs[formName].validate((valid) => {
         if (valid) {
           service({url: '/login',method: 'post',data: qs.stringify(this.loginData)})
             .then(response => {
               const { data } = response
               //Cookies.set('Authorization',data.data.token)
               alert('submit!!!' +'
'+ data.msg)
             })
             .catch(error => {
               console.log(error)
             })
         } else {
           console.log('illegad submit!!');
           return false;
        }
      })
    },
    testForm() {
      service({url: '/first',method: 'get'})
        .then(response => {
          const { data } = response
          alert('firstPage!!!' +'
'+ data.data.tips)
        })
        .catch(error => {
          console.log(error)
        })
    },
    logoutForm() {
      service({url: '/logout',method: 'get'})
        .then(response => {
          const { data } = response
          alert('logout!!!' +'
'+ data.data.tips)
        })
        .catch(error => {
          console.log(error)
        })
    },
    changePws() {
      service({url: '/changepws',method: 'get'})
        .then(response => {
          const { data } = response
          alert('changepws!!!' +'
'+ data.data)
        })
        .catch(error => {
          console.log(error)
        })
    },
  }
}
</script>

import axios from 'axios'
import Cookies from 'js-cookie'


/****** 创建axios实例 ******/
const service = axios.create({
  baseURL: 'http://localhost:5000',  // api的base_url
  timeout: 5000  // 请求超时时间
})

service.interceptors.request.use(
  config => {
    config.headers['Authorization'] = Cookies.get('Authorization')
    return config
  },
  error => {
    console.log(error)
    return Promise.reject(error)
  }
)

/****** respone拦截器==>对响应做处理 ******/
service.interceptors.response.use(
  response => {
    const { data } = response    
    if (isJSON(data.data)) {
      let jsonObj = JSON.parse(JSON.stringify(data.data))
      if (Object.prototype.hasOwnProperty.call(jsonObj,"token")){
        console.log('update token')
        Cookies.set('Authorization',jsonObj.token)
      }
    } else {
      console.log('not')
    }
    return response
  },
  error => {
    console.log(error);
    return Promise.reject(error)
  }
)

function isJSON(str) {
  let jsonData = JSON.stringify(str)
  try {
    if (typeof JSON.parse(jsonData) == "object") {
      return true;
    } else {
      return false;
    }
  } 
  catch(e) {
    return false;
  }
}

export default service;

后端代码，改动还是蛮大，

1.login.py 将更多无关逻辑下发给其他py文件，保持login.py的纯粹性，只解决登录相关问题

2.user.py 用户操作的所有代码，均由此单元负责。

3.jwt_token.py 仅仅是将超时时间，改为入参。

import time
import json
from flask import Blueprint,request
from flask_login import LoginManager,login_user,logout_user,login_required,current_user
from user import UserLogin,Operaters

login_page = Blueprint('login_page',__name__)

login_manager = LoginManager()
login_manager.login_view = None

@login_page.record_once
def on_load(state):
  login_manager.init_app(state.app)

# @login_manager.user_loader
# def load_user(user_id):
#   return User.get(user_id)

@login_manager.request_loader
def load_user_from_request(request):
  token = request.headers.get('Authorization')
  if token == None:
    return None

  payload = UserLogin.verfiyToken(token)
  if payload != None:
    alternativeID = payload['data']['alternativeID']
    sessionID = payload['data']['sessionID']
    user = UserLogin.queryUser(alternativeID=alternativeID,sessionID=sessionID)
  else:
    user = None
  return user


@login_page.route('/first')
@login_required
def firstPage():
  returnData = {'code': 0, 'msg': 'success', 'data': {'token':current_user.token,'tips':'First Blood(来自' + current_user.userName +')'}}
  return returnData,200

@login_page.route('/login', methods=['POST'])
def login():
  if request.method == 'POST':
    username = request.form['username']
    password = request.form['password']
    user = UserLogin.queryUser(userName = username)
    if (user != None) and (user.verifyPassword(password)) :
      login_user(user)
      returnData = {'code': 0, 'msg': 'success', 'data': {'token':user.token}}
      return json.dumps(returnData),200
    else :
      returnData = {'code': 1, 'msg': 'failed', 'data': {'tips':'username or password is not correct'} }
      return json.dumps(returnData),200  

@login_page.route('/logout') 
@login_required
def logout():
  userName = current_user.userName
  alternativeID = current_user.alternativeID
  sessionID = current_user.sessionID
  UserLogin.dropSessionID(alternativeID,sessionID)
  logout_user()
  returnData = {'code': 0, 'msg': 'success', 'data': {'tips':'Bye ' + userName} }
  return json.dumps(returnData),200  


@login_page.route('/changepws') 
@login_required
def changePws():
  user = UserLogin.queryUser(userID = current_user.id)  
  user.changePws()
  returnData = {'code': 0, 'msg': 'success', 'data': {'tips':'password was changed'} }
  return json.dumps(returnData),200  

import uuid
from flask_login import UserMixin
from werkzeug.security import check_password_hash,generate_password_hash
from jwt_token import genToken,verfiyToken
from datetime import datetime,timedelta

Operaters = [
    {
        "id": 1,
        "name": "admin",
        "password": generate_password_hash('123'),
        "alternativeID": generate_password_hash('1'+generate_password_hash('123')),
        "sessionIDs": []
    },
    {
        "id": 2,
        "name": "李四",
        "password": generate_password_hash('123'),
        "alternativeID": generate_password_hash('2'+generate_password_hash('123')),
        "sessionIDs": []
    },    
]

class UserLogin(UserMixin):
    def __init__(self,operater):
        self.id = operater.get("id")
        self.userName = operater.get("name")
        self.passwordHash = operater.get("password")
        self.alternativeID = operater.get("alternativeID")
        self.sessionID = None
        self.token = None
        self.oper = operater

        exp = datetime.utcnow() + timedelta(seconds=10)
        self.genSessionID(exp)
        self.genToken(exp)

        clearOvertimeSeesionID(operater)

    def genSessionID(self,exp,sessionID=None):
        if sessionID == None:
            self.sessionID = str(uuid.uuid4())            
            self.oper["sessionIDs"].append({'id':self.sessionID,'exp':exp})
        else:            
            self.sessionID = sessionID

    
    def verifyPassword(self,password):
        if self.passwordHash is None:
            return False
        return check_password_hash(self.passwordHash,password)

    def get_id(self):
        return self.id

    def get(user_id):
        if not user_id:
            return None
        for oper in Operaters:
            if str(oper.get('id')) == str(user_id) :
                return User(oper)
        return None

    def clearOvertimeSeesionID(operater):        
        for userSessionID in operater["sessionIDs"][::-1]:
            if userSessionID['exp'] < datetime.utcnow() :
                operater["sessionIDs"].remove(userSessionID)
    
    @staticmethod
    def queryUser(**kwargs):
        if 'userID' in kwargs:
            return UserLogin.queryUserByID( kwargs['userID'])
        elif 'userName' in kwargs:
            return UserLogin.queryUserByName(kwargs['userName'])
        elif ('alternativeID' in kwargs) and ('sessionID' in kwargs):
            return UserLogin.queryUserBySessionID(kwargs['alternativeID'],kwargs['sessionID'])
        else:
            return None
    

    @staticmethod
    def queryUserByID(userID):
        for oper in Operaters:
            if (oper.get('id') == userID) :
                user = UserLogin(oper)
                return user
        return None

    @staticmethod
    def queryUserByName(username):
        for oper in Operaters:
            if (oper.get('name') == username) :
                user = UserLogin(oper)
                return user
        return None

    @staticmethod
    def queryUserBySessionID(alternativeID,sessionID):
        exists = False
        for oper in Operaters:
            if (oper.get('alternativeID') == alternativeID) :
                sessionIDs = oper["sessionIDs"]
                exists = True
                break
        
        if exists:
            exists = False
            for userSessionID in sessionIDs:
                if (userSessionID['id'] == sessionID) :
                    exists = True
                    break

        if exists: 
            user = UserLogin(oper)
            return user
        else:
            return None

    @staticmethod
    def dropSessionID(alternativeID,sessionID):
        exists = False
        for oper in Operaters:
            if (oper.get('alternativeID') == alternativeID) :
                sessionIDs = oper["sessionIDs"]
                exists = True
                break
        
        if exists :
            exists = False
            for userSessionID in sessionIDs:
                if (userSessionID['id'] == sessionID) :
                    exists = True
                    break
        if exists :
            sessionIDs.remove(userSessionID)

    @staticmethod
    def verfiyToken(token):
        verfiyed = verfiyToken(token)
        if verfiyed :
            return True        

        for oper in Operaters:
            clearOvertimeSeesionID(oper)

        return verfiyed


    def changePws(self):
        for oper in Operaters:
            if (oper.get('id') == self.id) :              
                oper['password'] = generate_password_hash('456')
                oper['alternativeID'] = generate_password_hash('1'+generate_password_hash('456'))
                return true
        return False

    def genToken(self,exp):
        token = genToken(exp,{'alternativeID':self.alternativeID,'sessionID':self.sessionID})
        self.token = token
        return token

    

import jwt
from jwt import PyJWTError
from datetime import datetime,timedelta

SECRECT_KEY = b'x92R!x8exc6x9cxb3x89#xa6x0cxcbxf6xcbxd7xbc'


def genToken(exp,data):
  payload = {
    'exp': exp,
    'data': data 
    }
  token = jwt.encode(payload,key= SECRECT_KEY,algorithm= 'HS256')
  return bytes.decode(token)

def verfiyToken(tokenStr):
  try:
    tokenBytes =  tokenStr.encode('utf-8')
    payload = jwt.decode(tokenBytes,key= SECRECT_KEY,algorithm= 'HS256')
    return payload
  except PyJWTError as e:
    print("jwt验证失败: %s" % e)
    return None

题外话，学习python已经有几个月了，这个语言还是蛮有趣，相对来说，没有那么多约束，所以代码看起来比较随性。未来的日子，需要逐渐找出合适python的项目结构以及格式规范。

下一个话题，将步入数据库方面，我会把用户字典创建到数据库中，并用python+sql操作用户表。前端+后端+数据库，如此以来一套基本的项目就搭建完毕。

再有空就可以把缓存机制搭建起来。甚至是docker。现在程序员需要的知识量越来越繁杂，多而不精。这样真的好么？我没有答案。