fix(security): require CLOUD_AI_MEDIA_ROOT to be an existing directory (#1216 review)

.env.example

@@ -154,6 +154,20 @@ CACHE_DIR=youtube_processed_videos/markdown_analysis
 ENHANCED_ANALYSIS_DIR=youtube_processed_videos/enhanced_analysis
 FEEDBACK_DIR=youtube_processed_videos/feedback
 
+# ----------------------------------------------------------------------------
+# Cloud AI local media sandbox (AWS Rekognition / Azure Vision / Google Vision)
+# ----------------------------------------------------------------------------
+# The cloud AI providers accept an `image_url` that may be an s3:// URI, an
+# http(s) URL, or a local filesystem path. Local paths are DISABLED by default
+# (fail-closed) so that a caller-supplied path cannot be used to read arbitrary
+# files off the host.
+#
+# To enable local-path reads (dev/self-hosted only), set this to a directory
+# that contains ONLY media you are willing to expose. Paths are fully resolved,
+# so `../` traversal and symlinks that escape the root are rejected.
+# Leave empty in production: use s3:// or https:// inputs instead.
+CLOUD_AI_MEDIA_ROOT=
+
 # ============================================================================
 # SECURITY & AUTH
 # ============================================================================

src/youtube_extension/integrations/cloud_ai/__init__.py

@@ -12,8 +12,14 @@
     VideoAnalysisResult,
 )
 from .config import CloudAIConfig
-from .exceptions import CloudAIError, ConfigurationError, RateLimitError
+from .exceptions import (
+    CloudAIError,
+    ConfigurationError,
+    RateLimitError,
+    UnsafeMediaPathError,
+)
 from .integrator import CloudAIIntegrator
+from .media_paths import MEDIA_ROOT_ENV_VAR, get_media_root, resolve_local_media_path
 
 __all__ = [
     "BaseCloudAI",
@@ -25,5 +31,9 @@
     "CloudAIConfig",
     "CloudAIError",
     "RateLimitError",
-    "ConfigurationError"
+    "ConfigurationError",
+    "UnsafeMediaPathError",
+    "MEDIA_ROOT_ENV_VAR",
+    "get_media_root",
+    "resolve_local_media_path",
 ]

src/youtube_extension/integrations/cloud_ai/exceptions.py

@@ -57,3 +57,18 @@ def __init__(self, message: str, provider: Optional[str] = None,
                  quota_type: Optional[str] = None):
         super().__init__(message, provider, "QUOTA_EXCEEDED")
         self.quota_type = quota_type
+
+
+class UnsafeMediaPathError(CloudAIError):
+    """Exception raised when a caller-supplied local media path is rejected.
+
+    Raised instead of reading the file, so a traversal attempt fails loudly
+    rather than silently returning empty bytes. ``requested_path`` echoes only
+    the value the caller already supplied -- the resolved server-side path is
+    deliberately not included, to avoid disclosing the filesystem layout.
+    """
+
+    def __init__(self, message: str, provider: Optional[str] = None,
+                 requested_path: Optional[str] = None):
+        super().__init__(message, provider, "UNSAFE_MEDIA_PATH")
+        self.requested_path = requested_path

src/youtube_extension/integrations/cloud_ai/media_paths.py

