Filter collections with SQLAlchemy
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.
The Oso SQLAlchemy integration is available on
PyPI and can be installed using
$ pip install sqlalchemy-oso
sqlalchemy-oso works over your existing SQLAlchemy ORM models without
To get started, we need to:
- Make Oso aware of our SQLAlchemy model types so that we can write policies over them.
- Create a SQLAlchemy Session that uses Oso to authorize access to data.
Register models with Oso
sqlalchemy_oso.register_models() registers all models that descend from a
declarative base class as types that are available in the policy.
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
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
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
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:
- A user can read any public post.
- A user can read their own private posts.
- A user can read private posts for users they manage (defined through the
- 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);
The SQLAlchemy integration is deny by default. The final rule for
needed to allow access to user objects for any user.
If a query is made for a model that does not have an explicit rule in the policy, no results will be returned.
These rules are written over single model objects.
Trying it out
Let’s test out the policy in a REPL.
>>> 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
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
User(username="user") can see.
We’ll start by making an
>>> AuthorizedSession = authorized_sessionmaker(bind=engine, ... get_oso=lambda: oso, ... get_user=lambda: user, ... get_action=lambda: "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
Posts. We create a
new authorized session with user set to
>>> AuthorizedSession = authorized_sessionmaker(bind=engine, ... get_oso=lambda: oso, ... get_user=lambda: manager, ... get_action=lambda: "read") >>> manager_session = AuthorizedSession()
In a real application,
get_user would be a function returning the current
user based on the current request context. For example, in Flask this might
lambda: flask.g.current_user or some other proxy object.
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
>>> manager.manages.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.
This feature is still under active development. Not all policies that work in a non-partial setting will currently work with partials. More policies will be supported as we continue working on this feature. The SQLAlchemy adapter is ready for evaluation and testing. However, we recommend getting in touch with us on Slack before using it in production.
There are some operators and features that do not currently work with the SQLAlchemy adapter when used anywhere in the policy:
Rules that rely on ordered execution based on class inheritance
Negated queries using the
notoperator that contain a
matchesoperation within the negation or call a rule containing a specializer. For example:
# Not supported. allow(actor, action, resource) if not resource matches User; # Also not supported. is_user(user: User); allow(actor, action, resource) if not is_user(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
- Application method calls
- Arithmetic operators