SpringCloud+Tornado基于jwt實現(xiàn)請求安全校驗功能

更新時間：2020年12月21日 09:51:16 作者：95.8℃

這篇文章主要介紹了SpringCloud+Tornado基于jwt實現(xiàn)請求安全校驗,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下

項目背景

在實際項目中，Tornado項目作為一個微服務納入SpringCloud體系，該過程中涉及到Tornado與Spring體系的安全驗證，也就是權限調(diào)用校驗，在該項目中Tornado是通過SpringCloud中的Feign調(diào)用的，經(jīng)過一系列實驗，最后選用jwt來實現(xiàn)這個權限效驗的過程。

實現(xiàn)思路

用戶進行登陸認證(后臺微服務)，認證成功后調(diào)用Tornado項目的認證接口生成token，該值返回到后臺微服務保存在會話中，下一次請求時帶上該token值讓服務器進行校驗，校驗成功則返回正常的響應，否則返回錯誤信息。

項目結構

在這里插入圖片描述

common - authServer.py是認證接口
common - basicServer.py是示例接口
handlers - baseHandler.py中放了兩種校驗方式，會在basicServer.py的調(diào)用示例中貼出
utils - jwtUtils.py是生成與校驗token的
utils - responseUtils.py是返回結果工具類

具體實現(xiàn)

jwtUtils.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):
 """
 創(chuàng)建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')

baseHandler.py

# -*- 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):
  """
  這里調(diào)用的是 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)

authServer.py

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("參數(shù)異常"))
  else:
   token = jwtUtils.create_token({"username": username})
   print("token:" + token)
   self.write(JsonUtils.success(token))

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

basicServer.py

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')

responseUtils.py

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})

下面是一些調(diào)用的結果圖示：

在這里插入圖片描述