better range requests support (38f3510e) · Commits · certic / Jama

src/jama/settings.py

+6 −0

Changes for src/jama/settings.py: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -498,6 +498,12 @@ MIDDLEWARE = [
		]

		CORS_ALLOWED_ORIGINS = _env_list("JAMA_CORS_ALLOWED_ORIGINS")
		CORS_EXPOSE_HEADERS = [
		"Accept-Ranges",
		"Content-Length",
		"Content-Range",
		"Content-Disposition",
		]

		CORS_ALLOW_HEADERS = (
		*default_headers,

src/rpc/fileresponse.py

+10 −1

Changes for src/rpc/fileresponse.py: 10 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -17,7 +17,7 @@ class FileResponse(StreamingHttpResponse):
		super().__init__(args, *kwargs)

		def _set_streaming_content(self, value):
		if self.filename:
		if self.filename and "Content-Disposition" not in self.headers:
		self.headers["Content-Disposition"] = "attachment;filename={}".format(
		self.filename
		)
		@@ -197,8 +197,14 @@ class RangedFileResponse(FileResponse):
		self.ranged_file = RangedFileReader(file)
		super(RangedFileResponse, self).__init__(self.ranged_file, args, *kwargs)

		self["Accept-Ranges"] = "bytes"
		self["Content-Length"] = self.ranged_file.size
		self._resource_closers.append(file.close)

		if "HTTP_RANGE" in request.META:
		self.add_range_headers(request.META["HTTP_RANGE"])
		if request.method == "HEAD":
		self.ranged_file.stop = self.ranged_file.start

		def add_range_headers(self, range_header):
		"""
		@@ -222,6 +228,9 @@ class RangedFileResponse(FileResponse):
		if start >= size:
		# Requested range not satisfiable.
		self.status_code = 416
		self["Content-Range"] = f"bytes */{size}"
		self["Content-Length"] = 0
		self.ranged_file.stop = 0
		return

		if stop >= size:

src/rpc/tests.py

+121 −2

Changes for src/rpc/tests.py: 121 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		from django.test import RequestFactory, TestCase
		from django.test import RequestFactory, SimpleTestCase, TestCase
		from django.db import connection
		from django.test.utils import CaptureQueriesContext
		from django.utils import timezone
		@@ -1143,7 +1143,8 @@ class ServiceTestCase(TestCase):
		response = views.serve_file_public(request, file_instance.hash)

		self.assertTrue(response["Content-Disposition"].startswith("inline;"))
		self.assertNotIn("Content-Security-Policy", response)
		self.assertEqual(response["Content-Security-Policy"], "frame-ancestors 'self'")
		self.assertTrue(response.xframe_options_exempt)
		response.close()

		def test_public_serve_forces_dangerous_types_to_download(self):
		@@ -3433,3 +3434,121 @@ class ServiceTestCase(TestCase):
		{"id": child_collection.pk, "title": child_collection.title},
		],
		)


		class PdfServingTestCase(SimpleTestCase):
		def setUp(self):
		self.factory = RequestFactory()
		self.pdf = b"%PDF-1.7\nexample"

		def response(self, method="GET", **headers):
		from rpc.fileresponse import RangedFileResponse

		file = BytesIO(self.pdf)
		response = RangedFileResponse(
		self.factory.generic(method, "/rpc/serve/hash", **headers),
		file,
		filename="example.pdf",
		content_type="application/pdf",
		)
		self.addCleanup(response.close)
		return response, file

		def test_full_response_advertises_ranges_and_length(self):
		response, file = self.response()
		self.assertEqual(response.status_code, 200)
		self.assertEqual(response["Accept-Ranges"], "bytes")
		self.assertEqual(int(response["Content-Length"]), len(self.pdf))
		self.assertEqual(b"".join(response.streaming_content), self.pdf)
		response.close()
		self.assertTrue(file.closed)

		def test_partial_response_contains_only_requested_bytes(self):
		for header, expected, content_range in [
		("bytes=0-3", self.pdf[:4], "bytes 0-3/16"),
		("bytes=-4", self.pdf[-4:], "bytes 12-15/16"),
		("bytes=4-", self.pdf[4:], "bytes 4-15/16"),
		("bytes=12-999", self.pdf[12:], "bytes 12-15/16"),
		]:
		with self.subTest(header=header):
		response, _ = self.response(HTTP_RANGE=header)
		self.assertEqual(response.status_code, 206)
		self.assertEqual(response["Content-Range"], content_range)
		self.assertEqual(int(response["Content-Length"]), len(expected))
		self.assertEqual(b"".join(response.streaming_content), expected)

		def test_unsatisfiable_range_returns_empty_response(self):
		response, _ = self.response(HTTP_RANGE="bytes=100-200")
		self.assertEqual(response.status_code, 416)
		self.assertEqual(response["Content-Range"], "bytes */16")
		self.assertEqual(response["Content-Length"], "0")
		self.assertEqual(b"".join(response.streaming_content), b"")

		def test_head_has_full_headers_without_reading_body(self):
		response, file = self.response("HEAD")
		self.assertEqual(response["Accept-Ranges"], "bytes")
		self.assertEqual(response["Content-Length"], "16")
		self.assertEqual(b"".join(response.streaming_content), b"")
		self.assertEqual(file.tell(), 0)

		def test_invalid_range_falls_back_to_full_response(self):
		response, _ = self.response(HTTP_RANGE="bytes=invalid")
		self.assertEqual(response.status_code, 200)
		self.assertEqual(response["Content-Length"], "16")
		self.assertEqual(b"".join(response.streaming_content), self.pdf)

		def serve(self, mime, origin=None):
		file = Mock(new_filename="example.pdf")
		file.file_type.mime = mime
		from django.test import override_settings

		with (
		patch.object(views.settings, "JAMA_ENABLE_PUBLIC_DOWNLOAD_URLS", True),
		patch.object(
		views.settings, "CORS_ALLOWED_ORIGINS", ["https://front.example"]
		),
		override_settings(CORS_ALLOWED_ORIGINS=["https://front.example"]),
		patch.object(models.File.objects, "filter") as query,
		patch("rpc.views.open", return_value=BytesIO(self.pdf)),
		):
		query.return_value.first.return_value = file
		response = self.client.get(
		reverse("serve_file_public", args=["hash"]),
		**({"HTTP_ORIGIN": origin} if origin else {}),
		)
		self.addCleanup(response.close)
		return response

		def test_pdf_can_be_framed_by_same_origin_and_allowed_frontend(self):
		response = self.serve("application/pdf", "https://front.example")
		self.assertEqual(response.status_code, 200)
		self.assertEqual(response["Content-Type"], "application/pdf")
		self.assertTrue(response["Content-Disposition"].startswith("inline;"))
		self.assertNotIn("X-Frame-Options", response)
		self.assertEqual(
		response["Content-Security-Policy"],
		"frame-ancestors 'self' https://front.example",
		)
		self.assertEqual(
		response["Access-Control-Allow-Origin"], "https://front.example"
		)
		for header in [
		"Accept-Ranges",
		"Content-Length",
		"Content-Range",
		"Content-Disposition",
		]:
		self.assertIn(header, response["Access-Control-Expose-Headers"])

		def test_untrusted_origin_is_not_allowed(self):
		response = self.serve("application/pdf", "https://untrusted.example")
		self.assertNotIn("Access-Control-Allow-Origin", response)
		self.assertNotIn("untrusted.example", response["Content-Security-Policy"])

		def test_dangerous_content_retains_attachment_and_frame_protection(self):
		response = self.serve("text/html")
		self.assertTrue(response["Content-Disposition"].startswith("attachment;"))
		self.assertEqual(
		response["Content-Security-Policy"], "sandbox; default-src 'none'"
		)
		self.assertEqual(response["X-Frame-Options"], "DENY")

src/rpc/views.py

+5 −0

Changes for src/rpc/views.py: 5 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -880,6 +880,7 @@ def force_download_public(
		)


		@require_http_methods(["GET", "HEAD"])
		def serve_file_public(
		request: HttpRequest, file_hash: str
		) -> Union[HttpResponse, RangedFileResponse]:
		@@ -901,6 +902,10 @@ def serve_file_public(
		)
		if requires_attachment:
		response["Content-Security-Policy"] = "sandbox; default-src 'none'"
		else:
		frame_origins = " ".join(["'self'", *settings.CORS_ALLOWED_ORIGINS])
		response["Content-Security-Policy"] = f"frame-ancestors {frame_origins}"
		response.xframe_options_exempt = True
		return response