create embedded media in protected path + expose via url

api/directory_watcher.py

@@ -101,8 +101,8 @@ def handle_new_image(user, path, job_id):
             photo.save()
 
             file = File.create(path, user)
-            if has_embedded_media(path):
-                em_path = extract_embedded_media(path)
+            if has_embedded_media(file):
+                em_path = extract_embedded_media(file)
                 if em_path:
                     em_file = File.create(em_path, user)
                     file.embedded_media.add(em_file)
@@ -194,9 +194,7 @@ def handle_new_image(user, path, job_id):
             photo.files.add(file)
             photo.save()
             photo._check_files()
-            util.logger.warning(
-                "job {}: file {} exists already".format(job_id, path)
-            )
+            util.logger.warning("job {}: file {} exists already".format(job_id, path))
     except Exception as e:
         try:
             util.logger.exception(

api/migrations/0042_add_embedded_media.py → api/migrations/0044_add_embedded_media.py

@@ -2,9 +2,8 @@
 
 
 class Migration(migrations.Migration):
-
     dependencies = [
-        ("api", "0041_apply_user_enum_for_person"),
+        ("api", "0043_alter_photo_size"),
     ]
 
     operations = [

api/models/file.py

@@ -4,6 +4,7 @@
 
 import magic
 import pyvips
+from django.conf import settings
 from django.db import models
 
 import api.util as util
@@ -40,7 +41,7 @@ class File(models.Model):
         choices=FILE_TYPES,
     )
     missing = models.BooleanField(default=False)
-    embedded_media = models.ManyToManyField("self", related_name="+")
+    embedded_media = models.ManyToManyField("File")
 
     @staticmethod
     def create(path: str, user):
@@ -171,7 +172,8 @@ def _locate_embedded_video_samsung(data):
     return -1
 
 
-def has_embedded_media(path: str) -> bool:
+def has_embedded_media(file: File) -> bool:
+    path = str(file.path)
     with open(path, "rb+") as image:
         with mmap(image.fileno(), 0, access=ACCESS_READ) as mm:
             return (
@@ -180,15 +182,18 @@ def has_embedded_media(path: str) -> bool:
             )
 
 
-def extract_embedded_media(path: str) -> str | None:
-    with open(path, "rb") as image:
+def extract_embedded_media(file: File) -> str | None:
+    with open(str(file.path), "rb") as image:
         with mmap(image.fileno(), 0, access=ACCESS_READ) as mm:
             position = _locate_embedded_video_google(
                 mm
             ) or _locate_embedded_video_google(mm)
             if position == -1:
                 return None
-            output_path = f"{os.path.splitext(path)[0]}_embedded.mp4"
+            output_dir = f"{settings.MEDIA_ROOT}/embedded_media"
+            if not os.path.exists(output_dir):
+                os.makedirs(output_dir)
+            output_path = f"{output_dir}/{file.hash}_1.mp4"
             with open(output_path, "wb+") as video:
                 mm.seek(position)
                 data = mm.read(mm.size())

api/serializers/photos.py

@@ -9,7 +9,6 @@
 
 
 class PigPhotoSerilizer(serializers.ModelSerializer):
-
     id = serializers.SerializerMethodField()
     dominantColor = serializers.SerializerMethodField()
     aspectRatio = serializers.SerializerMethodField()
@@ -255,10 +254,14 @@ def serialize_file(file):
                 "type": "video" if file.type == File.VIDEO else "image",
             }
 
