aiohttp_security

The library provides security for aiohttp.web.

The current version is 0.4

Contents

Usage

First of all, what is aiohttp_security about?

aiohttp-security is a set of public API functions as well as a reference standard for implementation details for securing access to assets served by a wsgi server.

Assets are secured using authentication and authorization as explained below. aiohttp-security is part of the aio-libs project which takes advantage of asynchronous processing using Python’s asyncio library.

Public API

The API is agnostic to the low level implementation details such that all client code only needs to implement the endpoints as provided by the API (instead of calling policy code directly (see explanation below)).

Via the API an application can:

1. remember a user in a local session (remember()),
2. forget a user in a local session (forget()),
3. retrieve the userid (authorized_userid()) of a remembered user from an identity (discussed below), and
4. check the permission of a remembered user (permits()).

The library internals are built on top of two concepts:

1. authentication, and
2. authorization.

There are abstract base classes for both types as well as several pre-built implementations that are shipped with the library. However, the end user is free to build their own implementations.

The library comes with two pre-built identity policies; one that uses cookies, and one that uses sessions [1]. It is envisioned that in most use cases developers will use one of the provided identity policies (Cookie or Session) and implement their own authorization policy.

The workflow is as follows:

1. User is authenticated. This has to be implemented by the developer.
2. Once user is authenticated an identity string has to be created for that user. This has to be implemented by the developer.
3. The identity string is passed to the Identity Policy’s remember method and the user is now remembered (Cookie or Session if using built-in). Only once a user is remembered can the other API methods: permits(), forget(), and authorized_userid() be invoked .
4. If the user tries to access a restricted asset the permits() method is called. Usually assets are protected using the check_permission() helper. This should return True if permission is granted.

The permits() method is implemented by the developer as part of the AbstractAuthorizationPolicy and passed to the application at runtime via setup.

In addition a check_authorized() also exists that requires no permissions (i.e. doesn’t call permits() method) but only requires that the user is remembered (i.e. authenticated/logged in).

Authentication

Authentication is the process where a user’s identity is verified. It confirms who the user is. This is traditionally done using a user name and password (note: this is not the only way).

A authenticated user has no access rights, rather an authenticated user merely confirms that the user exists and that the user is who they say they are.

In aiohttp_security the developer is responsible for their own authentication mechanism. aiohttp_security only requires that the authentication result in a identity string which corresponds to a user’s id in the underlying system.

Note

identity is a string that is shared between the browser and the server. Therefore it is recommended that a random string such as a uuid or hash is used rather than things like a database primary key, user login/email, etc.

Identity Policy

Once a user is authenticated the aiohttp_security API is invoked for storing, retrieving, and removing a user’s identity. This is accommplished via AbstractIdentityPolicy’s remember(), identify(), and forget() methods. The Identity Policy is therefore the mechanism by which a authenticated user is persisted in the system.

aiohttp_security has two built in identity policy’s for this purpose. CookiesIdentityPolicy that uses cookies and SessionIdentityPolicy that uses sessions via aiohttp-session library.

Authorization

Once a user is authenticated (see above) it means that the user has an identity. This identity can now be used for checking access rights or permission using a authorization policy.

The authorization policy’s permits() method is used for this purpose.

When aiohttp.web.Request has an identity it means the user has been authenticated and therefore has an identity that can be checked by the authorization policy.

As noted above, identity is a string that is shared between the browser and the server. Therefore it is recommended that a random string such as a uuid or hash is used rather than things like a database primary key, user login/email, etc.

Footnotes

[1]	jwt - json web tokens in the works

Reference

Public API functions

aiohttp_security.setup(app, identity_policy, autz_policy)

Setup aiohttp application with security policies.

Parameters:

• app – aiohttp aiohttp.web.Application instance.
• identity_policy – indentification policy, an AbstractIdentityPolicy instance.
• autz_policy – authorization policy, an AbstractAuthorizationPolicy instance.

coroutine aiohttp_security.remember(request, response, identity, **kwargs)

Remember identity in response, e.g. by storing a cookie or saving info into session.

The action is performed by registered AbstractIdentityPolicy.remember().

Usually the identity is stored in user cookies somehow for using by authorized_userid() and permits().

Parameters:

• request – aiohttp.web.Request object.
• response – aiohttp.web.StreamResponse and descendants like aiohttp.web.Response.
• identity (str) – aiohttp.web.Request object.
• kwargs –

  additional arguments passed to AbstractIdentityPolicy.remember().

  They are policy-specific and may be used, e.g. for specifiying cookie lifetime.

coroutine aiohttp_security.forget(request, response)

Forget previously remembered identity.

The action is performed by registered AbstractIdentityPolicy.forget().

Parameters:

• request – aiohttp.web.Request object.
• response – aiohttp.web.StreamResponse and descendants like aiohttp.web.Response.

coroutine aiohttp_security.check_authorized(request)

Checker that doesn’t pass if user is not authorized by request.

Parameters:request – aiohttp.web.Request object. Return str:authorized user ID if success Raise:aiohttp.web.HTTPUnauthorized for anonymous users.

Usage:

async def handler(request):
    await check_authorized(request)
    # this line is never executed for anonymous users

coroutine aiohttp_security.check_permission(request, permission)

Checker that doesn’t pass if user has no requested permission.

Parameters:request – aiohttp.web.Request object. Raise:aiohttp.web.HTTPUnauthorized for anonymous users. Raise:aiohttp.web.HTTPForbidden if user is authorized but has no access rights.

Usage:

async def handler(request):
    await check_permission(request, 'read')
    # this line is never executed if a user has no read permission

coroutine aiohttp_security.authorized_userid(request)

Retrieve userid.

The user should be registered by remember() before the call.

Parameters:request – aiohttp.web.Request object. Returns:str userid or None for session without signed in user.

coroutine aiohttp_security.permits(request, permission, context=None)

Check user’s permission.

Return True if user remembered in request has specified permission.

Allowed permissions as well as context meaning are depends on AbstractAuthorizationPolicy implementation.

Actually it’s a wrapper around AbstractAuthorizationPolicy.permits() coroutine.

The user should be registered by remember() before the call.

Parameters:

• request – aiohttp.web.Request object.
• permission – Requested permission. str or enum.Enum object.
• context – additional object may be passed into AbstractAuthorizationPolicy.permission() coroutine.

Returns:

True if registered user has requested permission, False otherwise.

coroutine aiohttp_security.is_anonymous(request)

Checks if user is anonymous user.

Return True if user is not remembered in request, otherwise returns False.

Parameters:request – aiohttp.web.Request object.

Abstract policies

aiohttp_security is built on top of two abstract policies –

AbstractIdentityPolicy and AbstractAuthorizationPolicy.

The first one responds on remembering, retrieving and forgetting identity into some session storage, e.g. HTTP cookie or authorization token.

The second is responsible to return persistent userid for session-wide identity and check user’s permissions.

Most likely sofware developer reuses one of pre-implemented identity policies from aiohttp_security but build authorization policy from scratch for every application/project.

Identification policy

class aiohttp_security.AbstractIdentityPolicy

coroutine identify(request)

Extract identity from request.

Abstract method, should be overriden by descendant.

Parameters:request – aiohttp.web.Request object. Returns:the claimed identity of the user associated request or None if no identity can be found associated with the request.

coroutine remember(request, response, identity, **kwargs)

Remember identity.

May use request for accessing required data and response for storing identity (e.g. updating HTTP response cookies).

kwargs may be used by concrete implementation for passing additional data.

Abstract method, should be overriden by descendant.

Parameters:

• request – aiohttp.web.Request object.
• response – aiohttp.web.StreamResponse object or derivative.
• identity – identity to store.
• kwargs – optional additional arguments. An individual identity policy and its consumers can decide on the composition and meaning of the parameter.

coroutine forget(request, response)

Forget previously stored identity.

May use request for accessing required data and response for dropping identity (e.g. updating HTTP response cookies).

Abstract method, should be overriden by descendant.

Parameters:

• request – aiohttp.web.Request object.
• response – aiohttp.web.StreamResponse object or derivative.

Authorization policy

class aiohttp_security.AbstractAuthorizationPolicy

coroutine authorized_userid(identity)

Retrieve authorized user id.

Abstract method, should be overriden by descendant.

Parameters:identity – an identity used for authorization. Returns:the userid of the user identified by the identity or None if no user exists related to the identity.

coroutine permits(identity, permission, context=None)

Check user permissions.

Abstract method, should be overriden by descendant.

Parameters:

• identity – an identity used for authorization.
• permission – requested permission. The type of parameter is not fixed and depends on implementation.