@@ -0,0 +1,162 @@
+"""
+Safe resolution of local media paths for cloud AI providers.
+
+Every provider in this package exposes ``analyze_image(image_url, ...)`` and
+dispatches on the string's prefix. The remote schemes each provider recognises
+differ, and *anything else* used to be treated as a local filesystem path and
+opened verbatim. That final branch would happily read ``/etc/passwd`` or
+``../../secrets.env`` if a caller supplied it. The recognised remote schemes,
+per provider, are:
+
+* ``aws_rekognition``: ``s3://`` and ``http(s)://``.
+* ``azure_vision``: ``http(s)://`` only (``s3://`` falls through to the local
+  branch and is now rejected unless it names a real in-root path).
+* ``google_cloud``: ``http(s)://`` only (same ``s3://`` caveat as Azure).
+
+This module centralises the guard so all three providers share one policy:
+
+* Local reads are **opt-in**. With ``CLOUD_AI_MEDIA_ROOT`` unset, every local
+  path is rejected, so only each provider's recognised remote schemes above
+  remain usable. This is the production posture; the local branch is a
+  development convenience.
+* When a root *is* configured, a candidate path is fully resolved
+  (``Path.resolve()`` follows symlinks) and must live inside the equally
+  resolved root. That covers symlink escapes, not just lexical ``..`` segments.
+* Rejection raises :class:`UnsafeMediaPathError` rather than returning empty
+  bytes, so failures are loud.
+
+Callers should read from the returned resolved path rather than the original
+caller-supplied string: the returned path is the one that was validated, which
+narrows (though does not eliminate) the check-to-open race.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+from .exceptions import ConfigurationError, UnsafeMediaPathError
+
+logger = logging.getLogger(__name__)
+
+#: Environment variable naming the directory local media may be read from.
+#: Unset (the default) disables local reads entirely.
+MEDIA_ROOT_ENV_VAR = "CLOUD_AI_MEDIA_ROOT"
+
+__all__ = [
+    "MEDIA_ROOT_ENV_VAR",
+    "get_media_root",
+    "resolve_local_media_path",
+]
+
+
+def get_media_root() -> Path | None:
+    """Return the configured media root, or ``None`` when local reads are off.
+
+    A relative value is resolved against the process working directory. The
+    root is resolved with symlinks followed so that containment checks compare
+    real paths on both sides.
+
+    The resolved value must be an existing directory. Rejecting a root that
+    points at a regular file (e.g. ``CLOUD_AI_MEDIA_ROOT=/etc/passwd``) or at a
+    non-existent path is what keeps the guard fail-closed: without this check,
+    ``resolve_local_media_path`` would treat that single file as "inside" the
+    root via ``is_relative_to`` and hand it straight to the caller.
+
+    Raises:
+        ConfigurationError: if the variable is set to a value that cannot be
+            resolved, or that does not resolve to an existing directory.
+    """
+    raw = os.environ.get(MEDIA_ROOT_ENV_VAR)
+    if raw is None or not raw.strip():
+        return None
+
+    try:
+        root = Path(raw.strip()).expanduser().resolve()
+    except (OSError, RuntimeError) as exc:
+        # RuntimeError covers symlink loops on older resolvers; OSError covers
+        # unreadable path components and platform-specific failures.
+        raise ConfigurationError(
+            f"{MEDIA_ROOT_ENV_VAR} is not a resolvable directory path",
+            missing_config=MEDIA_ROOT_ENV_VAR,
+        ) from exc
+
+    if not root.is_dir():
+        raise ConfigurationError(
+            f"{MEDIA_ROOT_ENV_VAR} must point to an existing directory",
+            missing_config=MEDIA_ROOT_ENV_VAR,
+        )
+
+    return root
+
+
+def resolve_local_media_path(candidate: str, provider: str | None = None) -> Path:
+    """Validate a caller-supplied local media path and return its real path.
+
+    Args:
+        candidate: The path exactly as supplied by the caller.
+        provider: Provider name, attached to raised errors for context.
+
+    Returns:
+        The fully resolved path, guaranteed to sit inside the configured root.
+
+    Raises:
+        UnsafeMediaPathError: if local reads are disabled, the path escapes the
+            configured root (lexically or via symlink), or it resolves to
+            something that is not a regular file.
+        ConfigurationError: if ``CLOUD_AI_MEDIA_ROOT`` is set but unusable.
+    """
+    if not candidate or not candidate.strip():
+        raise UnsafeMediaPathError(
+            "Local media path is empty",
+            provider=provider,
+            requested_path=candidate,
+        )
+
+    root = get_media_root()
+    if root is None:
+        raise UnsafeMediaPathError(
+            "Local media reads are disabled. Use an s3:// or https:// source, "
+            f"or set {MEDIA_ROOT_ENV_VAR} to the directory local media may be "
+            "read from.",
+            provider=provider,
+            requested_path=candidate,
+        )
+
+    try:
+        resolved = Path(candidate).expanduser().resolve()
+    except (OSError, RuntimeError) as exc:
+        raise UnsafeMediaPathError(
+            "Local media path could not be resolved",
+            provider=provider,
+            requested_path=candidate,
+        ) from exc
+
+    if not resolved.is_relative_to(root):
+        # Log the resolution server-side for forensics; keep it out of the
+        # exception so the path is not echoed back to an untrusted caller.
+        logger.warning(
+            "Rejected local media path outside %s: %r resolved to %s",
+            MEDIA_ROOT_ENV_VAR,
+            candidate,
+            resolved,
+        )
+        raise UnsafeMediaPathError(
+            "Local media path is outside the permitted media root",
+            provider=provider,
+            requested_path=candidate,
+        )
+
+    # ``resolve()`` follows symlinks, so a link inside the root that points out
+    # of it has already been rejected above. What remains is to refuse
+    # non-regular files: a FIFO or character device placed inside the root
+    # would otherwise block a worker thread indefinitely on read.
+    if resolved.exists() and not resolved.is_file():
+        raise UnsafeMediaPathError(
+            "Local media path is not a regular file",
+            provider=provider,
+            requested_path=candidate,
+        )
+
+    return resolved

src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py

@@ -29,6 +29,7 @@
     ConfigurationError,
     RateLimitError,
 )
+from ..media_paths import resolve_local_media_path
 
 logger = logging.getLogger(__name__)
 
@@ -112,7 +113,6 @@ def _read_file_bytes(path: str) -> bytes:
         return handle.read()
 
 
-
 class AWSRekognition(BaseCloudAI):
     """Amazon Rekognition video and image analysis integration."""
 
@@ -300,6 +300,11 @@ async def analyze_image(self, image_url: str,
                 results, image_url, analysis_types, processing_time
             )
 
+        except CloudAIError:
+            # Typed errors (e.g. UnsafeMediaPathError from the local-path guard)
+            # already carry provider and error_code; re-wrapping them would
+            # flatten them into a generic CloudAIError and lose that type.
+            raise
         except Exception as e:
             raise CloudAIError(
                 f"AWS Rekognition image analysis failed: {e}",
@@ -483,8 +488,12 @@ async def _prepare_image_input(self, image_url: str) -> dict[str, Any]:
                 response = await client.get(image_url)
                 return {'Bytes': response.content}
         else:
-            # Local file - read off the event loop
-            return {'Bytes': await asyncio.to_thread(_read_file_bytes, image_url)}
+            # Local file. The path is caller-supplied, so it is validated
+            # against CLOUD_AI_MEDIA_ROOT first (raises UnsafeMediaPathError on
+            # traversal or symlink escape); the read then uses the resolved
+            # path, off the event loop.
+            safe_path = resolve_local_media_path(image_url, provider=self.provider.value)
+            return {'Bytes': await asyncio.to_thread(_read_file_bytes, str(safe_path))}
 
     def _process_video_results(self, results: dict[str, Any], video_id: str,
                              analysis_types: list[AnalysisType],

src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py

@@ -25,6 +25,7 @@
     ConfigurationError,
     RateLimitError,
 )
+from ..media_paths import resolve_local_media_path
 
 logger = logging.getLogger(__name__)
 
@@ -187,6 +188,11 @@ async def analyze_image(self, image_url: str,
                 results, image_url, analysis_types, processing_time
             )
 
+        except CloudAIError:
+            # Typed errors (e.g. UnsafeMediaPathError from the local-path guard)
+            # already carry provider and error_code; re-wrapping them would
+            # flatten them into a generic CloudAIError and lose that type.
+            raise
         except Exception as e:
             raise CloudAIError(
                 f"Azure AI Vision image analysis failed: {e}",
@@ -251,9 +257,11 @@ async def _prepare_image_input(self, image_url: str) -> Optional[bytes]:
             # For URL input, Azure can analyze directly
             return None
         else:
-            # For local files, read content
-            with open(image_url, 'rb') as image_file:
-                return image_file.read()
+            # Local file. The path is caller-supplied, so validate it against
+            # CLOUD_AI_MEDIA_ROOT before opening anything; the read then uses
+            # the resolved path rather than the raw string.
+            safe_path = resolve_local_media_path(image_url, provider=self.provider.value)
+            return safe_path.read_bytes()
 
     async def _await_ocr_call(self, deadline: float, func: Any, *args: Any, **kwargs: Any) -> Any:
         """Run a blocking Azure SDK call in a worker thread, bounded by a

