Compare commits

...
Author SHA1 Message Date
Josh Hawkins 4e61bcf82f 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.
2026-10-09 07:57:06 -05:00
Josh Hawkins a60601d7c9 require export image_path to belong to the exported camera
image_path was only checked for containment under the clips directory, which every camera shares, so a user restricted to one camera could start an export of that camera and have the server copy another camera's image, or a file from an admin-only directory like faces, into the export thumbnail. The path now also has to resolve to the camera being exported, using the same classifier that gates media requests.
2026-10-09 07:41:28 -05:00
2 changed files with 79 additions and 3 deletions
+19 -3
View File
@@ -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
+60
View File
@@ -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)