fix(security): stop proxy credentials leaking from subprocess/urlparse errors

shared/libs/youtube_proxy.py

@@ -12,6 +12,7 @@
 import logging
 import os
 import random
+import re
 import time
 import urllib.parse
 from dataclasses import dataclass
@@ -39,15 +40,42 @@
 logger = logging.getLogger("youtube_api_proxy")
 
 
+_ALLOWED_SCHEMES = ("http", "https", "socks5", "socks5h")
+
+# Matches the ``user[:password]@`` userinfo segment of any URL. The user and
+# password classes exclude "/" so a path containing "@" is never mistaken for
+# credentials. Kept in sync with youtube_extension.utils.proxy.
+_USERINFO_RE = re.compile(
+    r"(?P<scheme>[A-Za-z][A-Za-z0-9+.\-]*://)"
+    r"(?P<user>[^\s/:@]+)"
+    r"(?::(?P<password>[^\s/@]*))?"
+    r"@"
+)
+
+
 def _get_webshare_proxy_url() -> str | None:
-    """Return the validated WEBSHARE_PROXY_URL, or None for direct connection."""
+    """Return the validated WEBSHARE_PROXY_URL, or None for direct connection.
+
+    Mirrors ``youtube_extension.utils.proxy.get_proxy_url``. ``urllib.parse``
+    raises ``ValueError`` on several malformed inputs (an unterminated IPv6
+    literal at parse time; a non-numeric or out-of-range port when ``.port`` is
+    read). Contain it here so the exception -- which callers would log next to
+    the credential-bearing URL -- never escapes.
+    """
     url = os.getenv("WEBSHARE_PROXY_URL", "").strip()
     if not url:
         return None
-    parsed = urllib.parse.urlparse(url)
-    if parsed.scheme not in ("http", "https", "socks5") or not parsed.hostname:
+    try:
+        parsed = urllib.parse.urlparse(url)
+        valid = parsed.scheme in _ALLOWED_SCHEMES and bool(parsed.hostname)
+        if valid:
+            parsed.port  # noqa: B018 - validates port, raises ValueError if bad
+    except ValueError:
+        valid = False
+    if not valid:
         logger.warning(
-            "WEBSHARE_PROXY_URL is set but malformed — falling back to direct connection"
+            "WEBSHARE_PROXY_URL is set but malformed - falling back to direct "
+            "connection"
         )
         return None
     return url
@@ -65,27 +93,59 @@ def _get_transcript_proxy_config() -> GenericProxyConfig | None:
     if GenericProxyConfig is None:
         logger.warning(
             "WEBSHARE_PROXY_URL is set but youtube-transcript-api proxy support "
-            "is unavailable (requires youtube-transcript-api>=1.0) — falling back "
+            "is unavailable (requires youtube-transcript-api>=1.0) - falling back "
             "to direct connection"
         )
         return None
     return GenericProxyConfig(http_url=url, https_url=url)
 
 
-def _redact_proxy_credentials(text: str) -> str:
-    """Strip user:pass credentials of the configured proxy URL from text."""
+def _redact_proxy_credentials(text: Any) -> str:
+    """Strip URL userinfo (``user:pass@``) from ``text``.
+
+    Delegates to the canonical helper when the ``youtube_extension`` package is
+    importable; this module is also loaded standalone by path (see
+    ``src/agents/mcp_enhanced_video_processor.py``), so an equivalent local
+    implementation is retained as a fallback. Never raises -- it is called from
+    exception handlers, where a failure would mask the original error.
+    """
+    try:  # pragma: no cover - exercised only when the package is importable
+        from youtube_extension.utils.proxy import (
+            redact_proxy_credentials as _canonical,
+        )
+
+        return _canonical(text)
+    except ImportError:
+        pass
+
+    if not isinstance(text, str):
+        try:
+            text = str(text)
+        except Exception:
+            # "Never raises" contract: this fallback runs inside exception
+            # handlers, so a failing __str__ must not propagate and mask the
+            # original error. Return a fixed, non-sensitive placeholder.
+            return "<unprintable error>"
+
     url = os.getenv("WEBSHARE_PROXY_URL", "").strip()
-    if not url or url not in text:
-        return text
-    try:
-        parsed = urllib.parse.urlparse(url)
-        netloc = parsed.hostname or ""
-        if parsed.port:
-            netloc = f"{netloc}:{parsed.port}"
-        redacted = parsed._replace(netloc=netloc).geturl()
-    except (ValueError, AttributeError):
-        redacted = "<proxy-url>"
-    return text.replace(url, redacted)
+    if url and url in text:
+        try:
+            parsed = urllib.parse.urlparse(url)
+            netloc = parsed.hostname or ""
+            if parsed.port:
+                netloc = f"{netloc}:{parsed.port}"
+            redacted = parsed._replace(netloc=netloc).geturl()
+        except (ValueError, AttributeError):
+            redacted = "<proxy-url>"
+        text = text.replace(url, redacted)
+
+    def _mask(match: re.Match[str]) -> str:
+        if match.group("password") is None:
+            return f"{match.group('scheme')}***@"
+        return f"{match.group('scheme')}***:***@"
+
+    return _USERINFO_RE.sub(_mask, text)
+
 
 class YouTubeErrorType(Enum):
     """YouTube API specific error types"""