src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py

@@ -27,6 +27,7 @@
     ConfigurationError,
     RateLimitError,
 )
+from ..media_paths import resolve_local_media_path
 
 logger = logging.getLogger(__name__)
 
@@ -181,9 +182,12 @@ async def analyze_image(self, image_url: str,
             if image_url.startswith(('http://', 'https://')):
                 image.source.image_uri = image_url
             else:
-                # For local files
-                with open(image_url, 'rb') as image_file:
-                    image.content = image_file.read()
+                # Local file. The path is caller-supplied, so validate it
+                # against CLOUD_AI_MEDIA_ROOT before opening anything.
+                safe_path = resolve_local_media_path(
+                    image_url, provider=self.provider.value
+                )
+                image.content = safe_path.read_bytes()
 
             # Prepare features
             features = self._prepare_vision_features(analysis_types)
@@ -200,6 +204,11 @@ async def analyze_image(self, image_url: str,
                 response, image_url, analysis_types, processing_time
             )
 
+        except CloudAIError:
+            # Typed errors (e.g. UnsafeMediaPathError from the local-path guard)
+            # already carry provider and error_code; re-wrapping them would
+            # flatten them into a generic CloudAIError and lose that type.
+            raise
         except Exception as e:
             raise CloudAIError(
                 f"Google Cloud image analysis failed: {e}",

tests/unit/test_aws_rekognition_provider.py

@@ -337,9 +337,10 @@ async def test_s3_url_with_nested_key(self):
         result = await provider._prepare_image_input("s3://bucket/folder/image.jpg")
         assert result['S3Object']['Name'] == "folder/image.jpg"
 
-    async def test_local_file_returns_bytes(self, tmp_path):
+    async def test_local_file_returns_bytes(self, tmp_path, monkeypatch):
         img_file = tmp_path / "test.jpg"
         img_file.write_bytes(b"\xff\xd8\xff\xe0")
+        monkeypatch.setenv("CLOUD_AI_MEDIA_ROOT", str(tmp_path))
         provider = _make_provider()
         result = await provider._prepare_image_input(str(img_file))
         assert result == {'Bytes': b"\xff\xd8\xff\xe0"}
@@ -1163,7 +1164,9 @@ async def test_each_describe_collection_runs_off_the_loop(self, provider_method)
             f"describe_collection in {provider_method} blocked the event loop"
         )
 
-    async def test_local_image_read_runs_off_the_event_loop_thread(self, tmp_path):
+    async def test_local_image_read_runs_off_the_event_loop_thread(
+        self, tmp_path, monkeypatch
+    ):
         """
         Deterministic counterpart to the heartbeat tests.
 
@@ -1176,6 +1179,7 @@ async def test_local_image_read_runs_off_the_event_loop_thread(self, tmp_path):
 
         image = tmp_path / "frame.jpg"
         image.write_bytes(b"BINARY-IMAGE-PAYLOAD")
+        monkeypatch.setenv("CLOUD_AI_MEDIA_ROOT", str(tmp_path))
         provider = _make_provider()
 
         # Patch the exact namespace that _prepare_image_input resolves from.
@@ -1201,11 +1205,12 @@ def _recording_read(path):
             "being dispatched to a worker thread"
         )
 
-    async def test_local_image_bytes_are_read_correctly(self, tmp_path):
+    async def test_local_image_bytes_are_read_correctly(self, tmp_path, monkeypatch):
         """Guard: offloading must not change what is returned."""
         image = tmp_path / "frame.png"
         payload = bytes(range(256)) * 8
         image.write_bytes(payload)
+        monkeypatch.setenv("CLOUD_AI_MEDIA_ROOT", str(tmp_path))
         provider = _make_provider()
 
         result = await provider._prepare_image_input(str(image))