diff --git a/src/backend/core/recording/event/parsers.py b/src/backend/core/recording/event/parsers.py index 9e6a71ea..f99a07b0 100644 --- a/src/backend/core/recording/event/parsers.py +++ b/src/backend/core/recording/event/parsers.py @@ -6,6 +6,7 @@ import re from dataclasses import dataclass from functools import lru_cache from typing import Any, Dict, Optional, Protocol +from urllib.parse import quote_plus from django.conf import settings from django.utils.module_loading import import_string @@ -164,6 +165,8 @@ class S3Parser(BaseS3Parser): if not filepath: raise ParsingEventDataError("Missing object key name") filetype, _ = mimetypes.guess_type(filepath) + # Preserve already URL-encoded separators while encoding raw S3 object keys. + filepath = quote_plus(filepath, safe="%") return StorageEvent( filepath=filepath, filetype=filetype, diff --git a/src/backend/core/tests/recording/event/test_parsers.py b/src/backend/core/tests/recording/event/test_parsers.py index a7ae1916..bcb39d3a 100644 --- a/src/backend/core/tests/recording/event/test_parsers.py +++ b/src/backend/core/tests/recording/event/test_parsers.py @@ -360,6 +360,50 @@ def test_s3_parse_unrecognized_extension(s3_parser): s3_parser.parse(event_with_unknown_ext) +def test_s3_parser_keeps_encoded_filepath_compatible(settings): + settings.RECORDING_OUTPUT_FOLDER = "recordings" + + recording_id = "80ae9fe5-639a-438b-b86e-9e3dd2d55f4d" + parser = S3Parser(bucket_name="recordings-bucket") + + data = { + "Records": [ + { + "s3": { + "bucket": {"name": "recordings-bucket"}, + "object": { + "key": f"recordings%2F{recording_id}.mp4", + }, + } + } + ] + } + + assert parser.get_recording_id(data) == recording_id + + +def test_s3_parser_accepts_unencoded_filepath(settings): + settings.RECORDING_OUTPUT_FOLDER = "recordings" + + recording_id = "80ae9fe5-639a-438b-b86e-9e3dd2d55f4d" + parser = S3Parser(bucket_name="recordings-bucket") + + data = { + "Records": [ + { + "s3": { + "bucket": {"name": "recordings-bucket"}, + "object": { + "key": f"recordings/{recording_id}.mp4", + }, + } + } + ] + } + + assert parser.get_recording_id(data) == recording_id + + def test_s3_get_recording_id_success(s3_parser, valid_s3_event): """Test successful extraction of recording ID from S3 event.""" recording_id = s3_parser.get_recording_id(valid_s3_event)