sanitize user-supplied path components (#23990)

sanitize_filename leaves ".." intact and collapses variants like "..:" and "..*" to "..", so filesystem paths built from face names, classification model/category names, image ids, and trigger data could escape their base directory. Route every such site through new frigate/util/path.py helpers (safe_join, sanitize_path_component, sanitize_contained_path), which reject traversal and verify containment.

Worst case was DELETE /classification/{name}, which rmtree'd /media/frigate and /config while returning 200.

Important to note that all affected endpoints already require admin permission, so this sould be considered hardening rather than fixing exploitable code.

docs/docs/integrations/mqtt.md

@@ -292,7 +292,9 @@ Topic with the currently active profile name. Published value is the profile nam
 
 ### `frigate/notifications/set`
 
-Topic to turn notifications on and off. Expected values are `ON` and `OFF`.
+Topic to turn notifications on and off for all cameras. Expected values are `ON` and `OFF`.
+
+Only available when notifications are enabled in the config. Not persisted across Frigate restarts.
 
 ### `frigate/notifications/state`
 
@@ -570,16 +572,20 @@ Topic with current state of the Birdseye mode for a camera. Published values are
 
 ### `frigate/<camera_name>/notifications/set`
 
-Topic to turn notifications on and off. Expected values are `ON` and `OFF`.
+Topic to turn notifications for a camera on and off. Expected values are `ON` and `OFF`.
+
+`ON` is ignored unless notifications are enabled in the config for the camera. This is not persisted across Frigate restarts. It is the same control the UI labels **Suspend until restart**.
 
 ### `frigate/<camera_name>/notifications/state`
 
-Topic with current state of notifications. Published values are `ON` and `OFF`.
+Topic with current state of notifications. Published values are `ON` and `OFF`. This is the authoritative topic for whether a camera will notify.
 
 ### `frigate/<camera_name>/notifications/suspend`
 
-Topic to suspend notifications for a certain number of minutes. Expected value is an integer.
+Topic to suspend notifications for a certain number of minutes. Expected value is an integer. Separate from `notifications/set`: it does not change `notifications/state`, and is ignored while notifications are off.
 
 ### `frigate/<camera_name>/notifications/suspended`
 
-Topic with timestamp that notifications are suspended until. Published value is a UNIX timestamp, or 0 if notifications are not suspended.
+Topic with timestamp that notifications are suspended until. Published value is a UNIX timestamp, or 0 if there is no timed suspension.
+
+`0` does not mean notifications are enabled: `notifications/set` `OFF` clears the timed suspension, so this publishes `0` while `notifications/state` is `OFF`.

frigate/api/classification.py

@@ -11,7 +11,6 @@ from typing import Any
 import cv2
 from fastapi import APIRouter, Depends, Request, UploadFile
 from fastapi.responses import JSONResponse
-from pathvalidate import sanitize_filename
 from peewee import DoesNotExist
 from playhouse.shortcuts import model_to_dict
 
@@ -43,12 +42,21 @@ from frigate.util.classification import (
     write_training_metadata,
 )
 from frigate.util.file import get_event_snapshot
+from frigate.util.path import safe_join, sanitize_path_component
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(tags=[Tags.classification])
 
 
+def invalid_name_response(value: str) -> JSONResponse:
+    """Response for a name that cannot be used as a path component."""
+    return JSONResponse(
+        content={"success": False, "message": f"Invalid name: {value}"},
+        status_code=400,
+    )
+
+
 @router.get(
     "/faces",
     response_model=FacesResponse,
@@ -98,9 +106,7 @@ def reclassify_face(request: Request, body: dict = None):
         )
 
     json: dict[str, Any] = body or {}
-    training_file = os.path.join(
-        FACE_DIR, f"train/{sanitize_filename(json.get('training_file', ''))}"
-    )
+    training_file = safe_join(FACE_DIR, "train", json.get("training_file", ""))
 
     if not training_file or not os.path.isfile(training_file):
         return JSONResponse(
@@ -150,8 +156,10 @@ def train_face(request: Request, name: str, body: dict = None):
         )
 
     json: dict[str, Any] = body or {}
-    training_file_name = sanitize_filename(json.get("training_file", ""))
-    training_file = os.path.join(FACE_DIR, f"train/{training_file_name}")
+    training_file_name = json.get("training_file", "")
+    training_file = (
+        safe_join(FACE_DIR, "train", training_file_name) if training_file_name else None
+    )
     event_id = json.get("event_id")
 
     if not training_file_name and not event_id:
@@ -165,7 +173,9 @@ def train_face(request: Request, name: str, body: dict = None):
             status_code=400,
         )
 
-    if training_file_name and not os.path.isfile(training_file):
+    if training_file_name and (
+        training_file is None or not os.path.isfile(training_file)
+    ):
         return JSONResponse(
             content=(
                 {
@@ -176,9 +186,13 @@ def train_face(request: Request, name: str, body: dict = None):
             status_code=404,
         )
 
-    sanitized_name = sanitize_filename(name)
+    sanitized_name = sanitize_path_component(name)
+    new_file_folder = safe_join(FACE_DIR, name)
+
+    if sanitized_name is None or new_file_folder is None:
+        return invalid_name_response(name)
+
     new_name = f"{sanitized_name}-{datetime.datetime.now().timestamp()}.webp"
-    new_file_folder = os.path.join(FACE_DIR, f"{sanitized_name}")
 
     os.makedirs(new_file_folder, exist_ok=True)
 
@@ -261,9 +275,12 @@ async def create_face(request: Request, name: str):
             content={"message": "Face recognition is not enabled.", "success": False},
         )
 
-    os.makedirs(
-        os.path.join(FACE_DIR, sanitize_filename(name.replace(" ", "_"))), exist_ok=True
-    )
+    face_folder = safe_join(FACE_DIR, name.replace(" ", "_"))
+
+    if face_folder is None:
+        return invalid_name_response(name)
+
+    os.makedirs(face_folder, exist_ok=True)
     return JSONResponse(
         status_code=200,
         content={"success": False, "message": "Successfully created face folder."},
@@ -287,6 +304,9 @@ def register_face(request: Request, name: str, file: UploadFile):
             content={"message": "Face recognition is not enabled.", "success": False},
         )
 
+    if sanitize_path_component(name) is None:
+        return invalid_name_response(name)
+
     context: EmbeddingsContext = request.app.embeddings
     result = None if context is None else context.register_face(name, file.file.read())
 
@@ -356,8 +376,8 @@ def reclassify_face_image(request: Request, name: str, body: dict = None):
         )
 
     json: dict[str, Any] = body or {}
