Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

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

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.

like image 636
steve Avatar asked May 21 '10 20:05

steve


People also ask

How does the querying work with SQLAlchemy?

Python Flask and SQLAlchemy ORM All SELECT statements generated by SQLAlchemy ORM are constructed by Query object. It provides a generative interface, hence successive calls return a new Query object, a copy of the former with additional criteria and options associated with it.

Should I use SQLAlchemy core or ORM?

If you want to view your data in a more schema-centric view (as used in SQL), use Core. If you have data for which business objects are not needed, use Core. If you view your data as business objects, use ORM. If you are building a quick prototype, use ORM.

What does First () do in SQLAlchemy?

first() applies a limit of one within the generated SQL, so that only one primary entity row is generated on the server side (note this may consist of multiple result rows if join-loaded collections are present). Calling Query. first() results in an execution of the underlying query.


2 Answers

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.

like image 55
Denis Otkidach Avatar answered Oct 13 '22 21:10

Denis Otkidach


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)
like image 34
minhee Avatar answered Oct 13 '22 19:10

minhee