src/youtube_extension/backend/enhanced_video_processor.py

@@ -25,7 +25,11 @@
 # Load environment variables
 load_dotenv()
 
-from youtube_extension.utils.proxy import get_proxy_url, get_transcript_proxy_config
+from youtube_extension.utils.proxy import (
+    get_proxy_url,
+    get_transcript_proxy_config,
+    redact_proxy_credentials,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -316,8 +320,12 @@ async def _get_openai_whisper_transcript(self, video_id: str, video_url: str) ->
                 'processing_time': datetime.now().isoformat()
             }
         except Exception as e:
-            logger.warning(f"OpenAI Whisper failed: {e}")
-            return {'text': '', 'source': 'failed', 'error': str(e)}
+            # subprocess.run(check=True) raises CalledProcessError whose str()
+            # embeds the whole argv — including "--proxy <url-with-credentials>".
+            # Redact before it reaches the log or the returned error field.
+            detail = redact_proxy_credentials(e)
+            logger.warning(f"OpenAI Whisper failed: {detail}")
+            return {'text': '', 'source': 'failed', 'error': detail}
 
     async def _get_youtube_transcript_fallback(self, video_id: str) -> Dict[str, Any]:
         """Fallback to YouTube transcript API"""

src/youtube_extension/backend/services/youtube/adapters/robust.py

@@ -20,7 +20,11 @@
 
 import httpx
 
-from youtube_extension.utils.proxy import get_proxy_url, get_transcript_proxy_config
+from youtube_extension.utils.proxy import (
+    get_proxy_url,
+    get_transcript_proxy_config,
+    redact_proxy_credentials,
+)
 
 # Import our cost monitor
 try:
@@ -140,7 +144,9 @@ async def get_video_metadata(self, video_url: str) -> RobustYouTubeMetadata:
         try:
             return await self._get_metadata_ytdlp(video_url, video_id)
         except Exception as e:
-            logger.warning(f"yt-dlp fallback failed: {e}")
+            # subprocess.TimeoutExpired stringifies the whole argv, which carries
+            # "--proxy <url-with-credentials>"; redact before logging.
+            logger.warning(f"yt-dlp fallback failed: {redact_proxy_credentials(e)}")
 
         raise Exception("All YouTube metadata APIs failed")
 
@@ -167,7 +173,11 @@ def _get_ytdlp_metadata():
                 timeout=30,
             )
             if result.returncode != 0:
-                raise Exception(f"yt-dlp failed: {result.stderr}")
+                # yt-dlp echoes the --proxy value back on stderr for connection
+                # failures, so redact before the message is raised and logged.
+                raise Exception(
+                    f"yt-dlp failed: {redact_proxy_credentials(result.stderr)}"
+                )
             return json.loads(result.stdout)
 
         data = await asyncio.get_event_loop().run_in_executor(None, _get_ytdlp_metadata)

src/youtube_extension/utils/proxy.py

@@ -13,6 +13,7 @@
 
 import logging
 import os
+import re
 import urllib.parse
 from typing import Any
 
@@ -28,14 +29,41 @@
 
 _PROXY_ENV_VAR = "WEBSHARE_PROXY_URL"
 
+_ALLOWED_SCHEMES = ("http", "https", "socks5", "socks5h")
+
+# Matches the ``user[:password]@`` userinfo segment of any URL. The user and
+# password classes exclude "/" so a path containing "@" (e.g.
+# "https://example.com/a@b") is never mistaken for credentials.
+_USERINFO_RE = re.compile(
+    r"(?P<scheme>[A-Za-z][A-Za-z0-9+.\-]*://)"
+    r"(?P<user>[^\s/:@]+)"
+    r"(?::(?P<password>[^\s/@]*))?"
+    r"@"
+)
+
+_REDACTED = "***"
+
 
 def get_proxy_url() -> str | None:
-    """Return the validated Webshare proxy URL, or None for direct connection."""
+    """Return the validated Webshare proxy URL, or None for direct connection.
+
+    Never raises and never emits the URL (which carries credentials) into logs.
+    ``urllib.parse`` raises ``ValueError`` for several malformed inputs — an
+    unterminated IPv6 literal at parse time, a non-numeric or out-of-range port
+    when ``.port`` is read — so both are contained here rather than escaping to
+    callers that would log the exception alongside the offending URL.
+    """
     url = os.getenv(_PROXY_ENV_VAR, "").strip()
     if not url:
         return None
