check image_path against the caller's read access instead of the exported camera

Requiring the image to belong to the exported camera rejected custom images that a script saves under the clips directory for its own export pipeline. The path is now accepted whenever the caller's role could read it as media, so admins and unrestricted roles can use any image there and restricted roles are limited to their own cameras.
This commit is contained in:
Josh Hawkins
2026-10-09 07:57:06 -05:00
parent a60601d7c9
commit 4e61bcf82f
2 changed files with 49 additions and 47 deletions
+10 -13
View File
@@ -57,7 +57,7 @@ from frigate.api.defs.response.export_response import (
) )
from frigate.api.defs.response.generic_response import GenericResponse from frigate.api.defs.response.generic_response import GenericResponse
from frigate.api.defs.tags import Tags from frigate.api.defs.tags import Tags
from frigate.api.media_auth import MediaAuthResolution, resolve_media_uri from frigate.api.media_auth import deny_response_for_media_uri
from frigate.const import CLIPS_DIR, EXPORT_DIR from frigate.const import CLIPS_DIR, EXPORT_DIR
from frigate.jobs.export import ( from frigate.jobs.export import (
ExportJob, ExportJob,
@@ -134,22 +134,19 @@ def _validate_export_case(export_case_id: str | None) -> JSONResponse | None:
def _sanitize_existing_image( def _sanitize_existing_image(
request: Request, request: Request,
image_path: str | None, image_path: str | None,
camera: str,
) -> tuple[str | None, JSONResponse | None]: ) -> tuple[str | None, JSONResponse | None]:
if not image_path: if not image_path:
return None, None return None, None
existing_image = sanitize_contained_path(image_path, CLIPS_DIR) existing_image = sanitize_contained_path(image_path, CLIPS_DIR)
# CLIPS_DIR is shared by every camera, so the image must also belong to # CLIPS_DIR is shared by every camera, so the caller must also be allowed
# the camera being exported. # to read the image.
if existing_image is not None: if existing_image is not None and deny_response_for_media_uri(
resolution = resolve_media_uri( f"/clips/{quote(os.path.relpath(existing_image, CLIPS_DIR))}",
f"/clips/{os.path.relpath(existing_image, CLIPS_DIR)}", request.headers.get("remote-role"),
request.app.frigate_config, request.app.frigate_config,
) ):
if resolution != (MediaAuthResolution.CAMERA, camera):
existing_image = None existing_image = None
if existing_image is None: if existing_image is None:
@@ -695,7 +692,7 @@ def export_recordings_batch(
sanitized_images: list[str | None] = [] sanitized_images: list[str | None] = []
for item in body.items: for item in body.items:
existing_image, image_validation_error = _sanitize_existing_image( existing_image, image_validation_error = _sanitize_existing_image(
request, item.image_path, item.camera request, item.image_path
) )
if image_validation_error is not None: if image_validation_error is not None:
return image_validation_error return image_validation_error
@@ -843,7 +840,7 @@ def export_recording(
playback_source = body.source playback_source = body.source
friendly_name = body.name friendly_name = body.name
existing_image, image_validation_error = _sanitize_existing_image( existing_image, image_validation_error = _sanitize_existing_image(
request, body.image_path, camera_name request, body.image_path
) )
if image_validation_error is not None: if image_validation_error is not None:
return image_validation_error return image_validation_error
@@ -983,7 +980,7 @@ def export_recording_custom(
playback_source = body.source playback_source = body.source
friendly_name = body.name friendly_name = body.name
existing_image, image_validation_error = _sanitize_existing_image( existing_image, image_validation_error = _sanitize_existing_image(
request, body.image_path, camera_name request, body.image_path
) )
if image_validation_error is not None: if image_validation_error is not None:
return image_validation_error return image_validation_error
+36 -31
View File
@@ -950,17 +950,15 @@ class TestHttpExport(BaseTestHttp):
assert response.status_code == 400 assert response.status_code == 400
assert ExportCase.select().count() == 0 assert ExportCase.select().count() == 0
def test_batch_export_rejects_other_camera_image_path(self): def _batch_export_with_image(self, image_path: str, role: str):
self._insert_recording("rec-front", "front_door", 100, 400) with patch(
"frigate.api.export.start_export_job",
for image_path in ( side_effect=lambda _config, job: job.id,
f"{CLIPS_DIR}/review/thumb-backyard-123.456-abc.webp", ) as start_export_job:
f"{CLIPS_DIR}/thumbs/backyard/123.webp",
f"{CLIPS_DIR}/faces/someone/1.webp",
):
with AuthTestClient(self.app) as client: with AuthTestClient(self.app) as client:
response = client.post( response = client.post(
"/exports/batch", "/exports/batch",
headers={"remote-user": role, "remote-role": role},
json={ json={
"items": [ "items": [
{ {
@@ -973,36 +971,43 @@ class TestHttpExport(BaseTestHttp):
}, },
) )
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 assert response.status_code == 400, image_path
def test_batch_export_accepts_own_camera_image_path(self): def test_batch_export_restricted_role_accepts_own_camera_image_path(self):
self._insert_recording("rec-front", "front_door", 100, 400) 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"
with patch( response, start_export_job = self._batch_export_with_image(
"frigate.api.export.start_export_job", image_path, "limited_user"
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 response.status_code == 202
assert start_export_job.call_args.args[1].image_path == ( assert start_export_job.call_args.args[1].image_path == image_path
f"{CLIPS_DIR}/review/thumb-front_door-123.456-abc.webp"
) 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): def test_batch_export_non_admin_can_queue(self):
self._insert_recording("rec-front", "front_door", 100, 400) self._insert_recording("rec-front", "front_door", 100, 400)