针对aiohttp-server的跨站请求伪造(CSRF)保护
项目描述
aiohttp_csrf
此库为 aiohttp.web 提供跨服务器请求伪造(csrf/xsrf)保护。
这是对 https://github.com/bitnom/aiohttp-csrf 的分支,增加了现代Python类型注释,切换到aiohttp AppKey
标签,并修复了历史测试套件。
1.0.0版本新功能
- 整个库中添加了类型提示。
- 现在,
aiohttp_csrf.setup()
和@csrf_protect
装饰器现在接受单独的可选关键字参数exception=...
和error_renderer=...
以允许自定义csrf失败。之前这是重载的单个参数error_renderer
。 - 升级了对
blake3
、aiohttp
和aiohttp-session
的依赖。 - 取消了对Python 3.8的支持
0.1.0版本
- 默认使用Blake3散列。这意味着您必须传递
secret_phrase
到正在使用的任何存储后端以生成令牌,例如aiohttp_csrf.storage.SessionStorage
基本用法
此库允许您为请求实现csrf(xsrf)保护
基本用法示例
import aiohttp_csrf
from aiohttp import web
FORM_FIELD_NAME = '_csrf_token'
COOKIE_NAME = 'csrf_token'
def make_app():
csrf_policy = aiohttp_csrf.policy.FormPolicy(FORM_FIELD_NAME)
csrf_storage = aiohttp_csrf.storage.CookieStorage(COOKIE_NAME)
app = web.Application()
aiohttp_csrf.setup(app, policy=csrf_policy, storage=csrf_storage)
app.middlewares.append(aiohttp_csrf.csrf_middleware)
async def handler_get_form_with_token(request):
token = await aiohttp_csrf.generate_token(request)
body = '''
<html>
<head><title>Form with csrf protection</title></head>
<body>
<form method="POST" action="/">
<input type="hidden" name="{field_name}" value="{token}" />
<input type="text" name="name" />
<input type="submit" value="Say hello">
</form>
</body>
</html>
''' # noqa
body = body.format(field_name=FORM_FIELD_NAME, token=token)
return web.Response(
body=body.encode('utf-8'),
content_type='text/html',
)
async def handler_post_check(request):
post = await request.post()
body = 'Hello, {name}'.format(name=post['name'])
return web.Response(
body=body.encode('utf-8'),
content_type='text/html',
)
app.router.add_route(
'GET',
'/',
handler_get_form_with_token,
)
app.router.add_route(
'POST',
'/',
handler_post_check,
)
return app
web.run_app(make_app())
初始化
首先,您需要将 aiohttp_csrf
初始化到您的应用程序中
app = web.Application()
csrf_policy = aiohttp_csrf.policy.FormPolicy(FORM_FIELD_NAME)
csrf_storage = aiohttp_csrf.storage.CookieStorage(COOKIE_NAME)
aiohttp_csrf.setup(app, policy=csrf_policy, storage=csrf_storage)
中间件和装饰器
初始化后,您可以使用@aiohttp_csrf.csrf_protect
为需要保护的处理器添加保护。或者,您可以初始化aiohttp_csrf.csrf_middleware
,而不必担心装饰器的使用(完整中间件示例在此)
# ...
app.middlewares.append(aiohttp_csrf.csrf_middleware)
# ...
在这种情况下,您所有的处理器都将受到保护。
注意:我们强烈建议使用aiohttp_csrf.csrf_middleware
和@aiohttp_csrf.csrf_exempt
,而不是手动使用@aiohttp_csrf.csrf_protect
。但如果您更喜欢使用@aiohttp_csrf.csrf_protect
,别忘了为GET和POST方法都使用@aiohttp_csrf.csrf_protect
(手动保护示例)
如果您想使用中间件,但需要不保护的处理器,您可以使用@aiohttp_csrf.csrf_exempt
。使用此装饰器标记您的处理器,则此处理器将不会检查令牌
@aiohttp_csrf.csrf_exempt
async def handler_post_not_check(request):
...
生成令牌
要生成令牌,您需要在处理器中调用aiohttp_csrf.generate_token
@aiohttp_csrf.csrf_protect
async def handler_get(request):
token = await aiohttp_csrf.generate_token(request)
...
高级使用
策略
您可以使用不同的策略来检查令牌。库提供了3种类型的策略
- FormPolicy。此策略将在POST请求的正文(通常用于表单)或相同名称的GET变量中搜索令牌。您需要指定将被检查的字段名称。
- HeaderPolicy。此策略将在POST请求的头部中搜索令牌(通常用于AJAX请求)。您需要指定将被检查的头部名称。
- FormAndHeaderPolicy。此策略结合了FormPolicy和HeaderPolicy的行为。
如果需要,您可以实现自己的自定义策略。但请确保您的自定义策略实现了aiohttp_csrf.policy.AbstractPolicy
接口。
存储
您可以使用不同类型的存储来存储令牌。库提供了2种类型的存储
- CookieStorage。您的令牌将存储在cookie变量中。您需要指定cookie名称。
- SessionStorage。您的令牌将存储在会话中。您需要指定会话变量名称。
重要:如果您想使用会话存储,您需要在您的应用程序中设置aiohttp_session(会话存储示例)
如果需要,您可以实现自己的自定义存储。但请确保您的自定义存储实现了aiohttp_csrf.storage.AbstractStorage
接口。
令牌生成器
您可以在应用程序中使用不同的令牌生成器。默认存储使用aiohttp_csrf.token_generator.SimpleTokenGenerator
但如果您需要更安全的令牌生成器,您可以使用aiohttp_csrf.token_generator.HashedTokenGenerator
并且您可以实施自己的自定义令牌生成器,如果需要。但请确保您的自定义令牌生成器实现了aiohttp_csrf.token_generator.AbstractTokenGenerator
接口。
无效令牌行为
默认情况下,如果令牌无效,aiohttp_csrf
将引发aiohttp.web.HTTPForbidden
异常。
您有权指定自定义错误处理器。它可以是
- 可调用实例。输入参数 - aiohttp请求。
def custom_error_handler(request):
# do something
return aiohttp.web.Response(status=403)
# or
async def custom_async_error_handler(request):
# await do something
return aiohttp.web.Response(status=403)
它将被调用而不是受保护的处理器。
- Exception的子类。在这种情况下,将引发此异常。
class CustomException(Exception):
pass
您可以在初始化应用程序中的aiohttp_csrf
时指定自定义错误处理器
...
class CustomException(Exception):
pass
...
aiohttp_csrf.setup(app, policy=csrf_policy, storage=csrf_storage, error_renderer=CustomException)
...
在这种情况下,自定义错误处理器将应用于所有受保护的处理器。
或您可以为特定的处理器指定自定义错误处理器
...
class CustomException(Exception):
pass
...
@aiohttp_csrf.csrf_protect(error_renderer=CustomException)
def handler_with_custom_csrf_error(request):
...
在这种情况下,自定义错误处理器将仅应用于此处理器。对于其他处理器,将应用全局错误处理器。