Merge remote-tracking branch 'origin/master' into dev

Resolve conflicts in the export pipeline where dev's job-queue refactor
met master's chapter-metadata and security work.

- Unify chapter support under ChaptersEnum (none / recording_segments /
  review_items); the realtime stream-copy export selects the per-segment
  or per-review-item builder by the camera's configured mode. Thread
  chapters through ExportRecordingsBody -> _build_export_job -> ExportJob
  -> RecordingExporter.
- Keep master's creation_time/comment export metadata and fix a
  video_path duplication the textual merge introduced in the preview
  command.
- Move the chapters request field to ExportRecordingsBody (the single
  export endpoint) where it is actually honored.

Restore security fixes the automatic merge would have reverted:
- frigate/util/services.py: restore the #23493 rename to the public
  is_go2rtc_arbitrary_exec_allowed so create_config.py's dynamic-source
  exec guard imports and runs (the merge otherwise left a broken import).
- Preserve the export image-path ".." traversal check inside
  _sanitize_existing_image, applied to single/custom/batch exports.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

frigate/api/camera.py

@@ -147,6 +147,19 @@ def go2rtc_camera_stream(request: Request, stream_name: str):
 )
 def go2rtc_add_stream(request: Request, stream_name: str, src: str = ""):
     """Add or update a go2rtc stream configuration."""
+    if src and is_restricted_go2rtc_source(src):
+        logger.warning(
+            "Rejected go2rtc stream '%s' with restricted source type (echo/expr/exec)",
+            stream_name,
+        )
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "Restricted stream source type",
+            },
+            status_code=400,
+        )
+
     try:
         params = {"name": stream_name}
         if src:

frigate/api/defs/request/export_recordings_body.py

@@ -3,7 +3,10 @@ from typing import Optional, Union
 from pydantic import BaseModel, Field
 from pydantic.json_schema import SkipJsonSchema
 
-from frigate.record.export import PlaybackSourceEnum
+from frigate.record.export import (
+    ChaptersEnum,
+    PlaybackSourceEnum,
+)
 
 
 class ExportRecordingsBody(BaseModel):
@@ -18,6 +21,14 @@ class ExportRecordingsBody(BaseModel):
         max_length=30,
         description="ID of the export case to assign this export to",
     )
+    chapters: Optional[ChaptersEnum] = Field(
+        default=None,
+        title="Chapter mode",
+        description=(
+            "Optional chapter metadata to embed in the export. When omitted, "
+            "the camera's configured export chapter mode is used."
+        ),
+    )
 
 
 class ExportRecordingsCustomBody(BaseModel):

frigate/api/export.py