-    parsed = urllib.parse.urlparse(url)
-    if parsed.scheme not in ("http", "https", "socks5") or not parsed.hostname:
+    try:
+        parsed = urllib.parse.urlparse(url)
+        valid = parsed.scheme in _ALLOWED_SCHEMES and bool(parsed.hostname)
+        if valid:
+            parsed.port  # noqa: B018 - validates the port, raises ValueError if bad
+    except ValueError:
+        valid = False
+    if not valid:
         logger.warning(
             "%s is set but malformed — falling back to direct connection",
             _PROXY_ENV_VAR,
@@ -67,17 +95,46 @@ def get_transcript_proxy_config() -> Any | None:
     return GenericProxyConfig(http_url=url, https_url=url)
 
 
-def redact_proxy_credentials(text: str) -> str:
-    """Strip user:pass credentials of the configured proxy URL from text."""
+def redact_proxy_credentials(text: Any) -> str:
+    """Strip URL userinfo (``user:pass@``) from ``text``.
+
+    Two passes, because either alone is insufficient:
+
+    1. Exact replacement of the configured ``WEBSHARE_PROXY_URL`` — preserves
+       the host so operators can still tell *which* proxy was in play.
+    2. A generic ``scheme://user:pass@`` sweep — catches credentials that never
+       match the env value verbatim: yt-dlp echoing a normalised/percent-encoded
+       form back on stderr, a ``CalledProcessError`` repr of the argv, or a
+       different proxy variable (``HTTPS_PROXY`` and friends) entirely.
+
+    Always returns a string and never raises; it is called from exception
+    handlers, where a failure would mask the original error.
+    """
+    if not isinstance(text, str):
+        try:
+            text = str(text)
+        except Exception:
+            # The contract is "never raises": an object whose __str__ fails
+            # here is called from an exception handler, so propagating would
+            # mask the original error. Fall back to a fixed, non-sensitive
+            # placeholder instead.
+            return "<unprintable error>"
+
     url = os.getenv(_PROXY_ENV_VAR, "").strip()
-    if not url or url not in text:
-        return text
-    try:
-        parsed = urllib.parse.urlparse(url)
-        netloc = parsed.hostname or ""
-        if parsed.port:
-            netloc = f"{netloc}:{parsed.port}"
-        redacted = parsed._replace(netloc=netloc).geturl()
-    except (ValueError, AttributeError):
-        redacted = "<proxy-url>"
-    return text.replace(url, redacted)
+    if url and url in text:
+        try:
+            parsed = urllib.parse.urlparse(url)
+            netloc = parsed.hostname or ""
+            if parsed.port:
+                netloc = f"{netloc}:{parsed.port}"
+            redacted = parsed._replace(netloc=netloc).geturl()
+        except (ValueError, AttributeError):
+            redacted = "<proxy-url>"
+        text = text.replace(url, redacted)
+
+    def _mask(match: re.Match[str]) -> str:
+        if match.group("password") is None:
+            return f"{match.group('scheme')}{_REDACTED}@"
+        return f"{match.group('scheme')}{_REDACTED}:{_REDACTED}@"
+
+    return _USERINFO_RE.sub(_mask, text)

tests/unit/test_enhanced_video_processor.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import sys
 from contextlib import asynccontextmanager
 from pathlib import Path
@@ -639,6 +640,42 @@ async def test_yt_dlp_uses_canonical_url_after_option_terminator(self, tmp_path)
         assert hostile_url not in command
         assert result["text"] == "safe transcript"
 
+    async def test_subprocess_failure_never_leaks_proxy_credentials(
+        self, tmp_path, monkeypatch, caplog
+    ):
+        """Regression for #1113.
+
+        ``subprocess.run(..., check=True)`` raises ``CalledProcessError`` whose
+        ``str()`` renders the entire argv — which carries
+        ``--proxy http://user:pass@host``. That string reached both the warning
+        log (CWE-532) and the returned ``error`` field (CWE-209).
+        """
+        import subprocess
+
+        password = "sup3r-s3cret-pw"
+        proxy_url = f"http://wsuser:{password}@p.webshare.io:80"
+        monkeypatch.setenv("WEBSHARE_PROXY_URL", proxy_url)
+
+        proc = _make_processor()
+
+        def _raise(cmd, *args, **kwargs):
+            raise subprocess.CalledProcessError(returncode=1, cmd=cmd)
+
+        with patch.dict(sys.modules, {"openai": MagicMock()}):
+            with patch("tempfile.TemporaryDirectory") as temp_dir:
+                temp_dir.return_value.__enter__.return_value = str(tmp_path)
+                with patch("subprocess.run", side_effect=_raise):
+                    with caplog.at_level(logging.WARNING):
+                        result = await proc._get_openai_whisper_transcript(
+                            _VIDEO_ID, _VIDEO_URL
+                        )
+
+        assert result["source"] == "failed"
+        assert password not in result["error"]
+        assert "wsuser" not in result["error"]
+        assert password not in caplog.text
+        assert "wsuser" not in caplog.text
+
 
 # ===========================================================================
 # _get_gemini_transcript