cubicweb: comparison server/querier.py

equal deleted inserted replaced

-:643b19d79e4a
+:79d34ba48612
 from rql.stmts import Union
 from rql.nodes import ETYPE_PYOBJ_MAP, etype_from_pyobj, Relation, Exists, Not
 from yams import BASE_TYPES
 from cubicweb import ValidationError, Unauthorized, UnknownEid
+from cubicweb.rqlrewrite import RQLRelationRewriter
 from cubicweb import Binary, server
 from cubicweb.rset import ResultSet
 from cubicweb.utils import QueryCache, RepeatList
 from cubicweb.server.rqlannotation import SQLGenAnnotator, set_qdata
 try:
 return solution[term.name]
 except AttributeError:
 return cnx.entity_metas(term.eval(args))['type']
-def check_read_access(cnx, rqlst, solution, args):
+def check_relations_read_access(cnx, select, args):
+"""Raise :exc:`Unauthorized` if the given user doesn't have credentials to
+read relations used in the givel syntaxt tree
+"""
+# use `term_etype` since we've to deal with rewritten constants here,
+# when used as an external source by another repository.
+# XXX what about local read security w/ those rewritten constants...
+# XXX constants can also happen in some queries generated by req.find()
+DBG = (server.DEBUG & server.DBG_SEC) and 'read' in server._SECURITY_CAPS
+schema = cnx.repo.schema
+user = cnx.user
+if select.where is not None:
+for rel in select.where.iget_nodes(Relation):
+for solution in select.solutions:
+# XXX has_text may have specific perm ?
+if rel.r_type in READ_ONLY_RTYPES:
+continue
+rschema = schema.rschema(rel.r_type)
+if rschema.final:
+eschema = schema.eschema(term_etype(cnx, rel.children[0],
+solution, args))
+rdef = eschema.rdef(rschema)
+else:
+rdef = rschema.rdef(term_etype(cnx, rel.children[0],
+solution, args),
+term_etype(cnx, rel.children[1].children[0],
+solution, args))
+if not user.matching_groups(rdef.get_groups('read')):
+if DBG:
+print ('check_read_access: %s %s does not match %s' %
+(rdef, user.groups, rdef.get_groups('read')))
+# XXX rqlexpr not allowed
+raise Unauthorized('read', rel.r_type)
+if DBG:
+print ('check_read_access: %s %s matches %s' %
+(rdef, user.groups, rdef.get_groups('read')))
+def get_local_checks(cnx, rqlst, solution):
 """Check that the given user has credentials to access data read by the
 query and return a dict defining necessary "local checks" (i.e. rql
 expression in read permission defined in the schema) where no group grants
 him the permission.
 Returned dictionary's keys are variable names and values the rql expressions
 for this variable (with the given solution).
+Raise :exc:`Unauthorized` if access is known to be defined, i.e. if there is
+no matching group and no local permissions.
 """
-# use `term_etype` since we've to deal with rewritten constants here,
-# when used as an external source by another repository.
-# XXX what about local read security w/ those rewritten constants...
 DBG = (server.DEBUG & server.DBG_SEC) and 'read' in server._SECURITY_CAPS
 schema = cnx.repo.schema
-if rqlst.where is not None:
+user = cnx.user
-for rel in rqlst.where.iget_nodes(Relation):
-# XXX has_text may have specific perm ?
-if rel.r_type in READ_ONLY_RTYPES:
-continue
-rschema = schema.rschema(rel.r_type)
-if rschema.final:
-eschema = schema.eschema(term_etype(cnx, rel.children[0],
-solution, args))
-rdef = eschema.rdef(rschema)
-else:
-rdef = rschema.rdef(term_etype(cnx, rel.children[0],
-solution, args),
-term_etype(cnx, rel.children[1].children[0],
-solution, args))
-if not cnx.user.matching_groups(rdef.get_groups('read')):
-if DBG:
-print ('check_read_access: %s %s does not match %s' %
-(rdef, cnx.user.groups, rdef.get_groups('read')))
-# XXX rqlexpr not allowed
-raise Unauthorized('read', rel.r_type)
-if DBG:
-print ('check_read_access: %s %s matches %s' %
-(rdef, cnx.user.groups, rdef.get_groups('read')))
 localchecks = {}
 # iterate on defined_vars and not on solutions to ignore column aliases
 for varname in rqlst.defined_vars:
 eschema = schema.eschema(solution[varname])
 if eschema.final:
 continue
-if not cnx.user.matching_groups(eschema.get_groups('read')):
+if not user.matching_groups(eschema.get_groups('read')):
 erqlexprs = eschema.get_rqlexprs('read')
 if not erqlexprs:
 ex = Unauthorized('read', solution[varname])
 ex.var = varname
 if DBG:
 print ('check_read_access: %s %s %s %s' %
-(varname, eschema, cnx.user.groups, eschema.get_groups('read')))
+(varname, eschema, user.groups, eschema.get_groups('read')))
 raise ex
 # don't insert security on variable only referenced by 'NOT X relation Y' or
 # 'NOT EXISTS(X relation Y)'
 varinfo = rqlst.defined_vars[varname].stinfo
 if varinfo['selected'] or (
 len([r for r in varinfo['relations']
 if (not schema.rschema(r.r_type).final
 and ((isinstance(r.parent, Exists) and r.parent.neged(strict=True))
 or isinstance(r.parent, Not)))])
-!= len(varinfo['relations'])):
+!=
+len(varinfo['relations'])):
 localchecks[varname] = erqlexprs
 return localchecks
 # Plans #######################################################################
 localchecks = {}
 restricted_vars = set()
 newsolutions = []
 for solution in rqlst.solutions:
 try:
-localcheck = check_read_access(cnx, rqlst, solution, self.args)
+localcheck = get_local_checks(cnx, rqlst, solution)
 except Unauthorized as ex:
 msg = 'remove %s from solutions since %s has no %s access to %s'
 msg %= (solution, cnx.user.login, ex.args[0], ex.args[1])
 msgs.append(msg)
 LOGGER.info(msg)
 cachekey = None
 else:
 if cnx.read_security:
 for select in rqlst.children:
 check_no_password_selected(select)
+check_relations_read_access(cnx, select, args)
 # on select query, always copy the cached rqlst so we don't have to
 # bother modifying it. This is not necessary on write queries since
 # a new syntax tree is built from them.
 rqlst = rqlst.copy()
 self._annotate(rqlst)

changeset 9954	79d34ba48612
parent 9892	928732ec00dd
child 9955	60a9cd1b3a4b