diff --git a/frigate/api/export.py b/frigate/api/export.py index 817dbe7ba8..361300daa7 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 deny_response_for_media_uri from frigate.const import CLIPS_DIR, EXPORT_DIR from frigate.jobs.export import ( ExportJob, @@ -130,6 +132,7 @@ def _validate_export_case(export_case_id: str | None) -> JSONResponse | None: def _sanitize_existing_image( + request: Request, image_path: str | None, ) -> tuple[str | None, JSONResponse | None]: if not image_path: @@ -137,6 +140,15 @@ def _sanitize_existing_image( existing_image = sanitize_contained_path(image_path, CLIPS_DIR) + # CLIPS_DIR is shared by every camera, so the caller must also be allowed + # to read the image. + if existing_image is not None and deny_response_for_media_uri( + f"/clips/{quote(os.path.relpath(existing_image, CLIPS_DIR))}", + request.headers.get("remote-role"), + request.app.frigate_config, + ): + existing_image = None + if existing_image is None: return None, JSONResponse( content={"success": False, "message": "Invalid image path"}, @@ -680,7 +692,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 ) if image_validation_error is not None: return image_validation_error @@ -827,7 +839,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 + ) if image_validation_error is not None: return image_validation_error @@ -965,7 +979,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 + ) 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..e31a04ad2d 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,65 @@ class TestHttpExport(BaseTestHttp): assert response.status_code == 400 assert ExportCase.select().count() == 0 + def _batch_export_with_image(self, image_path: str, role: str): + 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", + headers={"remote-user": role, "remote-role": role}, + json={ + "items": [ + { + "camera": "front_door", + "start_time": 110, + "end_time": 150, + "image_path": image_path, + } + ], + }, + ) + + return response, start_export_job + + def test_batch_export_restricted_role_rejects_unreadable_image_path(self): + self._insert_recording("rec-front", "front_door", 100, 400) + self.app.frigate_config.auth.roles["limited_user"] = ["front_door"] + + 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", + f"{CLIPS_DIR}/custom/thumb.jpg", + ): + response, _ = self._batch_export_with_image(image_path, "limited_user") + + assert response.status_code == 400, image_path + + def test_batch_export_restricted_role_accepts_own_camera_image_path(self): + self._insert_recording("rec-front", "front_door", 100, 400) + self.app.frigate_config.auth.roles["limited_user"] = ["front_door"] + image_path = f"{CLIPS_DIR}/review/thumb-front_door-123.456-abc.webp" + + response, start_export_job = self._batch_export_with_image( + image_path, "limited_user" + ) + + assert response.status_code == 202 + assert start_export_job.call_args.args[1].image_path == image_path + + def test_batch_export_unrestricted_roles_accept_custom_image_path(self): + self._insert_recording("rec-front", "front_door", 100, 400) + image_path = f"{CLIPS_DIR}/custom/thumb.jpg" + + for role in ("admin", "viewer"): + response, start_export_job = self._batch_export_with_image(image_path, role) + + assert response.status_code == 202, role + assert start_export_job.call_args.args[1].image_path == image_path + def test_batch_export_non_admin_can_queue(self): self._insert_recording("rec-front", "front_door", 100, 400)