[#561] implementation: get_maximum_permissions_for_path

irods/access.py

@@ -1,5 +1,6 @@
 import collections
 import copy
+import enum
 import warnings
 
 from irods.collection import iRODSCollection
@@ -20,7 +21,6 @@
     "null",
 )
 
-
 class _Access_LookupMeta(type):
     @staticmethod
     def _codes():

irods/manager/access_manager.py

@@ -1,13 +1,14 @@
 """The access manager is a collection of methods useful for managing iRODS ACLs."""
 
 import logging
 from os.path import basename, dirname
 
 from irods.access import iRODSAccess
 from irods.api_number import api_number
 from irods.collection import iRODSCollection
 from irods.column import In
 from irods.data_object import irods_basename, irods_dirname, iRODSDataObject
+from irods.exception import CAT_NO_ROWS_FOUND
 from irods.manager import Manager
 from irods.message import JSON_Message, ModAclRequest, iRODSMessage
 from irods.models import (
@@ -16,9 +17,12 @@
     CollectionUser,
     DataAccess,
     DataObject,
+    Group,
     User,
 )
-from irods.user import iRODSUser
+from irods.query import SpecificQuery
+from irods.user import iRODSUser, iRODSGroup
+
 
 logger = logging.getLogger(__name__)
 
@@ -35,7 +39,133 @@
     return [iRODSUser(session.users, i) for i in session.query(User.id, User.name, User.type, User.zone).filter(*cond)]
 
 
+class QueryParameterError(ValueError):
+    pass
+
+# This is the alias where we expect to find a specific query
+alias_for_get_maximum_permission_for_path = 'get_maximum_permission_for_path'
+
 class AccessManager(Manager):
