安裝:
pip3 install Flask-JWT-Extended
什么是Flask-JWT-Extended
之前已經說過jwt是序列化並加密過的json串,那很明顯extend則是對之前功能的拓展。那下面我們就該看看拓展的強大之處。
app.py
from flask_jwt_extended import JWTManager app.config['JWT_SECRET_KEY'] = 'jose' # token密鑰 app.config['JWT_BLACKLIST_ENABLED'] = True # 黑名單管理 app.config['JWT_BLACKLIST_TOKEN_CHECKS'] = ['access', 'refresh'] # 允許將access and refresh tokens加入黑名單 #注冊jwt jwt = JWTManager(app)
接下來,因為此插件剔除了自動生成/auth,我們需要拓展user的功能,生成用戶有關的:注冊、登錄、登出等功能,以及用戶令牌認證以及刷新、失效等功能。
user.py
#關聯包介紹 #request約束 _user_parser = reqparse.RequestParser() _user_parser.add_argument('username', type=str, required=True, help="This field cannot be blank." ) _user_parser.add_argument('password', type=str, required=True, help="This field cannot be blank." ) #用戶登錄 class UserLogin(Resource): def post(self): #從parser獲取request資料 data = _user_parser.parse_args() #找到用戶資料 user = UserModel.find_by_username(data['username']) #進行用戶認證 if user and safe_str_cmp(user.password, data['password']): #生成token,認證令牌和刷新令牌 access_token = create_access_token(identity=user.id, fresh=True) refresh_token = create_refresh_token(user.id) return { 'access_token': access_token, 'refresh_token': refresh_token }, 200 #認證失敗 return {"message": "Invalid Credentials!"}, 401
claims
這是jwt的數據存儲機制
jwt = JWTManager(app) #user_claims_loader定義我們附加到jwt的有效數據 #在每一個收到jwt保護的端口,我們都可以使用get_jwt_claims檢索這些數據 #如下所示 @jwt.user_claims_loader def add_claims_to_jwt(identity): if identity == 1: # instead of hard-coding, we should read from a config file to get a list of admins instead return {'is_admin': True} return {'is_admin': False}
通過這個示例我們可以模擬用戶管理機制,如下對item.py的修改,非管理員用戶不可以刪除item:
注意下get_jwt_claims,說明在上文。
@jwt_required def delete(self, name): claims = get_jwt_claims() if not claims['is_admin']: return {'message': 'Admin privilege required.'}, 401 item = ItemModel.find_by_name(name) if item: item.delete_from_db() return {'message': 'Item deleted.'} return {'message': 'Item not found.'}, 404
接下來就是一個個介紹特色的時候了:
jwt_optional
官網介紹是,可以可選的保護節點,無論是否發送token,都可以進入節點,然后交給邏輯判斷如何處理。比較常見的判定如下:
如果請求中存在訪問令牌,則將調用
示例
jwt_optional
官網介紹是,可以可選的保護節點,無論是否發送token,都可以進入節點,然后交給邏輯判斷如何處理。比較常見的判定如下:
如果請求中存在訪問令牌,則將調用
get_jwt_identity()
具有訪問令牌標識的端點。如果請求中不存在訪問令牌,則仍將調用此端點,但
get_jwt_identity()
將返回None
示例
class ItemList(Resource): @jwt_optional def get(self): user_id = get_jwt_identity() items = [item.json() for item in ItemModel.find_all()] if user_id: return {'items': items}, 200 return { 'items': [item['name'] for item in items], 'message': 'More data available if you log in.' }, 200
create_refresh_token
這個要追溯到create_access_token。一般來說,我們都是將用戶權限和用戶基本資料存放在這個TOKEN中,但是當用戶權限?或者資料變更時,我們就要去刷新用戶的資料,或者說當access_token過期了,我們也要去更新access_token。這一部分可以細看 Web API與OAuth:既生access token,何生refresh token
其實這只是定義的問題(不過如果不是自己寫TOKEN規則,我們要遵守這個定義)。我們在書寫create_refresh_token的邏輯時,不在需要驗證用戶名和密碼即達到了這種效果,如果說個人不這么寫邏輯,你也看不出來效果,畢竟兩種TOKEN其實是一樣的。
這個要追溯到create_access_token。一般來說,我們都是將用戶權限和用戶基本資料存放在這個TOKEN中,但是當用戶權限?或者資料變更時,我們就要去刷新用戶的資料,或者說當access_token過期了,我們也要去更新access_token。這一部分可以細看 Web API與OAuth:既生access token,何生refresh token
其實這只是定義的問題(不過如果不是自己寫TOKEN規則,我們要遵守這個定義)。我們在書寫create_refresh_token的邏輯時,不在需要驗證用戶名和密碼即達到了這種效果,如果說個人不這么寫邏輯,你也看不出來效果,畢竟兩種TOKEN其實是一樣的。
class TokenRefresh(Resource): @jwt_refresh_token_required def post(self): current_user = get_jwt_identity() new_token = create_access_token(identity=current_user, fresh=False) return {'access_token': new_token}, 200
這里有另一個概念,令牌的新鮮度fresh=False/True。我們可以根據此處的變更靈活處理一些敏感數據。比如說有些數據只有剛輸入用戶名和密碼時的那段有效認證期才可以使用,當令牌刷新時即判定無效。此處匹配裝飾器
item.py
fresh_jwt_required()
item.py
POST /item/<name>的邏輯變更。
@fresh_jwt_required def post(self, name): if ItemModel.find_by_name(name): return {'message': "An item with name '{}' already exists.".format(name)}, 400 data = self.parser.parse_args() item = ItemModel(name, **data) try: item.save_to_db() except: return {"message": "An error occurred while inserting the item."}, 500 return item.json(), 201
expired_token_loader
invalid_token_loader
重新定義TOKEN回調錯誤內容,還有更多的請參考Changing Default Behaviors
#過期令牌 @jwt.expired_token_loader def expired_token_callback(): return jsonify({ 'message': 'The token has expired.', 'error': 'token_expired' }), 401 #無效令牌 @jwt.invalid_token_loader def invalid_token_callback(error): # we have to keep the argument here, since it's passed in by the caller internally return jsonify({ 'message': 'Signature verification failed.', 'error': 'invalid_token' }), 401
最后我們來介紹黑名單系統
token_in_blacklist_loader by app.py
app.config['JWT_BLACKLIST_ENABLED'] = True # enable blacklist feature app.config['JWT_BLACKLIST_TOKEN_CHECKS'] = ['access', 'refresh'] # allow blacklisting for access and refresh tokens # 檢查此令牌是否屬於黑名單,並在啟用黑名單時調用 @jwt.token_in_blacklist_loader def check_if_token_in_blacklist(decrypted_token): return decrypted_token['jti'] in BLACKLIST
文章摘要:https://www.jianshu.com/p/10d0da01244c
文章摘要: https://flask-jwt-extended.readthedocs.io/en/latest/api.html#flask_jwt_extended.JWTManager.expired_token_loader