diff --git a/frigate/embeddings/__init__.py b/frigate/embeddings/__init__.py index c604658457..5e713b3c16 100644 --- a/frigate/embeddings/__init__.py +++ b/frigate/embeddings/__init__.py @@ -5,11 +5,13 @@ import json import logging import multiprocessing as mp import os +import re import signal import threading from types import FrameType from typing import Optional, Union +from pathvalidate import ValidationError, sanitize_filename from setproctitle import setproctitle from frigate.comms.embeddings_updater import EmbeddingsRequestEnum, EmbeddingsRequestor @@ -240,6 +242,42 @@ class EmbeddingsContext: EmbeddingsRequestEnum.clear_face_classifier.value, None ) + def rename_face(self, old_name: str, new_name: str) -> None: + valid_name_pattern = r"^[a-zA-Z0-9_-]{1,50}$" + + try: + sanitized_old_name = sanitize_filename(old_name, replacement_text="_") + sanitized_new_name = sanitize_filename(new_name, replacement_text="_") + except ValidationError as e: + raise ValueError(f"Invalid face name: {str(e)}") + + if not re.match(valid_name_pattern, old_name): + raise ValueError(f"Invalid old face name: {old_name}") + if not re.match(valid_name_pattern, new_name): + raise ValueError(f"Invalid new face name: {new_name}") + if sanitized_old_name != old_name: + raise ValueError(f"Old face name contains invalid characters: {old_name}") + if sanitized_new_name != new_name: + raise ValueError(f"New face name contains invalid characters: {new_name}") + + old_path = os.path.normpath(os.path.join(FACE_DIR, old_name)) + new_path = os.path.normpath(os.path.join(FACE_DIR, new_name)) + + # Prevent path traversal + if not old_path.startswith( + os.path.normpath(FACE_DIR) + ) or not new_path.startswith(os.path.normpath(FACE_DIR)): + raise ValueError("Invalid path detected") + + if not os.path.exists(old_path): + raise ValueError(f"Face {old_name} not found.") + + os.rename(old_path, new_path) + + self.requestor.send_data( + EmbeddingsRequestEnum.clear_face_classifier.value, None + ) + def update_description(self, event_id: str, description: str) -> None: self.requestor.send_data( EmbeddingsRequestEnum.embed_description.value,