From 7a2de1ccddb5dfedf7462337bbc98e9e37fe99b7 Mon Sep 17 00:00:00 2001 From: Alex <25013571+alexhb1@users.noreply.github.com> Date: Sun, 3 May 2026 10:47:06 +0100 Subject: [PATCH] Fix SSRF access (#943) --- shelfmark/core/image_cache.py | 91 +++++++++++++++++++++++------- shelfmark/main.py | 1 + tests/core/test_cover_proxy_api.py | 28 +++++++++ tests/core/test_image_cache.py | 56 +++++++++++++++++- 4 files changed, 153 insertions(+), 23 deletions(-) create mode 100644 tests/core/test_cover_proxy_api.py diff --git a/shelfmark/core/image_cache.py b/shelfmark/core/image_cache.py index 1aeb0fdb..8587af68 100644 --- a/shelfmark/core/image_cache.py +++ b/shelfmark/core/image_cache.py @@ -8,7 +8,7 @@ import time from http import HTTPStatus from io import BytesIO from typing import TYPE_CHECKING, Any -from urllib.parse import urlparse +from urllib.parse import urljoin, urlparse import requests @@ -39,6 +39,7 @@ FETCH_HEADERS = { # Maximum image size to fetch (5 MB) MAX_IMAGE_SIZE = 5 * 1024 * 1024 +MAX_REDIRECTS = 5 # Negative cache TTL (for failed fetches) - 1 hour NEGATIVE_CACHE_TTL = 3600 @@ -482,29 +483,85 @@ class ImageCacheService: } @staticmethod - def _is_safe_url(url: str) -> bool: - """Check that a URL is safe to fetch (no SSRF to internal resources).""" + def _prepare_safe_url(url: str) -> str | None: + """Prepare and validate a URL before fetching it.""" + if "\\" in url or any(ord(char) < 32 for char in url): + return None + try: - parsed = urlparse(url) + prepared = requests.Request("GET", url).prepare() + prepared_url = prepared.url + if not isinstance(prepared_url, str): + return None + parsed = urlparse(prepared_url) hostname = parsed.hostname - except ValueError: - return False + except requests.exceptions.RequestException, ValueError: + return None + + if not prepared_url: + return None + + if "\\" in prepared_url or any(ord(char) < 32 for char in prepared_url): + return None + + netloc_lower = parsed.netloc.lower() + if "%2f" in netloc_lower or "%5c" in netloc_lower: + return None if parsed.scheme not in ("http", "https"): - return False + return None if not hostname: - return False + return None try: resolved = socket.getaddrinfo(hostname, None) for _, _, _, _, sockaddr in resolved: ip = ipaddress.ip_address(sockaddr[0]) if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: - return False + return None except socket.gaierror, ValueError: - return False + return None - return True + return prepared_url + + @staticmethod + def _is_safe_url(url: str) -> bool: + """Check that a URL is safe to fetch (no SSRF to internal resources).""" + return ImageCacheService._prepare_safe_url(url) is not None + + def _fetch_safe_response(self, url: str) -> requests.Response | None: + """Fetch a URL after validating the initial URL and each redirect.""" + current_url = self._prepare_safe_url(url) + if not current_url: + logger.warning("Blocked request to disallowed URL: %s", url) + return None + + for _ in range(MAX_REDIRECTS + 1): + response = requests.get( + current_url, + timeout=(5, 10), + headers=FETCH_HEADERS, + stream=True, + verify=get_ssl_verify(current_url), + allow_redirects=False, + ) + + if not response.is_redirect: + return response + + location = response.headers.get("location") + response.close() + if not location: + return None + + redirect_url = urljoin(current_url, location) + next_url = self._prepare_safe_url(redirect_url) + if not next_url: + logger.warning("Blocked redirect to disallowed URL: %s", redirect_url) + return None + current_url = next_url + + return None def fetch_and_cache(self, cache_id: str, url: str) -> tuple[bytes, str] | None: """Fetch an image from URL and cache it. @@ -519,17 +576,9 @@ class ImageCacheService: """ cached_data: tuple[bytes, str] | None = None try: - if not self._is_safe_url(url): - logger.warning("Blocked request to disallowed URL: %s", url) + response = self._fetch_safe_response(url) + if response is None: return None - - response = requests.get( - url, - timeout=(5, 10), - headers=FETCH_HEADERS, - stream=True, - verify=get_ssl_verify(url), - ) response.raise_for_status() # Validate content type diff --git a/shelfmark/main.py b/shelfmark/main.py index 66837114..81d0a692 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -1586,6 +1586,7 @@ def api_local_download() -> Response | tuple[Response, int]: @app.route("/api/covers/", methods=["GET"]) +@login_required def api_cover(cover_id: str) -> Response | tuple[Response, int]: """Serve a cached book cover image. diff --git a/tests/core/test_cover_proxy_api.py b/tests/core/test_cover_proxy_api.py new file mode 100644 index 00000000..dbb6ccac --- /dev/null +++ b/tests/core/test_cover_proxy_api.py @@ -0,0 +1,28 @@ +"""Cover proxy API security tests.""" + +from __future__ import annotations + +import importlib +from unittest.mock import patch + +import pytest + + +@pytest.fixture(scope="module") +def main_module(): + """Import `shelfmark.main` with background startup disabled.""" + with patch("shelfmark.download.orchestrator.start"): + import shelfmark.main as main + + importlib.reload(main) + return main + + +def test_cover_proxy_requires_authentication(main_module) -> None: + client = main_module.app.test_client() + + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + response = client.get("/api/covers/test-id") + + assert response.status_code == 401 + assert response.get_json() == {"error": "Unauthorized"} diff --git a/tests/core/test_image_cache.py b/tests/core/test_image_cache.py index 27783f08..fa915f89 100644 --- a/tests/core/test_image_cache.py +++ b/tests/core/test_image_cache.py @@ -5,15 +5,67 @@ import requests from shelfmark.core.image_cache import ImageCacheService +def test_fetch_and_cache_rejects_backslash_authority_bypass_before_request( + tmp_path, monkeypatch +) -> None: + cache = ImageCacheService(tmp_path) + calls = [] + + def fake_get(url, **_kwargs): + calls.append(url) + raise AssertionError("unsafe URL should not be requested") + + monkeypatch.setattr("shelfmark.core.image_cache.requests.get", fake_get) + + assert cache.fetch_and_cache("cover-ssrf", "http://127.0.0.1:6666\\@1.1.1.1") is None + assert calls == [] + assert "cover-ssrf" not in cache._index + + +def test_is_safe_url_rejects_encoded_separator_in_authority() -> None: + assert ImageCacheService._is_safe_url("http://127.0.0.1:6666%5c@1.1.1.1") is False + assert ImageCacheService._is_safe_url("http://127.0.0.1:6666%2f@1.1.1.1") is False + + def test_is_safe_url_rejects_invalid_ipv6_url() -> None: assert ImageCacheService._is_safe_url("http://[") is False +def test_fetch_and_cache_blocks_unsafe_redirect(tmp_path, monkeypatch) -> None: + cache = ImageCacheService(tmp_path) + + def fake_getaddrinfo(hostname, *_args, **_kwargs): + addresses = { + "example.com": "93.184.216.34", + "127.0.0.1": "127.0.0.1", + } + return [(None, None, None, None, (addresses[hostname], 0))] + + class RedirectResponse: + is_redirect = True + headers = {"location": "http://127.0.0.1/cover.jpg"} + + def close(self): + return None + + calls = [] + + def fake_get(url, **_kwargs): + calls.append(url) + return RedirectResponse() + + monkeypatch.setattr("shelfmark.core.image_cache.socket.getaddrinfo", fake_getaddrinfo) + monkeypatch.setattr("shelfmark.core.image_cache.requests.get", fake_get) + + assert cache.fetch_and_cache("cover-redirect", "https://example.com/cover.jpg") is None + assert calls == ["https://example.com/cover.jpg"] + assert "cover-redirect" not in cache._index + + def test_fetch_and_cache_returns_none_on_request_exception(tmp_path, monkeypatch) -> None: cache = ImageCacheService(tmp_path) - monkeypatch.setattr(cache, "_is_safe_url", lambda _url: True) - def fake_get(*args, **kwargs): + def fake_get(*_args, **_kwargs): raise requests.exceptions.TooManyRedirects("too many redirects") monkeypatch.setattr("shelfmark.core.image_cache.requests.get", fake_get)