Filter collections with SQLAlchemy

The sqlalchemy-oso library can enforce policies over SQLAlchemy models. This allows policies to control access to collections of objects without needing to authorize each object individually.

Installation

The Oso SQLAlchemy integration is available on PyPI and can be installed using pip:

$ pip install sqlalchemy-oso

Usage

sqlalchemy-oso works over your existing SQLAlchemy ORM models without modification.

To get started, we need to:

1. Make Oso aware of our SQLAlchemy model types so that we can write policies over them.
2. Create a SQLAlchemy Session that uses Oso to authorize access to data.

Register models with Oso

Alternatively, the oso.Oso.register_class method can be called on each SQLAlchemy model that you want to reference in your policy.

Create a SQLAlchemy Session that uses Oso

Oso performs authorization by integrating with SQLAlchemy sessions. Use the sqlalchemy_oso.authorized_sessionmaker() session factory instead of the default SQLAlchemy sessionmaker. Every query made using sessions from the authorized_sessionmaker() factory will have authorization applied.

Before executing a query, Oso consults the policy and obtains a list of conditions that must be met for a model to be authorized. These conditions are translated into SQLAlchemy expressions and applied to the query before retrieving objects from the database.

Using with Flask

sqlalchemy-oso has built-in support for the popular flask_sqlalchemy library.

Example

Let’s look at an example usage of this library. Our example is a social media app that allows users to create posts. There is a Post model and a User model:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

from sqlalchemy import Column, String, Integer, Boolean, ForeignKey, Enum, Table

Model = declarative_base(name="Model")

class Post(Model):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)

    contents = Column(String)
    access_level = Column(Enum("public", "private"), nullable=False)

    created_by_id = Column(Integer, ForeignKey("users.id"))
    created_by = relationship("User")

"""Represent a management relationship between users.  A record in this table
indicates that ``user_id``'s account can be managed by the user with ``manager_id``.
"""
user_manages = Table(
    "user_manages",
    Model.metadata,
    Column("managed_user_id", Integer, ForeignKey("users.id")),
    Column("manager_user_id", Integer, ForeignKey("users.id"))
)

class User(Model):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String, nullable=False)

    is_admin = Column(Boolean, nullable=False, default=False)

    manages = relationship("User",
        secondary="user_manages",
        primaryjoin=(id == user_manages.c.manager_user_id),
        secondaryjoin=(id == user_manages.c.managed_user_id),
        backref="managed_by"
    )

Now, we’ll write a policy over these models. Our policy contains the following rules:

1. A user can read any public post.
2. A user can read their own private posts.
3. A user can read private posts for users they manage (defined through the user.manages relationship).
4. A user can read all other users.

allow(_: User, "read", post: Post) if
    post.access_level = "public";

allow(user: User, "read", post: Post) if
    post.access_level = "private" and
    post.created_by = user;

allow(user: User, "read", post: Post) if
    post.access_level = "private" and
    post.created_by in user.manages;

allow(_: User, "read", _: User);

These rules are written over single model objects.

Trying it out

Let’s test out the policy in a REPL.

First, import sqlalchemy, oso, and sqlalchemy_oso:

>>> from sqlalchemy import create_engine
>>> from sqlalchemy.orm import Session
>>> from oso import Oso
>>> from sqlalchemy_oso import authorized_sessionmaker, register_models
>>> from sqlalchemy_example.models import Model, User, Post

Then, setup oso and register our models.

>>> oso = Oso()
>>> register_models(oso, Model)
>>> oso.load_file("sqlalchemy_example/policy.polar")

Next, setup some test data…

>>> user = User(username='user')
>>> manager = User(username='manager', manages=[user])
>>> public_user_post = Post(contents='public_user_post',
...                         access_level='public',
...                         created_by=user)
>>> private_user_post = Post(contents='private_user_post',
...                          access_level='private',
...                          created_by=user)
>>> private_manager_post = Post(contents='private_manager_post',
...                             access_level='private',
...                             created_by=manager)
>>> public_manager_post = Post(contents='public_manager_post',
...                            access_level='public',
...                            created_by=manager)

… and load that data into SQLAlchemy:

>>> engine = create_engine('sqlite:///:memory:')
>>> Model.metadata.create_all(engine)
>>> fixture_session = Session(bind=engine)
>>> fixture_session.add_all([
...     user, manager, public_user_post, private_user_post, private_manager_post,
...     public_manager_post])
>>> fixture_session.commit()

Authorizing a user’s posts

Now that we’ve setup some test data, let’s use oso to authorize Posts that User(username="user") can see.

We’ll start by making an authorized_sessionmaker():

>>> AuthorizedSession = authorized_sessionmaker(bind=engine,
...                                             get_oso=lambda: oso,
...                                             get_user=lambda: user,
...                                             get_checked_permissions=lambda: { Post: "read" })
>>> session = AuthorizedSession()

Then, issue a query for all posts:

>>> posts = session.query(Post).all()
>>> [p.contents for p in posts]
['public_user_post', 'private_user_post', 'public_manager_post']

Since we used authorized_sessionmaker(), the query only returned authorized posts based on the policy.

User(username="user") can see their own public and private posts and public posts made by other users.

Authorizing a manager’s posts

Now we’ll authorize access to User(username="manager")’s Posts. We create a new authorized session with user set to manager:

>>> AuthorizedSession = authorized_sessionmaker(bind=engine,
...                                             get_oso=lambda: oso,
...                                             get_user=lambda: manager,
...                                             get_checked_permissions=lambda: { Post: "read" })
>>> manager_session = AuthorizedSession()

And issue the same query as before…

>>> posts = manager_session.query(Post).all()
>>> [p.contents for p in posts]
['public_user_post', 'private_user_post', 'private_manager_post', 'public_manager_post']

This time, the query returned four posts! Since the manager user manages user, the private post of user is also authorized (based on our third rule above).

>>> manager.manages[0].username
'user'

This full example is available on GitHub.

How Oso authorizes SQLAlchemy Data

As you can see from the above example, the SQLAlchemy Oso integration allows regular SQLAlchemy queries to be executed with authorization applied.

Before compiling a SQLAlchemy query, the entities in the query are authorized with Oso. Oso returns authorization decisions for each entity that indicate what constraints must be met for the entity to be authorized. These constraints are then translated into filters on the SQLAlchemy query object.

For example, our above policy has the following code:

allow(user: User, "read", post: Post) if
    post.access_level = "private" and
    post.created_by = user;

The Oso library converts the constraints on Post expressed in this policy into a SQLAlchemy query like:

session.query(Post)
    .filter(Post.access_level == "private" & Post.created_by == user)

This translation makes the policy an effective abstraction for expressing authorization logic over collections.

Limitations

There are some operators and features that do not currently work with the SQLAlchemy adapter when used anywhere in the policy:

• The cut operator

• Rules that rely on ordered execution based on class inheritance

• Negated queries using the not operator that contain a matches operation within the negation or call a rule containing a specializer. For example:

  # Not supported.
  allow(_actor, _action, resource) if
      not resource matches Repository;
  
  # Also not supported.
  is_repo(r: Repository);
  allow(_actor, _action, resource) if
      not is_repo(resource);
  

Some operations cannot be performed on resources in allow rules used with the SQLAlchemy adapter. These operations can still be used on the actor or action:

• Application method calls
• Arithmetic operators

Connect with us on Slack

If you have any questions, or just want to talk something through, jump into Slack. An Oso engineer or one of the thousands of developers in the growing community will be happy to help.