-        embedded_media = obj.main_file.embedded_media.filter(
-            type__in=[File.VIDEO, File.IMAGE]
+        embedded_media = obj.main_file.embedded_media.all()
+        if len(embedded_media) == 0:
+            return []
+        return list(
+            map(
+                serialize_file, embedded_media.filter(type__in=[File.VIDEO, File.IMAGE])
+            )
         )
-        return list(map(serialize_file, embedded_media))
 
 
 class SharedFromMePhotoThroughSerializer(serializers.ModelSerializer):

api/tests/test_motion_photo.py

@@ -1,17 +1,23 @@
-from django.test import TestCase
+from django.conf import settings
+from django.test import TestCase, override_settings
+from rest_framework.test import APIClient
 
+from api.models import User
 from api.models.file import (
     GOOGLE_PIXEL_MOTION_PHOTO_MP4_SIGNATURES,
     JPEG_EOI_MARKER,
     SAMSUNG_MOTION_PHOTO_MARKER,
+    File,
     extract_embedded_media,
     has_embedded_media,
 )
+from api.tests.utils import create_test_photo, create_test_user
 
 
-def create_test_file(path: str, content: bytes):
-    with open(path, "w+b") as f:
+def create_test_file(path: str, user: User, content: bytes):
+    with open(path, "wb+") as f:
         f.write(content)
+    return File.create(path, user)
 
 
 JPEG = b"\xDE\xAD\xFA\xCE" + JPEG_EOI_MARKER
@@ -21,62 +27,82 @@ def create_test_file(path: str, content: bytes):
 RANDOM_BYTES = b"\x13\x37\xC0\xDE"
 
 
+@override_settings(MEDIA_ROOT="/tmp")
 class MotionPhotoTest(TestCase):
-    test_file_path = "/tmp/test_image.jpeg"
-    embedded_file_path = "/tmp/test_image_embedded.mp4"
+    def setUp(self):
+        self.test_file_path = "/tmp/test_image.jpeg"
+        self.user = create_test_user()
+        self.client = APIClient()
 
     def test_google_pixel_motion_photo_signatures(self):
         for signature in GOOGLE_PIXEL_MOTION_PHOTO_MP4_SIGNATURES:
             content = JPEG + MP4_PREFIX + signature + MP4_DATA
-            create_test_file(self.test_file_path, content)
-            actual = has_embedded_media(self.test_file_path)
+            file = create_test_file(self.test_file_path, self.user, content)
+            actual = has_embedded_media(file)
             self.assertTrue(actual)
 
     def test_samsung_motion_photo_signature(self):
         content = JPEG + SAMSUNG_MOTION_PHOTO_MARKER + MP4_DATA
-        create_test_file(self.test_file_path, content)
-        actual = has_embedded_media(self.test_file_path)
+        file = create_test_file(self.test_file_path, self.user, content)
+        actual = has_embedded_media(file)
         self.assertTrue(actual)
 
     def test_other_content_should_not_report_as_having_embedded_media(self):
-        create_test_file(self.test_file_path, RANDOM_BYTES)
-        actual = has_embedded_media(self.test_file_path)
+        file = create_test_file(self.test_file_path, self.user, RANDOM_BYTES)
+        actual = has_embedded_media(file)
         self.assertFalse(actual)
 
-    def test_should_throw_when_file_does_not_exist_when_checking_for_embedded_media(
-        self,
-    ):
-        def run():
-            has_embedded_media("/path/does/not/exist")
-
-        self.assertRaises(FileNotFoundError, run)
-
     def test_extract_embedded_media_from_google_motion_photo(self):
         for signature in GOOGLE_PIXEL_MOTION_PHOTO_MP4_SIGNATURES:
             content = JPEG + MP4_PREFIX + signature + MP4_DATA
-            create_test_file(self.test_file_path, content)
-            embedded_media_path = extract_embedded_media(self.test_file_path)
-            self.assertEqual(self.embedded_file_path, embedded_media_path)
-            with open(embedded_media_path, "rb+") as f:
+            file = create_test_file(self.test_file_path, self.user, content)
+            path = extract_embedded_media(file)
+            expected = f"{settings.MEDIA_ROOT}/embedded_media/{file.hash}_1.mp4"
+            self.assertEqual(path, expected)
+            with open(path, "rb") as f:
                 contents = f.read()
                 self.assertEqual(MP4_PREFIX + signature + MP4_DATA, contents)
 
     def test_extract_embedded_media_from_samsung_motion_photo(self):
         content = JPEG + SAMSUNG_MOTION_PHOTO_MARKER + MP4
-        create_test_file(self.test_file_path, content)
-        embedded_media_path = extract_embedded_media(self.test_file_path)
-        self.assertEqual(self.embedded_file_path, embedded_media_path)
-        with open(embedded_media_path, "rb+") as f:
+        file = create_test_file(self.test_file_path, self.user, content)
+        path = extract_embedded_media(file)
+        expected = f"{settings.MEDIA_ROOT}/embedded_media/{file.hash}_1.mp4"
+        self.assertEqual(expected, path)
+        with open(path, "rb+") as f:
             contents = f.read()
             self.assertEqual(MP4, contents)
 
-    def test_extract_from_file_that_does_not_have_embedded_media(self):
-        create_test_file(self.test_file_path, JPEG)
-        embedded_media_path = extract_embedded_media(self.test_file_path)
-        self.assertIsNone(embedded_media_path)
+    def test_fetch_embedded_media_as_owner(self):
+        self.client.force_authenticate(user=self.user)
+        embedded_media = create_test_file(self.test_file_path, self.user, MP4)
+        photo = create_test_photo(owner=self.user)
+        photo.main_file.embedded_media.add(embedded_media)
+
+        response = self.client.get(f"/media/embedded_media/{photo.pk}")
+        self.assertEqual(response.status_code, 200)
+
+    def test_fetch_embedded_media_as_anonymous_when_photo_is_public(self):
+        self.client.force_authenticate(user=None)
+        embedded_media = create_test_file(self.test_file_path, self.user, MP4)
+        photo = create_test_photo(owner=self.user, public=True)
+        photo.main_file.embedded_media.add(embedded_media)
+
+        response = self.client.get(f"/media/embedded_media/{photo.pk}")
+        self.assertEqual(response.status_code, 200)
+
+    def test_fetch_embedded_media_as_anonymous_when_photo_is_private(self):
+        self.client.force_authenticate(user=None)
+        embedded_media = create_test_file(self.test_file_path, self.user, MP4)
+        photo = create_test_photo(owner=self.user, public=False)
+        photo.main_file.embedded_media.add(embedded_media)
+
+        response = self.client.get(f"/media/embedded_media/{photo.pk}")
+        self.assertEqual(response.status_code, 404)
 
-    def test_should_throw_when_file_does_not_exist_when_extracting_embedded_media(self):
-        def run():
-            extract_embedded_media("/path/does/not/exist")
+    def test_fetch_embedded_media_when_photo_does_not_have_embedded_media(self):
+        self.client.force_authenticate(user=self.user)
+        photo = create_test_photo(owner=self.user)
 
-        self.assertRaises(Exception, run)
+        response = self.client.get(f"/media/embedded_media/{photo.pk}")
+        self.assertEqual(response.status_code, 404)

api/tests/test_user.py

@@ -103,13 +103,13 @@ def test_public_delete_user(self):
     @override_config(ALLOW_REGISTRATION=False)
     def test_public_user_create_successful_on_first_setup(self):
         User.objects.all().delete()
+        # That's a weird one. When deleting all users, the user with username "deleted" is not deleted.
+        User.objects.filter(username="deleted").delete()
         self.client.force_authenticate(user=None)
-        # You do not have to provide the is_superuser flag, because it is set to true by default, when there are no superusers in the database
         data = create_user_details()
         response = self.client.post("/api/user/", data=data)
         self.assertEqual(201, response.status_code)
-        # Changed to 2, because we also have a deleted user -> endpoint things it already has a user and does not create a new admin
-        self.assertEqual(2, len(User.objects.all()))
+        self.assertEqual(1, len(User.objects.all()))
         user = User.objects.get(username=data["username"])
         self.assertTrue(user.is_superuser)
 

api/tests/utils.py

@@ -7,6 +7,11 @@
 
 fake = Faker()
 
+ONE_PIXEL_PNG = (
+    b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00\xb1\x1e\x28"
+    b"\x00\x00\x00\x03PLTE\xff\xff\xff\xff\xff\xff\x00\x00\x00\x00IEND\xaeB`\x82"
+)
+
 
 def create_password():
     return secrets.token_urlsafe(10)
@@ -40,6 +45,8 @@ def create_test_user(is_admin=False, public_sharing=False, **kwargs):
 def create_test_photo(**kwargs):
     pk = fake.md5()
     photo = Photo(pk=pk, image_hash=pk, aspect_ratio=1, **kwargs)
+    file = create_test_file(f"/tmp/{pk}.png", photo.owner, ONE_PIXEL_PNG)
+    photo.main_file = file
     if "added_on" not in kwargs.keys():
         photo.added_on = timezone.now()
     photo.save()
@@ -50,10 +57,10 @@ def create_test_photos(number_of_photos=1, **kwargs):
     return [create_test_photo(**kwargs) for _ in range(0, number_of_photos)]
 
 
-def create_test_file(**kwargs):
-    file = File(type=File.IMAGE, **kwargs)
-    file.save()
-    return file
+def create_test_file(path: str, user: User, content: bytes):
+    with open(path, "wb+") as f:
+        f.write(content)
+    return File.create(path, user)
 
 
 def share_test_photos(photo_ids, user):

api/views/views.py

@@ -7,6 +7,7 @@
 import jsonschema
 import magic
 from constance import config as site_config
+from django.db.models import Q
 from django.http import HttpResponse, HttpResponseForbidden, StreamingHttpResponse
 from django.utils.decorators import method_decorator
 from django.utils.encoding import iri_to_uri
@@ -272,7 +273,6 @@ def get(self, request, path, fname, format=None):
 
 
 class VideoTranscoder:
-
     process = ""
 
     def __init__(self, path):
@@ -380,6 +380,22 @@ def get(self, request, path, fname, format=None):
                 return response
             except Exception:
                 return HttpResponse(status=404)
+        if path.lower() == "embedded_media":
+            try:
+                query = (
+                    Q(owner=request.user)
+                    if request.user.is_authenticated
+                    else Q(public=True)
+                )
+                photo = Photo.objects.filter(query, image_hash=fname).first()
+                if not photo or photo.main_file.embedded_media.count() < 1:
+                    raise Photo.DoesNotExist()
+            except Photo.DoesNotExist:
+                return HttpResponse(status=404)
+            response = HttpResponse()
+            response["Content-Type"] = "video/mp4"
+            response["X-Accel-Redirect"] = f"/protected_media/{path}/{fname}_1.mp4"
+            return response
         if path.lower() != "photos":
             jwt = request.COOKIES.get("jwt")
             image_hash = fname.split(".")[0].split("_")[0]