paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-11-03 13:33:24 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
818c30372f538149f01f112fcff36f238d81c1fe
zulip/zerver/tests/test_tusd.py
Alex Vandiver 818c30372f upload: Use tusd for resumable, larger uploads. 
Currently, it handles two hook types: 'pre-create' (to verify that the
user is authenticated and the file size is within the limit) and
'pre-finish' (which creates an attachment row).

No secret is shared between Django and tusd for authentication of the
hooks endpoints, because none is necessary -- tusd forwards the
end-user's credentials, and the hook checks them like it would any
end-user request.  An end-user gaining access to the endpoint would be
able to do no more harm than via tusd or the normal file upload API.

Regardless, the previous commit has restricted access to the endpoint
at the nginx layer.

Co-authored-by: Brijmohan Siyag <brijsiyag@gmail.com>
2024-09-19 11:37:29 -07:00

395 lines
14 KiB
Python
Raw Blame History

import os
import orjson
from django.conf import settings
from django.test import override_settings
from zerver.lib.cache import cache_delete, get_realm_used_upload_space_cache_key
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.upload import sanitize_name, upload_backend, upload_message_attachment
from zerver.lib.utils import assert_is_not_none
from zerver.models import Attachment
from zerver.views.tusd import TusEvent, TusHook, TusHTTPRequest, TusUpload
class TusdHooksTest(ZulipTestCase):
def test_non_localhost(self) -> None:
request = TusHook(
type="pre-create",
event=TusEvent(
http_request=TusHTTPRequest(
method="PATCH", uri="/api/v1/tus/thing", remote_addr="12.34.56.78", header={}
),
upload=TusUpload(
id="",
is_final=False,
is_partial=False,
meta_data={
"filename": "zulip.txt",
"filetype": "text/plain",
"name": "zulip.txt",
"type": "text/plain",
},
offset=0,
partial_uploads=None,
size=1234,
size_is_deferred=False,
storage=None,
),
),
)
result = self.client_post(
"/api/internal/tusd",
request.model_dump(),
content_type="application/json",
REMOTE_ADDR="1.2.3.4",
)
self.assertEqual(result.status_code, 403)
result = self.client_post(
"/api/internal/tusd",
request.model_dump(),
content_type="application/json",
REMOTE_ADDR="127.0.0.1",
)
self.assertEqual(result.status_code, 200)
def test_invalid_hook(self) -> None:
self.login("hamlet")
request = TusHook(
type="bogus",
event=TusEvent(
http_request=TusHTTPRequest(
method="PATCH", uri="/api/v1/tus/thing", remote_addr="12.34.56.78", header={}
),
upload=TusUpload(
id="",
is_final=False,
is_partial=False,
meta_data={
"filename": "zulip.txt",
"filetype": "text/plain",
"name": "zulip.txt",
"type": "text/plain",
},
offset=0,
partial_uploads=None,
size=1234,
size_is_deferred=False,
storage=None,
),
),
)
result = self.client_post(
"/api/internal/tusd", request.model_dump(), content_type="application/json"
)
self.assertEqual(result.status_code, 404)
def test_invalid_payload(self) -> None:
result = self.client_post(
"/api/internal/tusd",
{"type": "pre-create", "event": "moose"},
content_type="application/json",
)
self.assertEqual(result.status_code, 400)
class TusdPreCreateTest(ZulipTestCase):
def request(self) -> TusHook:
return TusHook(
type="pre-create",
event=TusEvent(
http_request=TusHTTPRequest(
method="PATCH", uri="/api/v1/tus/thing", remote_addr="12.34.56.78", header={}
),
upload=TusUpload(
id="",
is_final=False,
is_partial=False,
meta_data={
"filename": "zulip.txt",
"filetype": "text/plain",
"name": "zulip.txt",
"type": "text/plain",
},
offset=0,
partial_uploads=None,
size=1234,
size_is_deferred=False,
storage=None,
),
),
)
def test_tusd_pre_create_hook(self) -> None:
self.login("hamlet")
result = self.client_post(
"/api/internal/tusd",
self.request().model_dump(),
content_type="application/json",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json.get("HttpResponse", None), None)
self.assertEqual(result_json.get("RejectUpload", False), False)
self.assertEqual(list(result_json["ChangeFileInfo"].keys()), ["ID"])
self.assertTrue(result_json["ChangeFileInfo"]["ID"].endswith("/zulip.txt"))
def test_unauthed_rejected(self) -> None:
result = self.client_post(
"/api/internal/tusd",
self.request().model_dump(),
content_type="application/json",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json["HttpResponse"]["StatusCode"], 401)
self.assertEqual(
orjson.loads(result_json["HttpResponse"]["Body"]), {"message": "Unauthenticated upload"}
)
self.assertEqual(result_json["RejectUpload"], True)
def test_api_key_auth(self) -> None:
user_profile = self.example_user("hamlet")
result = self.client_post(
"/api/internal/tusd",
self.request().model_dump(),
content_type="application/json",
HTTP_AUTHORIZATION=self.encode_user(user_profile),
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json.get("HttpResponse", None), None)
self.assertEqual(result_json.get("RejectUpload", False), False)
self.assertEqual(list(result_json["ChangeFileInfo"].keys()), ["ID"])
self.assertTrue(result_json["ChangeFileInfo"]["ID"].endswith("/zulip.txt"))
def test_api_key_bad_auth(self) -> None:
result = self.client_post(
"/api/internal/tusd",
self.request().model_dump(),
content_type="application/json",
HTTP_AUTHORIZATION="Digest moose",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json["HttpResponse"]["StatusCode"], 401)
self.assertEqual(
orjson.loads(result_json["HttpResponse"]["Body"]), {"message": "Unauthenticated upload"}
)
self.assertEqual(result_json["RejectUpload"], True)
def test_sanitize_filename(self) -> None:
self.login("hamlet")
request = self.request()
request.event.upload.meta_data["filename"] = "some thing! ... like this?"
result = self.client_post(
"/api/internal/tusd",
request.model_dump(),
content_type="application/json",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertTrue(result_json["ChangeFileInfo"]["ID"].endswith("/some-thing-...-like-this"))
@override_settings(MAX_FILE_UPLOAD_SIZE=1) # In MB
def test_file_too_big_failure(self) -> None:
self.login("hamlet")
request = self.request()
request.event.upload.size = 1024 * 1024 * 5 # 5MB
result = self.client_post(
"/api/internal/tusd",
request.model_dump(),
content_type="application/json",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json["HttpResponse"]["StatusCode"], 413)
self.assertEqual(
orjson.loads(result_json["HttpResponse"]["Body"]),
{"message": "Uploaded file is larger than the allowed limit of 1 MiB"},
)
self.assertEqual(result_json["RejectUpload"], True)
def test_deferred_size(self) -> None:
self.login("hamlet")
request = self.request()
request.event.upload.size = None
request.event.upload.size_is_deferred = True
result = self.client_post(
"/api/internal/tusd",
request.model_dump(),
content_type="application/json",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json["HttpResponse"]["StatusCode"], 411)
self.assertEqual(
orjson.loads(result_json["HttpResponse"]["Body"]),
{"message": "SizeIsDeferred is not supported"},
)
self.assertEqual(result_json["RejectUpload"], True)
def test_quota_exceeded(self) -> None:
hamlet = self.example_user("hamlet")
self.login("hamlet")
# We fake being almost at quota
realm = hamlet.realm
realm.custom_upload_quota_gb = 1
realm.save(update_fields=["custom_upload_quota_gb"])
path_id = upload_message_attachment("zulip.txt", "text/plain", b"zulip!", hamlet)[
0
].removeprefix("/user_uploads/")
attachment = Attachment.objects.get(path_id=path_id)
attachment.size = assert_is_not_none(realm.upload_quota_bytes()) - 10
attachment.save(update_fields=["size"])
cache_delete(get_realm_used_upload_space_cache_key(realm.id))
result = self.client_post(
"/api/internal/tusd",
self.request().model_dump(),
content_type="application/json",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json["HttpResponse"]["StatusCode"], 413)
self.assertEqual(
orjson.loads(result_json["HttpResponse"]["Body"]),
{"message": "Upload would exceed your organization's upload quota."},
)
self.assertEqual(result_json["RejectUpload"], True)
class TusdPreFinishTest(ZulipTestCase):
def request(self, info: TusUpload) -> TusHook:
return TusHook(
type="pre-finish",
event=TusEvent(
upload=info,
http_request=TusHTTPRequest(
method="PATCH",
uri=f"/api/v1/tus/{info.id}",
remote_addr="12.34.56.78",
header={},
),
),
)
def test_tusd_pre_finish_hook(self) -> None:
self.login("hamlet")
hamlet = self.example_user("hamlet")
# Act like tusd does -- put the file and its .info in place
path_id = upload_backend.generate_message_upload_path(
str(hamlet.realm.id), sanitize_name("zulip.txt")
)
upload_backend.upload_message_attachment(
path_id, "zulip.txt", "text/plain", b"zulip!", hamlet
)
info = TusUpload(
id=path_id,
size=len("zulip!"),
offset=0,
size_is_deferred=False,
meta_data={
"filename": "zulip.txt",
"filetype": "text/plain",
"name": "zulip.txt",
"type": "text/plain",
},
is_final=False,
is_partial=False,
partial_uploads=None,
storage=None,
)
upload_backend.upload_message_attachment(
f"{path_id}.info",
"zulip.txt.info",
"application/octet-stream",
info.model_dump_json().encode(),
hamlet,
)
# Post the hook saying the file is in place
result = self.client_post(
"/api/internal/tusd",
self.request(info).model_dump(),
content_type="application/json",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json["HttpResponse"]["StatusCode"], 200)
self.assertEqual(
orjson.loads(result_json["HttpResponse"]["Body"]),
{"url": f"/user_uploads/{path_id}", "filename": "zulip.txt"},
)
self.assertEqual(
result_json["HttpResponse"]["Header"], {"Content-Type": "application/json"}
)
attachment = Attachment.objects.get(path_id=path_id)
self.assertEqual(attachment.size, len("zulip!"))
self.assertEqual(attachment.content_type, "text/plain")
assert settings.LOCAL_FILES_DIR is not None
self.assertTrue(os.path.exists(os.path.join(settings.LOCAL_FILES_DIR, path_id)))
self.assertFalse(os.path.exists(os.path.join(settings.LOCAL_FILES_DIR, f"{path_id}.info")))
def test_no_metadata(self) -> None:
self.login("hamlet")
hamlet = self.example_user("hamlet")
# Act like tusd does -- put the file and its .info in place
path_id = upload_backend.generate_message_upload_path(
str(hamlet.realm.id), sanitize_name("")
)
upload_backend.upload_message_attachment(path_id, "", "ignored", b"zulip!", hamlet)
info = TusUpload(
id=path_id,
size=len("zulip!"),
offset=0,
size_is_deferred=False,
meta_data={},
is_final=False,
is_partial=False,
partial_uploads=None,
storage=None,
)
upload_backend.upload_message_attachment(
f"{path_id}.info",
".info",
"ignored",
info.model_dump_json().encode(),
hamlet,
)
# Post the hook saying the file is in place
result = self.client_post(
"/api/internal/tusd",
self.request(info).model_dump(),
content_type="application/json",
)
self.assertEqual(result.status_code, 200)
result_json = result.json()
self.assertEqual(result_json["HttpResponse"]["StatusCode"], 200)
self.assertEqual(
orjson.loads(result_json["HttpResponse"]["Body"]),
{"url": f"/user_uploads/{path_id}", "filename": "uploaded-file"},
)
self.assertEqual(
result_json["HttpResponse"]["Header"], {"Content-Type": "application/json"}
)
attachment = Attachment.objects.get(path_id=path_id)
self.assertEqual(attachment.size, len("zulip!"))
self.assertEqual(attachment.content_type, "application/octet-stream")
assert settings.LOCAL_FILES_DIR is not None
self.assertTrue(os.path.exists(os.path.join(settings.LOCAL_FILES_DIR, path_id)))
self.assertFalse(os.path.exists(os.path.join(settings.LOCAL_FILES_DIR, f"{path_id}.info")))
Reference in New Issue View Git Blame Copy Permalink