From 8923366a6c6e8754cba0f3bce16c296d378ce9fd Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 07:13:18 -0500 Subject: [PATCH 1/2] fix(security): sandbox local media paths in cloud AI providers (#1209) All three cloud AI providers dispatch `analyze_image(image_url, ...)` on the string's prefix: `s3://` and `http(s)://` are treated as remote sources, and anything else fell through to an unguarded `open()`. A caller-supplied absolute path, `../` traversal, or symlink could therefore read any file readable by the service account. The same unguarded sink existed in all three providers, not just the one named in the issue: - aws_rekognition.py `_prepare_image_input` - azure_vision.py `_prepare_image_input` - google_cloud.py inline `open()` in `analyze_image` Introduce `cloud_ai/media_paths.py` as the single policy for local reads: - Local reads are opt-in via `CLOUD_AI_MEDIA_ROOT`. Unset (the default) disables them entirely, restricting providers to `s3://`/`https://`. This is fail-closed, and answers the issue's open question. - When a root is configured, both root and candidate are fully resolved (`Path.resolve()` follows symlinks) and the candidate must be contained by the root -- covering symlink escapes, not just lexical `..` segments. - Non-regular files (FIFO, device, directory) are rejected, so a FIFO placed inside the root cannot pin a `to_thread` worker forever. - Rejection raises the new typed `UnsafeMediaPathError(CloudAIError)` instead of silently returning empty bytes. Each provider re-raises `CloudAIError` subclasses unchanged so the type survives to the caller. - Providers read from the resolved path, not the caller string, narrowing the check-to-open race. - Error messages echo only the caller-supplied value; the resolved path is logged server-side for forensics rather than returned. Adds tests/unit/test_cloud_ai_media_paths.py (44 tests) covering absolute paths, `../` traversal, symlink escape, non-regular files, the disabled default, and per-provider propagation. Existing local-file tests now set `CLOUD_AI_MEDIA_ROOT`. Full cloud AI suite: 505 passed. Closes #1209 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .env.example | 14 + .../integrations/cloud_ai/__init__.py | 14 +- .../integrations/cloud_ai/exceptions.py | 15 + .../integrations/cloud_ai/media_paths.py | 141 ++++++ .../cloud_ai/providers/aws_rekognition.py | 15 +- .../cloud_ai/providers/azure_vision.py | 14 +- .../cloud_ai/providers/google_cloud.py | 15 +- tests/unit/test_aws_rekognition_provider.py | 11 +- tests/unit/test_azure_vision_provider.py | 3 +- tests/unit/test_cloud_ai_media_paths.py | 415 ++++++++++++++++++ 10 files changed, 642 insertions(+), 15 deletions(-) create mode 100644 src/youtube_extension/integrations/cloud_ai/media_paths.py create mode 100644 tests/unit/test_cloud_ai_media_paths.py diff --git a/.env.example b/.env.example index 0c3318bba..58d83d915 100644 --- a/.env.example +++ b/.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 # ============================================================================ diff --git a/src/youtube_extension/integrations/cloud_ai/__init__.py b/src/youtube_extension/integrations/cloud_ai/__init__.py index 5d24bed48..e9377dad1 100644 --- a/src/youtube_extension/integrations/cloud_ai/__init__.py +++ b/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", ] diff --git a/src/youtube_extension/integrations/cloud_ai/exceptions.py b/src/youtube_extension/integrations/cloud_ai/exceptions.py index 84c3c1c11..0ee3e2be6 100644 --- a/src/youtube_extension/integrations/cloud_ai/exceptions.py +++ b/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 diff --git a/src/youtube_extension/integrations/cloud_ai/media_paths.py b/src/youtube_extension/integrations/cloud_ai/media_paths.py new file mode 100644 index 000000000..426f4586a --- /dev/null +++ b/src/youtube_extension/integrations/cloud_ai/media_paths.py @@ -0,0 +1,141 @@ +""" +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: ``s3://`` and ``http(s)://`` are handled as +remote sources, 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. + +This module centralises the guard so all three providers share one policy: + +* Local reads are **opt-in**. With ``CLOUD_AI_MEDIA_ROOT`` unset, providers are + restricted to ``s3://``/``https://`` and every local path is rejected. 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. + + Raises: + ConfigurationError: if the variable is set to a value that cannot be + resolved to a path. + """ + raw = os.environ.get(MEDIA_ROOT_ENV_VAR) + if raw is None or not raw.strip(): + return None + + try: + return 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 + + +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 diff --git a/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py b/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py index 9b300bdb4..a7586bee2 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py +++ b/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], diff --git a/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py b/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py index 9f7cf296d..f0c603776 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py +++ b/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 diff --git a/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py b/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py index 9200d25ee..5af1a1f02 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py +++ b/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}", diff --git a/tests/unit/test_aws_rekognition_provider.py b/tests/unit/test_aws_rekognition_provider.py index ea88ae298..4d2e2faf7 100644 --- a/tests/unit/test_aws_rekognition_provider.py +++ b/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)) diff --git a/tests/unit/test_azure_vision_provider.py b/tests/unit/test_azure_vision_provider.py index 4b6839796..207b9b3a9 100644 --- a/tests/unit/test_azure_vision_provider.py +++ b/tests/unit/test_azure_vision_provider.py @@ -299,10 +299,11 @@ async def test_https_url_returns_none(self): result = await provider._prepare_image_input("https://example.com/img.jpg") assert result is None - async def test_local_file_reads_bytes_fixture(self, tmp_path): + async def test_local_file_reads_bytes_fixture(self, tmp_path, monkeypatch): provider = _make_provider() img_file = tmp_path / "test.jpg" img_file.write_bytes(b"\xff\xd8\xff") + monkeypatch.setenv("CLOUD_AI_MEDIA_ROOT", str(tmp_path)) result = await provider._prepare_image_input(str(img_file)) assert result == b"\xff\xd8\xff" diff --git a/tests/unit/test_cloud_ai_media_paths.py b/tests/unit/test_cloud_ai_media_paths.py new file mode 100644 index 000000000..f14e6e377 --- /dev/null +++ b/tests/unit/test_cloud_ai_media_paths.py @@ -0,0 +1,415 @@ +"""Security tests for the cloud AI local media path guard. + +Covers the acceptance criteria of issue #1209: a caller-supplied ``image_url`` +must not be able to read arbitrary files off the host via an absolute path, +``../`` traversal, or a symlink that escapes the permitted root. +""" + +from __future__ import annotations + +import os +import sys +import types as _types +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +_SRC = Path(__file__).resolve().parents[2] / "src" +sys.path.insert(0, str(_SRC)) + +from youtube_extension.integrations.cloud_ai.base import AnalysisType +from youtube_extension.integrations.cloud_ai.exceptions import ( + CloudAIError, + ConfigurationError, + UnsafeMediaPathError, +) +from youtube_extension.integrations.cloud_ai.media_paths import ( + MEDIA_ROOT_ENV_VAR, + get_media_root, + resolve_local_media_path, +) +from youtube_extension.integrations.cloud_ai.providers.aws_rekognition import ( + AWSRekognition, +) +from youtube_extension.integrations.cloud_ai.providers.azure_vision import AzureVision +from youtube_extension.integrations.cloud_ai.providers.google_cloud import GoogleCloudAI + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +AWS_CONFIG = { + "aws_access_key_id": "test-access-key-id", + "aws_secret_access_key": "test-secret-access-key", + "region": "us-east-1", +} + +AZURE_CONFIG = { + "subscription_key": "test-key-abc", + "endpoint": "https://eastus.api.cognitive.microsoft.com/", +} + +GOOGLE_CONFIG = {"project_id": "test-project"} + + +@pytest.fixture +def media_root(tmp_path, monkeypatch): + """A configured media root containing one legitimate image.""" + root = tmp_path / "media" + root.mkdir() + (root / "photo.jpg").write_bytes(b"\xff\xd8\xff\xe0") + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, str(root)) + return root + + +@pytest.fixture +def secret_file(tmp_path): + """A file that lives outside any media root -- the exfiltration target.""" + secret = tmp_path / "secrets.env" + secret.write_text("API_KEY=super-secret") + return secret + + +# =========================================================================== +# get_media_root +# =========================================================================== + + +class TestGetMediaRoot: + def test_unset_returns_none(self, monkeypatch): + monkeypatch.delenv(MEDIA_ROOT_ENV_VAR, raising=False) + assert get_media_root() is None + + def test_empty_string_returns_none(self, monkeypatch): + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, "") + assert get_media_root() is None + + def test_whitespace_only_returns_none(self, monkeypatch): + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, " ") + assert get_media_root() is None + + def test_returns_resolved_absolute_path(self, tmp_path, monkeypatch): + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, str(tmp_path)) + root = get_media_root() + assert root == tmp_path.resolve() + assert root.is_absolute() + + def test_relative_value_is_made_absolute(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "assets").mkdir() + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, "assets") + assert get_media_root() == (tmp_path / "assets").resolve() + + def test_surrounding_whitespace_is_stripped(self, tmp_path, monkeypatch): + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, f" {tmp_path} ") + assert get_media_root() == tmp_path.resolve() + + def test_unresolvable_value_raises_configuration_error(self, monkeypatch): + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, "/whatever") + with patch( + "youtube_extension.integrations.cloud_ai.media_paths.Path.resolve", + side_effect=OSError("boom"), + ): + with pytest.raises(ConfigurationError) as exc_info: + get_media_root() + assert exc_info.value.missing_config == MEDIA_ROOT_ENV_VAR + + +# =========================================================================== +# resolve_local_media_path +# =========================================================================== + + +class TestResolveLocalMediaPathDisabled: + """With no root configured, every local path must be refused.""" + + def test_local_reads_disabled_by_default(self, monkeypatch, tmp_path): + monkeypatch.delenv(MEDIA_ROOT_ENV_VAR, raising=False) + existing = tmp_path / "photo.jpg" + existing.write_bytes(b"data") + with pytest.raises(UnsafeMediaPathError) as exc_info: + resolve_local_media_path(str(existing)) + assert "disabled" in str(exc_info.value).lower() + assert exc_info.value.error_code == "UNSAFE_MEDIA_PATH" + + def test_disabled_error_names_the_env_var(self, monkeypatch): + monkeypatch.delenv(MEDIA_ROOT_ENV_VAR, raising=False) + with pytest.raises(UnsafeMediaPathError) as exc_info: + resolve_local_media_path("/etc/passwd") + assert MEDIA_ROOT_ENV_VAR in str(exc_info.value) + + +class TestResolveLocalMediaPathRejections: + def test_absolute_path_outside_root_is_rejected(self, media_root, secret_file): + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path(str(secret_file)) + + def test_etc_passwd_is_rejected(self, media_root): + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path("/etc/passwd") + + def test_dotdot_traversal_is_rejected(self, media_root, secret_file): + traversal = str(media_root / ".." / "secrets.env") + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path(traversal) + + def test_deep_dotdot_traversal_is_rejected(self, media_root): + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path( + str(media_root / ".." / ".." / ".." / "etc" / "passwd") + ) + + def test_symlink_escaping_root_is_rejected(self, media_root, secret_file): + """A symlink *inside* the root pointing outside it must not slip through. + + This is the case a purely lexical ``..`` check would miss. + """ + link = media_root / "innocent.jpg" + link.symlink_to(secret_file) + # Sanity: the link really does read the secret without the guard. + assert link.read_text() == "API_KEY=super-secret" + + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path(str(link)) + + def test_symlinked_directory_escape_is_rejected(self, media_root, tmp_path): + outside_dir = tmp_path / "outside" + outside_dir.mkdir() + (outside_dir / "loot.txt").write_text("loot") + (media_root / "shortcut").symlink_to(outside_dir, target_is_directory=True) + + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path(str(media_root / "shortcut" / "loot.txt")) + + def test_sibling_directory_prefix_is_rejected(self, media_root, tmp_path): + """``/tmp/media-evil`` must not pass a check against ``/tmp/media``.""" + sibling = tmp_path / "media-evil" + sibling.mkdir() + target = sibling / "photo.jpg" + target.write_bytes(b"nope") + + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path(str(target)) + + def test_empty_path_is_rejected(self, media_root): + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path("") + + def test_whitespace_path_is_rejected(self, media_root): + with pytest.raises(UnsafeMediaPathError): + resolve_local_media_path(" ") + + def test_directory_inside_root_is_rejected(self, media_root): + subdir = media_root / "album" + subdir.mkdir() + with pytest.raises(UnsafeMediaPathError) as exc_info: + resolve_local_media_path(str(subdir)) + assert "regular file" in str(exc_info.value) + + def test_fifo_inside_root_is_rejected(self, media_root): + """A FIFO would block a worker thread forever on read.""" + fifo = media_root / "pipe.jpg" + try: + os.mkfifo(fifo) + except (AttributeError, NotImplementedError, OSError): + pytest.skip("mkfifo unavailable on this platform") + with pytest.raises(UnsafeMediaPathError) as exc_info: + resolve_local_media_path(str(fifo)) + assert "regular file" in str(exc_info.value) + + def test_error_does_not_leak_resolved_server_path(self, media_root, secret_file): + with pytest.raises(UnsafeMediaPathError) as exc_info: + resolve_local_media_path(str(secret_file)) + # The message must not echo the resolved filesystem location. + assert str(secret_file.resolve()) not in str(exc_info.value) + + def test_error_carries_provider_and_requested_path(self, media_root): + with pytest.raises(UnsafeMediaPathError) as exc_info: + resolve_local_media_path("/etc/passwd", provider="aws_rekognition") + assert exc_info.value.provider == "aws_rekognition" + assert exc_info.value.requested_path == "/etc/passwd" + + def test_unsafe_media_path_error_is_a_cloud_ai_error(self, media_root): + with pytest.raises(CloudAIError): + resolve_local_media_path("/etc/passwd") + + +class TestResolveLocalMediaPathAccepts: + def test_file_in_root_is_accepted(self, media_root): + resolved = resolve_local_media_path(str(media_root / "photo.jpg")) + assert resolved == (media_root / "photo.jpg").resolve() + + def test_nested_file_in_root_is_accepted(self, media_root): + nested = media_root / "album" / "inner.jpg" + nested.parent.mkdir() + nested.write_bytes(b"ok") + assert resolve_local_media_path(str(nested)) == nested.resolve() + + def test_normalised_traversal_that_stays_inside_is_accepted(self, media_root): + """``root/album/../photo.jpg`` resolves back into the root -- allowed.""" + (media_root / "album").mkdir() + candidate = media_root / "album" / ".." / "photo.jpg" + assert ( + resolve_local_media_path(str(candidate)) + == (media_root / "photo.jpg").resolve() + ) + + def test_symlink_inside_root_is_accepted(self, media_root): + link = media_root / "alias.jpg" + link.symlink_to(media_root / "photo.jpg") + assert ( + resolve_local_media_path(str(link)) == (media_root / "photo.jpg").resolve() + ) + + def test_missing_file_inside_root_is_accepted_and_fails_at_open(self, media_root): + """Containment is the guard's job; existence is the caller's problem.""" + missing = media_root / "absent.jpg" + assert resolve_local_media_path(str(missing)) == missing.resolve() + with pytest.raises(FileNotFoundError): + missing.read_bytes() + + def test_returned_path_is_the_resolved_one(self, media_root): + """Providers read the returned path, not the raw caller string.""" + (media_root / "album").mkdir() + resolved = resolve_local_media_path( + str(media_root / "album" / ".." / "photo.jpg") + ) + assert ".." not in str(resolved) + assert resolved.is_absolute() + + +# =========================================================================== +# Provider integration -- the guard must apply to all three providers +# =========================================================================== + + +class TestAWSRekognitionMediaPathGuard: + async def test_traversal_rejected(self, media_root, secret_file): + provider = AWSRekognition(AWS_CONFIG) + with pytest.raises(UnsafeMediaPathError): + await provider._prepare_image_input(str(secret_file)) + + async def test_local_reads_disabled_by_default(self, monkeypatch, tmp_path): + monkeypatch.delenv(MEDIA_ROOT_ENV_VAR, raising=False) + img = tmp_path / "photo.jpg" + img.write_bytes(b"data") + provider = AWSRekognition(AWS_CONFIG) + with pytest.raises(UnsafeMediaPathError): + await provider._prepare_image_input(str(img)) + + async def test_symlink_escape_rejected(self, media_root, secret_file): + link = media_root / "innocent.jpg" + link.symlink_to(secret_file) + provider = AWSRekognition(AWS_CONFIG) + with pytest.raises(UnsafeMediaPathError): + await provider._prepare_image_input(str(link)) + + async def test_permitted_file_still_reads(self, media_root): + provider = AWSRekognition(AWS_CONFIG) + result = await provider._prepare_image_input(str(media_root / "photo.jpg")) + assert result == {"Bytes": b"\xff\xd8\xff\xe0"} + + async def test_s3_source_unaffected_by_guard(self, monkeypatch): + monkeypatch.delenv(MEDIA_ROOT_ENV_VAR, raising=False) + provider = AWSRekognition(AWS_CONFIG) + result = await provider._prepare_image_input("s3://bucket/key.jpg") + assert result["S3Object"]["Bucket"] == "bucket" + + async def test_analyze_image_propagates_typed_error(self, media_root, secret_file): + """The broad ``except Exception`` must not flatten the typed error.""" + provider = AWSRekognition(AWS_CONFIG) + provider._rekognition_client = MagicMock() + with pytest.raises(UnsafeMediaPathError): + await provider.analyze_image( + str(secret_file), [AnalysisType.LABEL_DETECTION] + ) + + +class TestAzureVisionMediaPathGuard: + async def test_traversal_rejected(self, media_root, secret_file): + provider = AzureVision(AZURE_CONFIG) + with pytest.raises(UnsafeMediaPathError): + await provider._prepare_image_input(str(secret_file)) + + async def test_local_reads_disabled_by_default(self, monkeypatch, tmp_path): + monkeypatch.delenv(MEDIA_ROOT_ENV_VAR, raising=False) + img = tmp_path / "photo.jpg" + img.write_bytes(b"data") + provider = AzureVision(AZURE_CONFIG) + with pytest.raises(UnsafeMediaPathError): + await provider._prepare_image_input(str(img)) + + async def test_symlink_escape_rejected(self, media_root, secret_file): + link = media_root / "innocent.jpg" + link.symlink_to(secret_file) + provider = AzureVision(AZURE_CONFIG) + with pytest.raises(UnsafeMediaPathError): + await provider._prepare_image_input(str(link)) + + async def test_permitted_file_still_reads(self, media_root): + provider = AzureVision(AZURE_CONFIG) + assert ( + await provider._prepare_image_input(str(media_root / "photo.jpg")) + == b"\xff\xd8\xff\xe0" + ) + + async def test_https_source_unaffected_by_guard(self, monkeypatch): + monkeypatch.delenv(MEDIA_ROOT_ENV_VAR, raising=False) + provider = AzureVision(AZURE_CONFIG) + assert ( + await provider._prepare_image_input("https://example.com/img.jpg") is None + ) + + async def test_analyze_image_propagates_typed_error(self, media_root, secret_file): + provider = AzureVision(AZURE_CONFIG) + provider._vision_client = MagicMock() + with pytest.raises(UnsafeMediaPathError): + await provider.analyze_image(str(secret_file), [AnalysisType.OCR]) + + +class TestGoogleCloudMediaPathGuard: + """The Google provider imports ``google.cloud.vision`` inside ``analyze_image``, + so the module is stubbed the same way the provider's own test suite does it.""" + + def _provider(self): + provider = GoogleCloudAI(GOOGLE_CONFIG) + provider._vision_client = MagicMock() + return provider + + @staticmethod + def _vision_modules(): + mock_image = MagicMock() + mock_image.return_value = MagicMock(source=MagicMock()) + mock_vision = MagicMock() + mock_vision.Image = mock_image + return { + "google": _types.ModuleType("google"), + "google.cloud": _types.ModuleType("google.cloud"), + "google.cloud.vision": mock_vision, + } + + async def test_traversal_rejected(self, media_root, secret_file): + provider = self._provider() + with patch.dict("sys.modules", self._vision_modules()): + with pytest.raises(UnsafeMediaPathError): + await provider.analyze_image( + str(secret_file), [AnalysisType.LABEL_DETECTION] + ) + + async def test_local_reads_disabled_by_default(self, monkeypatch, tmp_path): + monkeypatch.delenv(MEDIA_ROOT_ENV_VAR, raising=False) + img = tmp_path / "photo.jpg" + img.write_bytes(b"data") + provider = self._provider() + with patch.dict("sys.modules", self._vision_modules()): + with pytest.raises(UnsafeMediaPathError): + await provider.analyze_image(str(img), [AnalysisType.LABEL_DETECTION]) + + async def test_symlink_escape_rejected(self, media_root, secret_file): + link = media_root / "innocent.jpg" + link.symlink_to(secret_file) + provider = self._provider() + with patch.dict("sys.modules", self._vision_modules()): + with pytest.raises(UnsafeMediaPathError): + await provider.analyze_image(str(link), [AnalysisType.LABEL_DETECTION]) From 54161719dd7f192a30d0112a064fe824a5f48b1c Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 12:29:58 +0000 Subject: [PATCH 2/2] fix(security): reject non-directory CLOUD_AI_MEDIA_ROOT; close review gaps (#1216) Addresses the three unresolved Copilot review threads on #1216: 1. Fail-closed on a misconfigured root. get_media_root() left resolve() non-strict, so CLOUD_AI_MEDIA_ROOT=/etc/passwd (a regular file) was accepted as the root; that file then passed its own is_relative_to() containment check and was returned as a permitted read. Require the resolved root to be an existing directory, raising ConfigurationError otherwise. This also surfaces a nonexistent-directory typo loudly instead of silently rejecting every candidate. 2. Cover the Google permitted-file branch. AWS/Azure verified successful reads but the Google class only had rejection cases, while the PR's coverage table claimed the check for all three providers. Add an end-to-end analyze_image test asserting the resolved file's bytes are assigned to vision.Image().content. 3. Correct the module docstring. Remote-scheme handling is provider- specific: only AWS Rekognition recognises s3:// (Azure and Google treat it as a local path, rejected while local reads are disabled), and all three accept plain http:// as well as https://. Focused suite: 47 passed (44 + 3 new). Full cloud AI provider suites: 368 passed. ruff/mypy clean; black formatted. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01TFFcgtEzNyhrxJnHgimcd2 --- .../integrations/cloud_ai/media_paths.py | 45 ++++++++++---- tests/unit/test_cloud_ai_media_paths.py | 62 ++++++++++++++++++- 2 files changed, 93 insertions(+), 14 deletions(-) diff --git a/src/youtube_extension/integrations/cloud_ai/media_paths.py b/src/youtube_extension/integrations/cloud_ai/media_paths.py index 426f4586a..bdce24845 100644 --- a/src/youtube_extension/integrations/cloud_ai/media_paths.py +++ b/src/youtube_extension/integrations/cloud_ai/media_paths.py @@ -2,19 +2,24 @@ 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: ``s3://`` and ``http(s)://`` are handled as -remote sources, 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. +dispatches on the string's prefix. Remote sources are handled per provider -- +``http(s)://`` by all three, plus ``s3://`` by AWS Rekognition -- 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. This module centralises the guard so all three providers share one policy: -* Local reads are **opt-in**. With ``CLOUD_AI_MEDIA_ROOT`` unset, providers are - restricted to ``s3://``/``https://`` and every local path is rejected. 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. +* Local reads are **opt-in**. With ``CLOUD_AI_MEDIA_ROOT`` unset, every local + path is rejected and each provider is limited to the remote schemes it + recognises (see above). That set is provider-specific: an ``s3://`` URL is + only understood by AWS Rekognition -- Azure and Google treat it as a local + path, so it is rejected while local reads are disabled. This is the + production posture; the local branch is a development convenience. +* When a root *is* configured it must resolve to an existing directory, and 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. @@ -53,14 +58,15 @@ def get_media_root() -> Path | None: Raises: ConfigurationError: if the variable is set to a value that cannot be - resolved to a path. + resolved to a path, 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: - return Path(raw.strip()).expanduser().resolve() + 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. @@ -69,6 +75,21 @@ def get_media_root() -> Path | None: missing_config=MEDIA_ROOT_ENV_VAR, ) from exc + # Fail closed on a misconfigured root. ``resolve()`` is non-strict, so a + # typo or a value pointing at a regular file (e.g. ``/etc/passwd``) would + # otherwise be accepted -- and a file root passes its own ``is_relative_to`` + # check, letting that exact file through and defeating the whole guard. + # Requiring an existing directory keeps the "disabled unless deliberately + # configured" contract intact and surfaces the misconfiguration loudly + # instead of silently rejecting every candidate. + 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. diff --git a/tests/unit/test_cloud_ai_media_paths.py b/tests/unit/test_cloud_ai_media_paths.py index f14e6e377..b82d8625d 100644 --- a/tests/unit/test_cloud_ai_media_paths.py +++ b/tests/unit/test_cloud_ai_media_paths.py @@ -11,14 +11,17 @@ import sys import types as _types from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest _SRC = Path(__file__).resolve().parents[2] / "src" sys.path.insert(0, str(_SRC)) -from youtube_extension.integrations.cloud_ai.base import AnalysisType +from youtube_extension.integrations.cloud_ai.base import ( + AnalysisType, + VideoAnalysisResult, +) from youtube_extension.integrations.cloud_ai.exceptions import ( CloudAIError, ConfigurationError, @@ -115,6 +118,26 @@ def test_unresolvable_value_raises_configuration_error(self, monkeypatch): get_media_root() assert exc_info.value.missing_config == MEDIA_ROOT_ENV_VAR + def test_nonexistent_directory_raises_configuration_error( + self, tmp_path, monkeypatch + ): + """A root that does not exist must fail closed, not silently pass.""" + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, str(tmp_path / "missing")) + with pytest.raises(ConfigurationError) as exc_info: + get_media_root() + assert exc_info.value.missing_config == MEDIA_ROOT_ENV_VAR + + def test_root_pointing_at_a_file_raises_configuration_error( + self, tmp_path, monkeypatch + ): + """A file root (e.g. ``/etc/passwd``) would otherwise let that exact + file pass the ``is_relative_to`` containment check -- reject it.""" + a_file = tmp_path / "not-a-dir" + a_file.write_text("x") + monkeypatch.setenv(MEDIA_ROOT_ENV_VAR, str(a_file)) + with pytest.raises(ConfigurationError): + get_media_root() + # =========================================================================== # resolve_local_media_path @@ -413,3 +436,38 @@ async def test_symlink_escape_rejected(self, media_root, secret_file): with patch.dict("sys.modules", self._vision_modules()): with pytest.raises(UnsafeMediaPathError): await provider.analyze_image(str(link), [AnalysisType.LABEL_DETECTION]) + + async def test_permitted_file_still_reads(self, media_root): + """A permitted local file inside the root must reach the Vision call. + + AWS/Azure prove this through ``_prepare_image_input``; the Google + provider reads the file inline in ``analyze_image``, so exercise the + whole method and assert the resolved file's bytes are assigned to + ``vision.Image().content``. + """ + image_instance = MagicMock(source=MagicMock()) + mock_vision = MagicMock() + mock_vision.Image = MagicMock(return_value=image_instance) + + response = MagicMock() + response.label_annotations = [] + response.text_annotations = [] + response.logo_annotations = [] + response.localized_object_annotations = [] + + provider = GoogleCloudAI(GOOGLE_CONFIG) + provider._vision_client = MagicMock() + provider._vision_client.annotate_image = AsyncMock(return_value=response) + + modules = { + "google": _types.ModuleType("google"), + "google.cloud": _types.ModuleType("google.cloud"), + "google.cloud.vision": mock_vision, + } + with patch.dict("sys.modules", modules): + result = await provider.analyze_image( + str(media_root / "photo.jpg"), [AnalysisType.LABEL_DETECTION] + ) + + assert image_instance.content == b"\xff\xd8\xff\xe0" + assert isinstance(result, VideoAnalysisResult)