fix: verify audio magic bytes before extension-based short-circuit in convert_audio_format

tests/test_media_utils.py

@@ -871,3 +871,158 @@ async def test_wav_to_tencent_silk_skips_resample_for_supported_rate(
 
     assert len(fake.calls) == 1
     assert fake.calls[0]["sample_rate"] == 24000
+
+
+# ---------------------------------------------------------------------------
+# convert_audio_format: extension vs magic-byte mismatch (#9594, fixed in #9612)
+# These tests guard the magic-byte verification logic that prevents platforms
+# like NapCat from passing AMR content through with a .wav extension.
+# ---------------------------------------------------------------------------
+
+# Minimal AMR header (magic bytes: #!AMR)
+_AMR_HEADER = b"#!AMR\n"
+
+
+def _make_wav_bytes() -> bytes:
+    """Build a minimal valid WAV header (RIFF/WAVE)."""
+    return b"RIFF\x24\x00\x00\x00WAVEfmt " + b"\x00" * 16
+
+
+class _FakeFFmpegProcess:
+    """Minimal async subprocess mock for ffmpeg."""
+
+    def __init__(self, returncode: int = 0) -> None:
+        self.returncode = returncode
+
+    async def communicate(self) -> tuple[bytes, bytes]:
+        return (b"", b"")
+
+
+def _patch_ffmpeg_not_called(monkeypatch, reason: str) -> None:
+    """Monkeypatch ``asyncio.create_subprocess_exec`` to fail if ffmpeg runs.
+
+    Args:
+        monkeypatch: pytest monkeypatch fixture.
+        reason: scenario description included in the failure message.
+    """
+
+    async def fake_exec(*args, **kwargs):
+        raise AssertionError(f"ffmpeg should not be called {reason}")
+
+    monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec)
+
+
+def _patch_ffmpeg(monkeypatch, tmp_path, *, output_content: bytes = b""):
+    """Monkeypatch ``asyncio.create_subprocess_exec`` to simulate ffmpeg.
+
+    Args:
+        monkeypatch: pytest monkeypatch fixture.
+        tmp_path: pytest tmp_path fixture (used for temp dir patching).
+        output_content: Bytes written to the output path so
+            ``convert_audio_format`` can return it.
+
+    Returns:
+        A mutable list whose first element tracks whether ffmpeg was called.
+    """
+    called = [False]
+
+    async def fake_exec(*args, **kwargs):
+        called[0] = True
+        output_path = args[-1]
+        Path(output_path).write_bytes(output_content)
+        return _FakeFFmpegProcess()
+
+    monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec)
+    monkeypatch.setattr(media_utils, "get_astrbot_temp_path", lambda: str(tmp_path))
+    return called
+
+
+@pytest.mark.asyncio
+async def test_convert_audio_format_amr_with_wav_extension_does_not_short_circuit(
+    tmp_path, monkeypatch
+):
+    """AMR content saved with a .wav extension must still be converted (#9594)."""
+    audio_file = tmp_path / "voice.wav"
+    audio_file.write_bytes(_AMR_HEADER + b"\x00" * 50)
+
+    called = _patch_ffmpeg(monkeypatch, tmp_path, output_content=_make_wav_bytes())
+
+    result = await media_utils.convert_audio_format(str(audio_file), "wav")
+
+    assert called[0], "ffmpeg should have been invoked because content is AMR, not WAV"
+    assert result != str(audio_file)
+
+
+@pytest.mark.asyncio
+async def test_convert_audio_format_real_wav_with_wav_extension_short_circuits(
+    tmp_path, monkeypatch
+):
+    """Genuine WAV content with a .wav extension should skip conversion."""
+    audio_file = tmp_path / "voice.wav"
+    audio_file.write_bytes(_make_wav_bytes())
+
+    _patch_ffmpeg_not_called(monkeypatch, "for a real WAV file")
+
+    result = await media_utils.convert_audio_format(str(audio_file), "wav")
+    assert result == str(audio_file)
+
+
+@pytest.mark.asyncio
+async def test_convert_audio_format_unknown_content_with_matching_ext_converts(
+    tmp_path, monkeypatch
+):
+    """When magic bytes cannot be identified, proceed with conversion (safer)."""
+    audio_file = tmp_path / "voice.wav"
+    audio_file.write_bytes(b"\x00" * 64)
+
+    called = _patch_ffmpeg(monkeypatch, tmp_path, output_content=_make_wav_bytes())
+
+    result = await media_utils.convert_audio_format(str(audio_file), "wav")
+
+    assert called[0], (
+        "ffmpeg should be invoked for unrecognised content (safer to convert)"
+    )
+    assert result != str(audio_file)
+
+
+@pytest.mark.asyncio
+async def test_convert_audio_format_missing_file_with_matching_ext_returns_path(
+    tmp_path, monkeypatch
+):
+    """When the file does not exist yet (e.g. NapCat race), return the path as-is."""
+    audio_file = tmp_path / "voice.wav"  # never created
+
+    _patch_ffmpeg_not_called(monkeypatch, "for a missing file")
+
+    result = await media_utils.convert_audio_format(str(audio_file), "wav")
+    assert result == str(audio_file)
+
+
+@pytest.mark.asyncio
+async def test_convert_audio_format_real_amr_with_amr_extension_short_circuits(
+    tmp_path, monkeypatch
+):
+    """AMR content with a matching .amr extension should skip conversion."""
+    audio_file = tmp_path / "voice.amr"
+    audio_file.write_bytes(_AMR_HEADER + b"\x00" * 50)
+
+    _patch_ffmpeg_not_called(monkeypatch, "for a real AMR file")
+
+    result = await media_utils.convert_audio_format(str(audio_file), "amr")
+    assert result == str(audio_file)
+
+
+@pytest.mark.asyncio
+async def test_convert_audio_format_wav_with_ogg_extension_does_not_short_circuit(
+    tmp_path, monkeypatch
+):
+    """WAV content saved with a .ogg extension must still be converted."""
+    audio_file = tmp_path / "voice.ogg"
+    audio_file.write_bytes(_make_wav_bytes())
+
+    called = _patch_ffmpeg(monkeypatch, tmp_path, output_content=b"OggS" + b"\x00" * 60)
+
+    result = await media_utils.convert_audio_format(str(audio_file), "ogg")
+
+    assert called[0], "ffmpeg should have been invoked because content is WAV, not OGG"
+    assert result != str(audio_file)