# 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.