mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-26 16:34:43 +03:00
Josh HawkinsandGitHub
94e570017c
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / Synaptics Build (push) Blocked by required conditions
CI / Assemble and push default build (push) Blocked by required conditions
* check ownership before stopping a camera watch from chat The stop_camera_watch chat tool cancelled the active VLM watch job without checking who started it, so any authenticated user could cancel another user's job by calling /chat/completion directly. The tool now uses the same admin-or-creator check as DELETE /vlm/monitor, and chat-started jobs record their creator so non-admins can still stop their own. * Added a test that calls `_execute_start_camera_watch` and checks `start_vlm_watch_job` receives the caller's username, so removing that argument now fails it.
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
"""Tests for camera watch ownership in the chat tools."""
|
|
|
|
import asyncio
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from frigate.api.chat import _execute_start_camera_watch, _execute_stop_camera_watch
|
|
from frigate.jobs.vlm_watch import VLMWatchJob
|
|
|
|
|
|
def _request(user: str, role: str) -> SimpleNamespace:
|
|
return SimpleNamespace(headers={"remote-user": user, "remote-role": role})
|
|
|
|
|
|
class TestStopCameraWatch(unittest.TestCase):
|
|
def setUp(self):
|
|
job = VLMWatchJob(camera="front", condition="person", username="alice")
|
|
patch("frigate.api.chat.get_vlm_watch_job", return_value=job).start()
|
|
self.stop = patch(
|
|
"frigate.api.chat.stop_vlm_watch_job", return_value=True
|
|
).start()
|
|
self.addCleanup(patch.stopall)
|
|
|
|
def test_other_user_cannot_cancel(self):
|
|
result = _execute_stop_camera_watch(_request("bob", "viewer"))
|
|
|
|
self.assertFalse(result["success"])
|
|
self.stop.assert_not_called()
|
|
|
|
def test_creator_can_cancel(self):
|
|
result = _execute_stop_camera_watch(_request("alice", "viewer"))
|
|
|
|
self.assertTrue(result["success"])
|
|
self.stop.assert_called_once()
|
|
|
|
def test_admin_can_cancel(self):
|
|
result = _execute_stop_camera_watch(_request("bob", "admin"))
|
|
|
|
self.assertTrue(result["success"])
|
|
self.stop.assert_called_once()
|
|
|
|
|
|
class TestStartCameraWatch(unittest.TestCase):
|
|
@patch("frigate.api.chat.require_camera_access", new_callable=AsyncMock)
|
|
@patch("frigate.api.chat.start_vlm_watch_job", return_value="job1")
|
|
def test_records_caller_as_owner(self, start, _access):
|
|
request = _request("alice", "viewer")
|
|
request.app = MagicMock()
|
|
request.app.frigate_config.cameras = {"front": None}
|
|
request.app.genai_manager.chat_client.supports_vision = True
|
|
|
|
asyncio.run(
|
|
_execute_start_camera_watch(
|
|
request, {"camera": "front", "condition": "person"}
|
|
)
|
|
)
|
|
|
|
self.assertEqual(start.call_args.kwargs["username"], "alice")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|