+    def get_maximum_permission_for_path(self,
+        user : [iRODSUser|iRODSGroup],
+        coll : [str|iRODSCollection|None]=None,
+        data : [str|iRODSDataObject|None]=None,
+        specific_query : [SpecificQuery|str|None]='',
+        return_type : type[int]|type[str]=int,
+    ) -> int|str:
+        """
+        Return the highest permission the named user/group has with respect to the
+        given data object or collection.  In calculating this result, all ACLs and
+        group memberships are factored in.
+
+        Only one of 'data' or 'coll' may be specified.
+
+        Args:
+            user: an iRODSUser or iRODSGroup object for whom the permission is to be calculated.
+            coll: an iRODS logical path or iRODSCollection.
+            data: an iRODS logical path or iRODSDataObject.
+            specific_query (default ""): a SpecificQuery object or an alias specifying 
+                the specific query used for the calculation.  The default value of "" will be
+                interpreted to mean the designated specific query preconfigured for this express
+                purpose.  If the server does not cache a query under that alias, or if None is
+                given as a value, then the calculation will be done via a general query.
+            return_type: the type expected for the return value, i.e. str or int.  Defaults to int.
+        """
+        ses = self.sess
+
+        # Convert user or group to object.  Need to know if we are dealing with a rodsadmin
+        if isinstance(user, (iRODSUser, iRODSGroup)):
+            pass
+        else:
+            msg = "'user' argument should be of type iRODSUser or iRODSGroup."
+            raise QueryParameterError(msg)
+
+        _specific_query = None
+
+        if specific_query == '':
+            specific_query = alias_for_get_maximum_permission_for_path
+
+        if isinstance(specific_query, SpecificQuery):
+            _specific_query = specific_query
+        elif isinstance(specific_query, str):
+            _specific_query = SpecificQuery(ses, alias=specific_query)
+
+        # Dispatch to the available/chosen query method.
+        if _specific_query is not None:
+            def denullify(s):
+                return s if s else "x"
+            _specific_query.reset_for_new_query(
+                args=[
+                    user.name,
+                    getattr(user,'zone',ses.zone),
+                    denullify(coll.path if isinstance(coll, iRODSCollection) else coll),
+                    denullify(data.path if isinstance(data, iRODSDataObject) else data),
+                ]
+            )
+            max_permission = self._get_maximum_permission_for_path_via_SpecificQuery(_specific_query)
+            res = [max_permission]
+        else:
+            res = None
+            groups = {g[Group.id] for g in ses.query(Group, User).filter(Group.id != user.id, User.id == user.id)}
+            groups |= {user.id}
+
+            if data:
+                if isinstance(data, str):
+                    collN = irods_dirname(data)
+                    dataN = irods_basename(data)
+                    join_criteria = [DataObject.name == dataN, Collection.name == collN]
+                elif isinstance(data, iRODSDataObject):
+                    join_criteria = [ DataAccess.data_id == data.id]
+                else:
+                    msg="'data' parameter must be full logical path or data object"
+                    raise ValueError(msg)
+
+                q = ses.query(DataAccess.type).filter(
+                    In(DataAccess.user_id, groups),
+                    *join_criteria
+                ).order_by(DataAccess.type,order='desc')
+
+                res = [_[DataAccess.type] for _ in q]
+
+            elif coll:
+                if isinstance(coll, str):
+                    join_criteria = [Collection.name == coll]
+                elif isinstance(coll, iRODSCollection):
+                    join_criteria = [CollectionAccess.access_id == coll.id] # TODO: access_id is not an accurate name, should be coll_id
+                else:
+                    msg="'coll' parameter must be full logical path or collection object"
+                    raise ValueError(msg)
+
+                q=ses.query(CollectionAccess).filter(
+                        In(CollectionAccess.user_id, groups),
+                        *join_criteria
+                    ).order_by(CollectionAccess.type,order='desc')
+
+                res = [_[CollectionAccess.type] for _ in q]
+
+            else:
+                msg = 'Need a collection or data object'
+                raise QueryParameterError(msg)
+
+        retval_transform = (
+            (lambda _: iRODSAccess.strings[_]) if return_type is str else (lambda _:_)
+        )
+        res = res[0] if res else None
+        return retval_transform(res) if isinstance(res, int) else None
+
+
+    @staticmethod
+    def _get_maximum_permission_for_path_via_SpecificQuery(spec_query):
+        try:
+            row = next(iter(
+                spec_query.execute()
+            ))
+        except StopIteration:
+            row = {}
+
+        highest_perm = row.get(0)
+        return None if highest_perm is None else int(highest_perm)
+
     @staticmethod
     def _to_acl_operation_json(op_input: iRODSAccess):
         return {

irods/query.py

@@ -1,4 +1,6 @@
 from collections import OrderedDict
+import copy
+import uuid
 
 from irods import MAX_SQL_ROWS
 from irods.models import Model
@@ -290,6 +292,11 @@
         self._columns = columns
         self._args = args or []
 
+    def reset_for_new_query(self, args=None):
+        self._continue_index = 0
+        if args is not None:
+            self._args = args
+
     def register(self):
         if not self._sql:
             raise ValueError("Empty query")
@@ -313,7 +320,10 @@
             response = conn.recv()
         return response
 
-    def execute(self, limit=MAX_SQL_ROWS, offset=0, options=0, conditions=None):
+    def execute(self, limit=MAX_SQL_ROWS, offset=0, options=0, conditions=None, instance=None):
+        if instance is not None:
+            self = instance
+
         target = self._alias or self._sql
 
         if conditions is None:
@@ -363,7 +373,9 @@
                 yield result
 
 
+
+
 # Record a copy of the original value of IRODS_QUERY_LIMIT for possible future access.
 import irods.helpers as helpers
 
 cached_values = helpers.create_value_cache(globals())

irods/test/access_test.py

@@ -1,9 +1,11 @@
 #! /usr/bin/env python
-
+ 
 import os
 import sys
+import textwrap
 import unittest
 
+import irods.access
 from irods.access import ACLOperation, iRODSAccess
 from irods.collection import iRODSCollection
 from irods.column import In, Like
@@ -14,6 +16,50 @@
 from irods.test import helpers
 from irods.user import iRODSUser
 
+from irods.manager.access_manager import alias_for_get_maximum_permission_for_path
+
+SQL = textwrap.dedent('''
+    WITH target_user AS (
+        SELECT user_id
+        FROM r_user_main
+        WHERE user_name = ?
+          AND zone_name = ?
+    ),
+    resolved_object AS (
+        -- The path names a collection
+        SELECT coll_id AS object_id
+        FROM r_coll_main
+        WHERE coll_name = ?
+
+        UNION
+
+        -- The path names a data object: collection path + '/' + data name
+        SELECT d.data_id AS object_id
+        FROM r_data_main AS d
+        JOIN r_coll_main AS c ON c.coll_id = d.coll_id
+        WHERE CONCAT(c.coll_name, '/', d.data_name) = ?
+    ),
+    grants AS (
+        -- Direct grants
+        SELECT a.access_type_id
+        FROM r_objt_access AS a
+        JOIN target_user AS u ON u.user_id = a.user_id
+        JOIN resolved_object AS o ON o.object_id = a.object_id
+
+        UNION ALL
+
+        -- Grants to groups the user belongs to
+        SELECT a.access_type_id
+        FROM r_user_group AS ug
+        JOIN target_user AS u ON u.user_id = ug.user_id
+        JOIN r_objt_access AS a ON a.user_id = ug.group_user_id
+        JOIN resolved_object AS o ON o.object_id = a.object_id
+    )
+    SELECT access_type_id
+    FROM grants
+    ORDER BY access_type_id DESC
+    LIMIT 1;'''
+)
 
 class TestAccess(unittest.TestCase):
     def setUp(self):
@@ -532,6 +578,72 @@
             if zone:
                 zone.remove()
 
+    def test_get_maximum_permission_for_path__issue_561(self):
+        user = group = coll = None
+        ses = self.sess
+        try:
+            coll = ses.collections.create('/tempZone/home/public/folder_561')
+            data = ses.data_objects.create('/tempZone/home/public/folder_561/file')
+            user = ses.users.create('bob','rodsuser')
+            group = ses.groups.create('lab')
+
+            # Set up test scenario with a new user and group, the group having deliberately higher permission
+            # so as to be able to account for the effect of group membership in the query results.
+            USER_PERM = 'modify_metadata'
+            GROUP_PERM = 'modify_object'
+            test_perms = { user.name: USER_PERM, group.name: GROUP_PERM }
+            for obj in (coll, data):
+                ses.acls.set(iRODSAccess(USER_PERM, obj.path, user.name))
+                ses.acls.set(iRODSAccess(GROUP_PERM, obj.path, group.name))
+
+            options = {}
+            try:
+                SpecificQuery(ses, alias='findQueryByAlias', args=(alias_for_get_maximum_permission_for_path,)).execute()
+            except CAT_NO_ROWS_FOUND:
+                options = dict(
+                    query_alias = str(uuid.uuid1()),
+                    sql = SQL
+                )
+
+            specific_query = SpecificQuery(ses, alias=query_alias, **options)
+
+#   def test_get_maximum_permission_for_path__issue_561(self):
+
+            for is_member in (False, True):
+                if is_member:
+                    group.addmember(user.name,user.zone)
+                for usr in (user, group):
+                    max_user = usr if usr not in group.members else group
+                    expected_permission = test_perms[max_user.name]
+                    for obj_type, obj in (
+                        ("coll", coll),
+                        ("coll", coll.path),
+                        ("data", data),
+                        ("data", data.path),
+                    ):
+                        for s_query in (None, specific_query):
+                            with self.subTest(f"{is_member=} testing {usr=} {obj=} {s_query=}"):
+                                force_genquery_option = {'general': True, 'specific': False}[query_type]
+                                feedback = []
+                                permission = ses.acls.get_maximum_permission_for_path(
+                                    usr,
+                                    return_type=irods.access.return_type.STRING,
+                                    specific_query=s_query,
+                                    **{obj_type: obj},
+                                )
+                                self.assertEqual(permission, expected_permission)
+                                self.assertEqual(feedback[0], query_type)
+        finally:
+            if coll:
+                # Removes test data object as well.
+                coll.remove(force=True, recurse=True)
+            if user:
+                user.remove()
+            if group:
+                group.remove()
+            if query_alias:
+                specific_query.remove()
+
 
 if __name__ == "__main__":
     # let the tests find the parent irods lib

irods/user.py

@@ -17,6 +17,10 @@ def remove_quota(self, resource="total"):
     def set_quota(self, amount, resource="total"):
         self.manager.set_quota(self.name, amount, resource=resource)
 
+    # TODO: add issue for iRODSUser & iRODSGroup equality operators
+    def __eq__(self, other_user):
+        return isinstance(other_user, iRODSUser) and (self.name, self.zone) == (other_user.name, other_user.zone)
+
     def __init__(self, manager, result=None):
         self.manager = manager
         if result:
@@ -82,6 +86,9 @@ def remove_quota(self, resource="total"):
     def set_quota(self, amount, resource="total"):
         self.manager.set_quota(self.name, amount, resource=resource)
 
+    def __eq__(self, other_group):
+        return self.name == other_group.name
+
     def __init__(self, manager, result=None):
         self.manager = manager
         if result: