[Python知识库] flask token

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> Python知识库 -> flask token -> 正文阅读

[Python知识库]flask token

jwt pyjwt会导致jwt.encode()方法报错
```
  pip install pyjwt
```

  def generate_jwt(payload: dict, expiry: datetime, secret=None) -> str:
      """
      生成jwt
      :param payload: dict, 载荷
      :param expiry: datetime 有效期
      :param secret: 密钥
      :return: jwt
      """
  
      # 将过期时间加入payload
      _payload = {"exp": expiry}
      _payload.update(payload)
  
      if not secret:
          secret = current_app.config['JWT_SECRET']
  
      token = jwt.encode(_payload, secret, algorithm='HS256')
      return token

生成jwt，三个部分组成：载荷、密钥、加密方法

载荷中可以包含用户的信息以及exp有效期

  def verify_jwt(token: str, secret=None) -> dict:
      """
      检验token
      :param token: jwt
      :param secret: 密钥
      :return: payload
      """
  
      if not secret:
          secret = current_app.config["JWT_SECRET"]
  
      try:
          payload = jwt.decode(token, secret, algorithms=['HS256'])
      except jwt.PyJWTError as e:
          current_app.logger.error(e)
          payload = None
  
      return payload

检验jwt，就是decode之前生成的jwt，然后验证其中的载荷部分的用户信息

  def generate_token(user_id: int, refresh=True) -> tuple:
      """
      生成token和refresh_token
      :param user_id: 用户id
      :param refresh:
      :return:
      """
      secret = current_app.config["JWT_SECRET"]
  
      expiry = datetime.utcnow() + timedelta(hours=current_app.config["JWT_EXPIRY_HOURS"])
  
      token = generate_jwt({"user_id": user_id}, expiry, secret)
  
      if refresh:
          expiry = datetime.utcnow() + timedelta(days=current_app.config["JWT_REFRESH_DAYS"])
          refresh_token = generate_jwt({"user_id": user_id, "is_refresh": True}, expiry, secret)
      else:
          refresh_token = None
  
      return token, refresh_token

生成token的函数，token和refresh_token就是调用generate_jwt而生成的

  def jwt_authentication():
      """
      获取请求头中的token
      :return:
      """
      # 对于进入的请求先设定g中的user_id,is_refresh 来判定是否登录过
      g.user_id = None
      g.is_refresh = False
  
      token = request.headers.get("Authorization")
      current_app.logger.info(f"token is {token}")
  
      if token is not None and token.startswith("Bearer "):
          token = token[7:]
  
          # 验证token
          payload = verify_jwt(token)
          current_app.logger.info(f"payload is {payload}")
          if payload is not None:
              g.user_id = payload.get("user_id")
              g.is_refresh = payload.get("is_refresh", False)

app.before_request(jwt_authentication)

给flask添加钩子函数，当请求中包含token时，设定g中的user_id 和 is_refresh

  def login_required(func):
      @wraps(func)
      def wrapper(*args, **kwargs):
          if g.user_id is not None and g.is_refresh is False:
              return func(*args, **kwargs)
          else:
              current_app.logger.warn("Invalid token")
              return {
                         "msg": "Invalid token"
                     }, 401
  
      return wrapper