From ccc3bdb6ff4aba1263d590e8444db747cdb8ec71 Mon Sep 17 00:00:00 2001 From: Miker Date: Sun, 14 Jun 2026 09:16:59 -0400 Subject: [PATCH] security(transcribe): cap chunk count to bound Whisper cost/output transcribe() looped over every chunk with no limit, so a multi-hour or hostile source ran up unbounded Whisper API spend and flooded the agent context. Cap at MAX_CHUNKS (~4h) and append a clear truncation marker when exceeded. Co-Authored-By: Claude Opus 4.8 (1M context) --- agent_reach/transcribe.py | 16 +++++++++++++++- tests/test_transcribe.py | 27 +++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/agent_reach/transcribe.py b/agent_reach/transcribe.py index 2ffd6124..df9008da 100644 --- a/agent_reach/transcribe.py +++ b/agent_reach/transcribe.py @@ -26,6 +26,9 @@ # Whisper API limit is 25MB; leave headroom for multipart overhead. SIZE_LIMIT_BYTES = 24 * 1024 * 1024 CHUNK_SECONDS = 600 # 10 min — small enough that boundary cuts rarely lose meaning +# Bound the number of chunks transcribed per call so a multi-hour or hostile +# source cannot run up unbounded Whisper API cost / output. ~4 hours of audio. +MAX_CHUNKS = 24 PROVIDERS = { "groq": { @@ -239,11 +242,22 @@ def transcribe( else: chunks = chunk_audio(compressed, work_dir) + truncated = len(chunks) > MAX_CHUNKS + if truncated: + chunks = chunks[:MAX_CHUNKS] + pieces: List[str] = [] for chunk in chunks: text = _transcribe_with_fallback(chunk, order, cfg) pieces.append(text.strip()) - return "\n".join(p for p in pieces if p) + result = "\n".join(p for p in pieces if p) + if truncated: + minutes = MAX_CHUNKS * CHUNK_SECONDS // 60 + result += ( + f"\n\n[transcript truncated: source exceeded the {MAX_CHUNKS}-chunk " + f"(~{minutes} min) limit]" + ) + return result def _transcribe_with_fallback(chunk: Path, order: List[str], config: Config) -> str: diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py index bf8afdf8..17cff6e0 100644 --- a/tests/test_transcribe.py +++ b/tests/test_transcribe.py @@ -216,6 +216,33 @@ def test_chunks_concatenated_with_newlines( ) assert text == "part one\npart two" + def test_chunk_count_is_capped(self, monkeypatch, fake_config, tmp_path, chunk_file): + fake_config.set("groq_api_key", "gsk_test") + big = tmp_path / "compressed.m4a" + big.write_bytes(b"x" * (tr.SIZE_LIMIT_BYTES + 1)) + monkeypatch.setattr(tr, "compress_audio", lambda src, out_dir: big) + + # Produce more chunks than the cap allows. + many = [] + for i in range(tr.MAX_CHUNKS + 5): + c = tmp_path / f"chunk_{i:03d}.m4a" + c.write_bytes(b"a") + many.append(c) + monkeypatch.setattr(tr, "chunk_audio", lambda src, out_dir: many) + + calls = {"n": 0} + + def fake_post(*a, **k): + calls["n"] += 1 + return FakeResponse(200, "seg") + + monkeypatch.setattr(tr.requests, "post", fake_post) + + text = tr.transcribe(str(chunk_file), out_dir=tmp_path / "work", config=fake_config) + # Only MAX_CHUNKS Whisper calls were made — cost is bounded. + assert calls["n"] == tr.MAX_CHUNKS + assert "truncated" in text + def test_no_provider_configured_fails_fast(self, fake_config, chunk_file): with pytest.raises(tr.NoProviderConfigured): tr.transcribe(str(chunk_file), config=fake_config)