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.
This commit is contained in:
Josh Hawkins
2026-10-09 07:41:28 -05:00
parent 2273cf2d50
commit a60601d7c9
2 changed files with 77 additions and 3 deletions
+22 -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 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
+55
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,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)