How to Make a Simple Server With Authorization

Simple example:

from aiohttp import web
from aiohttp_session import SimpleCookieStorage, session_middleware
from aiohttp_security import check_permission, \
    is_anonymous, remember, forget, \
    setup as setup_security, SessionIdentityPolicy
from aiohttp_security.abc import AbstractAuthorizationPolicy


# Demo authorization policy for only one user.
# User 'jack' has only 'listen' permission.
# For more complicated authorization policies see examples
# in the 'demo' directory.
class SimpleJack_AuthorizationPolicy(AbstractAuthorizationPolicy):
    async def authorized_userid(self, identity):
        """Retrieve authorized user id.
        Return the user_id of the user identified by the identity
        or 'None' if no user exists related to the identity.
        """
        if identity == 'jack':
            return identity

    async def permits(self, identity, permission, context=None):
        """Check user permissions.
        Return True if the identity is allowed the permission
        in the current context, else return False.
        """
        return identity == 'jack' and permission in ('listen',)


async def handler_root(request):
    is_logged = not await is_anonymous(request)
    return web.Response(text='''<html><head></head><body>
            Hello, I'm Jack, I'm {logged} logged in.<br /><br />
            <a href="/login">Log me in</a><br />
            <a href="/logout">Log me out</a><br /><br />
            Check my permissions,
            when i'm logged in and logged out.<br />
            <a href="/listen">Can I listen?</a><br />
            <a href="/speak">Can I speak?</a><br />
        </body></html>'''.format(
            logged='' if is_logged else 'NOT',
        ), content_type='text/html')


async def handler_login_jack(request):
    redirect_response = web.HTTPFound('/')
    await remember(request, redirect_response, 'jack')
    raise redirect_response


async def handler_logout(request):
    redirect_response = web.HTTPFound('/')
    await forget(request, redirect_response)
    raise redirect_response


async def handler_listen(request):
    await check_permission(request, 'listen')
    return web.Response(body="I can listen!")


async def handler_speak(request):
    await check_permission(request, 'speak')
    return web.Response(body="I can speak!")


async def make_app():
    #
    # WARNING!!!
    # Never use SimpleCookieStorage on production!!!
    # It’s highly insecure!!!
    #

    # make app
    middleware = session_middleware(SimpleCookieStorage())
    app = web.Application(middlewares=[middleware])

    # add the routes
    app.add_routes([
        web.get('/', handler_root),
        web.get('/login', handler_login_jack),
        web.get('/logout', handler_logout),
        web.get('/listen', handler_listen),
        web.get('/speak', handler_speak)])

    # set up policies
    policy = SessionIdentityPolicy()
    setup_security(app, policy, SimpleJack_AuthorizationPolicy())

    return app


if __name__ == '__main__':
    web.run_app(make_app(), port=9000)

Permissions with database-based storage

We use SimpleCookieStorage and an in-memory SQLite DB to make it easy to try out the demo. When developing an application, you should use EncryptedCookieStorage or RedisStorage and a production-ready database. If you want the full source code in advance or for comparison, check out the demo source.

Database

When the application runs, we initialise the DB with sample data using SQLAlchemy ORM:

async def init_db(db_engine: AsyncEngine, db_session: async_sessionmaker[AsyncSession]) -> None:
    """Initialise DB with sample data."""
    async with db_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with db_session.begin() as sess:
        pw = "$5$rounds=535000$2kqN9fxCY6Xt5/pi$tVnh0xX87g/IsnOSuorZG608CZDFbWIWBr58ay6S4pD"
        sess.add(User(username="admin", password=pw, is_superuser=True))
        moderator = User(username="moderator", password=pw)
        user = User(username="user", password=pw)
        sess.add(moderator)
        sess.add(user)
    async with db_session.begin() as sess:
        sess.add(Permission(user_id=moderator.id, name="protected"))
        sess.add(Permission(user_id=moderator.id, name="public"))
        sess.add(Permission(user_id=user.id, name="public"))

This will consist of 2 tables/models created in db.py:

Users:

class User(Base):
    """A user and their credentials."""

    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(sa.String(256), unique=True, index=True)
    password: Mapped[str] = mapped_column(sa.String(256))
    is_superuser: Mapped[bool] = mapped_column(
        default=False, server_default=sa.sql.expression.false())
    disabled: Mapped[bool] = mapped_column(
        default=False, server_default=sa.sql.expression.false())
    permissions = relationship("Permission", cascade="all, delete")

And their permissions:

class Permission(Base):
    """A permission that grants a user access to something."""

    __tablename__ = "permissions"

    user_id: Mapped[int] = mapped_column(
        sa.ForeignKey(User.id, ondelete="CASCADE"), primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(64), primary_key=True)

Writing policies

You need to implement two entities: IdentityPolicy and AuthorizationPolicy. First one should have these methods: identify, remember and forget. For the second one: authorized_userid and permits. We will use the included SessionIdentityPolicy and write our own database-based authorization policy.

In our example we will lookup a user login in the database and, if present, return the identity.

    async def authorized_userid(self, identity: str) -> str | None:
        where = _where_authorized(identity)
        async with self.dbsession() as sess:
            user_id = await sess.scalar(sa.select(User.id).where(*where))
        return str(user_id) if user_id else None

For permission checking, we will fetch the user first, check if he is superuser (all permissions are allowed), otherwise check if the permission is explicitly set for that user.

    async def permits(self, identity: str | None, permission: str | Enum,
                      context: dict[str, object] | None = None) -> bool:
        if identity is None:
            return False

        where = _where_authorized(identity)
        stmt = sa.select(User).options(selectinload(User.permissions)).where(*where)
        async with self.dbsession() as sess:
            user = await sess.scalar(stmt)

        if user is None:
            return False
        if user.is_superuser:
            return True
        return any(p.name == permission for p in user.permissions)

Setup

Once we have all the code in place we can install it for our application:

async def init_app() -> web.Application:
    app = web.Application()

    db_engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    app["db_session"] = async_sessionmaker(db_engine, expire_on_commit=False)

    await init_db(db_engine, app["db_session"])

    setup_session(app, SimpleCookieStorage())
    setup_security(app, SessionIdentityPolicy(), DBAuthorizationPolicy(app["db_session"]))

    web_handlers = Web()
    web_handlers.configure(app)

    return app

Now we have authorization and can decorate every other view with access rights based on permissions. There are two helpers included for this:

from aiohttp_security import check_authorized, check_permission

For each view you need to protect - just apply the decorator on it.

    async def protected_page(self, request: web.Request) -> web.Response:
        await check_permission(request, 'protected')
        return web.Response(text="You are on protected page")

or

    async def logout(self, request: web.Request) -> web.Response:
        await check_authorized(request)
        response = web.Response(text="You have been logged out")
        await forget(request, response)
        return response

If someone tries to access that protected page he will see:

403: Forbidden

The best part of it - you can implement any logic you want following the API conventions.

Launch application

For working with passwords there is a good library passlib. Once you’ve created some users you want to check their credentials on login. A similar function may do what you are trying to accomplish:

from passlib.hash import sha256_crypt

async def check_credentials(db_session: async_sessionmaker[AsyncSession],
                            username: str, password: str) -> bool:
    where = _where_authorized(username)
    async with db_session() as sess:
        hashed_pw = await sess.scalar(sa.select(User.password).where(*where))

    if hashed_pw is None:
        return False

    return sha256_crypt.verify(password, hashed_pw)

Final step is to launch your application:

python -m database_auth

Try to login with admin/moderator/user accounts (with password password) and access /public or /protected endpoints.

Glossary

aiohttp

asyncio based library for making web servers.

asyncio

The library for writing single-threaded concurrent code using coroutines, multiplexing I/O access over sockets and other resources, running network clients and servers, and other related primitives.

Reference implementation of PEP 3156

https://pypi.python.org/pypi/asyncio/

authentication

Actions related to retrieving, storing and removing user’s identity.

Authenticated user has no access rights, the system even has no knowledge is there the user still registered in DB.

If Request has an identity it means the user has some ID that should be checked by authorization policy.

authorization

Checking actual permissions for identified user along with getting userid.

identity

Session-wide str for identifying user.

Stored in local storage (client-side cookie or server-side storage).

Use remember() for saving identity (sign in) and forget() for dropping it (sign out).

identity is used for getting userid and permission.

permission

Permission required for access to resource.

Permissions are just strings, and they have no required composition: you can name permissions whatever you like.

userid

User’s ID, most likely his login or email

License

aiohttp_security is offered under the Apache 2 license.

Indices and tables

• Index
• Module Index
• Search Page