-    image_id = sanitize_filename(json.get("id", ""))
-    new_name = sanitize_filename(json.get("new_name", ""))
+    image_id = sanitize_path_component(json.get("id", ""))
+    new_name = sanitize_path_component(json.get("new_name", ""))
 
     if not image_id or not new_name:
         return JSONResponse(
@@ -381,7 +401,12 @@ def reclassify_face_image(request: Request, name: str, body: dict = None):
             status_code=400,
         )
 
-    source_folder = os.path.join(FACE_DIR, sanitize_filename(name))
+    source_folder = safe_join(FACE_DIR, name)
+    target_folder = safe_join(FACE_DIR, new_name)
+
+    if source_folder is None or target_folder is None:
+        return invalid_name_response(name)
+
     source_file = os.path.join(source_folder, image_id)
 
     if not os.path.isfile(source_file):
@@ -396,7 +421,6 @@ def reclassify_face_image(request: Request, name: str, body: dict = None):
         )
 
     target_filename = f"{new_name}-{datetime.datetime.now().timestamp()}.webp"
-    target_folder = os.path.join(FACE_DIR, new_name)
 
     os.makedirs(target_folder, exist_ok=True)
     shutil.move(source_file, os.path.join(target_folder, target_filename))
@@ -430,8 +454,19 @@ def deregister_faces(request: Request, name: str, body: DeleteFaceImagesBody):
             content={"message": "Face recognition is not enabled.", "success": False},
         )
 
+    sanitized_name = sanitize_path_component(name)
+
+    if sanitized_name is None:
+        return invalid_name_response(name)
+
+    sanitized_ids = [
+        component
+        for component in map(sanitize_path_component, body.ids)
+        if component is not None
+    ]
+
     context: EmbeddingsContext = request.app.embeddings
