add permission

seatable_thumbnail/constants.py

@@ -11,6 +11,11 @@
 JWT_ISSUER = None
 
 
+# permission
+PERMISSION_READ = 'r'
+PERMISSION_READ_WRITE = 'rw'
+
+
 # file type
 EMPTY_BYTES = b''
 THUMBNAIL_EXTENSION = 'png'

seatable_thumbnail/models.py

@@ -70,3 +70,17 @@ class DTableViewGroupShare(Base):
     permission = Column(String(15))
     table_id = Column(String(36), index=True)
     view_id = Column(String(36), index=True)
+
+
+class DTableExternalLinks(Base):
+    __tablename__ = 'dtable_external_link'
+    id = Column(BigInteger, primary_key=True)
+    dtable_id = Column(Integer, ForeignKey('dtables.id'))
+    creator = Column(String(255))
+    token = Column(String(100), unique=True)
+    permission = Column(String(50))
+    view_cnt = Column(Integer)
+    create_at = Column(DateTime)
+    is_custom = Column(Boolean)
+    password = Column(String(128), nullable=True)
+    expire_date = Column(DateTime)

seatable_thumbnail/permissions.py

@@ -1,7 +1,9 @@
 from seaserv import ccnet_api
 from seatable_thumbnail import session
-from seatable_thumbnail.models import Workspaces, DTables, DTableShare, \
-    DTableGroupShare, DTableViewUserShare, DTableViewGroupShare
+from seatable_thumbnail.models import DTables, DTableShare, \
+    DTableGroupShare, DTableViewUserShare, DTableViewGroupShare, \
+    DTableExternalLinks
+from seatable_thumbnail.constants import PERMISSION_READ, PERMISSION_READ_WRITE
 
 
 class ThumbnailPermission(object):
@@ -10,105 +12,139 @@ def __init__(self, **info):
         self.check()
 
     def check(self):
-        pass
+        self.dtable = session.query(
+            DTables).filter_by(uuid=self.dtable_uuid).first()
 
+        if not self.has_dtable_asset_read_permission():
+            raise AssertionError(403, 'Permission denied.')
 
-def has_dtable_asset_read_permission(request, workspace, dtable, asset_name):
-    # three ways to access asset
-    # 1. through external link to get image
-    # 2. through dtable perm, including dtable share
-    # 3. through view share perm
-    username = request.user.username
+    def has_dtable_asset_read_permission(self):
+        # three ways to access asset
+        # 1. through external link to get image
+        # 2. through dtable perm, including dtable share
+        # 3. through view share perm
 
-    if can_access_image_through_external_link(request, dtable, asset_name):
-        return True
-    if check_dtable_permission(username, workspace, dtable):
-        return True
-    if get_view_share_permision(username, dtable) in [PERMISSION_READ, PERMISSION_READ_WRITE]:
-        return True
-    return False
-
-
-def can_access_image_through_external_link(request, dtable, asset_name):
-    if not is_image_asset_type(asset_name):
-        return False
+        if not hasattr(self, 'username'):
+            if self.can_access_image_through_external_link():
+                return True
+        else:
+            if 'r' in self.check_dtable_permission():
+                return True
+            if 'r' in self.get_view_share_permission():
+                return True
 
-    external_link = request.session.get('external_link')
-    if not external_link:
         return False
 
-    return external_link['dtable_uuid'] == dtable.uuid.hex
-
-
-def get_view_share_permision(username, dtable):
-    """
-        return 'r' or 'rw' or ''
-    """
-    user_view_share_perm = get_user_view_share_permission(username, dtable)
-    if user_view_share_perm:
-        return user_view_share_perm
-    group_view_share_perm = get_group_view_share_permission(username, dtable)
-    return group_view_share_perm
-
-
-def check_dtable_permission(username, workspace, dtable=None, org_id=None):
-    """Check workspace/dtable access permission of a user.
-    """
-    owner = workspace.owner
-
-    if '@seafile_group' in owner:
-        group_id = int(owner.split('@')[0])
-        if is_group_member(group_id, username):
-            return PERMISSION_READ_WRITE
-    else:
-        if username == owner:
-            return PERMISSION_READ_WRITE
-
-    if dtable:  # check user's all permissions from `share`, `group-share` and checkout higher one
-        dtable_share = DTableShare.objects.get_by_dtable_and_to_user(
-            dtable, username)
-        if dtable_share and dtable_share.permission == PERMISSION_READ_WRITE:
-            return dtable_share.permission
-        permission = dtable_share.permission if dtable_share else None
-
-        if org_id and org_id > 0:
-            groups = ccnet_api.get_org_groups_by_user(org_id, username)
+    def can_access_image_through_external_link(self):
+        external_link = session.query(
+            DTableExternalLinks).filter_by(token=self.external_link_token).first()
+        if not external_link:
+            return False
+
+        external_link_dtable = session.query(
+            DTables).filter_by(id=external_link.dtable_id).first()
+        if not external_link_dtable:
+            return False
+
+        return external_link_dtable.uuid == self.dtable_uuid
+
+    def check_dtable_permission(self):
+        """Check workspace/dtable access permission of a user.
+        """
+        owner = self.workspace_owner
+        username = self.username
+        dtable = self.dtable
+        if not hasattr(self, 'org_id'):
+            self.org_id = -1
+        org_id = self.org_id
+
+        if '@seafile_group' in owner:
+            group_id = int(owner.split('@')[0])
+            if self.is_group_member(group_id, username):
+                return PERMISSION_READ_WRITE
         else:
