diff --git a/seahub/api2/urls.py b/seahub/api2/urls.py index 3caf44d5ad..b920234f05 100644 --- a/seahub/api2/urls.py +++ b/seahub/api2/urls.py @@ -32,6 +32,10 @@ urlpatterns = patterns('', url(r'^repos/(?P[-0-9a-f]{36})/download-info/$', DownloadRepo.as_view()), url(r'^repos/(?P[-0-9a-f]{36})/owner/$', RepoOwner.as_view(), name="api2-repo-owner"), url(r'^repos/(?P[-0-9a-f]{36})/public/$', RepoPublic.as_view()), + url(r'^repos/(?P[-0-9a-f]{36})/download-shared-links/$', RepoDownloadSharedLinks.as_view(), name="api2-repo-download-shared-links"), + url(r'^repos/(?P[-0-9a-f]{36})/download-shared-links/(?P[a-f0-9]{10})/$', RepoDownloadSharedLink.as_view(), name="api2-repo-download-shared-link"), + url(r'^repos/(?P[-0-9a-f]{36})/upload-shared-links/$', RepoUploadSharedLinks.as_view(), name="api2-repo-upload-shared-links"), + url(r'^repos/(?P[-0-9a-f]{36})/upload-shared-links/(?P[a-f0-9]{10})/$', RepoUploadSharedLink.as_view(), name="api2-repo-upload-shared-link"), url(r'^repos/(?P[-0-9a-f]{36})/upload-link/$', UploadLinkView.as_view()), url(r'^repos/(?P[-0-9a-f]{36})/update-link/$', UpdateLinkView.as_view()), url(r'^repos/(?P[-0-9a-f]{36})/upload-blks-link/$', UploadBlksLinkView.as_view()), diff --git a/seahub/api2/views.py b/seahub/api2/views.py index 48663fd2c3..505f85b8f1 100644 --- a/seahub/api2/views.py +++ b/seahub/api2/views.py @@ -29,6 +29,7 @@ from django.template.defaultfilters import filesizeformat from django.shortcuts import render_to_response from django.utils import timezone from django.utils.translation import ugettext as _ +from django.utils.dateformat import DateFormat from .throttling import ScopedRateThrottle, AnonRateThrottle, UserRateThrottle from .authentication import TokenAuthentication @@ -4199,3 +4200,180 @@ class OrganizationView(APIView): except Exception as e: logger.error(e) return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, "Internal error") + +class RepoDownloadSharedLinks(APIView): + authentication_classes = (TokenAuthentication, SessionAuthentication) + permission_classes = (IsAuthenticated, ) + throttle_classes = (UserRateThrottle, ) + + def get(self, request, repo_id, format=None): + repo = get_repo(repo_id) + if not repo: + error_msg = 'Library %s not found.' % repo_id + return api_error(status.HTTP_404_NOT_FOUND, error_msg) + + org_id = None + if is_org_context(request): + org_id = request.user.org.org_id + + # check permission + if org_id: + repo_owner = seafile_api.get_org_repo_owner(repo_id) + else: + repo_owner = seafile_api.get_repo_owner(repo_id) + + if request.user.username != repo_owner or repo.is_virtual: + error_msg = 'Permission denied.' + return api_error(status.HTTP_403_FORBIDDEN, error_msg) + + shared_links = [] + fileshares = FileShare.objects.filter(repo_id=repo_id) + for fs in fileshares: + size = None + shared_link = {} + if fs.is_file_share_link(): + path = fs.path.rstrip('/') # Normalize file path + if seafile_api.get_file_id_by_path(repo.id, fs.path) is None: + continue + + obj_id = seafile_api.get_file_id_by_path(repo_id, path) + size = seafile_api.get_file_size(repo.store_id, repo.version, obj_id) + else: + path = fs.path + if path[-1] != '/': # Normalize dir path + path += '/' + + if seafile_api.get_dir_id_by_path(repo.id, fs.path) is None: + continue + + shared_link['create_by'] = fs.username + shared_link['creator_name'] = email2nickname(fs.username) + # return time with time zone: 2016-01-18T15:03:10+0800 + shared_link['create_time'] = fs.ctime.strftime("%Y-%m-%dT%H:%M:%S") + DateFormat(fs.ctime).format('O') + shared_link['token'] = fs.token + shared_link['path'] = path + shared_link['name'] = os.path.basename(path.rstrip('/')) if path != '/' else '/' + shared_link['view_count'] = fs.view_cnt + shared_link['share_type'] = fs.s_type + shared_link['size'] = size if size else '' + shared_links.append(shared_link) + + return Response(shared_links) + +class RepoDownloadSharedLink(APIView): + authentication_classes = (TokenAuthentication, SessionAuthentication) + permission_classes = (IsAuthenticated, ) + throttle_classes = (UserRateThrottle, ) + + def delete(self, request, repo_id, token, format=None): + repo = get_repo(repo_id) + if not repo: + error_msg = 'Library %s not found.' % repo_id + return api_error(status.HTTP_404_NOT_FOUND, error_msg) + + org_id = None + if is_org_context(request): + org_id = request.user.org.org_id + + # check permission + if org_id: + repo_owner = seafile_api.get_org_repo_owner(repo_id) + else: + repo_owner = seafile_api.get_repo_owner(repo_id) + + if request.user.username != repo_owner or repo.is_virtual: + error_msg = 'Permission denied.' + return api_error(status.HTTP_403_FORBIDDEN, error_msg) + + try: + link = FileShare.objects.get(token=token) + except FileShare.DoesNotExist: + error_msg = 'Link %s not found.' % token + return api_error(status.HTTP_404_NOT_FOUND, error_msg) + + link.delete() + result = {'success': True} + return Response(result) + +class RepoUploadSharedLinks(APIView): + authentication_classes = (TokenAuthentication, SessionAuthentication) + permission_classes = (IsAuthenticated, ) + throttle_classes = (UserRateThrottle, ) + + def get(self, request, repo_id, format=None): + repo = get_repo(repo_id) + if not repo: + error_msg = 'Library %s not found.' % repo_id + return api_error(status.HTTP_404_NOT_FOUND, error_msg) + + org_id = None + if is_org_context(request): + org_id = request.user.org.org_id + + # check permission + if org_id: + repo_owner = seafile_api.get_org_repo_owner(repo_id) + else: + repo_owner = seafile_api.get_repo_owner(repo_id) + + if request.user.username != repo_owner or repo.is_virtual: + error_msg = 'Permission denied.' + return api_error(status.HTTP_403_FORBIDDEN, error_msg) + + shared_links = [] + fileshares = UploadLinkShare.objects.filter(repo_id=repo_id) + for fs in fileshares: + shared_link = {} + path = fs.path + if path[-1] != '/': # Normalize dir path + path += '/' + + if seafile_api.get_dir_id_by_path(repo.id, fs.path) is None: + continue + + shared_link['create_by'] = fs.username + shared_link['creator_name'] = email2nickname(fs.username) + # return time with time zone: 2016-01-18T15:03:10+0800 + shared_link['create_time'] = fs.ctime.strftime("%Y-%m-%dT%H:%M:%S") + DateFormat(fs.ctime).format('O') + shared_link['token'] = fs.token + shared_link['path'] = path + shared_link['name'] = os.path.basename(path.rstrip('/')) if path != '/' else '/' + shared_link['view_count'] = fs.view_cnt + shared_links.append(shared_link) + + return Response(shared_links) + +class RepoUploadSharedLink(APIView): + authentication_classes = (TokenAuthentication, SessionAuthentication) + permission_classes = (IsAuthenticated, ) + throttle_classes = (UserRateThrottle, ) + + def delete(self, request, repo_id, token, format=None): + repo = get_repo(repo_id) + if not repo: + error_msg = 'Library %s not found.' % repo_id + return api_error(status.HTTP_404_NOT_FOUND, error_msg) + + org_id = None + if is_org_context(request): + org_id = request.user.org.org_id + + # check permission + if org_id: + repo_owner = seafile_api.get_org_repo_owner(repo_id) + else: + repo_owner = seafile_api.get_repo_owner(repo_id) + + if request.user.username != repo_owner or repo.is_virtual: + error_msg = 'Permission denied.' + return api_error(status.HTTP_403_FORBIDDEN, error_msg) + + try: + link = UploadLinkShare.objects.get(token=token) + except FileShare.DoesNotExist: + error_msg = 'Link %s not found.' % token + return api_error(status.HTTP_404_NOT_FOUND, error_msg) + + link.delete() + result = {'success': True} + return Response(result) diff --git a/tests/api/test_repo_shared_links.py b/tests/api/test_repo_shared_links.py new file mode 100644 index 0000000000..3e397cff7f --- /dev/null +++ b/tests/api/test_repo_shared_links.py @@ -0,0 +1,107 @@ +"""seahub/api2/views.py::Repo api tests. +""" +import json +from django.core.urlresolvers import reverse + +from seahub.test_utils import BaseTestCase +from seahub.share.models import UploadLinkShare, FileShare + +class RepoSharedLinksTest(BaseTestCase): + + def setUp(self): + self.repo_id = self.repo.id + + upload_link_share = UploadLinkShare.objects.create_upload_link_share( + self.user.username, self.repo_id, self.folder) + + file_download_link_share = FileShare.objects.create_file_link(self.user.username, + self.repo_id, self.file) + + dir_download_link_share = FileShare.objects.create_dir_link(self.user.username, + self.repo_id, self.folder) + + self.upload_token = upload_link_share.token + self.file_download_token = file_download_link_share.token + self.dir_download_token = dir_download_link_share.token + + def tearDown(self): + self.remove_repo() + + def test_can_get_download_share_link(self): + self.login_as(self.user) + resp = self.client.get(reverse("api2-repo-download-shared-links", args=[self.repo_id])) + json_resp = json.loads(resp.content) + self.assertEqual(200, resp.status_code) + json_resp = json.loads(resp.content) + assert len(json_resp) == 2 + + assert json_resp[0]['token'] == self.file_download_token + assert json_resp[0]['create_by'] == self.user.email + assert json_resp[1]['token'] == self.dir_download_token + assert json_resp[1]['create_by'] == self.user.email + + def test_can_delete_download_share_link(self): + self.login_as(self.user) + resp = self.client.delete(reverse("api2-repo-download-shared-link", args=[self.repo_id, self.file_download_token])) + self.assertEqual(200, resp.status_code) + + resp = self.client.get(reverse("api2-repo-download-shared-links", args=[self.repo_id])) + json_resp = json.loads(resp.content) + self.assertEqual(200, resp.status_code) + json_resp = json.loads(resp.content) + assert len(json_resp) == 1 + + resp = self.client.delete(reverse("api2-repo-download-shared-link", args=[self.repo_id, self.dir_download_token])) + self.assertEqual(200, resp.status_code) + + resp = self.client.get(reverse("api2-repo-download-shared-links", args=[self.repo_id])) + json_resp = json.loads(resp.content) + self.assertEqual(200, resp.status_code) + json_resp = json.loads(resp.content) + assert len(json_resp) == 0 + + def test_can_get_upload_share_link(self): + self.login_as(self.user) + resp = self.client.get(reverse("api2-repo-upload-shared-links", args=[self.repo_id])) + json_resp = json.loads(resp.content) + self.assertEqual(200, resp.status_code) + json_resp = json.loads(resp.content) + assert len(json_resp) == 1 + + assert json_resp[0]['create_by'] == self.user.email + assert json_resp[0]['token'] == self.upload_token + + def test_can_delete_upload_share_link(self): + self.login_as(self.user) + resp = self.client.delete(reverse("api2-repo-upload-shared-link", args=[self.repo_id, self.upload_token])) + self.assertEqual(200, resp.status_code) + + resp = self.client.get(reverse("api2-repo-upload-shared-links", args=[self.repo_id])) + json_resp = json.loads(resp.content) + self.assertEqual(200, resp.status_code) + json_resp = json.loads(resp.content) + assert len(json_resp) == 0 + + def test_can_not_get_download_share_link_if_not_repo_owner(self): + self.login_as(self.admin) + resp = self.client.get(reverse("api2-repo-download-shared-links", args=[self.repo_id])) + self.assertEqual(403, resp.status_code) + + def test_can_not_delete_download_share_link_if_not_repo_owner(self): + self.login_as(self.admin) + + resp = self.client.delete(reverse("api2-repo-download-shared-link", args=[self.repo_id, self.file_download_token])) + self.assertEqual(403, resp.status_code) + + resp = self.client.delete(reverse("api2-repo-download-shared-link", args=[self.repo_id, self.dir_download_token])) + self.assertEqual(403, resp.status_code) + + def test_can_not_get_upload_share_link_if_not_repo_owner(self): + self.login_as(self.admin) + resp = self.client.get(reverse("api2-repo-upload-shared-links", args=[self.repo_id])) + self.assertEqual(403, resp.status_code) + + def test_can_not_delete_upload_share_link_if_not_repo_owner(self): + self.login_as(self.admin) + resp = self.client.delete(reverse("api2-repo-upload-shared-link", args=[self.repo_id, self.upload_token])) + self.assertEqual(403, resp.status_code)