diff --git a/frigate/api/classification.py b/frigate/api/classification.py index 6dd7055092..5ddc3d34c7 100644 --- a/frigate/api/classification.py +++ b/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) diff --git a/frigate/api/event.py b/frigate/api/event.py index 835e7918cc..1597c18856 100644 --- a/frigate/api/event.py +++ b/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}." ) diff --git a/frigate/api/export.py b/frigate/api/export.py index 741a0bb5dc..27f85a2cde 100644 --- a/frigate/api/export.py +++ b/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, diff --git a/frigate/data_processing/real_time/face.py b/frigate/data_processing/real_time/face.py index ec1fccb45e..a55c2566e1 100644 --- a/frigate/data_processing/real_time/face.py +++ b/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) diff --git a/frigate/embeddings/__init__.py b/frigate/embeddings/__init__.py index cb50188893..d9b45e7c65 100644 --- a/frigate/embeddings/__init__.py +++ b/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: diff --git a/frigate/test/http_api/test_http_classification_traversal.py b/frigate/test/http_api/test_http_classification_traversal.py new file mode 100644 index 0000000000..ec74b1e061 --- /dev/null +++ b/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)) diff --git a/frigate/test/test_util_path.py b/frigate/test/test_util_path.py new file mode 100644 index 0000000000..56951a76cd --- /dev/null +++ b/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) diff --git a/frigate/util/path.py b/frigate/util/path.py new file mode 100644 index 0000000000..c5e1c388cf --- /dev/null +++ b/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")