j3d1/toolshed
1
0
Fork 0
Code Issues 9 Pull requests 1 Projects Releases Packages Wiki Activity

use check permissions in /media endpoint
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse source
This commit is contained in:
j3d1 2023-11-01 04:32:03 +01:00
parent 90d1149c07
commit e819700bb0
7 changed files with 56 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

17
backend/files/media_urls.py
View file

@ -2,22 +2,27 @@ from django.http import HttpResponse
from django.urls import path
from drf_yasg.utils import swagger_auto_schema
from rest_framework import status
from rest_framework.decorators import api_view
from rest_framework.decorators import api_view, permission_classes, authentication_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from authentication.signature_auth import SignatureAuthentication
from files.models import File
# TODO check file permissions here
@swagger_auto_schema(method='GET', auto_schema=None)
@api_view(['GET'])
def media_urls(request, id, format=None):
@permission_classes([IsAuthenticated])
@authentication_classes([SignatureAuthentication])
def media_urls(request, hash_path):
try:
file = File.objects.get(file=id)
file = File.objects.filter(connected_items__owner__in=request.user.friends_or_self()).distinct().get(
file=hash_path)
return HttpResponse(status=status.HTTP_200_OK,
content_type=file.mime_type,
headers={
'X-Accel-Redirect': f'/redirect_media/{id}',
'X-Accel-Redirect': f'/redirect_media/{hash_path}',
'Access-Control-Allow-Origin': '*',
}) # TODO Expires and Cache-Control
@ -26,5 +31,5 @@ def media_urls(request, id, format=None):
urlpatterns = [
path('<path:id>', media_urls),
path('<path:hash_path>', media_urls),
]

34
backend/files/tests.py
View file

@ -3,6 +3,7 @@ from django.core.files.storage import DefaultStorage
from django.db import IntegrityError, transaction
from django.test import Client
from authentication.tests import SignatureAuthClient, ToolshedTestCase, UserTestMixin
from toolshed.tests import InventoryTestMixin
from nacl.hash import sha256
from nacl.encoding import HexEncoder
import base64
@ -105,11 +106,19 @@ class FilesTestCase(FilesTestMixin, ToolshedTestCase):
self.assertEqual(countdir(DefaultStorage(), ''), 3)
class MediaUrlTestCase(FilesTestMixin, UserTestMixin, ToolshedTestCase):
class MediaUrlTestCase(FilesTestMixin, UserTestMixin, InventoryTestMixin, ToolshedTestCase):
def setUp(self):
super().setUp()
self.prepare_files()
self.prepare_users()
self.prepare_categories()
self.prepare_tags()
self.prepare_properties()
self.prepare_inventory()
self.f['item1'].files.add(self.f['test_file1'])
self.f['item1'].files.add(self.f['test_file2'])
self.f['item2'].files.add(self.f['test_file1'])
def test_file_url(self):
reply = client.get(
@ -126,10 +135,33 @@ class MediaUrlTestCase(FilesTestMixin, UserTestMixin, ToolshedTestCase):
self.assertEqual(reply.headers['X-Accel-Redirect'],
f"/redirect_media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}")
self.assertEqual(reply.headers['Content-Type'], self.f['test_file2'].mime_type)
reply = client.get(
f"/media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}",
self.f['local_user2'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(reply.headers['X-Accel-Redirect'],
f"/redirect_media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}")
self.assertEqual(reply.headers['Content-Type'], self.f['test_file2'].mime_type)
def test_file_url_fail(self):
reply = client.get('/media/{}/'.format('nonexistent'), self.f['local_user1'])
self.assertEqual(reply.status_code, 404)
self.assertTrue('X-Accel-Redirect' not in reply.headers)
def test_file_url_anonymous(self):
reply = anonymous_client.get(
f"/media/{self.f['hash1'][:2]}/{self.f['hash1'][2:4]}/{self.f['hash1'][4:6]}/{self.f['hash1'][6:]}")
self.assertEqual(reply.status_code, 403)
self.assertTrue('X-Accel-Redirect' not in reply.headers)
def test_file_url_wrong_user(self):
reply = client.get(
f"/media/{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}",
self.f['local_user1'])
self.assertEqual(reply.status_code, 404)
self.assertTrue('X-Accel-Redirect' not in reply.headers)
reply = client.get(
f"/media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}",
self.f['ext_user1'])
self.assertEqual(reply.status_code, 404)
self.assertTrue('X-Accel-Redirect' not in reply.headers)