Don't allow path traversal

frigate/api/media.py

@@ -59,19 +59,6 @@ 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,
@@ -393,9 +380,7 @@ async def submit_recording_snapshot_to_plus(
             )
 
         nd = cv2.imdecode(np.frombuffer(image_data, dtype=np.int8), cv2.IMREAD_COLOR)
-        await asyncio.to_thread(
-            request.app.frigate_config.plus_api.upload_image, nd, camera_name
-        )
+        request.app.frigate_config.plus_api.upload_image(nd, camera_name)
 
         return JSONResponse(
             content={
@@ -1228,7 +1213,7 @@ async def event_thumbnail(
         thumbnail_bytes,
         media_type=extension.get_mime_type(),
         headers={
-            "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}"
+            "Cache-Control": f"private, max-age={max_cache_age}"
             if event_complete
             else "no-store",
         },
@@ -1532,14 +1517,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 await preview_gif(request, event.camera, start_ts, end_ts)
+    return 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)],
 )
-async def preview_gif(
+def preview_gif(
     request: Request,
     camera_name: str,
     start_ts: float,
@@ -1602,8 +1587,7 @@ async def preview_gif(
             "-",
         ]
 
-        process = await asyncio.to_thread(
-            sp.run,
+        process = sp.run(
             ffmpeg_cmd,
             capture_output=True,
         )
@@ -1670,8 +1654,7 @@ async def preview_gif(
             "-",
         ]
 
-        process = await asyncio.to_thread(
-            sp.run,
+        process = sp.run(
             ffmpeg_cmd,
             input=str.encode("\n".join(selected_previews)),
             capture_output=True,
@@ -1690,7 +1673,7 @@ async def preview_gif(
         gif_bytes,
         media_type="image/gif",
         headers={
-            "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}",
+            "Cache-Control": f"private, max-age={max_cache_age}",
             "Content-Type": "image/gif",
         },
     )
@@ -1700,7 +1683,7 @@ async def preview_gif(
     "/{camera_name}/start/{start_ts}/end/{end_ts}/preview.mp4",
     dependencies=[Depends(require_camera_access)],
 )
-async def preview_mp4(
+def preview_mp4(
     request: Request,
     camera_name: str,
     start_ts: float,
@@ -1780,8 +1763,7 @@ async def preview_mp4(
             path,
         ]
 
-        process = await asyncio.to_thread(
-            sp.run,
+        process = sp.run(
             ffmpeg_cmd,
             capture_output=True,
         )
@@ -1845,8 +1827,7 @@ async def preview_mp4(
             path,
         ]
 
-        process = await asyncio.to_thread(
-            sp.run,
+        process = sp.run(
             ffmpeg_cmd,
             input=str.encode("\n".join(selected_previews)),
             capture_output=True,
@@ -1861,7 +1842,7 @@ async def preview_mp4(
 
     headers = {
         "Content-Description": "File Transfer",
-        "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}",
+        "Cache-Control": f"private, max-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
@@ -1899,9 +1880,9 @@ async def review_preview(
     )
 
     if format == "gif":
-        return await preview_gif(request, review.camera, start_ts, end_ts)
+        return preview_gif(request, review.camera, start_ts, end_ts)
     else:
-        return await preview_mp4(request, review.camera, start_ts, end_ts)
+        return preview_mp4(request, review.camera, start_ts, end_ts)
 
 
 @router.get(

frigate/comms/ws.py

@@ -34,7 +34,6 @@ from frigate.const import (
     UPDATE_REVIEW_DESCRIPTION,
     UPSERT_REVIEW_SEGMENT,
 )
-from frigate.models import User
 
 logger = logging.getLogger(__name__)
 
@@ -70,16 +69,11 @@ _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.
 
@@ -87,10 +81,6 @@ 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.
@@ -100,33 +90,16 @@ def _check_ws_authorization(
         return False
 
     # No role header: default to viewer (fail-closed)
-    roles = [r.strip() for r in role_header.split(separator)] if role_header else []
+    if role_header is None:
+        return topic in _WS_VIEWER_TOPICS
 
-    # Admin can send anything
+    # Check if any role is admin
+    roles = [r.strip() for r in role_header.split(separator)]
     if "admin" in roles:
         return True
 
-    # 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
+    # Non-admin: only viewer topics allowed
+    return topic in _WS_VIEWER_TOPICS
 
 
 class WebSocket(WebSocket_):  # type: ignore[misc]
@@ -158,8 +131,6 @@ 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:
@@ -181,11 +152,7 @@ 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,
-                    self.roles_config,
-                    self.camera_names,
+                    topic, role_header, self.role_separator
                 ):
                     logger.warning(
                         "Blocked unauthorized WebSocket message: topic=%s, role=%s",

frigate/test/test_ws_auth.py

@@ -11,16 +11,6 @@ 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):
@@ -171,124 +161,6 @@ 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()