fix broken upload with X-origin-dir for user with limited access (3b66ad24) · Commits · certic / Jama

src/jama/init.py

+1 −1

Original line number	Diff line number	Diff line
		__version__ = "0.6.8"
		__version__ = "0.6.9"

src/rpc/methods.py

+14 −14

Original line number	Diff line number	Diff line
		@@ -618,6 +618,9 @@ def add_collection(user: User, title: str, parent_id: int) -> Dict:
		}
		```
		"""
		title = title.strip()
		if not title:
		raise ServiceException("missing collection title")
		# fetch parent, check parent's project add access
		parent = (
		Collection.objects.filter(
		@@ -629,12 +632,17 @@ def add_collection(user: User, title: str, parent_id: int) -> Dict:
		)
		if not parent:
		raise ServiceException(NO_SUCH_COLLECTION)
		UserAccess(user, parent.project).check_create(parent)
		collection_instance = Collection.objects.filter(
		title=title, parent=parent, project_id=parent.project.pk
		).first()
		if not collection_instance:
		UserAccess(user, parent.project).check_update(parent)
		collection_instance, created = Collection.objects.get_or_create(
		title=title, parent=parent, project_id=parent.project.pk
		)
		# collection was previously soft-deleted, reactivate it.
		if collection_instance.deleted_at:
		UserAccess(user, parent.project).check_update(collection_instance)
		collection_instance.deleted_at = None
		collection_instance.save()
		return serializers.collection(collection_instance, cache=SerializerCache())
		@@ -707,20 +715,12 @@ def add_collection_from_path(user: User, path: str, project_id: int) -> List[Dic
		if not project:
		raise ServiceException(NO_SUCH_PROJECT)

		UserAccess(user, project).check_create("collection")

		hierarchy = []
		previous_dir = project.root_collection
		serializer_cache = SerializerCache()
		for dir_name in _ascii_collection_path_segments(path):
		previous_dir, created = Collection.objects.get_or_create(
		title=dir_name, parent=previous_dir, project=project
		)
		# if has been soft-deleted, undelete.
		if previous_dir.deleted_at:
		previous_dir.deleted_at = None
		previous_dir.save()
		hierarchy.append(serializers.collection(previous_dir, cache=serializer_cache))
		serialized_collection = add_collection(user, dir_name, previous_dir.pk)
		hierarchy.append(serialized_collection)
		previous_dir = Collection.objects.get(pk=serialized_collection["id"])
		return hierarchy

src/rpc/tests.py

+0 −19

Original line number	Diff line number	Diff line
		@@ -543,25 +543,6 @@ class ServiceTestCase(TestCase):
		{"status": "not available", "id": None, "available_chunks": []},
		)

		def test_upload_partial_origin_dir_requires_collection_create_access(self):
		file_instance = self._public_download_test_file()
		models.ObjectPermission.objects.filter(
		role=self.admin_role, object_class="collection", object_pk=None
		).update(object_create=False)

		response = views.upload_partial(
		self._upload_partial_request(
		file_instance.hash,
		"1/1",
		origin_dir="private collection",
		)
		)

		self.assertEqual(response.status_code, 403)
		self.assertFalse(
		models.Collection.objects.filter(title="private collection").exists()
		)

		def test_upload_partial_origin_dir_requires_collection_update_access(self):
		file_instance = self._public_download_test_file()
		models.ObjectPermission.objects.filter(

src/rpc/views.py

+13 −19

Original line number	Diff line number	Diff line
		@@ -110,23 +110,10 @@ def _file_hash256(file_path: str) -> str:
		def _collection_from_origin_dir(
		origin_dir: str, project: models.Project, acl: UserAccess
		) -> Union[models.Collection, None]:
		acl.check_create("collection")
		# root dir always first
		previous_dir = project.root_collection
		for dir_name in origin_dir.split("/"):
		dir_name = dir_name.strip()
		if not dir_name:
		continue
		dir_name = dir_name.strip()
		if dir_name:
		previous_dir, created = models.Collection.objects.get_or_create(
		project=project, title=dir_name, parent=previous_dir
		)
		# this was soft-deleted. Un-delete !
		if not created and previous_dir.deleted_at:
		previous_dir.deleted_at = None
		previous_dir.save()
		return previous_dir
		if not origin_dir or not any(segment.strip() for segment in origin_dir.split("/")):
		return project.root_collection
		hierarchy = rpc_methods.add_collection_from_path(acl.user, origin_dir, project.pk)
		return models.Collection.objects.get(pk=hierarchy[-1]["id"])


		def _add_file_to_origin_collection(
		@@ -137,7 +124,7 @@ def _add_file_to_origin_collection(
		) -> models.Collection:
		collection = _collection_from_origin_dir(origin_dir, project, acl)
		acl.check_update(collection)
		acl.check_read(file_instance)
		# acl.check_read(file_instance)
		collection.resources.add(file_instance)
		return collection

		@@ -472,6 +459,10 @@ def upload_partial(request: HttpRequest) -> HttpResponse:
		acl.check_create("resource")
		except ServiceException:
		return HttpResponse("Forbidden", status=403)

		##
		## First pass is trying to find an existing file
		##
		try:
		# optional header, used to create collections
		origin_dir = request.headers.get("X-origin-dir", None)
		@@ -518,8 +509,11 @@ def upload_partial(request: HttpRequest) -> HttpResponse:
		except (KeyError, ValueError, UnicodeDecodeError, binascii.Error):
		return HttpResponse(status=400)
		except models.File.DoesNotExist:
		pass
		pass # File does not exist, just go on with the upload

		##
		## Existing file not found, go on with the upload
		##
		partials_dir = _partial_upload_dir(user, project, file_hash)
		lock_file_path = "{}/.lock".format(partials_dir)
		if os.path.isfile(lock_file_path):