diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 10ee3e3e109..def4d4d09e8 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -844,6 +844,24 @@ def get_audio_cache_dir() -> Path: return d +def _sniff_audio_ext(data: bytes, fallback_ext: str) -> str: + """Prefer a container-matching extension when audio magic bytes are obvious.""" + fallback = fallback_ext if fallback_ext.startswith(".") else f".{fallback_ext}" + if len(data) >= 8 and data[4:8] == b"ftyp": + return ".m4a" + if data.startswith(b"OggS"): + return ".ogg" + if data.startswith(b"fLaC"): + return ".flac" + if len(data) >= 12 and data.startswith(b"RIFF") and data[8:12] == b"WAVE": + return ".wav" + if data.startswith(b"ID3") or data[:2] in {b"\xff\xfb", b"\xff\xf3", b"\xff\xf2"}: + return ".mp3" + if data.startswith(b"\x1a\x45\xdf\xa3"): + return ".webm" + return fallback + + def cache_audio_from_bytes(data: bytes, ext: str = ".ogg") -> str: """ Save raw audio bytes to the cache and return the absolute file path. @@ -857,7 +875,8 @@ def cache_audio_from_bytes(data: bytes, ext: str = ".ogg") -> str: """ validate_inbound_media_size(len(data), media_type="audio") cache_dir = get_audio_cache_dir() - filename = f"audio_{uuid.uuid4().hex[:12]}{ext}" + sniffed_ext = _sniff_audio_ext(data, ext) + filename = f"audio_{uuid.uuid4().hex[:12]}{sniffed_ext}" filepath = cache_dir / filename filepath.write_bytes(data) return str(filepath) diff --git a/tests/gateway/test_platform_base.py b/tests/gateway/test_platform_base.py index 786a8f69c52..e4d12438f92 100644 --- a/tests/gateway/test_platform_base.py +++ b/tests/gateway/test_platform_base.py @@ -18,6 +18,7 @@ from gateway.platforms.base import ( validate_inbound_media_size, _log_safe_path, _prefix_within_utf16_limit, + cache_audio_from_bytes, ) @@ -120,6 +121,26 @@ class TestSafeUrlForLog: assert safe_url_for_log(url, max_len=0) == "" +class TestCacheAudioFromBytes: + def test_sniffs_mp4_quicktime_audio_even_when_ext_is_ogg(self, tmp_path): + payload = b"\x00\x00\x00\x14ftypqt " + b"\x00" * 32 + with patch("gateway.platforms.base.AUDIO_CACHE_DIR", tmp_path): + result = cache_audio_from_bytes(payload, ext=".ogg") + + saved = tmp_path / os.path.basename(result) + assert saved.suffix == ".m4a" + assert saved.read_bytes() == payload + + def test_preserves_fallback_ext_when_audio_header_is_unknown(self, tmp_path): + payload = b"not-a-known-audio-header" + with patch("gateway.platforms.base.AUDIO_CACHE_DIR", tmp_path): + result = cache_audio_from_bytes(payload, ext=".aac") + + saved = tmp_path / os.path.basename(result) + assert saved.suffix == ".aac" + assert saved.read_bytes() == payload + + # --------------------------------------------------------------------------- # MessageEvent — command parsing # ---------------------------------------------------------------------------