@@ -68,6 +68,7 @@ from frigate.jobs.export import (
 from frigate.models import Export, ExportCase, Previews, Recordings
 from frigate.record.export import (
     DEFAULT_TIME_LAPSE_FFMPEG_ARGS,
+    ChaptersEnum,
     PlaybackSourceEnum,
     validate_ffmpeg_args,
 )
@@ -128,6 +129,15 @@ def _validate_export_case(export_case_id: Optional[str]) -> Optional[JSONRespons
 def _sanitize_existing_image(
     image_path: Optional[str],
 ) -> tuple[Optional[str], Optional[JSONResponse]]:
+    # sanitize_filepath normalizes "\" to "/" but leaves ".." intact, so a path
+    # like "clips\..\..\etc/passwd" passes the CLIPS_DIR prefix check yet still
+    # escapes the directory once resolved. A valid snapshot path never uses "..".
+    if image_path and ".." in image_path:
+        return None, JSONResponse(
+            content={"success": False, "message": "Invalid image path"},
+            status_code=400,
+        )
+
     existing_image = sanitize_filepath(image_path) if image_path else None
 
     if existing_image and not existing_image.startswith(CLIPS_DIR):
@@ -254,6 +264,7 @@ def _build_export_job(
     ffmpeg_input_args: Optional[str] = None,
     ffmpeg_output_args: Optional[str] = None,
     cpu_fallback: bool = False,
+    chapters: Optional[ChaptersEnum] = None,
 ) -> ExportJob:
     return ExportJob(
         id=_generate_export_id(camera_name),
@@ -267,6 +278,7 @@ def _build_export_job(
         ffmpeg_input_args=ffmpeg_input_args,
         ffmpeg_output_args=ffmpeg_output_args,
         cpu_fallback=cpu_fallback,
+        chapters=chapters,
     )
 
 
@@ -725,6 +737,9 @@ def export_recordings_batch(
             sanitized_images[index],
             PlaybackSourceEnum.recordings,
             export_case_id,
+            chapters=request.app.frigate_config.cameras[
+                item.camera
+            ].record.export.chapters,
         )
         try:
             start_export_job(request.app.frigate_config, export_job)
@@ -803,6 +818,14 @@ def export_recording(
 
     export_case_id = body.export_case_id
 
+    # a chapters value in the request body overrides the camera's export config
+    camera_config = request.app.frigate_config.cameras[camera_name]
+    chapters = (
+        body.chapters
+        if body.chapters is not None
+        else camera_config.record.export.chapters
+    )
+
     # Attaching to an existing case requires admin. Single-export for
     # cameras the user can access is otherwise non-admin; we only gate
     # the case-attachment side effect.
@@ -839,6 +862,7 @@ def export_recording(
         existing_image,
         playback_source,
         export_case_id,
+        chapters=chapters,
     )
     try:
         start_export_job(request.app.frigate_config, export_job)

frigate/api/media.py

@@ -60,6 +60,19 @@ logger = logging.getLogger(__name__)
 router = APIRouter(tags=[Tags.media])
 
 
+def _resolve_cache_age(max_cache_age: int) -> int:
+    """Return max_cache_age as an int.
+
+    When a media handler is invoked directly by another handler instead of
+    through its route, FastAPI doesn't resolve the Query() default and
+    max_cache_age arrives as the Query object; fall back to its int default.
+    """
+    if isinstance(max_cache_age, int):
+        return max_cache_age
+
+    return max_cache_age.default
+
+
 @router.get("/{camera_name}", dependencies=[Depends(require_camera_access)])
 async def mjpeg_feed(
     request: Request,
@@ -413,7 +426,9 @@ async def submit_recording_snapshot_to_plus(
             )
 
         nd = cv2.imdecode(np.frombuffer(image_data, dtype=np.int8), cv2.IMREAD_COLOR)
-        request.app.frigate_config.plus_api.upload_image(nd, camera_name)
+        await asyncio.to_thread(
+            request.app.frigate_config.plus_api.upload_image, nd, camera_name
+        )
 
         return JSONResponse(
             content={
@@ -936,7 +951,7 @@ async def event_thumbnail(
         thumbnail_bytes,
         media_type=extension.get_mime_type(),
         headers={
-            "Cache-Control": f"private, max-age={max_cache_age}"
+            "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}"
             if event_complete
             else "no-store",
         },
@@ -1270,14 +1285,14 @@ async def event_preview(request: Request, event_id: str):
     end_ts = start_ts + (
         min(event.end_time - event.start_time, 20) if event.end_time else 20
     )
-    return preview_gif(request, event.camera, start_ts, end_ts)
+    return await preview_gif(request, event.camera, start_ts, end_ts)
 
 
 @router.get(
     "/{camera_name}/start/{start_ts}/end/{end_ts}/preview.gif",
     dependencies=[Depends(require_camera_access)],
 )
-def preview_gif(
+async def preview_gif(
     request: Request,
     camera_name: str,
     start_ts: float,
@@ -1340,7 +1355,8 @@ def preview_gif(
             "-",
         ]
 
-        process = sp.run(
+        process = await asyncio.to_thread(
+            sp.run,
             ffmpeg_cmd,
             capture_output=True,
         )
@@ -1419,7 +1435,8 @@ def preview_gif(
             "-",
         ]
 
-        process = sp.run(
+        process = await asyncio.to_thread(
+            sp.run,
             ffmpeg_cmd,
             input=str.encode("\n".join(selected_previews)),
             capture_output=True,
@@ -1438,7 +1455,7 @@ def preview_gif(
         gif_bytes,
         media_type="image/gif",
         headers={
-            "Cache-Control": f"private, max-age={max_cache_age}",
+            "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}",
             "Content-Type": "image/gif",
         },
     )
@@ -1448,7 +1465,7 @@ def preview_gif(
     "/{camera_name}/start/{start_ts}/end/{end_ts}/preview.mp4",
     dependencies=[Depends(require_camera_access)],
 )
-def preview_mp4(
+async def preview_mp4(
     request: Request,
     camera_name: str,
     start_ts: float,
@@ -1528,7 +1545,8 @@ def preview_mp4(
             path,
         ]
 
-        process = sp.run(
+        process = await asyncio.to_thread(
+            sp.run,
             ffmpeg_cmd,
             capture_output=True,
         )
@@ -1604,7 +1622,8 @@ def preview_mp4(
             path,
         ]
 
-        process = sp.run(
+        process = await asyncio.to_thread(
+            sp.run,
             ffmpeg_cmd,
             input=str.encode("\n".join(selected_previews)),
             capture_output=True,
@@ -1619,7 +1638,7 @@ def preview_mp4(
 
     headers = {
         "Content-Description": "File Transfer",
-        "Cache-Control": f"private, max-age={max_cache_age}",
+        "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}",
         "Content-Type": "video/mp4",
         "Content-Length": str(os.path.getsize(path)),
         # nginx: https://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_ignore_headers
@@ -1657,9 +1676,9 @@ async def review_preview(
     )
 
     if format == "gif":
-        return preview_gif(request, review.camera, start_ts, end_ts)
+        return await preview_gif(request, review.camera, start_ts, end_ts)
     else:
-        return preview_mp4(request, review.camera, start_ts, end_ts)
+        return await preview_mp4(request, review.camera, start_ts, end_ts)
 
 
 @router.get(

frigate/comms/ws.py

@@ -72,11 +72,16 @@ _WS_VIEWER_TOPICS = frozenset(
     }
 )
 
+# Camera-scoped command topics a camera-authorized (non-admin) user may send.
+_WS_CAMERA_COMMAND_TOPICS = frozenset({"ptz"})
+
 
 def _check_ws_authorization(
     topic: str,
     role_header: str | None,
     separator: str,
+    roles_config: dict[str, list[str]] | None = None,
+    camera_names: set[str] | None = None,
 ) -> bool:
     """Check if a WebSocket message is authorized.
 
@@ -84,6 +89,10 @@ def _check_ws_authorization(
         topic: The message topic.
         role_header: The HTTP_REMOTE_ROLE header value, or None.
         separator: The role separator character from proxy config.
+        roles_config: The auth.roles mapping (role -> allowed cameras), used to
+            authorize camera-scoped commands for non-admin users.
+        camera_names: All configured camera names, used to resolve a role's
+            allowed cameras.
 
     Returns:
         True if authorized, False if blocked.
@@ -93,16 +102,33 @@ def _check_ws_authorization(
         return False
 
     # No role header: default to viewer (fail-closed)
-    if role_header is None:
-        return topic in _WS_VIEWER_TOPICS
+    roles = [r.strip() for r in role_header.split(separator)] if role_header else []
 
-    # Check if any role is admin
-    roles = [r.strip() for r in role_header.split(separator)]
+    # Admin can send anything
     if "admin" in roles:
         return True
 
-    # Non-admin: only viewer topics allowed
-    return topic in _WS_VIEWER_TOPICS
+    # Read-only topics any authenticated user can send
+    if topic in _WS_VIEWER_TOPICS:
+        return True
+
+    # Camera-scoped command like "<camera>/ptz": allow when the user's role(s)
+    # grant access to that camera.
+    parts = topic.split("/")
+    if (
+        roles_config is not None
+        and len(parts) == 2
+        and parts[1] in _WS_CAMERA_COMMAND_TOPICS
+    ):
+        allowed: set[str] = set()
+        # No role header maps to the default viewer role (e.g. proxy-only setups)
+        for role in roles or ["viewer"]:
+            allowed.update(
+                User.get_allowed_cameras(role, roles_config, camera_names or set())
+            )
+        return parts[0] in allowed
+
+    return False
 
 
 # ---- Outbound filtering ---------------------------------------------------
@@ -449,6 +475,8 @@ class WebSocketClient(Communicator):
         class _WebSocketHandler(WebSocket):
             receiver = self._dispatcher
             role_separator = self.config.proxy.separator or ","
+            roles_config = self.config.auth.roles
+            camera_names = set(self.config.cameras.keys())
 
             def received_message(self, message: WebSocket.received_message) -> None:  # type: ignore[name-defined]
                 try:
@@ -470,7 +498,11 @@ class WebSocketClient(Communicator):
                     self.environ.get("HTTP_REMOTE_ROLE") if self.environ else None
                 )
                 if self.environ is not None and not _check_ws_authorization(
-                    topic, role_header, self.role_separator
+                    topic,
+                    role_header,
+                    self.role_separator,
+                    self.roles_config,
+                    self.camera_names,
                 ):
                     logger.warning(
                         "Blocked unauthorized WebSocket message: topic=%s, role=%s",

frigate/config/camera/record.py

@@ -9,6 +9,7 @@ from frigate.review.types import SeverityEnum
 from ..base import FrigateBaseModel
 
 __all__ = [
+    "ChaptersEnum",
     "RecordConfig",
     "RecordExportConfig",
     "RecordPreviewConfig",
@@ -86,6 +87,12 @@ class RecordPreviewConfig(FrigateBaseModel):
     )
 
 
+class ChaptersEnum(str, Enum):
+    none = "none"
+    recording_segments = "recording_segments"
+    review_items = "review_items"
+
+
 class RecordExportConfig(FrigateBaseModel):
     hwaccel_args: Union[str, list[str]] = Field(
         default="auto",
@@ -98,6 +105,10 @@ class RecordExportConfig(FrigateBaseModel):
         title="Maximum concurrent exports",
         description="Maximum number of export jobs to process at the same time.",
     )
+    chapters: ChaptersEnum = Field(
+        default=ChaptersEnum.review_items,
+        title="Chapter metadata to embed in exported recordings",
+    )
 
 
 class RecordConfig(FrigateBaseModel):

frigate/jobs/export.py

@@ -13,6 +13,7 @@ from peewee import DoesNotExist
 
 from frigate.comms.inter_process import InterProcessRequestor
 from frigate.config import FrigateConfig
+from frigate.config.camera.record import ChaptersEnum
 from frigate.const import UPDATE_JOB_STATE
 from frigate.jobs.job import Job
 from frigate.models import Export
@@ -55,6 +56,7 @@ class ExportJob(Job):
     ffmpeg_input_args: Optional[str] = None
     ffmpeg_output_args: Optional[str] = None
     cpu_fallback: bool = False
+    chapters: Optional[ChaptersEnum] = None
     current_step: str = "queued"
     progress_percent: float = 0.0
 
@@ -343,6 +345,7 @@ class ExportJobManager:
             job.ffmpeg_input_args,
             job.ffmpeg_output_args,
             job.cpu_fallback,
+            job.chapters,
             on_progress=self._make_progress_callback(job),
         )
 

frigate/record/export.py

@@ -17,6 +17,7 @@ import pytz  # type: ignore[import-untyped]
 from peewee import DoesNotExist
 
 from frigate.config import FfmpegConfig, FrigateConfig
+from frigate.config.camera.record import ChaptersEnum
 from frigate.const import (
     CACHE_DIR,
     CLIPS_DIR,
@@ -217,6 +218,7 @@ class RecordingExporter(threading.Thread):
         ffmpeg_input_args: Optional[str] = None,
         ffmpeg_output_args: Optional[str] = None,
         cpu_fallback: bool = False,
+        chapters: Optional[ChaptersEnum] = None,
         on_progress: Optional[Callable[[str, float], None]] = None,
     ) -> None:
         super().__init__()
@@ -232,6 +234,7 @@ class RecordingExporter(threading.Thread):
         self.ffmpeg_input_args = ffmpeg_input_args
         self.ffmpeg_output_args = ffmpeg_output_args
         self.cpu_fallback = cpu_fallback
+        self.chapters = chapters
         self.on_progress = on_progress
 
         # ensure export thumb dir
@@ -509,6 +512,74 @@ class RecordingExporter(threading.Thread):
 
         return meta_path
 
+    def _build_recording_segment_chapter_metadata_file(
+        self, recordings: list
+    ) -> Optional[str]:
+        """Write an FFmpeg metadata file with one chapter per recording segment.
+
+        Each chapter's title is the segment's wallclock start time in
+        strict ISO 8601 form so a viewer can map any point in the
+        export's playback timeline back to real-world time without
+        OCR-ing a burnt-in timestamp. Chapter offsets are computed in
+        *output time*: the VOD endpoint concatenates recording clips
+        back-to-back, so wall-clock gaps between recordings collapse in
+        the produced video. Returns ``None`` when there are no
+        recordings or every segment is empty after clipping.
+        """
+        if not recordings:
+            return None
+
+        tz_name = self.config.ui.timezone
+        tz: Optional[datetime.tzinfo] = None
+        if tz_name:
+            try:
+                tz = pytz.timezone(tz_name)
+            except pytz.UnknownTimeZoneError:
+                tz = None
+        if tz is None:
+            tz = datetime.timezone.utc
+
+        chapter_blocks: list[str] = []
+        output_offset_ms = 0
+        for rec in recordings:
+            clipped_start = max(float(rec.start_time), float(self.start_time))
+            clipped_end = min(float(rec.end_time), float(self.end_time))
+            if clipped_end <= clipped_start:
+                continue
+
+            duration_ms = int(round((clipped_end - clipped_start) * 1000))
+            if duration_ms <= 0:
+                continue
+
+            title = datetime.datetime.fromtimestamp(clipped_start, tz=tz).isoformat(
+                timespec="seconds"
+            )
+            chapter_blocks.append(
+                "[CHAPTER]\n"
+                "TIMEBASE=1/1000\n"
+                f"START={output_offset_ms}\n"
+                f"END={output_offset_ms + duration_ms}\n"
+                f"title={title}"
+            )
+            output_offset_ms += duration_ms
+
+        if not chapter_blocks:
+            return None
+
+        meta_path = self._chapter_metadata_path()
+        try:
+            with open(meta_path, "w", encoding="utf-8") as f:
+                f.write(";FFMETADATA1\n")
+                f.write("\n".join(chapter_blocks))
+                f.write("\n")
+        except OSError:
+            logger.exception(
+                "Failed to write chapter metadata file for export %s", self.export_id
+            )
+            return None
+
+        return meta_path
+
     def save_thumbnail(self, id: str) -> str:
         thumb_path = os.path.join(CLIPS_DIR, f"export/{id}.webp")
 
@@ -672,7 +743,18 @@ class RecordingExporter(threading.Thread):
                 )
             ).split(" ")
         else:
-            chapters_path = self._build_chapter_metadata_file(recordings)
+            # Realtime/stream-copy export. Embed chapter metadata according to
+            # the camera's configured chapter mode: per-recording-segment
+            # timestamps or per-review-item titles.
+            if self.chapters == ChaptersEnum.recording_segments:
+                chapters_path = self._build_recording_segment_chapter_metadata_file(
+                    recordings
+                )
+            elif self.chapters == ChaptersEnum.review_items:
+                chapters_path = self._build_chapter_metadata_file(recordings)
+            else:
+                chapters_path = None
+
             chapter_args = (
                 f" -i {chapters_path} -map 0 -dn -map_metadata 1"
                 if chapters_path
@@ -684,7 +766,19 @@ class RecordingExporter(threading.Thread):
 
         # add metadata
         title = f"Frigate Recording for {self.camera}, {self.get_datetime_from_timestamp(self.start_time)} - {self.get_datetime_from_timestamp(self.end_time)}"
-        ffmpeg_cmd.extend(["-metadata", f"title={title}"])
+        creation_time = datetime.datetime.fromtimestamp(
+            self.start_time, tz=datetime.timezone.utc
+        ).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+        ffmpeg_cmd.extend(
+            [
+                "-metadata",
+                f"title={title}",
+                "-metadata",
+                f"creation_time={creation_time}",
+                "-metadata",
+                f"comment=Camera: {self.camera}",
+            ]
+        )
 
         ffmpeg_cmd.append(video_path)
 
@@ -770,18 +864,32 @@ class RecordingExporter(threading.Thread):
                     self.config.ffmpeg.ffmpeg_path,
                     hwaccel_args,
                     f"{self.ffmpeg_input_args} {TIMELAPSE_DATA_INPUT_ARGS} {ffmpeg_input}".strip(),
-                    f"{self.ffmpeg_output_args} -movflags +faststart {video_path}".strip(),
+                    f"{self.ffmpeg_output_args} -movflags +faststart".strip(),
                     EncodeTypeEnum.timelapse,
                 )
             ).split(" ")
         else:
             ffmpeg_cmd = (
-                f"{self.config.ffmpeg.ffmpeg_path} -hide_banner {ffmpeg_input} {codec} -movflags +faststart {video_path}"
+                f"{self.config.ffmpeg.ffmpeg_path} -hide_banner {ffmpeg_input} {codec} -movflags +faststart"
             ).split(" ")
 
         # add metadata
         title = f"Frigate Preview for {self.camera}, {self.get_datetime_from_timestamp(self.start_time)} - {self.get_datetime_from_timestamp(self.end_time)}"
-        ffmpeg_cmd.extend(["-metadata", f"title={title}"])
+        creation_time = datetime.datetime.fromtimestamp(
+            self.start_time, tz=datetime.timezone.utc
+        ).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+        ffmpeg_cmd.extend(
+            [
+                "-metadata",
+                f"title={title}",
+                "-metadata",
+                f"creation_time={creation_time}",
+                "-metadata",
+                f"comment=Camera: {self.camera}",
+            ]
+        )
+
+        ffmpeg_cmd.append(video_path)
 
         return ffmpeg_cmd, playlist_lines
 

frigate/record/maintainer.py

@@ -650,6 +650,8 @@ class RecordingMaintainer(threading.Thread):
                     "copy",
                     "-movflags",
                     "+faststart",
+                    "-metadata",
+                    f"creation_time={start_time.strftime('%Y-%m-%dT%H:%M:%S.%fZ')}",
                     file_path,
                     stderr=asyncio.subprocess.PIPE,
                     stdout=asyncio.subprocess.DEVNULL,

frigate/test/test_ws_auth.py

@@ -11,6 +11,16 @@ class TestCheckWsAuthorization(unittest.TestCase):
 
     DEFAULT_SEPARATOR = ","
 
+    # admin/viewer are reserved and always map to all cameras (empty list);
+    # custom roles map to a specific set of cameras.
+    ROLES_CONFIG = {
+        "admin": [],
+        "viewer": [],
+        "yard": ["front_door", "backyard"],
+        "garage_only": ["garage"],
+    }
+    CAMERA_NAMES = {"front_door", "backyard", "garage"}
+
     # --- IPC topic blocking (unconditional, regardless of role) ---
 
     def test_ipc_topic_blocked_for_admin(self):
@@ -161,6 +171,124 @@ class TestCheckWsAuthorization(unittest.TestCase):
             _check_ws_authorization("onConnect", None, self.DEFAULT_SEPARATOR)
         )
 
+    # --- Camera-scoped PTZ access (non-admin with camera access) ---
+
+    def test_viewer_can_ptz_camera_with_access(self):
+        # viewer maps to all cameras, so PTZ is allowed
+        self.assertTrue(
+            _check_ws_authorization(
+                "front_door/ptz",
+                "viewer",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_custom_role_can_ptz_assigned_camera(self):
+        self.assertTrue(
+            _check_ws_authorization(
+                "front_door/ptz",
+                "yard",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_custom_role_blocked_from_ptz_unassigned_camera(self):
+        self.assertFalse(
+            _check_ws_authorization(
+                "garage/ptz",
+                "yard",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_multiple_roles_union_camera_access_for_ptz(self):
+        # "yard" covers front_door/backyard, "garage_only" covers garage
+        self.assertTrue(
+            _check_ws_authorization(
+                "garage/ptz",
+                "yard,garage_only",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_unknown_role_blocked_from_ptz(self):
+        self.assertFalse(
+            _check_ws_authorization(
+                "front_door/ptz",
+                "nonexistent",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_no_role_header_treated_as_viewer_for_ptz(self):
+        # proxy-only / auth-disabled setups default to the viewer role
+        self.assertTrue(
+            _check_ws_authorization(
+                "front_door/ptz",
+                None,
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_camera_access_does_not_grant_set_commands(self):
+        # camera access enables PTZ only, not config-changing "set" commands
+        self.assertFalse(
+            _check_ws_authorization(
+                "front_door/detect/set",
+                "yard",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_ptz_autotracker_stays_admin_only(self):
+        # ptz_autotracker is a config toggle, not a live-view action
+        self.assertFalse(
+            _check_ws_authorization(
+                "front_door/ptz_autotracker/set",
+                "viewer",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_admin_can_ptz_any_camera_with_config(self):
+        self.assertTrue(
+            _check_ws_authorization(
+                "garage/ptz",
+                "admin",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
+    def test_ipc_topic_still_blocked_with_camera_access(self):
+        # IPC topics are blocked unconditionally, even with camera access
+        self.assertFalse(
+            _check_ws_authorization(
+                UPDATE_CAMERA_ACTIVITY,
+                "viewer",
+                self.DEFAULT_SEPARATOR,
+                self.ROLES_CONFIG,
+                self.CAMERA_NAMES,
+            )
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

frigate/util/services.py

@@ -790,7 +790,7 @@ def get_hailo_temps() -> dict[str, float]:
     return temps
 
 
-def _go2rtc_arbitrary_exec_allowed() -> bool:
+def is_go2rtc_arbitrary_exec_allowed() -> bool:
     """Read the GO2RTC_ALLOW_ARBITRARY_EXEC override from env, docker
     secrets, or the Home Assistant add-on options file."""
     raw: Optional[str] = None
@@ -822,7 +822,7 @@ def is_restricted_go2rtc_source(stream_source: str) -> bool:
     and the GO2RTC_ALLOW_ARBITRARY_EXEC override is not set."""
     if not stream_source.strip().startswith(("echo:", "expr:", "exec:")):
         return False
-    return not _go2rtc_arbitrary_exec_allowed()
+    return not is_go2rtc_arbitrary_exec_allowed()
 
 
 def ffprobe_stream(ffmpeg, path: str, detailed: bool = False) -> sp.CompletedProcess: