diff --git a/frigate/api/chat.py b/frigate/api/chat.py index fa4510cf14..127d0ea741 100644 --- a/frigate/api/chat.py +++ b/frigate/api/chat.py @@ -45,6 +45,7 @@ from frigate.genai.prompts import ( ) from frigate.genai.utils import build_assistant_message_for_conversation from frigate.jobs.vlm_watch import ( + VLMWatchJob, get_vlm_watch_job, start_vlm_watch_job, stop_vlm_watch_job, @@ -765,7 +766,7 @@ async def _execute_tool_internal( elif tool_name == "start_camera_watch": return await _execute_start_camera_watch(request, arguments) elif tool_name == "stop_camera_watch": - return _execute_stop_camera_watch() + return _execute_stop_camera_watch(request) elif tool_name == "get_profile_status": return _execute_get_profile_status(request) elif tool_name == "get_recap": @@ -819,6 +820,7 @@ async def _execute_start_camera_watch( dispatcher=request.app.dispatcher, labels=labels, zones=zones, + username=request.headers.get("remote-user", ""), ) except RuntimeError as e: logger.exception("Failed to start VLM watch job: %s", e) @@ -834,7 +836,18 @@ async def _execute_start_camera_watch( } -def _execute_stop_camera_watch() -> dict[str, Any]: +def _can_cancel_watch_job(request: Request, job: VLMWatchJob) -> bool: + """Admins can cancel any watch job; other users only jobs they started.""" + role = request.headers.get("remote-role", "viewer") + username = request.headers.get("remote-user", "") + return role == "admin" or username == job.username + + +def _execute_stop_camera_watch(request: Request) -> dict[str, Any]: + job = get_vlm_watch_job() + if job is not None and not _can_cancel_watch_job(request, job): + return {"success": False, "message": "Not authorized to cancel this watch job."} + cancelled = stop_vlm_watch_job() if cancelled: return {"success": True, "message": "Watch job cancelled."} @@ -1548,11 +1561,7 @@ async def cancel_vlm_monitor(request: Request) -> JSONResponse: status_code=404, ) - role = request.headers.get("remote-role", "viewer") - username = request.headers.get("remote-user", "") - - # Admin can cancel any job; other users can only cancel jobs they started. - if role != "admin" and username != job.username: + if not _can_cancel_watch_job(request, job): return JSONResponse( content={ "success": False, diff --git a/frigate/test/test_chat_stop_camera_watch.py b/frigate/test/test_chat_stop_camera_watch.py new file mode 100644 index 0000000000..9b78ef1c64 --- /dev/null +++ b/frigate/test/test_chat_stop_camera_watch.py @@ -0,0 +1,44 @@ +"""Tests for the stop_camera_watch chat tool.""" + +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from frigate.api.chat import _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() + + +if __name__ == "__main__": + unittest.main()