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..bdce24845 --- /dev/null +++ b/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. 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, 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. + +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, 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 + + # 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. + + 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 b05e4c82f..8d4705198 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py +++ b/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py @@ -26,6 +26,7 @@ ConfigurationError, RateLimitError, ) +from ..media_paths import resolve_local_media_path logger = logging.getLogger(__name__) @@ -194,6 +195,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}", @@ -258,8 +264,11 @@ async def _prepare_image_input(self, image_url: str) -> Optional[bytes]: # For URL input, Azure can analyze directly return None else: - # Local file - read off the event loop - return await asyncio.to_thread(_read_file_bytes, image_url) + # 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, off the event loop. + safe_path = resolve_local_media_path(image_url, provider=self.provider.value) + return await asyncio.to_thread(_read_file_bytes, str(safe_path)) 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 98e6de8c3..e15c2c639 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__) @@ -187,8 +188,15 @@ async def analyze_image(self, image_url: str, if image_url.startswith(('http://', 'https://')): image.source.image_uri = image_url else: - # Local file - read off the event loop - image.content = await asyncio.to_thread(_read_file_bytes, image_url) + # Local file. The path is caller-supplied, so validate it + # against CLOUD_AI_MEDIA_ROOT before opening anything; read the + # resolved path off the event loop. + safe_path = resolve_local_media_path( + image_url, provider=self.provider.value + ) + image.content = await asyncio.to_thread( + _read_file_bytes, str(safe_path) + ) # Prepare features features = self._prepare_vision_features(analysis_types) @@ -205,6 +213,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 98d6c2d7b..54b811d0e 100644 --- a/tests/unit/test_azure_vision_provider.py +++ b/tests/unit/test_azure_vision_provider.py @@ -301,10 +301,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" @@ -1250,7 +1251,8 @@ def __call__(self, file, *args, **kwargs): class TestAzureVisionImageReadOffEventLoop: - async def test_local_file_read_runs_on_worker_thread(self, tmp_path): + async def test_local_file_read_runs_on_worker_thread(self, tmp_path, monkeypatch): + monkeypatch.setenv("CLOUD_AI_MEDIA_ROOT", str(tmp_path)) provider = _make_provider() img_file = tmp_path / "frame.jpg" img_file.write_bytes(b"\xff\xd8\xff\xe0") @@ -1280,8 +1282,9 @@ async def test_http_url_performs_no_disk_read(self, tmp_path): assert result is None assert recorder.threads == [] - async def test_missing_file_still_raises_file_not_found(self, tmp_path): + async def test_missing_file_still_raises_file_not_found(self, tmp_path, monkeypatch): """Offloading must not swallow or re-wrap I/O errors.""" + monkeypatch.setenv("CLOUD_AI_MEDIA_ROOT", str(tmp_path)) provider = _make_provider() with pytest.raises(FileNotFoundError): await provider._prepare_image_input(str(tmp_path / "does-not-exist.jpg")) 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..b82d8625d --- /dev/null +++ b/tests/unit/test_cloud_ai_media_paths.py @@ -0,0 +1,473 @@ +"""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 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, + VideoAnalysisResult, +) +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 + + 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 +# =========================================================================== + + +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]) + + 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) diff --git a/tests/unit/test_google_cloud_provider.py b/tests/unit/test_google_cloud_provider.py index 22d50820b..e25c5a17d 100644 --- a/tests/unit/test_google_cloud_provider.py +++ b/tests/unit/test_google_cloud_provider.py @@ -1196,7 +1196,8 @@ def _client(self): client.annotate_image = AsyncMock(return_value=_make_vision_response()) return client - async def test_local_file_read_runs_on_worker_thread(self, tmp_path): + async def test_local_file_read_runs_on_worker_thread(self, tmp_path, monkeypatch): + monkeypatch.setenv("CLOUD_AI_MEDIA_ROOT", str(tmp_path)) provider = _make_provider() provider._vision_client = self._client() img_file = tmp_path / "frame.jpg" @@ -1232,7 +1233,7 @@ async def test_http_url_performs_no_disk_read(self, tmp_path): assert recorder.threads == [] assert mock_vision.Image.return_value.source.image_uri == "https://example.com/img.jpg" - async def test_missing_local_file_wrapped_in_cloud_ai_error(self, tmp_path): + async def test_missing_local_file_wrapped_in_cloud_ai_error(self, tmp_path, monkeypatch): """A missing local image surfaces as ``CloudAIError`` from the public API. Unlike Azure's private ``_prepare_image_input`` (which propagates @@ -1241,6 +1242,7 @@ async def test_missing_local_file_wrapped_in_cloud_ai_error(self, tmp_path): that wrapper contract so moving the read off the loop cannot silently alter how a missing file is reported. """ + monkeypatch.setenv("CLOUD_AI_MEDIA_ROOT", str(tmp_path)) provider = _make_provider() provider._vision_client = self._client() missing = tmp_path / "does-not-exist.jpg"