Add ability to rename faces in the Face Library (#17879)

* api endpoint

* embeddings rename function

* frontend and i18n keys

* lazy load train tab images

* only log exception to make codeql happy

frigate/api/classification.py

@@ -14,6 +14,7 @@ from peewee import DoesNotExist
 from playhouse.shortcuts import model_to_dict
 
 from frigate.api.auth import require_role
+from frigate.api.defs.request.classification_body import RenameFaceBody
 from frigate.api.defs.tags import Tags
 from frigate.config.camera import DetectConfig
 from frigate.const import FACE_DIR
@@ -260,6 +261,35 @@ def deregister_faces(request: Request, name: str, body: dict = None):
     )
 
 
+@router.put("/faces/{old_name}/rename", dependencies=[Depends(require_role(["admin"]))])
+def rename_face(request: Request, old_name: str, body: RenameFaceBody):
+    if not request.app.frigate_config.face_recognition.enabled:
+        return JSONResponse(
+            status_code=400,
+            content={"message": "Face recognition is not enabled.", "success": False},
+        )
+
+    context: EmbeddingsContext = request.app.embeddings
+    try:
+        context.rename_face(old_name, body.new_name)
+        return JSONResponse(
+            content={
+                "success": True,
+                "message": f"Successfully renamed face to {body.new_name}.",
+            },
+            status_code=200,
+        )
+    except ValueError as e:
+        logger.error(e)
+        return JSONResponse(
+            status_code=400,
+            content={
+                "message": "Error renaming face. Check Frigate logs.",
+                "success": False,
+            },
+        )
+
+
 @router.put("/lpr/reprocess")
 def reprocess_license_plate(request: Request, event_id: str):
     if not request.app.frigate_config.lpr.enabled:

frigate/api/defs/request/classification_body.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class RenameFaceBody(BaseModel):
+    new_name: str

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,