sysadmin reconstruct links backend api (#4149)

2025-09-02 07:27:04 +00:00 · 2019-10-15 18:21:09 +08:00
parent 63c65c1615
commit 790d9305d5
6 changed files with 245 additions and 6 deletions
--- a/seahub/api2/endpoints/admin/share_links.py
+++ b/seahub/api2/endpoints/admin/share_links.py
@@ -28,6 +28,7 @@ from seahub.utils import gen_file_get_url, gen_dir_zip_download_url, \
 from seahub.utils.timeutils import timestamp_to_isoformat_timestr, \
        datetime_to_isoformat_timestr
 from seahub.views.file import send_file_access_msg
+from seahub.wiki.models import Wiki

 logger = logging.getLogger(__name__)

@@ -86,6 +87,53 @@ def get_share_link_info(fileshare):
    return data


+class AdminShareLinks(APIView):
+    authentication_classes = (TokenAuthentication, SessionAuthentication)
+    permission_classes = (IsAdminUser,)
+    throttle_classes = (UserRateThrottle,)
+
+    def get(self, request):
+        """ Get all share links.
+
+        Permission checking:
+        1. only admin can perform this action.
+        """
+        try:
+            current_page = int(request.GET.get('page', '1'))
+            per_page = int(request.GET.get('per_page', '100'))
+        except ValueError:
+            current_page = 1
+            per_page = 100
+
+        start = (current_page - 1) * per_page
+        end = start + per_page
+
+        share_links = FileShare.objects.all().order_by('ctime')[start:end]
+        count = FileShare.objects.all().count()
+
+        # Use dict to reduce memcache fetch cost in large for-loop.
+        nickname_dict = {}
+        owner_email_set = set([link.username for link in share_links])
+        for e in owner_email_set:
+            if e not in nickname_dict:
+                nickname_dict[e] = email2nickname(e)
+
+        share_links_info = []
+        for link in share_links:
+            link_info = {}
+            link_info['obj_name'] = link.get_obj_name()
+            link_info['token'] = link.token
+
+            owner_email = link.username
+            link_info['creator_email'] = owner_email
+            link_info['creator_name'] = nickname_dict.get(owner_email, '')
+            link_info['ctime'] = datetime_to_isoformat_timestr(link.ctime)
+            link_info['view_cnt'] = link.view_cnt
+            share_links_info.append(link_info)
+
+        return Response({"share_link_list": share_links_info, "count": count})
+
+
 class AdminShareLink(APIView):

    authentication_classes = (TokenAuthentication, SessionAuthentication)
@@ -115,12 +163,24 @@ class AdminShareLink(APIView):
        1. only admin can perform this action.
        """
        try:
-            fs = FileShare.objects.get(token=token)
+            share_link = FileShare.objects.get(token=token)
        except FileShare.DoesNotExist:
            return Response({'success': True})

+        has_published_library = False
+        if share_link.path == '/':
+            try:
+                Wiki.objects.get(repo_id=share_link.repo_id)
+                has_published_library = True
+            except Wiki.DoesNotExist:
+                pass
+
+        if has_published_library:
+            error_msg = 'There is an associated published library.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
        try:
-            fs.delete()
+            share_link.delete()
        except Exception as e:
            logger.error(e)
            error_msg = 'Internal Server Error'
--- a/seahub/api2/endpoints/admin/upload_links.py
+++ b/seahub/api2/endpoints/admin/upload_links.py
@@ -63,6 +63,52 @@ def get_upload_link_info(uls):

    return data

+class AdminUploadLinks(APIView):
+    authentication_classes = (TokenAuthentication, SessionAuthentication)
+    permission_classes = (IsAdminUser,)
+    throttle_classes = (UserRateThrottle,)
+
+    def get(self, request):
+        """ Get all upload links.
+
+        Permission checking:
+        1. only admin can perform this action.
+        """
+        try:
+            current_page = int(request.GET.get('page', '1'))
+            per_page = int(request.GET.get('per_page', '100'))
+        except ValueError:
+            current_page = 1
+            per_page = 100
+
+        start = (current_page - 1) * per_page
+        end = start + per_page
+
+        upload_links = UploadLinkShare.objects.all().order_by('ctime')[start:end]
+        count = UploadLinkShare.objects.all().count()
+
+        # Use dict to reduce memcache fetch cost in large for-loop.
+        nickname_dict = {}
+        owner_email_set = set([link.username for link in upload_links])
+        for e in owner_email_set:
+            if e not in nickname_dict:
+                nickname_dict[e] = email2nickname(e)
+
+        upload_links_info = []
+        for link in upload_links:
+            link_info = {}
+            link_info['path'] = link.path
+            link_info['token'] = link.token
+
+            owner_email = link.username
+            link_info['creator_email'] = owner_email
+            link_info['creator_name'] = nickname_dict.get(owner_email, '')
+            link_info['ctime'] = datetime_to_isoformat_timestr(link.ctime)
+            link_info['view_cnt'] = link.view_cnt
+            upload_links_info.append(link_info)
+
+        return Response({"upload_link_list": upload_links_info, "count": count})
+

 class AdminUploadLink(APIView):

@@ -93,12 +139,12 @@ class AdminUploadLink(APIView):
        1. only admin can perform this action.
        """
        try:
-            fs = UploadLinkShare.objects.get(token=token)
+            upload_link = UploadLinkShare.objects.get(token=token)
        except UploadLinkShare.DoesNotExist:
            return Response({'success': True})

        try:
-            fs.delete()
+            upload_link.delete()
        except Exception as e:
            logger.error(e)
            error_msg = 'Internal Server Error'
--- a/seahub/share/models.py
+++ b/seahub/share/models.py
@@ -2,6 +2,7 @@
 import operator
 import datetime
 import logging
+import os

 from django.db import models
 from django.db.models import Q
@@ -366,6 +367,11 @@ class FileShare(models.Model):
            assert False
        return perm_dict

+    def get_obj_name(self):
+        if self.path:
+            return '/' if self.path == '/' else os.path.basename(self.path.rstrip('/'))
+        return ''
+

 class OrgFileShareManager(models.Manager):
    def set_org_file_share(self, org_id, file_share):
--- a/seahub/urls.py
+++ b/seahub/urls.py
@@ -120,10 +120,10 @@ from seahub.api2.endpoints.admin.groups import AdminGroups, AdminGroup
 from seahub.api2.endpoints.admin.group_libraries import AdminGroupLibraries, AdminGroupLibrary
 from seahub.api2.endpoints.admin.group_members import AdminGroupMembers, AdminGroupMember
 from seahub.api2.endpoints.admin.shares import AdminShares
-from seahub.api2.endpoints.admin.share_links import AdminShareLink, \
+from seahub.api2.endpoints.admin.share_links import AdminShareLinks, AdminShareLink, \
        AdminShareLinkDownload, AdminShareLinkCheckPassword, \
        AdminShareLinkDirents
-from seahub.api2.endpoints.admin.upload_links import AdminUploadLink, \
+from seahub.api2.endpoints.admin.upload_links import AdminUploadLinks, AdminUploadLink, \
        AdminUploadLinkUpload, AdminUploadLinkCheckPassword
 from seahub.api2.endpoints.admin.users_batch import AdminUsersBatch, AdminAdminUsersBatch
 from seahub.api2.endpoints.admin.operation_logs import AdminOperationLogs
@@ -501,6 +501,7 @@ urlpatterns = [
    url(r'^api/v2.1/admin/admin-login-logs/$', AdminLoginLogs.as_view(), name='api-v2.1-admin-admin-login-logs'),

    ## admin::share-links
+    url(r'^api/v2.1/admin/share-links/$', AdminShareLinks.as_view(), name='api-v2.1-admin-share-links'),
    url(r'^api/v2.1/admin/share-links/(?P<token>[a-f0-9]+)/$', AdminShareLink.as_view(), name='api-v2.1-admin-share-link'),
    url(r'^api/v2.1/admin/share-links/(?P<token>[a-f0-9]+)/download/$',
            AdminShareLinkDownload.as_view(), name='api-v2.1-admin-share-link-download'),
@@ -510,6 +511,7 @@ urlpatterns = [
            AdminShareLinkDirents.as_view(), name='api-v2.1-admin-share-link-dirents'),

    ## admin::upload-links
+    url(r'^api/v2.1/admin/upload-links/$', AdminUploadLinks.as_view(), name='api-v2.1-admin-upload-links'),
    url(r'^api/v2.1/admin/upload-links/(?P<token>[a-f0-9]+)/$', AdminUploadLink.as_view(), name='api-v2.1-admin-upload-link'),
    url(r'^api/v2.1/admin/upload-links/(?P<token>[a-f0-9]+)/upload/$',
            AdminUploadLinkUpload.as_view(), name='api-v2.1-admin-upload-link-upload'),
--- a/tests/api/endpoints/admin/test_share_links.py
+++ b/tests/api/endpoints/admin/test_share_links.py
@@ -13,6 +13,54 @@ try:
 except ImportError:
    LOCAL_PRO_DEV_ENV = False

+class AdminShareLinksTest(BaseTestCase):
+
+    def setUp(self):
+        self.repo_id = self.repo.id
+        self.file_path= self.file
+        self.folder_path= self.folder
+        self.invalid_token = '00000000000000000000'
+
+    def tearDown(self):
+        self.remove_repo()
+
+    def _add_file_share_link(self, password=None):
+        fs = FileShare.objects.create_file_link(
+                self.user.username, self.repo.id, self.file, password, None)
+
+        return fs.token
+
+    def _add_dir_share_link(self, password=None):
+        fs = FileShare.objects.create_dir_link(
+                self.user.username, self.repo.id, self.folder, password, None)
+
+        return fs.token
+
+    def _remove_share_link(self, token):
+        link = FileShare.objects.get(token=token)
+        link.delete()
+
+    def test_get_share_links(self):
+        self.login_as(self.admin)
+        token1 = self._add_file_share_link()
+        token2 = self._add_dir_share_link()
+
+        url = reverse('api-v2.1-admin-share-links')
+        resp = self.client.get(url)
+        self.assertEqual(200, resp.status_code)
+
+        self._remove_share_link(token1)
+        self._remove_share_link(token2)
+
+    def test_get_share_links_with_invalid_permission(self):
+        self.login_as(self.user)
+        token = self._add_file_share_link()
+
+        url = reverse('api-v2.1-admin-share-links')
+        resp = self.client.get(url)
+        self.assertEqual(403, resp.status_code)
+
+        self._remove_share_link(token)

 class AdminShareLinkTest(BaseTestCase):

@@ -90,6 +138,24 @@ class AdminShareLinkTest(BaseTestCase):
        resp = self.client.get(url)
        self.assertEqual(404, resp.status_code)

+    def test_can_delete_share_link_by_token(self):
+        self.login_as(self.admin)
+        token = self._add_dir_share_link()
+
+        url = reverse('api-v2.1-admin-share-link', args=[token])
+        resp = self.client.delete(url)
+        self.assertEqual(200, resp.status_code)
+
+    def test_delete_share_link_with_invalid_permission(self):
+        self.login_as(self.user)
+        token = self._add_dir_share_link()
+
+        url = reverse('api-v2.1-admin-share-link', args=[token])
+        resp = self.client.delete(url)
+        self.assertEqual(403, resp.status_code)
+
+        self._remove_share_link(token)
+

 class AdminShareLinkDirentsTest(BaseTestCase):

--- a/tests/api/endpoints/admin/test_upload_links.py
+++ b/tests/api/endpoints/admin/test_upload_links.py
@@ -11,6 +11,47 @@ try:
 except ImportError:
    LOCAL_PRO_DEV_ENV = False

+
+class AdminUploadLinksTest(BaseTestCase):
+
+    def setUp(self):
+        self.repo_id = self.repo.id
+        self.folder_path= self.folder
+        self.invalid_token = '00000000000000000000'
+
+    def tearDown(self):
+        self.remove_repo()
+
+    def _add_upload_link(self, password=None):
+        fs = UploadLinkShare.objects.create_upload_link_share(
+                self.user.username, self.repo.id, self.folder_path, password, None)
+
+        return fs.token
+
+    def _remove_upload_link(self, token):
+        link = UploadLinkShare.objects.get(token=token)
+        link.delete()
+
+    def test_get_share_links(self):
+        self.login_as(self.admin)
+        token = self._add_upload_link()
+
+        url = reverse('api-v2.1-admin-upload-links')
+        resp = self.client.get(url)
+        self.assertEqual(200, resp.status_code)
+
+        self._remove_upload_link(token)
+
+    def test_get_share_links_with_invalid_permission(self):
+        self.login_as(self.user)
+        token = self._add_upload_link()
+
+        url = reverse('api-v2.1-admin-upload-links')
+        resp = self.client.get(url)
+        self.assertEqual(403, resp.status_code)
+
+        self._remove_upload_link(token)
+
 class AdminUploadLinkTest(BaseTestCase):

    def setUp(self):
@@ -63,6 +104,24 @@ class AdminUploadLinkTest(BaseTestCase):
        resp = self.client.get(url)
        self.assertEqual(404, resp.status_code)

+    def test_can_delete_upload_link_by_token(self):
+        self.login_as(self.admin)
+        token = self._add_upload_link()
+
+        url = reverse('api-v2.1-admin-upload-link', args=[token])
+        resp = self.client.delete(url)
+        self.assertEqual(200, resp.status_code)
+
+    def test_delete_upload_link_with_invalid_permission(self):
+        self.login_as(self.user)
+        token = self._add_upload_link()
+
+        url = reverse('api-v2.1-admin-upload-link', args=[token])
+        resp = self.client.delete(url)
+        self.assertEqual(403, resp.status_code)
+
+        self._remove_upload_link(token)
+

 class AdminUploadLinkUploadTest(BaseTestCase):