Qual è il modo migliore per filtrare l'accesso utente per le query sqlalchemy?

Question

Ho guardato le ricette di sqlalchemy sul loro wiki, ma non so quale sia il migliore per implementare quello che sto cercando di fare.

Ogni riga nelle mie tabelle ha un ID utente associato. In questo momento, per ogni query, ho interrogato l'id dell'utente che ha attualmente effettuato l'accesso, quindi eseguo la ricerca in base ai criteri a cui sono interessato. La mia preoccupazione è che gli sviluppatori possano dimenticare di aggiungere questo filtro alla query (un enorme rischio per la sicurezza). Pertanto, vorrei impostare un filtro globale basato sui diritti di amministratore dell'utente corrente per filtrare ciò che l'utente loggato poteva vedere.

Apprezzo il tuo aiuto. Grazie.

Answer 1

Di seguito è stato semplificato il costruttore di query ridefinito per filtrare tutte le query del modello (incluse le relazioni). È possibile passarlo come parametro query_cls a sessionmaker. Non è necessario che il parametro ID utente sia globale per quanto riguarda la sessione quando è già disponibile.

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 

Funziona solo per le query con modello singolo, e c'è molto lavoro per renderlo adatto ad altri casi. Mi piacerebbe vedere una versione elaborata poiché DEVE AVERE funzionalità per la maggior parte delle applicazioni web. Utilizza condizioni fisse memorizzate in ogni classe di modello, quindi è necessario modificarlo in base alle proprie esigenze.

Answer 2

Questa è un'implementazione molto ingenua che presuppone l'esistenza di un attributo/proprietà self.current_user che l'utente ha registrato.

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) 

Answer 3

ho scritto un'estensione SQLAlchemy che penso fa quello che si sta descrivendo: https://github.com/mwhite/multialchemy

Lo fa con il proxy modifiche alle proprietà Query._from_obj e QueryContext._froms, che è dove i tavoli tra cui scegliere in ultima analisi, preparatevi .