mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-09 16:22:48 +03:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4e61bcf82f | ||
|
|
a60601d7c9 |
+19
-3
@@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
import datetime
|
import datetime
|
||||||
import logging
|
import logging
|
||||||
|
import os
|
||||||
import random
|
import random
|
||||||
import string
|
import string
|
||||||
import time
|
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.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 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,
|
||||||
@@ -130,6 +132,7 @@ def _validate_export_case(export_case_id: str | None) -> JSONResponse | None:
|
|||||||
|
|
||||||
|
|
||||||
def _sanitize_existing_image(
|
def _sanitize_existing_image(
|
||||||
|
request: Request,
|
||||||
image_path: str | None,
|
image_path: str | None,
|
||||||
) -> tuple[str | None, JSONResponse | None]:
|
) -> tuple[str | None, JSONResponse | None]:
|
||||||
if not image_path:
|
if not image_path:
|
||||||
@@ -137,6 +140,15 @@ def _sanitize_existing_image(
|
|||||||
|
|
||||||
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 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:
|
if existing_image is None:
|
||||||
return None, JSONResponse(
|
return None, JSONResponse(
|
||||||
content={"success": False, "message": "Invalid image path"},
|
content={"success": False, "message": "Invalid image path"},
|
||||||
@@ -680,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(
|
||||||
item.image_path
|
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
|
||||||
@@ -827,7 +839,9 @@ 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(body.image_path)
|
existing_image, image_validation_error = _sanitize_existing_image(
|
||||||
|
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
|
||||||
|
|
||||||
@@ -965,7 +979,9 @@ 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(body.image_path)
|
existing_image, image_validation_error = _sanitize_existing_image(
|
||||||
|
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
|
||||||
ffmpeg_input_args = body.ffmpeg_input_args
|
ffmpeg_input_args = body.ffmpeg_input_args
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import tempfile
|
|||||||
import zipfile
|
import zipfile
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from frigate.const import CLIPS_DIR
|
||||||
from frigate.jobs.export import (
|
from frigate.jobs.export import (
|
||||||
ExportJob,
|
ExportJob,
|
||||||
get_export_job_manager,
|
get_export_job_manager,
|
||||||
@@ -949,6 +950,65 @@ class TestHttpExport(BaseTestHttp):
|
|||||||
assert response.status_code == 400
|
assert response.status_code == 400
|
||||||
assert ExportCase.select().count() == 0
|
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):
|
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)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user