-    context.delete_face_ids(name, map(lambda file: sanitize_filename(file), body.ids))
+    context.delete_face_ids(sanitized_name, sanitized_ids)
     return JSONResponse(
         content=({"success": True, "message": "Successfully deleted faces."}),
         status_code=200,
@@ -642,7 +677,11 @@ def transcribe_audio(request: Request, body: AudioTranscriptionBody):
 def get_classification_dataset(name: str):
     dataset_dict: dict[str, list[str]] = {}
 
-    dataset_dir = os.path.join(CLIPS_DIR, sanitize_filename(name), "dataset")
+    sanitized_name = sanitize_path_component(name)
+    dataset_dir = safe_join(CLIPS_DIR, name, "dataset")
+
+    if sanitized_name is None or dataset_dir is None:
+        return invalid_name_response(name)
 
     if not os.path.exists(dataset_dir):
         return JSONResponse(
@@ -664,8 +703,8 @@ def get_classification_dataset(name: str):
             dataset_dict[category_name].append(file)
 
     # Get training metadata
-    metadata = read_training_metadata(sanitize_filename(name))
-    current_image_count = get_dataset_image_count(sanitize_filename(name))
+    metadata = read_training_metadata(sanitized_name)
+    current_image_count = get_dataset_image_count(sanitized_name)
 
     if metadata is None:
         training_metadata = {
@@ -729,8 +768,8 @@ def get_custom_attributes(
         if object_type is not None and object_type not in model_objects:
             continue
 
-        dataset_dir = os.path.join(CLIPS_DIR, sanitize_filename(model_key), "dataset")
-        if not os.path.exists(dataset_dir):
+        dataset_dir = safe_join(CLIPS_DIR, model_key, "dataset")
+        if dataset_dir is None or not os.path.exists(dataset_dir):
             continue
 
         attributes = []
@@ -760,7 +799,10 @@ def get_custom_attributes(
     The name must exist in the classification models. Returns a success message or an error if the name is invalid.""",
 )
 def get_classification_images(name: str):
-    train_dir = os.path.join(CLIPS_DIR, sanitize_filename(name), "train")
+    train_dir = safe_join(CLIPS_DIR, name, "train")
+
+    if train_dir is None:
+        return invalid_name_response(name)
 
     if not os.path.exists(train_dir):
         return JSONResponse(status_code=200, content=[])
@@ -831,15 +873,17 @@ def delete_classification_dataset_images(
 
     json: dict[str, Any] = body or {}
     list_of_ids = json.get("ids", "")
-    folder = os.path.join(
-        CLIPS_DIR, sanitize_filename(name), "dataset", sanitize_filename(category)
-    )
+    sanitized_name = sanitize_path_component(name)
+    folder = safe_join(CLIPS_DIR, name, "dataset", category)
+
+    if sanitized_name is None or folder is None:
+        return invalid_name_response(name)
 
     deleted_count = 0
     for id in list_of_ids:
-        file_path = os.path.join(folder, sanitize_filename(id))
+        file_path = safe_join(folder, id)
 
-        if os.path.isfile(file_path):
+        if file_path and os.path.isfile(file_path):
             os.unlink(file_path)
             deleted_count += 1
 
@@ -850,7 +894,6 @@ def delete_classification_dataset_images(
     # This ensures the dataset is marked as changed after deletion
     # (even if the total count happens to be the same after adding and deleting)
     if deleted_count > 0:
-        sanitized_name = sanitize_filename(name)
         metadata = read_training_metadata(sanitized_name)
         if metadata:
             last_count = metadata.get("last_training_image_count", 0)
@@ -888,8 +931,8 @@ def reclassify_classification_image(
         )
 
     json: dict[str, Any] = body or {}
-    image_id = sanitize_filename(json.get("id", ""))
-    new_category = sanitize_filename(json.get("new_category", ""))
+    image_id = sanitize_path_component(json.get("id", ""))
+    new_category = sanitize_path_component(json.get("new_category", ""))
 
     if not image_id or not new_category:
         return JSONResponse(
@@ -913,10 +956,13 @@ def reclassify_classification_image(
             status_code=400,
         )
 
-    sanitized_name = sanitize_filename(name)
-    source_folder = os.path.join(
-        CLIPS_DIR, sanitized_name, "dataset", sanitize_filename(category)
-    )
+    sanitized_name = sanitize_path_component(name)
+    source_folder = safe_join(CLIPS_DIR, name, "dataset", category)
+    target_folder = safe_join(CLIPS_DIR, name, "dataset", new_category)
+
+    if sanitized_name is None or source_folder is None or target_folder is None:
+        return invalid_name_response(name)
+
     source_file = os.path.join(source_folder, image_id)
 
     if not os.path.isfile(source_file):
@@ -933,7 +979,6 @@ def reclassify_classification_image(
     random_id = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
     timestamp = datetime.datetime.now().timestamp()
     new_name = f"{new_category}-{timestamp}-{random_id}.png"
-    target_folder = os.path.join(CLIPS_DIR, sanitized_name, "dataset", new_category)
 
     os.makedirs(target_folder, exist_ok=True)
 
@@ -983,7 +1028,7 @@ def rename_classification_category(
         )
 
     json: dict[str, Any] = body or {}
-    new_category = sanitize_filename(json.get("new_category", ""))
+    new_category = sanitize_path_component(json.get("new_category", ""))
 
     if not new_category:
         return JSONResponse(
@@ -996,12 +1041,12 @@ def rename_classification_category(
             status_code=400,
         )
 
-    old_folder = os.path.join(
-        CLIPS_DIR, sanitize_filename(name), "dataset", sanitize_filename(old_category)
-    )
-    new_folder = os.path.join(
-        CLIPS_DIR, sanitize_filename(name), "dataset", new_category
-    )
+    sanitized_name = sanitize_path_component(name)
+    old_folder = safe_join(CLIPS_DIR, name, "dataset", old_category)
+    new_folder = safe_join(CLIPS_DIR, name, "dataset", new_category)
+
+    if sanitized_name is None or old_folder is None or new_folder is None:
+        return invalid_name_response(name)
 
     if not os.path.exists(old_folder):
         return JSONResponse(
@@ -1030,7 +1075,6 @@ def rename_classification_category(
 
         # Mark dataset as ready to train by resetting training metadata
         # This ensures the dataset is marked as changed after renaming
-        sanitized_name = sanitize_filename(name)
         write_training_metadata(sanitized_name, 0)
 
         return JSONResponse(
@@ -1078,13 +1122,20 @@ def categorize_classification_image(request: Request, name: str, body: dict = No
         )
 
     json: dict[str, Any] = body or {}
-    category = sanitize_filename(json.get("category", ""))
-    training_file_name = sanitize_filename(json.get("training_file", ""))
-    training_file = os.path.join(
-        CLIPS_DIR, sanitize_filename(name), "train", training_file_name
+    category = sanitize_path_component(json.get("category", ""))
+    training_file_name = json.get("training_file", "")
+    training_file = (
+        safe_join(CLIPS_DIR, name, "train", training_file_name)
+        if training_file_name
+        else None
     )
 
-    if training_file_name and not os.path.isfile(training_file):
+    if category is None:
+        return invalid_name_response(json.get("category", ""))
+
+    if training_file_name and (
+        training_file is None or not os.path.isfile(training_file)
+    ):
         return JSONResponse(
             content=(
                 {
@@ -1098,9 +1149,10 @@ def categorize_classification_image(request: Request, name: str, body: dict = No
     random_id = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
     timestamp = datetime.datetime.now().timestamp()
     new_name = f"{category}-{timestamp}-{random_id}.png"
-    new_file_folder = os.path.join(
-        CLIPS_DIR, sanitize_filename(name), "dataset", category
-    )
+    new_file_folder = safe_join(CLIPS_DIR, name, "dataset", category)
+
+    if new_file_folder is None:
+        return invalid_name_response(name)
 
     os.makedirs(new_file_folder, exist_ok=True)
 
@@ -1138,9 +1190,10 @@ def create_classification_category(request: Request, name: str, category: str):
             status_code=404,
         )
 
-    category_folder = os.path.join(
-        CLIPS_DIR, sanitize_filename(name), "dataset", sanitize_filename(category)
-    )
+    category_folder = safe_join(CLIPS_DIR, name, "dataset", category)
+
+    if category_folder is None:
+        return invalid_name_response(category)
 
     os.makedirs(category_folder, exist_ok=True)
 
@@ -1179,12 +1232,15 @@ def delete_classification_train_images(request: Request, name: str, body: dict =
 
     json: dict[str, Any] = body or {}
     list_of_ids = json.get("ids", "")
-    folder = os.path.join(CLIPS_DIR, sanitize_filename(name), "train")
+    folder = safe_join(CLIPS_DIR, name, "train")
+
+    if folder is None:
+        return invalid_name_response(name)
 
     for id in list_of_ids:
-        file_path = os.path.join(folder, sanitize_filename(id))
+        file_path = safe_join(folder, id)
 
-        if os.path.isfile(file_path):
+        if file_path and os.path.isfile(file_path):
             os.unlink(file_path)
 
     return JSONResponse(
@@ -1201,7 +1257,11 @@ def delete_classification_train_images(request: Request, name: str, body: dict =
 )
 async def generate_state_examples(request: Request, body: GenerateStateExamplesBody):
     """Generate examples for state classification."""
-    model_name = sanitize_filename(body.model_name)
+    model_name = sanitize_path_component(body.model_name)
+
+    if model_name is None:
+        return invalid_name_response(body.model_name)
+
     cameras_normalized = {
         camera_name: tuple(crop)
         for camera_name, crop in body.cameras.items()
@@ -1224,7 +1284,11 @@ async def generate_state_examples(request: Request, body: GenerateStateExamplesB
 )
 async def generate_object_examples(request: Request, body: GenerateObjectExamplesBody):
     """Generate examples for object classification."""
-    model_name = sanitize_filename(body.model_name)
+    model_name = sanitize_path_component(body.model_name)
+
+    if model_name is None:
+        return invalid_name_response(body.model_name)
+
     collect_object_classification_examples(model_name, body.label)
 
     return JSONResponse(
@@ -1243,10 +1307,16 @@ async def generate_object_examples(request: Request, body: GenerateObjectExample
     Returns a success message.""",
 )
 def delete_classification_model(request: Request, name: str):
-    sanitized_name = sanitize_filename(name)
+    # This endpoint intentionally accepts models that are not in the config, so
+    # there is no allow list to fall back on. Both paths below are recursive
+    # deletes, so an unusable name has to be rejected outright.
+    data_dir = safe_join(CLIPS_DIR, name)
+    model_dir = safe_join(MODEL_CACHE_DIR, name)
+
+    if data_dir is None or model_dir is None:
+        return invalid_name_response(name)
 
     # Delete the classification model's data directory in clips
-    data_dir = os.path.join(CLIPS_DIR, sanitized_name)
     if os.path.exists(data_dir):
         try:
             shutil.rmtree(data_dir)
@@ -1255,7 +1325,6 @@ def delete_classification_model(request: Request, name: str):
             logger.debug(f"Failed to delete data directory for {name}: {e}")
 
     # Delete the classification model's files in model_cache
-    model_dir = os.path.join(MODEL_CACHE_DIR, sanitized_name)
     if os.path.exists(model_dir):
         try:
             shutil.rmtree(model_dir)

frigate/api/event.py

@@ -16,7 +16,6 @@ import numpy as np
 from fastapi import APIRouter, Request
 from fastapi.params import Depends
 from fastapi.responses import JSONResponse
-from pathvalidate import sanitize_filename
 from peewee import JOIN, DoesNotExist, fn, operator
 from playhouse.shortcuts import model_to_dict
 
@@ -56,11 +55,12 @@ from frigate.api.defs.response.generic_response import GenericResponse
 from frigate.api.defs.tags import Tags
 from frigate.comms.event_metadata_updater import EventMetadataTypeEnum
 from frigate.config.classification import ObjectClassificationType
-from frigate.const import CLIPS_DIR, TRIGGER_DIR
+from frigate.const import CLIPS_DIR
 from frigate.embeddings import EmbeddingsContext
 from frigate.models import Event, ReviewSegment, Timeline, Trigger
 from frigate.track.object_processing import TrackedObject
 from frigate.util.file import get_event_thumbnail_bytes, load_event_snapshot_image
+from frigate.util.path import get_trigger_thumbnail_path, safe_join
 from frigate.util.time import get_dst_transitions, get_tz_modifiers
 
 logger = logging.getLogger(__name__)
@@ -1452,10 +1452,10 @@ async def set_attributes(
             continue
 
         # Get available labels from dataset directory
-        dataset_dir = os.path.join(CLIPS_DIR, sanitize_filename(model_key), "dataset")
+        dataset_dir = safe_join(CLIPS_DIR, model_key, "dataset")
         available_labels = set()
 
-        if os.path.exists(dataset_dir):
+        if dataset_dir and os.path.exists(dataset_dir):
             for category_name in os.listdir(dataset_dir):
                 category_dir = os.path.join(dataset_dir, category_name)
                 if os.path.isdir(category_dir):
@@ -1959,18 +1959,13 @@ def create_trigger_embedding(
         if body.type == "thumbnail":
             # Save image to the triggers directory
             try:
-                os.makedirs(
-                    os.path.join(TRIGGER_DIR, sanitize_filename(camera_name)),
-                    exist_ok=True,
-                )
-                with open(
-                    os.path.join(
-                        TRIGGER_DIR,
-                        sanitize_filename(camera_name),
-                        f"{sanitize_filename(body.data)}.webp",
-                    ),
-                    "wb",
-                ) as f:
+                webp_path = get_trigger_thumbnail_path(camera_name, body.data)
+
+                if webp_path is None:
+                    raise ValueError(f"Invalid trigger thumbnail path for {body.data}")
+
+                os.makedirs(os.path.dirname(webp_path), exist_ok=True)
+                with open(webp_path, "wb") as f:
                     f.write(thumbnail)
                 logger.debug(
                     f"Writing thumbnail for trigger with data {body.data} in {camera_name}."
@@ -2042,10 +2037,16 @@ def update_trigger_embedding(
         if body.type == "description":
             embedding = context.generate_description_embedding(body.data)
         elif body.type == "thumbnail":
-            webp_file = sanitize_filename(body.data) + ".webp"
-            webp_path = os.path.join(
-                TRIGGER_DIR, sanitize_filename(camera_name), webp_file
-            )
+            webp_path = get_trigger_thumbnail_path(camera_name, body.data)
+
+            if webp_path is None:
+                return JSONResponse(
+                    content={
+                        "success": False,
+                        "message": f"Invalid data for {body.type} trigger",
+                    },
+                    status_code=400,
+                )
 
             try:
                 event: Event = Event.get(Event.id == body.data)
@@ -2102,13 +2103,14 @@ def update_trigger_embedding(
             # Update existing trigger
             if trigger.data != body.data:  # Delete old thumbnail only if data changes
                 try:
-                    os.remove(
-                        os.path.join(
-                            TRIGGER_DIR,
-                            sanitize_filename(camera_name),
-                            f"{trigger.data}.webp",
+                    old_path = get_trigger_thumbnail_path(camera_name, trigger.data)
+
+                    if old_path is None:
+                        raise ValueError(
+                            f"Invalid trigger thumbnail path for {trigger.data}"
                         )
-                    )
+
+                    os.remove(old_path)
                     logger.debug(
                         f"Deleted thumbnail for trigger with data {trigger.data} in {camera_name}."
                     )
@@ -2142,12 +2144,13 @@ def update_trigger_embedding(
         if body.type == "thumbnail":
             # Save image to the triggers directory
             try:
-                camera_path = os.path.join(TRIGGER_DIR, sanitize_filename(camera_name))
-                os.makedirs(camera_path, exist_ok=True)
-                with open(
-                    os.path.join(camera_path, f"{sanitize_filename(body.data)}.webp"),
-                    "wb",
-                ) as f:
+                thumbnail_path = get_trigger_thumbnail_path(camera_name, body.data)
+
+                if thumbnail_path is None:
+                    raise ValueError(f"Invalid trigger thumbnail path for {body.data}")
+
+                os.makedirs(os.path.dirname(thumbnail_path), exist_ok=True)
+                with open(thumbnail_path, "wb") as f:
                     f.write(thumbnail)
                 logger.debug(
                     f"Writing thumbnail for trigger with data {body.data} in {camera_name}."
@@ -2218,11 +2221,12 @@ def delete_trigger_embedding(
             )
 
         try:
-            os.remove(
-                os.path.join(
-                    TRIGGER_DIR, sanitize_filename(camera_name), f"{trigger.data}.webp"
-                )
-            )
+            thumbnail_path = get_trigger_thumbnail_path(camera_name, trigger.data)
+
+            if thumbnail_path is None:
+                raise ValueError(f"Invalid trigger thumbnail path for {trigger.data}")
+
+            os.remove(thumbnail_path)
             logger.debug(
                 f"Deleted thumbnail for trigger with data {trigger.data} in {camera_name}."
             )

frigate/api/export.py

@@ -13,7 +13,7 @@ from pathlib import Path
 import psutil
 from fastapi import APIRouter, Depends, Query, Request
 from fastapi.responses import JSONResponse, StreamingResponse
-from pathvalidate import sanitize_filename, sanitize_filepath
+from pathvalidate import sanitize_filename
 from peewee import DoesNotExist
 from playhouse.shortcuts import model_to_dict
 
@@ -72,6 +72,7 @@ from frigate.record.export import (
     PlaybackSourceEnum,
     validate_ffmpeg_args,
 )
+from frigate.util.path import sanitize_contained_path
 from frigate.util.time import is_current_hour
 
 logger = logging.getLogger(__name__)
@@ -129,18 +130,12 @@ def _validate_export_case(export_case_id: str | None) -> JSONResponse | None:
 def _sanitize_existing_image(
     image_path: str | None,
 ) -> tuple[str | None, JSONResponse | None]:
-    # 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,
-        )
+    if not image_path:
+        return None, None
 
-    existing_image = sanitize_filepath(image_path) if image_path else None
+    existing_image = sanitize_contained_path(image_path, CLIPS_DIR)
 
-    if existing_image and not existing_image.startswith(CLIPS_DIR):
+    if existing_image is None:
         return None, JSONResponse(
             content={"success": False, "message": "Invalid image path"},
             status_code=400,

frigate/data_processing/real_time/face.py

@@ -28,6 +28,7 @@ from frigate.data_processing.common.face.model import (
 from frigate.types import TrackedObjectUpdateTypesEnum
 from frigate.util.builtin import EventsPerSecond, InferenceSpeed
 from frigate.util.image import area
+from frigate.util.path import safe_join, sanitize_path_component
 
 from ..types import DataProcessorMetrics
 from .api import RealTimeProcessorApi
@@ -409,9 +410,17 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
                 )
 
             # write face to library
-            folder = os.path.join(FACE_DIR, label)
+            sanitized_label = sanitize_path_component(label)
+            folder = safe_join(FACE_DIR, label)
+
+            if sanitized_label is None or folder is None:
+                return {
+                    "message": f"Invalid face name: {label}",
+                    "success": False,
+                }
+
             file = os.path.join(
-                folder, f"{label}_{datetime.datetime.now().timestamp()}.webp"
+                folder, f"{sanitized_label}_{datetime.datetime.now().timestamp()}.webp"
             )
             os.makedirs(folder, exist_ok=True)
 

frigate/embeddings/__init__.py

@@ -21,6 +21,7 @@ from frigate.db.sqlitevecq import SqliteVecQueueDatabase
 from frigate.models import Event
 from frigate.util.builtin import serialize
 from frigate.util.classification import kickoff_model_training
+from frigate.util.path import safe_join
 from frigate.util.process import FrigateProcess
 
 from .maintainer import EmbeddingMaintainer
@@ -234,11 +235,16 @@ class EmbeddingsContext:
         )
 
     def delete_face_ids(self, face: str, ids: list[str]) -> None:
-        folder = os.path.join(FACE_DIR, face)
-        for id in ids:
-            file_path = os.path.join(folder, id)
+        folder = safe_join(FACE_DIR, face)
 
-            if os.path.isfile(file_path):
+        if folder is None:
+            logger.warning("Not deleting faces for invalid name %s", face)
+            return
+
+        for id in ids:
+            file_path = safe_join(folder, id)
+
+            if file_path and os.path.isfile(file_path):
                 os.unlink(file_path)
 
         if face != "train" and len(os.listdir(folder)) == 0:

frigate/test/http_api/test_http_classification_traversal.py

@@ -0,0 +1,73 @@
+"""End to end checks that classification endpoints cannot escape their base dir."""
+
+import os
+import shutil
+import tempfile
+from unittest.mock import patch
+
+from frigate.models import Event
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+# Percent encodings that survive nginx normalization. nginx collapses a bare
+# ".." segment, but "..:" and friends are not relative segments to nginx while
+# pathvalidate still reduces them to exactly "..".
+TRAVERSAL_NAMES = ["..%3A", "..%2A", "..%3C", "..%7C", "..%20", ".."]
+
+
+class TestHttpClassificationTraversal(BaseTestHttp):
+    def setUp(self):
+        super().setUp([Event])
+        self.app = super().create_app()
+
+        self.root = tempfile.mkdtemp()
+        self.clips = os.path.join(self.root, "clips")
+        self.model_cache = os.path.join(self.root, "model_cache")
+        os.makedirs(os.path.join(self.clips, "model1"))
+        os.makedirs(os.path.join(self.model_cache, "model1"))
+        os.makedirs(os.path.join(self.root, "recordings"))
+
+        # Sibling data that a "/.." escape from clips would reach.
+        self.canary = os.path.join(self.root, "recordings", "seg.mp4")
+
+        with open(self.canary, "w") as f:
+            f.write("recording")
+
+        clips_patch = patch("frigate.api.classification.CLIPS_DIR", self.clips)
+        cache_patch = patch(
+            "frigate.api.classification.MODEL_CACHE_DIR", self.model_cache
+        )
+        clips_patch.start()
+        cache_patch.start()
+        self.addCleanup(clips_patch.stop)
+        self.addCleanup(cache_patch.stop)
+
+    def tearDown(self):
+        shutil.rmtree(self.root, ignore_errors=True)
+        self.app.dependency_overrides.clear()
+        super().tearDown()
+
+    def test_delete_model_rejects_traversal_names(self):
+        client = AuthTestClient(self.app)
+
+        for name in TRAVERSAL_NAMES:
+            with self.subTest(name=name):
+                response = client.delete(f"/classification/{name}")
+
+                # Either the router never matches it or the handler rejects it,
+                # but the sibling directory must survive either way.
+                self.assertNotEqual(response.status_code, 200)
+                self.assertTrue(
+                    os.path.exists(self.canary),
+                    f"{name} deleted data outside the clips directory",
+                )
+                self.assertTrue(os.path.exists(os.path.join(self.root, "recordings")))
+
+    def test_delete_model_still_removes_its_own_directories(self):
+        client = AuthTestClient(self.app)
+
+        response = client.delete("/classification/model1")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertFalse(os.path.exists(os.path.join(self.clips, "model1")))
+        self.assertFalse(os.path.exists(os.path.join(self.model_cache, "model1")))
+        self.assertTrue(os.path.exists(self.canary))

frigate/test/test_util_path.py

@@ -0,0 +1,197 @@
+"""Tests for safe filesystem path construction."""
+
+import os
+import shutil
+import tempfile
+import unittest
+
+from frigate.const import TRIGGER_DIR
+from frigate.util.path import (
+    get_trigger_thumbnail_path,
+    is_contained_in,
+    safe_join,
+    sanitize_contained_path,
+    sanitize_path_component,
+)
+
+# Values that pathvalidate's sanitize_filename reduces to exactly "..", because
+# it strips reserved characters but leaves relative markers intact. nginx only
+# normalizes a bare ".." segment, so the decorated variants reach the app.
+DOT_DOT_VARIANTS = ["..", "..:", "..*", "..?", '.."', "..<", "..>", "..|", ".. ", " .."]
+
+
+class TestSanitizePathComponent(unittest.TestCase):
+    def test_rejects_dot_dot_variants(self):
+        for value in DOT_DOT_VARIANTS:
+            with self.subTest(value=value):
+                self.assertIsNone(sanitize_path_component(value))
+
+    def test_rejects_relative_markers_and_empty(self):
+        for value in [".", "", None, "   ", "/", "//", "\\"]:
+            with self.subTest(value=value):
+                self.assertIsNone(sanitize_path_component(value))
+
+    def test_strips_separators(self):
+        component = sanitize_path_component("a/b/c")
+        self.assertIsNotNone(component)
+        self.assertNotIn("/", component)
+
+    def test_allows_ordinary_names(self):
+        for value in ["model1", "front-door", "My Model", "café", "a.b_c-1"]:
+            with self.subTest(value=value):
+                self.assertEqual(sanitize_path_component(value), value)
+
+
+class TestSafeJoin(unittest.TestCase):
+    base = "/media/frigate/clips"
+
+    def test_rejects_dot_dot_variants(self):
+        for value in DOT_DOT_VARIANTS:
+            with self.subTest(value=value):
+                self.assertIsNone(safe_join(self.base, value))
+
+    def test_rejects_dot_dot_in_any_segment(self):
+        self.assertIsNone(safe_join(self.base, "model", "dataset", ".."))
+        self.assertIsNone(safe_join(self.base, "..", "dataset", ".."))
+
+    def test_result_stays_inside_base(self):
+        for value in ["model1", "a/../..", "....//", "..\\..", "%2e%2e"]:
+            with self.subTest(value=value):
+                joined = safe_join(self.base, value)
+
+                if joined is not None:
+                    self.assertTrue(is_contained_in(joined, self.base))
+
+    def test_joins_multiple_segments(self):
+        self.assertEqual(
+            safe_join(self.base, "model1", "dataset", "none"),
+            "/media/frigate/clips/model1/dataset/none",
+        )
+
+    def test_rejects_empty_segment(self):
+        self.assertIsNone(safe_join(self.base, "model1", "", "none"))
+
+
+class TestIsContainedIn(unittest.TestCase):
+    def test_rejects_sibling_sharing_a_name_prefix(self):
+        self.assertFalse(
+            is_contained_in("/media/frigate/clips_evil/x.webp", "/media/frigate/clips")
+        )
+
+    def test_accepts_base_itself_and_children(self):
+        self.assertTrue(is_contained_in("/media/frigate/clips", "/media/frigate/clips"))
+        self.assertTrue(
+            is_contained_in("/media/frigate/clips/a/b.webp", "/media/frigate/clips")
+        )
+
+    def test_rejects_parent(self):
+        self.assertFalse(is_contained_in("/media/frigate", "/media/frigate/clips"))
+
+    def test_handles_a_root_base(self):
+        # A prefix test would compare against "//" here and wrongly report that
+        # the root directory contains nothing.
+        self.assertTrue(is_contained_in("/child", "/"))
+        self.assertEqual(safe_join("/", "child"), "/child")
+
+    def test_rejects_uncomparable_paths(self):
+        self.assertFalse(is_contained_in("relative/x", "/media/frigate/clips"))
+
+
+class TestSanitizeContainedPath(unittest.TestCase):
+    base = "/media/frigate/clips"
+
+    def test_rejects_dot_dot_anywhere(self):
+        for value in [
+            "/media/frigate/clips/../../etc/passwd",
+            "clips\\..\\..\\etc/passwd",
+            "/media/frigate/clips/a/../../../x",
+        ]:
+            with self.subTest(value=value):
+                self.assertIsNone(sanitize_contained_path(value, self.base))
+
+    def test_rejects_sibling_sharing_a_name_prefix(self):
+        self.assertIsNone(
+            sanitize_contained_path("/media/frigate/clips_evil/x.webp", self.base)
+        )
+
+    def test_rejects_outside_base(self):
+        self.assertIsNone(sanitize_contained_path("/etc/passwd", self.base))
+
+    def test_rejects_empty(self):
+        self.assertIsNone(sanitize_contained_path("", self.base))
+        self.assertIsNone(sanitize_contained_path(None, self.base))
+
+    def test_keeps_a_valid_nested_path(self):
+        self.assertEqual(
+            sanitize_contained_path("/media/frigate/clips/a/b.webp", self.base),
+            "/media/frigate/clips/a/b.webp",
+        )
+
+
+class TestTriggerThumbnailPath(unittest.TestCase):
+    def test_stays_inside_the_trigger_dir(self):
+        for camera, data in [
+            ("cam", "../../../../etc/passwd"),
+            ("cam", "../../../../config/config.yml"),
+            ("cam", "normal-event-id"),
+        ]:
+            with self.subTest(camera=camera, data=data):
+                path = get_trigger_thumbnail_path(camera, data)
+
+                self.assertIsNotNone(path)
+                self.assertTrue(is_contained_in(path, TRIGGER_DIR))
+
+    def test_rejects_traversal_camera_names(self):
+        for camera in DOT_DOT_VARIANTS:
+            with self.subTest(camera=camera):
+                self.assertIsNone(get_trigger_thumbnail_path(camera, "data"))
+
+    def test_builds_the_expected_path(self):
+        self.assertEqual(
+            get_trigger_thumbnail_path("front_door", "abc"),
+            os.path.join(TRIGGER_DIR, "front_door", "abc.webp"),
+        )
+
+
+class TestRmtreeContainment(unittest.TestCase):
+    """A recursive delete built through safe_join must not reach a parent.
+
+    shutil.rmtree on a path ending in ".." deletes the parent's contents before
+    failing on the final rmdir, so the guard has to run before the call.
+    """
+
+    def setUp(self):
+        self.root = tempfile.mkdtemp()
+        self.clips = os.path.join(self.root, "clips")
+        os.makedirs(os.path.join(self.clips, "model1"))
+        os.makedirs(os.path.join(self.root, "recordings"))
+
+        with open(os.path.join(self.root, "recordings", "seg.mp4"), "w") as f:
+            f.write("recording")
+
+    def tearDown(self):
+        shutil.rmtree(self.root, ignore_errors=True)
+
+    def test_traversal_name_never_yields_a_path_to_delete(self):
+        for value in DOT_DOT_VARIANTS:
+            with self.subTest(value=value):
+                self.assertIsNone(safe_join(self.clips, value))
+
+        self.assertTrue(
+            os.path.exists(os.path.join(self.root, "recordings", "seg.mp4"))
+        )
+
+    def test_ordinary_name_still_deletes_its_own_directory(self):
+        target = safe_join(self.clips, "model1")
+        self.assertIsNotNone(target)
+
+        shutil.rmtree(target)
+
+        self.assertFalse(os.path.exists(os.path.join(self.clips, "model1")))
+        self.assertTrue(
+            os.path.exists(os.path.join(self.root, "recordings", "seg.mp4"))
+        )
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

frigate/util/path.py

@@ -0,0 +1,134 @@
+"""Helpers for building filesystem paths out of user supplied values."""
+
+import os
+
+from pathvalidate import ValidationError, sanitize_filename, sanitize_filepath
+
+from frigate.const import TRIGGER_DIR
+
+# Components that name a directory relative to its parent instead of a child.
+# pathvalidate strips separators and reserved characters but leaves these
+# intact, and it collapses values like "..:" down to "..", so they have to be
+# rejected after sanitizing rather than before.
+RELATIVE_COMPONENTS = {"", ".", ".."}
+
+
+def sanitize_path_component(value: str | None) -> str | None:
+    """Reduce a user supplied value to a single path component.
+
+    Args:
+        value: The untrusted value, such as a path parameter or body field
+
+    Returns:
+        A component that is safe to join onto a base directory, or None when
+        nothing usable remains so the caller can reject the request.
+    """
+    if not value:
+        return None
+
+    try:
+        component = sanitize_filename(value)
+    except (ValidationError, ValueError):
+        return None
+
+    if component.strip() in RELATIVE_COMPONENTS:
+        return None
+
+    if os.sep in component or (os.altsep and os.altsep in component):
+        return None
+
+    return component
+
+
+def is_contained_in(path: str, base: str) -> bool:
+    """Check that a path sits inside a base directory.
+
+    Compares whole path components, so a sibling directory that merely shares a
+    name prefix with base is not treated as contained.
+    """
+    resolved = os.path.normpath(path)
+    root = os.path.normpath(base)
+
+    try:
+        # commonpath compares components, and unlike a prefix test it stays
+        # correct for a base that already ends in a separator such as "/".
+        return os.path.commonpath([resolved, root]) == root
+    except ValueError:
+        # Raised when the paths cannot be compared, such as one relative and
+        # one absolute, or two different Windows drives.
+        return False
+
+
+def safe_join(base: str, *parts: str | None) -> str | None:
+    """Join user supplied parts beneath a trusted base directory.
+
+    Args:
+        base: Trusted base directory the result must stay inside of
+        parts: Untrusted values, each becoming one path component
+
+    Returns:
+        The joined path, or None if any part is unusable or the result would
+        land outside base.
+    """
+    components: list[str] = []
+
+    for part in parts:
+        component = sanitize_path_component(part)
+
+        if component is None:
+            return None
+
+        components.append(component)
+
+    resolved = os.path.normpath(os.path.join(base, *components))
+
+    # normpath rather than realpath so symlinked media roots keep working; the
+    # per component checks above are what actually prevent traversal.
+    if not is_contained_in(resolved, base):
+        return None
+
+    return resolved
+
+
+def sanitize_contained_path(path: str | None, base: str) -> str | None:
+    """Validate a whole user supplied path that must already sit under base.
+
+    Unlike safe_join this keeps the directory structure the caller sent, so it
+    suits values that name an existing file rather than one component.
+
+    Args:
+        path: The untrusted path
+        base: Directory the path has to stay inside of
+
+    Returns:
+        The sanitized path, or None if it is unusable or escapes base.
+    """
+    if not path:
+        return None
+
+    # sanitize_filepath normalizes "\" to "/" but leaves ".." intact, so a path
+    # like "clips\..\..\etc/passwd" would pass the containment check yet still
+    # escape once resolved. A valid path here never uses "..".
+    if ".." in path:
+        return None
+
+    sanitized = sanitize_filepath(path)
+
+    if not is_contained_in(sanitized, base):
+        return None
+
+    return sanitized
+
+
+def get_trigger_thumbnail_path(camera_name: str, data: str) -> str | None:
+    """Path of the thumbnail stored for a semantic search trigger.
+
+    Args:
+        camera_name: Camera the trigger belongs to
+        data: The trigger's data value, which is free-form text supplied by the
+            client and persisted verbatim
+
+    Returns:
+        The thumbnail path, or None if it cannot be built safely.
+    """
+    return safe_join(TRIGGER_DIR, camera_name, f"{data}.webp")

web/src/api/ws.ts

@@ -205,7 +205,7 @@ function applyCameraActivity(payload: string) {
     );
     applyTopicUpdate(
       `${name}/notifications/suspended`,
-      notifications_suspended || 0,
+      String(notifications_suspended ?? 0),
     );
     applyTopicUpdate(
       `${name}/ptz_autotracker/state`,
@@ -806,7 +806,7 @@ export function useNotificationSuspend(camera: string): {
     `${camera}/notifications/suspended`,
     `${camera}/notifications/suspend`,
   );
-  return { payload: payload as string, send };
+  return { payload: String(payload ?? 0), send };
 }
 
 export function useNotificationTest(): {

web/src/components/config-form/sectionExtras/NotificationsSettingsExtras.tsx

@@ -746,18 +746,10 @@ export function CameraNotificationSwitch({
     useNotifications(camera);
   const { payload: notificationSuspendUntil, send: sendNotificationSuspend } =
     useNotificationSuspend(camera);
-  const [isSuspended, setIsSuspended] = useState<boolean>(false);
-
-  useEffect(() => {
-    if (notificationSuspendUntil) {
-      setIsSuspended(
-        notificationSuspendUntil !== "0" || notificationState === "OFF",
-      );
-    }
-  }, [notificationSuspendUntil, notificationState]);
+  const isSuspended =
+    notificationSuspendUntil !== "0" || notificationState === "OFF";
 
   const handleSuspend = (duration: string) => {
-    setIsSuspended(true);
     if (duration == "off") {
       sendNotification("OFF");
     } else {

web/src/components/menu/LiveContextMenu.tsx

@@ -228,15 +228,8 @@ export default function LiveContextMenu({
     useNotifications(camera);
   const { payload: notificationSuspendUntil, send: sendNotificationSuspend } =
     useNotificationSuspend(camera);
-  const [isSuspended, setIsSuspended] = useState<boolean>(false);
-
-  useEffect(() => {
-    if (notificationSuspendUntil) {
-      setIsSuspended(
-        notificationSuspendUntil !== "0" || notificationState === "OFF",
-      );
-    }
-  }, [notificationSuspendUntil, notificationState]);
+  const isSuspended =
+    notificationSuspendUntil !== "0" || notificationState === "OFF";
 
   const handleSuspend = (duration: string) => {
     if (duration === "off") {