Skip to content

Overview

Introduction

respo states for resource policy and is tiny, user friendly tool for building RBAC systems based on static yml file, mainly with FastAPI framework in mind. In most cases – for even large set of roles – single file would be enough to provide restricting system access.

Features:

  • It provides custom fields for SQLAlchemy and Django to store users roles in database.

  • Implements R. Sandhu Role-based access control text.

  • Dead simple, fast and can be trusted – 100% coverage.

  • No issues with mutlithreading and multiprocessing – you just pass around already prepared, compiled respo_model (from file) in your app that is readonly.

  • Generates your roles, permissions offline and compile it to pickle file for superfast access in an app.

  • Detailed documentation and error messages in CLI command.

  • 100% autocompletion and typing support with optional code generation for even better typing support.

Note, every piece of code in the docs is a tested python/yml file, feel free to use it.

Usage in FastAPI

The goal is to use simple and reusable dependency factory user_have_permission("some permission") that will verify just having User database instance if user have access to resoruce. Single endpoint must have single permission for it, and thanks to respo compilation step, every "stronger" permissions and roles would include "weaker" so we don't need to have the if statements everywhere around application.

1
2
3
4
5
6
7
8
from .dependencies import user_have_permission

...


@router.get("/users/read_all/")
def users_read_all(user = Depends(user_have_permission("users.read_all"))):
    return user

Declaring YML file with permissions, roles.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
# paste this file to "respo-example.yml"

permissions:
  - user.read_basic
  - user.read_all

principles:
  - when: user.read_all
    then: [user.read_basic]

roles:
  - name: default
    permissions:
      - user.read_basic

  - name: admin
    include: [default]
    permissions:
      - user.read_all

There are 3 sections:

  • permissions, list of double labels "{collection}.{label}", they represent single permission that user can be given.
  • principles, list of principles where one can declare rules for permissions to contain others. For example, we want to some one that (through some role) have "more powerful" permission user.read_all to also have access to resources that require user.read_basic permission.
  • roles, list of roles objects, note that they have unique name, set of permissions and optional list include of other roles, so we can give admin role the same set of rules as default user (and add more powerful ones).

Parsing YML file to readonly pickle using respo CLI interface.

Thanks to Click, respo has powerful cli interface.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
$ respo --help  # prints all available commands

$ respo create respo-example.yml  # create model from file

### Result ###
$ INFO: Validating respo model from respo-example.yml...
$ INFO: Saved binary file to .respo_cache/__auto__respo_model.bin
$ INFO: Saved yml file to .respo_cache/__auto__respo_model.yml
$ INFO: Saved python file to respo_model.py
$ INFO: Processed in 0.0058s. Bin file size: 0.0007 mb.
$ INFO: Success!

That powerful command create serveral of things:

  • .respo_cache/__auto__respo_model.bin file (pickle format of model).

  • .respo_cache/__auto__respo_model.yml file (yml format of model).

  • respo_model.py file (python file of model for better autocompletion).

(Refer to User Guide for more information about first two of them.)

respo_model.py

Autogenerated file with class based on respo.RespoModel with bonus autocompletion.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
"""
Auto generated using respo create command
Docs: https://rafsaf.github.io/respo/
"""

import typing

import respo


class RespoModel(respo.RespoModel):
    if typing.TYPE_CHECKING:

        class _ROLES(respo.ROLESContainer):
            ADMIN: str
            DEFAULT: str

        class _PERMS(respo.PERMSContainer):
            USER__ALL: str
            USER__READ_ALL: str
            USER__READ_BASIC: str

        PERMS: _PERMS
        ROLES: _ROLES

        @staticmethod
        def get_respo_model() -> "RespoModel":
            return respo.RespoModel.get_respo_model()  # type: ignore

Usage in FastAPI and SQLAlchemy.

To interact with stateless, readonly respo_model created above, respo provides abstraction called RespoClient that can be stored in database as a custom respo SQLAlchemyRespoField field, be given or removed a role that can also check using respo_model instance its permissions.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import asyncio
from dataclasses import dataclass, field

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import registry

from respo import RespoClient
from respo.fields.sqlalchemy import SQLAlchemyRespoField

from .respo_model import RespoModel

Base = registry()


@Base.mapped
@dataclass
class ExampleModel:
    __tablename__ = "example_model"
    __sa_dataclass_metadata_key__ = "sa"

    id: int = field(init=False, metadata={"sa": Column(Integer, primary_key=True)})
    respo_field: RespoClient = field(
        default_factory=RespoClient,
        metadata={
            "sa": Column(SQLAlchemyRespoField, nullable=False, server_default="")
        },
    )
    name: str = field(
        default="Ursula",
        metadata={"sa": Column(String(128), nullable=False, server_default="Ursula")},
    )


async def main():
    respo_model = RespoModel.get_respo_model()  # loads respo model from pickled file

    async_engine = create_async_engine(
        "sqlite+aiosqlite:///db.sqlite3", pool_pre_ping=True
    )
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

        async with AsyncSession(bind=conn) as session:
            new_obj = ExampleModel(name="Respo")
            new_obj.respo_field.add_role(respo_model.ROLES.ADMIN, respo_model)

            session.add(new_obj)
            await session.commit()  # respo_field is stored as a string!
            await session.refresh(new_obj)

            assert new_obj.respo_field.has_permission(
                respo_model.PERMS.USER__READ_BASIC, respo_model
            )


asyncio.run(main())

Note, thanks to smart types in RespoField for SQLAlchemy and auto generated typed file respo_model we have powerful autocompletions:

autocompletion-respo-client

autocompletion-respo-client

autocompletion-respo-client