diff --git a/lib/galaxy_test/api/test_proxy.py b/lib/galaxy_test/api/test_proxy.py index 19a33d9bd4d..b930b009e5b 100644 --- a/lib/galaxy_test/api/test_proxy.py +++ b/lib/galaxy_test/api/test_proxy.py @@ -1,9 +1,3 @@ -from unittest.mock import ( - AsyncMock, - MagicMock, - patch, -) - import pytest import requests @@ -137,82 +131,41 @@ class TestProxyApi(ApiTestCase): # Verify content-encoding header was properly filtered out (no double decompression) assert "content-encoding" not in response.headers - @patch("galaxy.webapps.galaxy.api.proxy.httpx.AsyncClient") - def test_proxy_validates_redirects(self, mock_client_class): - """Test that redirects are validated.""" - # Create mock responses - redirect to a local file (invalid scheme) - redirect_response = MagicMock() - redirect_response.status_code = 302 - redirect_response.headers = {"location": "file://internal-server/secret-files"} - redirect_response.aclose = AsyncMock() - - # Setup mock client - mock_client = MagicMock() - mock_client.build_request = MagicMock(return_value=MagicMock()) - mock_client.send = AsyncMock(return_value=redirect_response) - mock_client.aclose = AsyncMock() - mock_client_class.return_value = mock_client - - # Attempt to proxy a URL that redirects to file:// (should be blocked) - response = self._get("proxy?url=https://evil.com/redirect") - - # Should fail with 400 Bad Request due to invalid redirect URL scheme + def test_proxy_validates_redirects(self, mock_http_server): + """Test that redirects to invalid schemes are blocked.""" + url = mock_http_server.get_url( + remote_url="https://evil.com/redirect", + status=302, + response_headers={"Location": "file://internal-server/secret-files"}, + ) + response = self._get(f"proxy?url={url}") self._assert_status_code_is(response, 400) assert "Invalid URL format" in response.json()["err_msg"] - @patch("galaxy.webapps.galaxy.api.proxy.httpx.AsyncClient") - def test_proxy_follows_valid_redirects(self, mock_client_class): + def test_proxy_follows_valid_redirects(self, mock_http_server): """Test that valid redirects are followed after validation.""" - # Create mock responses - redirect_response = MagicMock() - redirect_response.status_code = 301 - redirect_response.headers = {"location": "https://example.com/final"} - redirect_response.aclose = AsyncMock() - - final_response = MagicMock() - final_response.status_code = 200 - final_response.headers = {"content-type": "text/plain"} - final_response.aclose = AsyncMock() - - # Create async generator for streaming - async def mock_stream(): - yield b"test content" - - final_response.aiter_bytes = mock_stream - - # Setup mock client to return redirect first, then final response - mock_client = MagicMock() - mock_client.build_request = MagicMock(return_value=MagicMock()) - mock_client.send = AsyncMock(side_effect=[redirect_response, final_response]) - mock_client.aclose = AsyncMock() - mock_client_class.return_value = mock_client - - # Proxy a URL that redirects to a valid external URL - response = self._get("proxy?url=https://example.com/redirect") - - # Should succeed and follow the redirect + final_url = mock_http_server.get_url( + remote_url="https://example.com/final", + status=200, + body="test content", + content_type="text/plain", + ) + redirect_url = mock_http_server.get_url( + remote_url="https://example.com/redirect", + status=301, + response_headers={"Location": final_url}, + ) + response = self._get(f"proxy?url={redirect_url}") self._assert_status_code_is_ok(response) assert b"test content" in response.content - @patch("galaxy.webapps.galaxy.api.proxy.httpx.AsyncClient") - def test_proxy_blocks_too_many_redirects(self, mock_client_class): + def test_proxy_blocks_too_many_redirects(self, mock_http_server): """Test that excessive redirects are blocked to prevent redirect loops.""" - # Create a mock response that always redirects - redirect_response = MagicMock() - redirect_response.status_code = 302 - redirect_response.headers = {"location": "https://example.com/loop"} - redirect_response.aclose = AsyncMock() - - # Setup mock client - mock_client = MagicMock() - mock_client.build_request = MagicMock(return_value=MagicMock()) - mock_client.send = AsyncMock(return_value=redirect_response) - mock_client.aclose = AsyncMock() - mock_client_class.return_value = mock_client - - # Attempt to proxy a URL that loops redirects - response = self._get("proxy?url=https://example.com/loop") - - # Should fail with 400 Bad Request due to too many redirects + url = mock_http_server.get_url( + remote_url="https://example.com/loop", + status=302, + redirect_to_self=True, + ) + response = self._get(f"proxy?url={url}") self._assert_status_code_is(response, 400) assert "Too many redirects" in response.json()["err_msg"] diff --git a/lib/galaxy_test/base/mock_http_server.py b/lib/galaxy_test/base/mock_http_server.py index 4d9899ffa5b..7dd6b2bd086 100644 --- a/lib/galaxy_test/base/mock_http_server.py +++ b/lib/galaxy_test/base/mock_http_server.py @@ -141,6 +141,7 @@ class MockHttpServer: request_method: str = "GET", support_head: bool = False, support_ranges: bool = False, + redirect_to_self: bool = False, ) -> str: """Register a mock endpoint and return its URL, or return remote_url with skip-if-down. @@ -155,6 +156,7 @@ class MockHttpServer: request_method: HTTP method to match (GET, POST, etc.). support_head: Whether to also respond to HEAD requests. support_ranges: Whether to support Range request headers (HTTP 206). + redirect_to_self: Whether to add a Location header pointing to this route's own URL. """ if self.is_remote: if not is_site_up(remote_url): @@ -194,7 +196,10 @@ class MockHttpServer: support_head=support_head, support_ranges=support_ranges, ) - return f"{self.base_url}{path}" + url = f"{self.base_url}{path}" + if redirect_to_self: + headers["Location"] = url + return url def start_mock_http_server(host: str = "127.0.0.1", port: int = 0) -> tuple[HTTPServer, str]: