nik/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-08-31 07:27:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

sanitize user-supplied path components (#23990)

Browse Source 
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.
This commit is contained in:
Josh Hawkins 2026-08-13 21:59:46 -05:00 committed by GitHub
parent 812e5308a3
commit 11f8786459
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 604 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

195
frigate/api/classification.py
View File

@ -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)

78
frigate/api/event.py
View File

@ -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}."
)

17
frigate/api/export.py
View File

@ -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,

13
frigate/data_processing/real_time/face.py
View File

@ -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)

14
frigate/embeddings/__init__.py
View File

@ -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:

73
frigate/test/http_api/test_http_classification_traversal.py Normal file
View File

@ -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))

197
frigate/test/test_util_path.py Normal file
View File

@ -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)

134
frigate/util/path.py Normal file
View File

@ -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")