What is the best way pre filter user access for sqlalchemy queries?

Question

I have been looking at the sqlalchemy recipes on their wiki, but don't know which one is best to implement what I am trying to do.

Every row on in my tables have an user_id associated with it. Right now, for every query, I queried by the id of the user that's currently logged in, then query by the criteria I am interested in. My concern is that the developers might forget to add this filter to the query (a huge security risk). Therefore, I would like to set a global filter based on the current user's admin rights to filter what the logged in user could see.

Appreciate your help. Thanks.

Agenesis

Have you thought about using the database security system to achieve this? For example, Oracle offers this functionality with its VPD feature (oracle.com/technology/deploy/security/database-security/…). You can achieve similar functionality in other databases, too (e.g. ddj.com/database/215900773 and technet.microsoft.com/en-gb/library/cc966395.aspx). Just google for "row level security RDBSName". – Roldan

or set up views or rules on all the tables that add the appropriate limitation, and strip users of the ability to directly read the underlying table. postgresql.org/docs/8.4/interactive/rules.html – Proponent

Answer 1

Below is simplified redefined query constructor to filter all model queries (including relations). You can pass it to as query_cls parameter to sessionmaker. User ID parameter don't need to be global as far as session is constructed when it's already available.

class HackedQuery(Query):

    def get(self, ident):
        # Use default implementation when there is no condition
        if not self._criterion:
            return Query.get(self, ident)
        # Copied from Query implementation with some changes.
        if hasattr(ident, '__composite_values__'):
            ident = ident.__composite_values__()
        mapper = self._only_mapper_zero(
                    "get() can only be used against a single mapped class.")
        key = mapper.identity_key_from_primary_key(ident)
        if ident is None:
            if key is not None:
                ident = key[1]
        else:
            from sqlalchemy import util
            ident = util.to_list(ident)
        if ident is not None:
            columns = list(mapper.primary_key)
            if len(columns)!=len(ident):
                raise TypeError("Number of values doen't match number "
                                'of columns in primary key')
            params = {}
            for column, value in zip(columns, ident):
                params[column.key] = value
            return self.filter_by(**params).first()


def QueryPublic(entities, session=None):
    # It's not directly related to the problem, but is useful too.
    query = HackedQuery(entities, session).with_polymorphic('*')
    # Version for several entities needs thorough testing, so we 
    # don't use it yet.
    assert len(entities)==1, entities
    cls = _class_to_mapper(entities[0]).class_
    public_condition = getattr(cls, 'public_condition', None)
    if public_condition is not None:
        query = query.filter(public_condition)
    return query

It works for single model queries only, and there is a lot of work to make it suitable for other cases. I'd like to see an elaborated version since it's MUST HAVE functionality for most web applications. It uses fixed condition stored in each model class, so you have to modify it to your needs.

Seaworthy

So i ended up following your answer with a combination of sqlalchemy.org/trac/wiki/UsageRecipes/PreFilteredQuery. Is there a way to specify the query_cls at the model class level as suppose to the sessionmaker? Not all models have this column that I want to filter by. – Agenesis

Ooh, I did query = Session.query_property(FilterByUserQuery) in the model. Many thanks. – Agenesis

Answer 2

Here is a very naive implementation that assumes there is the attribute/property self.current_user logged in user has stored.

class YourBaseRequestHandler(object):

    @property
    def current_user(self):
        """The current user logged in."""
        pass

    def query(self, session, entities):
        """Use this method instead of :method:`Session.query()
        <sqlalchemy.orm.session.Session.query>`.

        """
        return session.query(entities).filter_by(user_id=self.current_user.id)

Faina

Answer 3

I wrote an SQLAlchemy extension that I think does what you are describing: https://github.com/mwhite/multialchemy

It does this by proxying changes to the Query._from_obj and QueryContext._froms properties, which is where the tables to select from ultimately get set.

Adalia