添加了一个后路由器，以提高spa和自动刷新机密的安全级别。古德曼文/aiohttpjwtplus
aiohttp-jwtplus的Python项目详细描述
aiohttp jwtplus公司
用于处理JSON web令牌（签名）的Aiohttp中间件和helper实用程序。 添加了一个后路由器，以提高spa的安全级别和自动刷新机密。在
- 秘密自动刷新。在
- 完全分离的内容。在
- 在Python3.7+上工作
要求
安装
pip install aiohttp-jwtplus
使用
- 您需要创建一个SecretManager对象，该对象首先管理关于jwt的信息（secret\scheme\algorithm\exp\uinterval\auto\u refresh_interval等）。在
- 然后，您需要创建一个JWTHelper，在它的槽中您可以确定您的业务逻辑，例如您从哪里获得令牌，您在identify process中做了什么等等。如果您不将它们传入，JWTHelper将为您提供token getter&identifier的基本版本，它简单地从headers['Authorization']值获取令牌，然后检查解码后的字典是否有键值“username”。在
- 最后你可以创造aiohttp.web.Application应用程序然后把中间商送进来。如果要激活post router，这是传递pre_jwt_identifier（）和post_jwt_router（）的必要条件。不需要先通过decorator注册中间件。在
不同授权下的路由行为
| path | remarks | authorized destination | unauthorized destination |
|---|---|---|---|
| /index.html | Entry of main functional SPA | /index.html | /login.html |
| /login.html | Entry of login page. Independent SPA | /index.html | /login.html |
| /login_api | Login api , one in jwt whitelist. | /login_api | /login_api |
| /setattr_api | One of protected apis. | /setattr_api | 403 or 401 |
| /404 | Undefined page | /index.html | /login.html |
/*状态代码404将在SPA*/中处理
示例
server_basic.py
#下面是一个基本的aiohttp hello world服务器，分别有四种路由要求。在
server.py
#添加几行代码，以便使用jwtplus插件轻松启动服务器。在
importasynciofromaiohttpimportwebfromaiohttp_jwtplusimport(SecretManager,JWTHelper,basic_identifier,basic_token_getter,show_request_info)routes=web.RouteTableDef()@routes.get('/index.html')asyncdefmain_spa_page(request):show_request_info(request)returnweb.Response(text="this is index.html")@routes.get('/login.html')asyncdeflogin_spa_page(request):show_request_info(request)returnweb.Response(text="this is login.html")@routes.get('/authentication')asyncdefloginapi(request):show_request_info(request)returnweb.Response(text="loginapi called")@routes.get('/setattr')asyncdefsetattr(request):show_request_info(request)returnweb.Response(text='this is a procted api')secret_manager=SecretManager(secret='testsecret',# default empty, will generate a random string.refresh_interval='30d',# default 0 ,represents secret auto refresh disabled. Accept string or int scheme="Bearer",# default.algorithm='HS256',# default.exptime='30d',# default.)jwt=JWTHelper(unauthorized_return_route='/login.html',# this's an exception which means if you've alreadly logined ,you cannot access to this page. unauthorized_return_route_handler=login_spa_page,authorized_return_page_handler=main_spa_page,secret_manager=secret_manager,token_getter=basic_token_getter,# defaultidentifier=basic_identifier,# defaultwhitelist=('/authentication',),# must be a tuple ,accepts regular expresion.protected_apis=['/setattr',])app=web.Application(middlewares=[jwt.pre_jwt_identifier(),jwt.post_jwt_router(),])app.add_routes(routes)loop=asyncio.get_event_loop()loop.create_task(secret_manager.auto_refresh())# Explicit trigger eventloop since we starts a secret-auto-refresh thread. loop.run_until_complete(web._run_app(app))
client.py
#为了快速测试python假装的前端。在
importasynciofromaiohttpimportClientSessionfromaiohttp_jwtplusimport(SecretManager,JWTHelper,basic_identifier,# it's a coroutinebasic_token_getter,# it's a coroutineshow_request_info# print function)secret_manager=SecretManager(secret='testsecret',# default empty, will generate a random string.refresh_interval=0,# default 0 , no auto refresh.algorithm='HS256',# default.exptime='30d',# default.)url_pattern='http://localhost:8080{}'url_exts=['/index.html','/login.html','/authentication','/setattr','/404',]# Simulate you've alreadly got token feedback from server.# If you would like to generate token(without scheme) ,it accepts a dictionary# and items in which would be updated to jwt source payload. jwt=secret_manager.encode({'username':'jacky'})headers={'Authorization':"Bearer "+jwt.decode()}asyncdefmain():asyncwithClientSession()assession:print(f'{"#"*22}\nWith authentication')forurlextinurl_exts:asyncwithsession.get(url_pattern.format(urlext),headers=headers)asresponse:text=awaitresponse.text()print(f"called {urlext} ,\n\tget statuscode {response.status} , \n\treturn text \"{text}\"")print(f'{"#"*22}\nWithout authentication')forurlextinurl_exts:asyncwithsession.get(url_pattern.format(urlext),headers={'Authorization':'None'})asresponse:text=awaitresponse.text()print(f"called {urlext} ,\n\tget statuscode {response.status} , \n\treturn text \"{text}\"")asyncio.run(main())
modify_identifier.py
#自修改标识符和令牌。在
fromaiohttpimportwebfromaiohttp_jwtplusimport(SecretManager,JWTHelper)asyncdefidentifier_mod(payload):''' An identifier accepts a payload(as dictionary of jwt decoded result), and whose return value will be stored as one of request's property named 'auth_carry' If you would like to make identification fail in middleware(before handle requests), return False. You don't need to worry about exceptions. '''if'username'inpayload:returnpayload['username']else:returnFalse@routes.get('/index.html')asyncdefauthorised(request):username=request['auth_carry']['username']ifusername=='admin':returnweb.Response(text='pass')else:returnweb.Response(text='fail')secret_manager=SecretManager(secret='testsecret')jwt=JWTHelper(unauthorized_return_route='',unauthorized_return_route_handler=authorised,authorized_return_page_handler=authorised,secret_manager=secret_manager)app=web.Application(middlewares=[jwt.pre_jwt_identifier(),jwt.post_jwt_router(),])app.add_routes(routes)web.run_app(app)# Then you shall start a simulate client and encode a header with jwt authenrized payload with 'usernaem : admin' in it# and test if you got the corret response.
- 项目
标签：
欢迎加入QQ群-->： 979659372
