sqlalchemy-oso API Reference

sqlalchemy_oso

class sqlalchemy_oso.SQLAlchemyOso(sqlalchemy_base: type)

The central object to manage application policy state, e.g. the policy data, and verify requests when using Oso with SQLAlchemy.

Supports SQLAlchemy-specific functionality, including data filtering.

Accepts a SQLAlchemy declarative_base on initialization, which is used to register all relevant SQLAlchemy models with Oso.

>>> from sqlalchemy_oso import SQLAlchemyOso
>>> from sqlalchemy.ext.declarative import declarative_base
>>> Base = declarative_base(name="MyBaseModel")
>>> SQLAlchemyOso(Base)
<sqlalchemy_oso.oso.SQLAlchemyOso object at 0x...>
sqlalchemy_oso.authorized_sessionmaker(get_oso: Callable[[], Oso], get_user: Callable[[], Any], get_checked_permissions: Callable[[], Optional[Dict[Type[Any], Any]]], class_: Optional[Type[Session]] = None, **kwargs: Any) → sessionmaker

Session factory for sessions with Oso authorization applied.

Parameters:

  • get_oso – Callable that returns the Oso instance to use for authorization.

  • get_user – Callable that returns the user for an authorization request.

  • get_checked_permissions – Callable that returns an optional map of permissions (resource-action pairs) to authorize for the session. If the callable returns None, no authorization will be applied to the session. If a map of permissions is provided, querying for a SQLAlchemy model present in the map will authorize results according to the action specified as the value in the map. E.g., providing a map of {Post: "read", User: "view"} where Post and User are SQLAlchemy models will apply authorization to session.query(Post) and session.query(User) such that only Post objects that the user can "read" and User objects that the user can "view" are fetched from the database.

  • class – Base class to use for sessions.

All other keyword arguments are passed through to sqlalchemy.orm.session.sessionmaker() unchanged.

Invariant: the values returned by the get_oso(), get_user(), and get_checked_permissions() callables provided to this function must remain fixed for a given session. This prevents authorization responses from changing, ensuring that the session’s identity map never contains unauthorized objects.

NOTE: _baked_queries are disabled on SQLAlchemy 1.3 since the caching

mechanism can bypass authorization by using queries from the cache that were previously baked without authorization applied. Note that _baked_queries are deprecated as of SQLAlchemy 1.4.

sqlalchemy_oso.register_models(oso: Oso, base_or_registry)

Register all models in registry (SQLAlchemy 1.4) or declarative base class (1.3 and 1.4) base_or_registry with Oso as classes.

sqlalchemy_oso.session

SQLAlchemy session classes and factories for oso.

class sqlalchemy_oso.session.AuthorizedSession(oso: Oso, user: object, checked_permissions: Optional[Dict[Type[Any], Any]], **options: Any)

SQLAlchemy session that uses oso for authorization.

Queries on this session only return authorized objects.

Usually authorized_sessionmaker() is used instead of directly instantiating the session.

NOTE: _baked_queries are disabled on SQLAlchemy 1.3 since the caching

mechanism can bypass authorization by using queries from the cache that were previously baked without authorization applied. Note that _baked_queries are deprecated as of SQLAlchemy 1.4.

class sqlalchemy_oso.session.AuthorizedSessionBase(oso: Oso, user: object, checked_permissions: Optional[Dict[Type[Any], Any]], **options: Any)

Mixin for SQLAlchemy Session that uses oso authorization for queries.

Can be used to create a custom session class that uses oso:

class MySession(AuthorizedSessionBase, sqlalchemy.orm.Session):
    pass
NOTE: _baked_queries are disabled on SQLAlchemy 1.3 since the caching

mechanism can bypass authorization by using queries from the cache that were previously baked without authorization applied. Note that _baked_queries are deprecated as of SQLAlchemy 1.4.

sqlalchemy_oso.session.authorized_sessionmaker(get_oso: Callable[[], Oso], get_user: Callable[[], Any], get_checked_permissions: Callable[[], Optional[Dict[Type[Any], Any]]], class_: Optional[Type[Session]] = None, **kwargs: Any) → sessionmaker

Session factory for sessions with Oso authorization applied.

Parameters:

  • get_oso – Callable that returns the Oso instance to use for authorization.

  • get_user – Callable that returns the user for an authorization request.

  • get_checked_permissions – Callable that returns an optional map of permissions (resource-action pairs) to authorize for the session. If the callable returns None, no authorization will be applied to the session. If a map of permissions is provided, querying for a SQLAlchemy model present in the map will authorize results according to the action specified as the value in the map. E.g., providing a map of {Post: "read", User: "view"} where Post and User are SQLAlchemy models will apply authorization to session.query(Post) and session.query(User) such that only Post objects that the user can "read" and User objects that the user can "view" are fetched from the database.

  • class – Base class to use for sessions.

All other keyword arguments are passed through to sqlalchemy.orm.session.sessionmaker() unchanged.

Invariant: the values returned by the get_oso(), get_user(), and get_checked_permissions() callables provided to this function must remain fixed for a given session. This prevents authorization responses from changing, ensuring that the session’s identity map never contains unauthorized objects.

NOTE: _baked_queries are disabled on SQLAlchemy 1.3 since the caching

mechanism can bypass authorization by using queries from the cache that were previously baked without authorization applied. Note that _baked_queries are deprecated as of SQLAlchemy 1.4.

sqlalchemy_oso.session.scoped_session(get_oso: Callable[[], Oso], get_user: Callable[[], Any], get_checked_permissions: Callable[[], Optional[Dict[Type[Any], Any]]], scopefunc: Optional[Callable[[...], Any]] = None, **kwargs: Any) → scoped_session

Return a scoped session maker that uses the Oso instance, user, and checked permissions (resource-action pairs) as part of the scope function.

Use in place of sqlalchemy’s scoped_session.

Uses authorized_sessionmaker() as the factory.

Parameters:

  • get_oso – Callable that returns the Oso instance to use for authorization.

  • get_user – Callable that returns the user for an authorization request.

  • get_checked_permissions – Callable that returns an optional map of permissions (resource-action pairs) to authorize for the session. If the callable returns None, no authorization will be applied to the session. If a map of permissions is provided, querying for a SQLAlchemy model present in the map will authorize results according to the action specified as the value in the map. E.g., providing a map of {Post: "read", User: "view"} where Post and User are SQLAlchemy models will apply authorization to session.query(Post) and session.query(User) such that only Post objects that the user can "read" and User objects that the user can "view" are fetched from the database.

  • scopefunc – Additional scope function to use for scoping sessions. Output will be combined with the Oso, permissions (resource-action pairs), and user objects.

  • kwargs – Additional keyword arguments to pass to authorized_sessionmaker().

NOTE: _baked_queries are disabled on SQLAlchemy 1.3 since the caching

mechanism can bypass authorization by using queries from the cache that were previously baked without authorization applied. Note that _baked_queries are deprecated as of SQLAlchemy 1.4.

sqlalchemy_oso.session.set_get_session(oso: Oso, get_session_func: Callable[[], Session]) → None

Set the function that oso uses to expose a SQLAlchemy session to the policy

Parameters:

  • oso (Oso) – The Oso instance used to evaluate the policy.

  • get_session_func (lambda) – A function that returns a SQLAlchemy session

The session can be accessed from polar via the OsoSession constant. E.g.,

OsoSession.get().query(...)

sqlalchemy_oso.flask

Wrappers for using sqlalchemy_oso with the flask_sqlalchemy library.

class sqlalchemy_oso.flask.AuthorizedSQLAlchemy(get_oso: Callable[[], Oso], get_user: Callable[[], object], get_checked_permissions: Callable[[], Optional[Dict[Type[Any], Any]]], **kwargs: Any)

flask_sqlalchemy SQLAlchemy subclass that uses oso.

Creates sessions with oso authorization applied. See flask_sqlalchemy documentation for more information on using flask_sqlalchemy.

Parameters:

  • get_oso – Callable that returns the oso.Oso instance to use for authorization.

  • get_user – Callable that returns the user to authorize for the current request.

  • get_checked_permissions – Callable that returns the permissions to authorize for the current request.

>>> from sqlalchemy_oso.flask import AuthorizedSQLAlchemy
>>> db = AuthorizedSQLAlchemy(
...    get_oso=lambda: flask.current_app.oso,
...    get_user=lambda: flask_login.current_user,
...    get_checked_permissions=lambda: {Post: flask.request.method}
... )
create_scoped_session(options: Optional[MutableMapping[str, Any]] = None) → scoped_session

Create a scoped_session on the factory from create_session().

An extra key 'scopefunc' can be set on the options dict to specify a custom scope function. If it’s not provided, Flask’s app context stack identity is used. This will ensure that sessions are created and removed with the request/response cycle, and should be fine in most cases.

Parameters:

options – dict of keyword arguments passed to session class in create_session

create_session(options: Mapping[str, Any]) → sessionmaker

Create the session factory used by create_scoped_session().

The factory must return an object that SQLAlchemy recognizes as a session, or registering session events may raise an exception.

Valid factories include a Session class or a sessionmaker.

The default implementation creates a sessionmaker for SignallingSession.

Parameters:

options – dict of keyword arguments passed to session class