SpringCloud+Tornado基于jwt实现请求安全校验功能

common - authServer.py是认证接口
common - basicServer.py是示例接口
handlers - baseHandler.py中放了两种校验方式，会在basicServer.py的调用示例中贴出
utils - jwtUtils.py是生成与校验token的
utils - responseUtils.py是返回结果工具类

# -*- coding:utf-8 -*-
import jwt
import datetime
from jwt import exceptions
from utils.responseUtils import JsonUtils

JWT_SALT = '1qazxdr5'


def create_token(payload, timeout=12):
 """
 创建token
 :param payload: 例如：{'user_id':1,'username':'xxx@xxx.xx'}用户信息
 :param timeout: token的过期时间，默认20分钟
 :return:
 """
 headers = {
  'typ': 'jwt',
  'alg': 'HS256'
 }
 payload['exp'] = datetime.datetime.utcnow() + datetime.timedelta(minutes=timeout)
 result = jwt.encode(payload=payload, key=JWT_SALT, algorithm="HS256", headers=headers).decode('utf-8')
 return result


def parse_payload(token):
 """
 对token进行校验并获取payload
 :param token:
 :return:
 """
 try:
  verified_payload = jwt.decode(token, JWT_SALT, True)
  print(verified_payload)
  return JsonUtils.success('认证通过')
 except exceptions.ExpiredSignatureError:
  return JsonUtils.noAuth('token已失效')
 except jwt.DecodeError:
  return JsonUtils.noAuth('token认证失败')
 except jwt.InvalidTokenError:
  return JsonUtils.noAuth('非法的token')

# -*- coding:utf-8 -*-
import functools
import json
import tornado.web
from utils import jwtUtils
from utils.responseUtils import JsonUtils


# 方式一：authenticated 装饰器
def authenticated(method):
 @functools.wraps(method)
 def wrapper(self, *args, **kwargs):
  """
  这里调用的是 current_user 的 get 方法(property装饰),
  """
  # 通过token请求头传递token
  head = self.request.headers
  token = head.get("token", "")
  if not token:
   self.write(JsonUtils.noAuth("未获取到token请求头"))
   self.set_header('Content-Type', 'application/json')
   return

  result = json.loads(jwtUtils.parse_payload(token)) # 将json解码
  print(result)
  token_msg = json.dumps(result)

  if result['sta'] != '00':
   self.write(token_msg)
   self.set_header('Content-Type', 'application/json')
   return

  return method(self, *args, **kwargs)
 return wrapper


# 方式二：进行预设 继承tornado的RequestHandler
class BaseHandler(tornado.web.RequestHandler):
 def prepare(self):
  super(BaseHandler, self).prepare()

 def set_default_headers(self):
  super().set_default_headers()


# 进行token校验，继承上面的BaseHandler
class TokenHandler(BaseHandler):
 def prepare(self):
  # 通过token请求头传递token
  head = self.request.headers
  token = head.get("token","")
  if not token:
   self.authMsg = json.dumps(JsonUtils.noAuth("未获取到token请求头"))

  result = json.loads(jwtUtils.parse_payload(token)) # 将json解码
  print(result)

  if result['sta'] != '00':
   self.isAuth = False
  else:
   self.isAuth = True

  self.authMsg = json.dumps(result)

import tornado.web
from utils import jwtUtils
from utils.responseUtils import JsonUtils


class authHandler(tornado.web.RequestHandler):
 def post(self, *args, **kwargs):
  """
  安全认证接口
  :param args:
  :param kwargs:
  :return:
  """
  username = self.get_argument("username")
  print("authHandler:" + username)

  if not username:
   self.write(JsonUtils.error("参数异常"))
  else:
   token = jwtUtils.create_token({"username": username})
   print("token:" + token)
   self.write(JsonUtils.success(token))

  self.set_header('Content-Type', 'application/json')

import tornado.web
import json
from pandas.core.frame import DataFrame
from handlers import baseHandler
from utils.responseUtils import JsonUtils
from handlers.baseHandler import authenticated


class StringHandler(baseHandler.TokenHandler, tornado.web.RequestHandler):
 """
 *** TokenHandler验证，对应baseHandler.py中的方式二 ***
 """
 def get(self):
  username = self.get_argument('username', 'Hello')

  # 权限认证通过
  if self.isAuth:
   self.write(JsonUtils.success(username))
  else:
   self.write(self.authMsg))

  self.set_header('Content-Type', 'application/json')


class TestHandler(tornado.web.RequestHandler):
 """
 *** authenticated验证，对应baseHandler.py中的方式一 ***
 """
 @authenticated
 def post(self):
  username = self.get_argument('username', 'Hello')
  self.write(JsonUtils.success(username))
  self.set_header('Content-Type', 'application/json')

from tornado.escape import json_encode, utf8


class JsonUtils(object):
 @staticmethod
 def success(response):
  """
  正确返回
  :param response: 返回结果
  :return: string, {"message": "ok", "sta": "00", "data": }
  """
  return json_encode({"message": "ok", "sta": "00", "data": response})

 @staticmethod
 def info(message):
  """
  提示返回
  :param message: 提示信息
  :return: string,
  """
  return json_encode({"message": str(message), "sta": "99001", "data": None})

 @staticmethod
 def error(message):
  """
  错误返回
  :param message: 错误信息
  :return: string,
  """
  return json_encode({"message": str(message), "sta": "9999", "data": None})

 @staticmethod
 def noAuth(message):
  """
  无权限返回
  :param message: 错误信息
  :return: string,
  """
  return json_encode({"message": str(message), "sta": "403", "data": None})