diff --git a/frigate/api/export.py b/frigate/api/export.py index 817dbe7ba8..e9de2f0ddc 100644 --- a/frigate/api/export.py +++ b/frigate/api/export.py @@ -2,6 +2,7 @@ import datetime import logging +import os import random import string import time @@ -56,6 +57,7 @@ from frigate.api.defs.response.export_response import ( ) from frigate.api.defs.response.generic_response import GenericResponse from frigate.api.defs.tags import Tags +from frigate.api.media_auth import MediaAuthResolution, resolve_media_uri from frigate.const import CLIPS_DIR, EXPORT_DIR from frigate.jobs.export import ( ExportJob, @@ -130,13 +132,26 @@ def _validate_export_case(export_case_id: str | None) -> JSONResponse | None: def _sanitize_existing_image( + request: Request, image_path: str | None, + camera: str, ) -> tuple[str | None, JSONResponse | None]: if not image_path: return None, None existing_image = sanitize_contained_path(image_path, CLIPS_DIR) + # CLIPS_DIR is shared by every camera, so the image must also belong to + # the camera being exported. + if existing_image is not None: + resolution = resolve_media_uri( + f"/clips/{os.path.relpath(existing_image, CLIPS_DIR)}", + request.app.frigate_config, + ) + + if resolution != (MediaAuthResolution.CAMERA, camera): + existing_image = None + if existing_image is None: return None, JSONResponse( content={"success": False, "message": "Invalid image path"}, @@ -680,7 +695,7 @@ def export_recordings_batch( sanitized_images: list[str | None] = [] for item in body.items: existing_image, image_validation_error = _sanitize_existing_image( - item.image_path + request, item.image_path, item.camera ) if image_validation_error is not None: return image_validation_error @@ -827,7 +842,9 @@ def export_recording( playback_source = body.source friendly_name = body.name - existing_image, image_validation_error = _sanitize_existing_image(body.image_path) + existing_image, image_validation_error = _sanitize_existing_image( + request, body.image_path, camera_name + ) if image_validation_error is not None: return image_validation_error @@ -965,7 +982,9 @@ def export_recording_custom( playback_source = body.source friendly_name = body.name - existing_image, image_validation_error = _sanitize_existing_image(body.image_path) + existing_image, image_validation_error = _sanitize_existing_image( + request, body.image_path, camera_name + ) if image_validation_error is not None: return image_validation_error ffmpeg_input_args = body.ffmpeg_input_args diff --git a/frigate/test/http_api/test_http_export.py b/frigate/test/http_api/test_http_export.py index 44eb0c2c4a..d065e954ab 100644 --- a/frigate/test/http_api/test_http_export.py +++ b/frigate/test/http_api/test_http_export.py @@ -4,6 +4,7 @@ import tempfile import zipfile from unittest.mock import patch +from frigate.const import CLIPS_DIR from frigate.jobs.export import ( ExportJob, get_export_job_manager, @@ -949,6 +950,60 @@ class TestHttpExport(BaseTestHttp): assert response.status_code == 400 assert ExportCase.select().count() == 0 + def test_batch_export_rejects_other_camera_image_path(self): + self._insert_recording("rec-front", "front_door", 100, 400) + + for image_path in ( + f"{CLIPS_DIR}/review/thumb-backyard-123.456-abc.webp", + f"{CLIPS_DIR}/thumbs/backyard/123.webp", + f"{CLIPS_DIR}/faces/someone/1.webp", + ): + with AuthTestClient(self.app) as client: + response = client.post( + "/exports/batch", + json={ + "items": [ + { + "camera": "front_door", + "start_time": 110, + "end_time": 150, + "image_path": image_path, + } + ], + }, + ) + + assert response.status_code == 400, image_path + + def test_batch_export_accepts_own_camera_image_path(self): + self._insert_recording("rec-front", "front_door", 100, 400) + + with patch( + "frigate.api.export.start_export_job", + side_effect=lambda _config, job: job.id, + ) as start_export_job: + with AuthTestClient(self.app) as client: + response = client.post( + "/exports/batch", + json={ + "items": [ + { + "camera": "front_door", + "start_time": 110, + "end_time": 150, + "image_path": ( + f"{CLIPS_DIR}/review/thumb-front_door-123.456-abc.webp" + ), + } + ], + }, + ) + + assert response.status_code == 202 + assert start_export_job.call_args.args[1].image_path == ( + f"{CLIPS_DIR}/review/thumb-front_door-123.456-abc.webp" + ) + def test_batch_export_non_admin_can_queue(self): self._insert_recording("rec-front", "front_door", 100, 400)