專案背景

在實際專案中，Tornado專案作為一個微服務納入SpringCloud體系，該過程中涉及到Tornado與Spring體系的安全驗證，也就是許可權呼叫校驗，在該專案中Tornado是通過SpringCloud中的Feign呼叫的，經過一系列實驗，最後選用jwt來實現這個許可權效驗的過程。

實現思路

使用者進行登陸認證(後臺微服務)，認證成功後呼叫Tornado專案的認證介面生成token，該值返回到後臺微服務儲存在會話中，下一次請求時帶上該token值讓伺服器進行校驗，校驗成功則返回正常的響應，否則返回錯誤資訊。

專案結構

common - authServer.py是認證介面
common - basicServer.py是示例介面
handlers - baseHandler.py中放了兩種校驗方式，會在basicServer.py的呼叫示例中貼出
utils - jwtUtils.py是生成與校驗token的
utils - responseUtils.py是返回結果工具類

具體實現

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):
 """
 建立token
 :param payload: 例如：{'user_id':1,'username':'[email protected]'}使用者資訊
 :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):
  """
  這裡呼叫的是 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,**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,**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')

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","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,"sta": "9999","data": None})

 @staticmethod
 def noAuth(message):
  """
  無許可權返回
  :param message: 錯誤資訊
  :return: string,"sta": "403","data": None})

下面是一些呼叫的結果圖示：

.end

到此這篇關於SpringCloud+Tornado基於jwt實現請求安全校驗的文章就介紹到這了,更多相關SpringCloud+Tornado實現請求安全校驗內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！