-            groups = ccnet_api.get_groups(username, return_ancestors=True)
-        group_ids = [group.id for group in groups]
-        group_permissions = DTableGroupShare.objects.filter(
-            dtable=dtable, group_id__in=group_ids).values_list('permission', flat=True)
-        for group_permission in group_permissions:
-            permission = permission if permission else group_permission
-            if group_permission == PERMISSION_READ_WRITE:
-                return group_permission
-        return permission
-
-    return None
-
-
-def get_user_view_share_permission(username, dtable):
-    # if multiple view share is in same dtable, get highest
-    # 'rw' is lower than 'r' in lexical order, reverse to get 'rw' first
-    view_share = DTableViewUserShare.objects.filter(
-        to_user=username, dtable=dtable).order_by('-permission').first()
-    if not view_share:
-        return ''
-    return view_share.permission
+            if username == owner:
+                return PERMISSION_READ_WRITE
+
+        if dtable:  # check user's all permissions from `share`, `group-share` and checkout higher one
+            dtable_share = session.query(
+                DTableShare).filter_by(dtable_id=dtable.id, to_user=username).first()
+            if dtable_share and dtable_share.permission == PERMISSION_READ_WRITE:
+                return dtable_share.permission
+            permission = dtable_share.permission if dtable_share else ''
+
+            if org_id and org_id > 0:
+                groups = ccnet_api.get_org_groups_by_user(org_id, username)
+            else:
+                groups = ccnet_api.get_groups(username, return_ancestors=True)
+            group_ids = [group.id for group in groups]
+            group_permissions = session.query(
+                DTableGroupShare.permission).filter(DTableGroupShare.dtable_id==dtable.id, DTableGroupShare.group_id.in_(group_ids)).all()
+
+            for group_permission in group_permissions:
+                permission = permission if permission else group_permission[0]
+                if group_permission[0] == PERMISSION_READ_WRITE:
+                    return group_permission[0]
+            return permission
 
-
-def get_group_view_share_permission(username, dtable):
-    # if multiple view share is in same dtable, get highest
-    # 'rw' is lower than 'r' in lexical order, reverse to get 'rw' first
-    view_shares = DTableViewGroupShare.objects.filter(
-        dtable=dtable).order_by('-permission')
-
-    target_view_share = None
-    for view_share in view_shares:
-        if is_group_member(view_share.to_group_id, username):
-            target_view_share = view_share
-            break
-
-    if not target_view_share:
         return ''
-    return target_view_share.permission
+
+    def get_view_share_permission(self):
+        """return 'r' or 'rw' or ''
+        """
+        user_view_share_perm = self.get_user_view_share_permission()
+        if user_view_share_perm:
+            return user_view_share_perm
+        group_view_share_perm = self.get_group_view_share_permission()
+        return group_view_share_perm
+
+    def get_user_view_share_permission(self):
+        # if multiple view share is in same dtable, get highest
+        # 'rw' is lower than 'r' in lexical order, reverse to get 'rw' first
+        username = self.username
+        dtable = self.dtable
+
+        view_share = session.query(
+            DTableViewUserShare).filter_by(dtable_id=dtable.id, to_user=username).order_by(DTableViewUserShare.permission.desc()).first()
+        if not view_share:
+            return ''
+        return view_share.permission
+
+    def get_group_view_share_permission(self):
+        # if multiple view share is in same dtable, get highest
+        # 'rw' is lower than 'r' in lexical order, reverse to get 'rw' first
+        username = self.username
+        dtable = self.dtable
+
+        view_shares = session.query(
+            DTableViewGroupShare).filter_by(dtable_id=dtable.id).order_by(DTableViewGroupShare.permission.desc()).all()
+
+        target_view_share = None
+        for view_share in view_shares:
+            if self.is_group_member(view_share.to_group_id, username):
+                target_view_share = view_share
+                break
+
+        if not target_view_share:
+            return ''
+        return target_view_share.permission
+
+    def is_group_member(self, group_id, email, in_structure=None):
+
+        group_id = int(group_id)
+
+        if in_structure in (True, False):
+            return ccnet_api.is_group_user(group_id, email, in_structure)
+
+        group = ccnet_api.get_group(group_id)
+        if not group:
+            return False
+
+        if group.parent_group_id == 0:
+            # -1: top address book group
+            #  0: group not in address book
+            # >0: sub group in address book
+            # if `in_structure` is False, NOT check sub groups in address book
+            return ccnet_api.is_group_user(group_id, email, in_structure=False)
+        else:
+            return ccnet_api.is_group_user(group_id, email)

seatable_thumbnail/serializers.py

@@ -50,10 +50,6 @@ def jwt_check(self):
         jwt_token = self.request.query_dict['token'][0]
         self.payload = self.jwt_decode_handler(jwt_token)
 
-        # permission check
-        if 'r' not in self.payload['permission']:
-            raise AssertionError(403, 'permission denied.')
-
     def get_enable_file_type(self):
         enable_file_type = [IMAGE]
         if settings.ENABLE_PSD_THUMBNAIL: