跳转到主要内容

针对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
  • 升级了对 blake3aiohttpaiohttp-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。此策略结合了FormPolicyHeaderPolicy的行为。

如果需要,您可以实现自己的自定义策略。但请确保您的自定义策略实现了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):
    ...

在这种情况下,自定义错误处理器将仅应用于此处理器。对于其他处理器,将应用全局错误处理器。

项目详情


下载文件

下载您平台的文件。如果您不确定选择哪一个,请了解有关 安装包 的更多信息。

源代码分发

aiohttp_csrf2-1.0.1.tar.gz (16.0 kB 查看散列值)

上传时间 源代码

构建分发

aiohttp_csrf2-1.0.1-py3-none-any.whl (9.1 kB 查看散列值)

上传时间 Python 3

支持