fix(p0-3): redact signed URL + reason in bcut transcribe

autosearch/skills/tools/video-to-text-bcut/transcribe.py

@@ -19,6 +19,8 @@
 import httpx
 import structlog
 
+from autosearch.core.redact import redact, redact_url
+
 LOGGER = structlog.get_logger(__name__).bind(component="tool", skill="video-to-text-bcut")
 
 _BCUT_BASE = "https://member.bilibili.com/x/bcut/rubick-interface"
@@ -258,8 +260,13 @@ async def transcribe(source: str, timeout: float = _DEFAULT_TIMEOUT) -> BcutResu
         {ok, text, segments, duration_seconds, source}
         or {ok: False, reason, source}
     """
-    if not source or not source.strip():
-        return {"ok": False, "source": source, "reason": "empty source"}
+    if not isinstance(source, str):
+        return {"ok": False, "source": "", "reason": "source must be a string"}
+
+    if not source.strip():
+        return {"ok": False, "source": "", "reason": "empty source"}
+
+    redacted_source = redact_url(source)
 
     try:
         with tempfile.TemporaryDirectory() as tmpdir:
@@ -268,30 +275,50 @@ async def transcribe(source: str, timeout: float = _DEFAULT_TIMEOUT) -> BcutResu
             try:
                 _extract_audio_to_wav(source, wav_path)
             except Exception as exc:
-                LOGGER.warning("bcut_audio_extraction_failed", source=source, reason=str(exc))
-                return {"ok": False, "source": source, "reason": f"audio_extraction: {exc}"}
+                reason = redact(str(exc).replace(source, redacted_source))
+                LOGGER.warning(
+                    "bcut_audio_extraction_failed",
+                    source=redacted_source,
+                    reason=reason,
+                )
+                return {
+                    "ok": False,
+                    "source": redacted_source,
+                    "reason": f"audio_extraction: {reason}",
+                }
 
             try:
                 utterances = await asyncio.wait_for(
                     _bcut_transcribe(wav_path),
                     timeout=timeout,
                 )
             except Exception as exc:
-                LOGGER.warning("bcut_api_failed", source=source, reason=str(exc))
-                return {"ok": False, "source": source, "reason": f"bcut_api: {exc}"}
+                reason = redact(str(exc).replace(source, redacted_source))
+                LOGGER.warning(
+                    "bcut_api_failed",
+                    source=redacted_source,
+                    reason=reason,
+                )
+                return {"ok": False, "source": redacted_source, "reason": f"bcut_api: {reason}"}
 
             segments = _build_segments(utterances)
             full_text = "".join(s["text"] for s in segments)
             duration = max((s["end"] for s in segments), default=0.0)
 
             return {
                 "ok": True,
-                "source": source,
+                "source": redacted_source,
                 "text": full_text,
                 "segments": segments,
                 "duration_seconds": duration,
             }
 
     except Exception as exc:
-        LOGGER.exception("bcut_unexpected_error", source=source)
-        return {"ok": False, "source": source, "reason": f"unexpected: {exc}"}
+        reason = redact(str(exc).replace(source, redacted_source))
+        LOGGER.warning(
+            "bcut_unexpected_error",
+            source=redacted_source,
+            error_type=type(exc).__name__,
+            reason=reason,
+        )
+        return {"ok": False, "source": redacted_source, "reason": f"unexpected: {reason}"}

tests/tools/test_url_leak_prevention.py

@@ -9,8 +9,10 @@
 
 import httpx
 import pytest
+from structlog.testing import capture_logs
 
 from autosearch.core.models import SubQuery
+from autosearch.core.redact import redact_url
 
 ROOT = Path(__file__).resolve().parents[2]
 SECRET_TOKEN = "P0_SECRET_TOKEN_1234567890"
@@ -148,6 +150,116 @@ def transcribe(*_args: object, **_kwargs: object) -> dict[str, object]:
     return module.transcribe(SIGNED_URL, mlx_whisper_module=MlxWhisper)
 
 
+BCUT_SIGNED_URL = "https://cdn.example.com/audio.mp3?Signature=ABCD&Expires=999"
+
+
+def _load_bcut_tool() -> ModuleType:
+    return _load_tool("video_bcut", "autosearch/skills/tools/video-to-text-bcut/transcribe.py")
+
+
+def _assert_no_bcut_signature(value: Any) -> None:
+    serialized = json.dumps(value, sort_keys=True, default=str)
+    assert "Signature=" not in serialized
+    assert "ABCD" not in serialized
+
+
+def _configure_bcut_audio_extraction_failed(
+    module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    def fail_extract(source: str, _output_wav: Path) -> None:
+        raise RuntimeError(f"yt-dlp failed for {source}")
+
+    monkeypatch.setattr(module, "_extract_audio_to_wav", fail_extract)
+
+
+def _configure_bcut_api_failed(module: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None:
+    def extract_audio(_source: str, output_wav: Path) -> None:
+        output_wav.write_bytes(b"fake wav")
+
+    async def fail_transcribe(_audio_path: Path) -> list[dict[str, object]]:
+        raise RuntimeError(f"bcut failed for {BCUT_SIGNED_URL}")
+
+    monkeypatch.setattr(module, "_extract_audio_to_wav", extract_audio)
+    monkeypatch.setattr(module, "_bcut_transcribe", fail_transcribe)
+
+
+def _configure_bcut_unexpected_error(module: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None:
+    def extract_audio(_source: str, output_wav: Path) -> None:
+        output_wav.write_bytes(b"fake wav")
+
+    async def transcribe_audio(_audio_path: Path) -> list[dict[str, object]]:
+        return [{"transcript": "hello", "words": []}]
+
+    def fail_build_segments(_utterances: list[dict[str, object]]) -> list[dict[str, object]]:
+        raise RuntimeError(f"unexpected failure for {BCUT_SIGNED_URL}")
+
+    monkeypatch.setattr(module, "_extract_audio_to_wav", extract_audio)
+    monkeypatch.setattr(module, "_bcut_transcribe", transcribe_audio)
+    monkeypatch.setattr(module, "_build_segments", fail_build_segments)
+
+
+BCUT_FAILURE_CONFIGURERS = (
+    _configure_bcut_audio_extraction_failed,
+    _configure_bcut_api_failed,
+    _configure_bcut_unexpected_error,
+)
+
+
+def test_bcut_log_redacts_source(monkeypatch: pytest.MonkeyPatch) -> None:
+    for configure_failure in BCUT_FAILURE_CONFIGURERS:
+        module = _load_bcut_tool()
+        configure_failure(module, monkeypatch)
+
+        with capture_logs() as logs:
+            asyncio.run(module.transcribe(BCUT_SIGNED_URL))
+
+        _assert_no_bcut_signature(logs)
+
+
+def test_bcut_result_reason_redacts_source(monkeypatch: pytest.MonkeyPatch) -> None:
+    for configure_failure in BCUT_FAILURE_CONFIGURERS:
+        module = _load_bcut_tool()
+        configure_failure(module, monkeypatch)
+
+        result = asyncio.run(module.transcribe(BCUT_SIGNED_URL))
+
+        assert result["ok"] is False
+        _assert_no_bcut_signature(result["reason"])
+
+
+def test_bcut_structured_output_redacts_source(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = _load_bcut_tool()
+    _configure_bcut_audio_extraction_failed(module, monkeypatch)
+
+    result = asyncio.run(module.transcribe(BCUT_SIGNED_URL))
+
+    assert result["source"] == redact_url(BCUT_SIGNED_URL)
+    _assert_no_bcut_signature(result["source"])
+
+
+def test_bcut_handles_non_string_source() -> None:
+    module = _load_bcut_tool()
+
+    result = asyncio.run(module.transcribe(None))
+
+    assert result == {"ok": False, "source": "", "reason": "source must be a string"}
+
+
+def test_bcut_unexpected_error_no_traceback_leak(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = _load_bcut_tool()
+    _configure_bcut_unexpected_error(module, monkeypatch)
+
+    with capture_logs() as logs:
+        result = asyncio.run(module.transcribe(BCUT_SIGNED_URL))
+
+    assert result["ok"] is False
+    assert any(log.get("event") == "bcut_unexpected_error" for log in logs)
+    for log in logs:
+        serialized = json.dumps(log, sort_keys=True, default=str)
+        assert "Signature=" not in serialized
+        assert not log.get("exc_info")
+
+
 @pytest.mark.parametrize(
     ("tool_key", "runner"),
     [