test(proxy): harden transcript-proxy guard against two silent bypasses (hardens #1129)

src/agents/interactive_metadata_extractor.py

@@ -16,6 +16,25 @@
 from dotenv import load_dotenv
 from youtube_transcript_api import YouTubeTranscriptApi
 
+try:
+    from youtube_extension.utils.proxy import get_transcript_proxy_config
+except ImportError:  # pragma: no cover - standalone execution outside the package
+    import sys as _sys
+    from pathlib import Path as _Path
+
+    # Running this module by path (the documented CLI entry point) leaves the
+    # repository's ``src`` directory off sys.path. Bootstrap it so the canonical
+    # helper resolves instead of silently disabling the proxy.
+    _sys.path.insert(0, str(_Path(__file__).resolve().parents[1]))
+    try:
+        from youtube_extension.utils.proxy import get_transcript_proxy_config
+    except ImportError as _exc:  # pragma: no cover - helper genuinely unreachable
+        raise ImportError(
+            "youtube_extension.utils.proxy is required so transcript requests "
+            "honour WEBSHARE_PROXY_URL; refusing to continue with unproxied "
+            "egress."
+        ) from _exc
+
 load_dotenv()
 logger = logging.getLogger(__name__)
 
@@ -84,7 +103,10 @@ async def extract_transcript(self, video_id: str) -> list[dict[str, Any]]:
             # event loop free.
             loop = asyncio.get_event_loop()
             transcript = await loop.run_in_executor(
-                None, lambda: YouTubeTranscriptApi().fetch(video_id).to_raw_data()
+                None,
+                lambda: YouTubeTranscriptApi(proxy_config=get_transcript_proxy_config())
+                .fetch(video_id)
+                .to_raw_data(),
             )
 
             for i, entry in enumerate(transcript):

src/agents/process_video_with_mcp.py

@@ -49,6 +49,25 @@
     HAS_YTA = False
     YouTubeTranscriptApi = None  # type: ignore
 
+try:
+    from youtube_extension.utils.proxy import get_transcript_proxy_config
+except ImportError:  # pragma: no cover - standalone execution outside the package
+    import sys as _sys
+    from pathlib import Path as _Path
+
+    # Running this module by path (the documented CLI entry point) leaves the
+    # repository's ``src`` directory off sys.path. Bootstrap it so the canonical
+    # helper resolves instead of silently disabling the proxy.
+    _sys.path.insert(0, str(_Path(__file__).resolve().parents[1]))
+    try:
+        from youtube_extension.utils.proxy import get_transcript_proxy_config
+    except ImportError as _exc:  # pragma: no cover - helper genuinely unreachable
+        raise ImportError(
+            "youtube_extension.utils.proxy is required so transcript requests "
+            "honour WEBSHARE_PROXY_URL; refusing to continue with unproxied "
+            "egress."
+        ) from _exc
+
 try:
     import yt_dlp  # type: ignore
 except Exception:  # Provide a minimal stub so tests can patch attribute
@@ -219,7 +238,12 @@ async def _extract_transcript_with_rotation(self, video_id: str) -> list[dict[st
             if YouTubeTranscriptApi is not None:
                 # youtube-transcript-api >=1.0 instance API
                 transcript = await loop.run_in_executor(
-                    None, lambda: YouTubeTranscriptApi().fetch(video_id).to_raw_data()
+                    None,
+                    lambda: YouTubeTranscriptApi(
+                        proxy_config=get_transcript_proxy_config()
+                    )
+                    .fetch(video_id)
+                    .to_raw_data(),
                 )
                 if transcript:
                     return transcript
@@ -230,7 +254,10 @@ async def _extract_transcript_with_rotation(self, video_id: str) -> list[dict[st
         try:
             if YouTubeTranscriptApi is not None:
                 transcript_list = await loop.run_in_executor(
-                    None, lambda: YouTubeTranscriptApi().list(video_id)  # type: ignore[union-attr]
+                    None,
+                    lambda: YouTubeTranscriptApi(  # type: ignore[union-attr]
+                        proxy_config=get_transcript_proxy_config()
+                    ).list(video_id),
                 )
                 fetch_tasks = [
                     loop.run_in_executor(None, lambda t=t: t.fetch().to_raw_data())

src/integration/youtube_api.py

@@ -12,6 +12,25 @@
 import httpx
 from youtube_transcript_api import YouTubeTranscriptApi
 
+try:
+    from youtube_extension.utils.proxy import get_transcript_proxy_config
+except ImportError:  # pragma: no cover - standalone execution outside the package
+    import sys as _sys
+    from pathlib import Path as _Path
+
+    # Running this module by path (the documented CLI entry point) leaves the
+    # repository's ``src`` directory off sys.path. Bootstrap it so the canonical
+    # helper resolves instead of silently disabling the proxy.
+    _sys.path.insert(0, str(_Path(__file__).resolve().parents[1]))
+    try:
+        from youtube_extension.utils.proxy import get_transcript_proxy_config
+    except ImportError as _exc:  # pragma: no cover - helper genuinely unreachable
+        raise ImportError(
+            "youtube_extension.utils.proxy is required so transcript requests "
+            "honour WEBSHARE_PROXY_URL; refusing to continue with unproxied "
+            "egress."
+        ) from _exc
+
 
 @dataclass
 class VideoMetadata:
@@ -96,7 +115,9 @@ async def get_transcript(
         loop = asyncio.get_event_loop()
         transcript = await loop.run_in_executor(
             None,
-            lambda: YouTubeTranscriptApi().fetch(video_id, languages=languages).to_raw_data()
+            lambda: YouTubeTranscriptApi(proxy_config=get_transcript_proxy_config())
+            .fetch(video_id, languages=languages)
+            .to_raw_data()
         )
 
         return [

src/mcp/mcp_video_processor.py

@@ -77,6 +77,25 @@ class IpBlocked(Exception):
 except Exception:
     HAS_YT_PROXY = False
 
+try:
+    from youtube_extension.utils.proxy import get_transcript_proxy_config
+except ImportError:  # pragma: no cover - standalone execution outside the package
+    import sys as _sys
+    from pathlib import Path as _Path
+
+    # Running this module by path (the documented CLI entry point) leaves the
+    # repository's ``src`` directory off sys.path. Bootstrap it so the canonical
+    # helper resolves instead of silently disabling the proxy.
+    _sys.path.insert(0, str(_Path(__file__).resolve().parents[1]))
+    try:
+        from youtube_extension.utils.proxy import get_transcript_proxy_config
+    except ImportError as _exc:  # pragma: no cover - helper genuinely unreachable
+        raise ImportError(
+            "youtube_extension.utils.proxy is required so transcript requests "
+            "honour WEBSHARE_PROXY_URL; refusing to continue with unproxied "
+            "egress."
+        ) from _exc
+
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
@@ -691,7 +710,7 @@ async def _direct_extraction():
             # executor — otherwise it stalls the event loop and defeats the
             # @timeout_protection / circuit-breaker hanging protection.
             loop = asyncio.get_event_loop()
-            yt_api = YouTubeTranscriptApi()
+            yt_api = YouTubeTranscriptApi(proxy_config=get_transcript_proxy_config())
             transcript = await loop.run_in_executor(
                 None,
                 lambda: yt_api.fetch(
@@ -715,7 +734,7 @@ async def _routed_extraction():
             # These are blocking network calls — run them in an executor to keep
             # the event loop free and let the timeout protection work.
             loop = asyncio.get_event_loop()
-            yt_api = YouTubeTranscriptApi()
+            yt_api = YouTubeTranscriptApi(proxy_config=get_transcript_proxy_config())
             transcript_list = await loop.run_in_executor(
                 None, lambda: yt_api.list(video_id)
             )