diff --git a/web/pgadmin/llm/__init__.py b/web/pgadmin/llm/__init__.py index bf035b59b52..6480a6d5dbe 100644 --- a/web/pgadmin/llm/__init__.py +++ b/web/pgadmin/llm/__init__.py @@ -22,7 +22,7 @@ from pgadmin.utils.constants import MIMETYPE_APP_JS from pgadmin.utils.csrf import pgCSRFProtect import config -from pgadmin.llm.utils import LLMApiError +from pgadmin.llm.utils import LLMApiError, urlopen_no_redirect from pgadmin.tools.user_management.PgAdminPermissions import \ AllPermissionTypes @@ -807,7 +807,7 @@ def _fetch_anthropic_models(api_key, api_url=''): req = urllib.request.Request(url, headers=headers) try: - with urllib.request.urlopen( + with urlopen_no_redirect( req, timeout=30, context=SSL_CONTEXT ) as response: data = json.loads(response.read().decode('utf-8')) @@ -882,7 +882,7 @@ def _fetch_openai_models(api_key, api_url=''): req = urllib.request.Request(url, headers=headers) try: - with urllib.request.urlopen( + with urlopen_no_redirect( req, timeout=30, context=SSL_CONTEXT ) as response: data = json.loads(response.read().decode('utf-8')) @@ -947,7 +947,7 @@ def _fetch_ollama_models(api_url): req = urllib.request.Request(url) try: - with urllib.request.urlopen( + with urlopen_no_redirect( req, timeout=30, context=SSL_CONTEXT ) as response: data = json.loads(response.read().decode('utf-8')) @@ -1008,7 +1008,7 @@ def _fetch_docker_models(api_url): req = urllib.request.Request(url) try: - with urllib.request.urlopen( + with urlopen_no_redirect( req, timeout=30, context=SSL_CONTEXT ) as response: data = json.loads(response.read().decode('utf-8')) diff --git a/web/pgadmin/llm/providers/anthropic.py b/web/pgadmin/llm/providers/anthropic.py index c7b91a56e8e..bd5ee2d67ee 100644 --- a/web/pgadmin/llm/providers/anthropic.py +++ b/web/pgadmin/llm/providers/anthropic.py @@ -33,6 +33,7 @@ Message, Tool, ToolCall, LLMResponse, LLMError, Role, StopReason, Usage ) +from pgadmin.llm.utils import urlopen_no_redirect # Default model if none specified @@ -215,7 +216,7 @@ def _make_request(self, payload: dict) -> dict: ) try: - with urllib.request.urlopen( + with urlopen_no_redirect( request, timeout=120, context=SSL_CONTEXT ) as response: return json.loads(response.read().decode('utf-8')) @@ -343,7 +344,7 @@ def _process_stream( ) try: - response = urllib.request.urlopen( + response = urlopen_no_redirect( request, timeout=120, context=SSL_CONTEXT ) except urllib.error.HTTPError as e: diff --git a/web/pgadmin/llm/providers/docker.py b/web/pgadmin/llm/providers/docker.py index 52132827e67..7407a69766a 100644 --- a/web/pgadmin/llm/providers/docker.py +++ b/web/pgadmin/llm/providers/docker.py @@ -38,6 +38,7 @@ Message, Tool, ToolCall, LLMResponse, LLMError, Role, StopReason, Usage ) +from pgadmin.llm.utils import urlopen_no_redirect # Default configuration @@ -237,7 +238,7 @@ def _make_request(self, payload: dict) -> dict: try: # Use longer timeout for local models which can be slower - with urllib.request.urlopen( + with urlopen_no_redirect( request, timeout=300, context=SSL_CONTEXT ) as response: return json.loads(response.read().decode('utf-8')) @@ -436,7 +437,7 @@ def _process_stream( ) try: - response = urllib.request.urlopen( + response = urlopen_no_redirect( request, timeout=300, context=SSL_CONTEXT ) except urllib.error.HTTPError as e: diff --git a/web/pgadmin/llm/providers/ollama.py b/web/pgadmin/llm/providers/ollama.py index 1706e3d8ead..adc34824b98 100644 --- a/web/pgadmin/llm/providers/ollama.py +++ b/web/pgadmin/llm/providers/ollama.py @@ -22,6 +22,7 @@ Message, Tool, ToolCall, LLMResponse, LLMError, Role, StopReason, Usage ) +from pgadmin.llm.utils import urlopen_no_redirect # Default configuration @@ -72,7 +73,7 @@ def is_available(self) -> bool: try: # Check if Ollama is running req = urllib.request.Request(f'{self._api_url}/api/tags') - with urllib.request.urlopen(req, timeout=5) as response: + with urlopen_no_redirect(req, timeout=5) as response: data = json.loads(response.read().decode('utf-8')) # Check if our model is available models = [m.get('name', '') for m in data.get('models', [])] @@ -215,7 +216,7 @@ def _make_request(self, payload: dict) -> dict: ) try: - with urllib.request.urlopen(request, timeout=300) as response: + with urlopen_no_redirect(request, timeout=300) as response: return json.loads(response.read().decode('utf-8')) except urllib.error.HTTPError as e: error_body = e.read().decode('utf-8') @@ -348,7 +349,7 @@ def _process_stream( ) try: - response = urllib.request.urlopen(request, timeout=300) + response = urlopen_no_redirect(request, timeout=300) except urllib.error.HTTPError as e: error_body = e.read().decode('utf-8') try: diff --git a/web/pgadmin/llm/providers/openai.py b/web/pgadmin/llm/providers/openai.py index 38662de1511..c55ee642bf4 100644 --- a/web/pgadmin/llm/providers/openai.py +++ b/web/pgadmin/llm/providers/openai.py @@ -33,6 +33,7 @@ Message, Tool, ToolCall, LLMResponse, LLMError, Role, StopReason, Usage ) +from pgadmin.llm.utils import urlopen_no_redirect # Default model if none specified @@ -373,7 +374,7 @@ def _make_request(self, payload: dict) -> dict: ) try: - with urllib.request.urlopen( + with urlopen_no_redirect( request, timeout=120, context=SSL_CONTEXT ) as response: return json.loads(response.read().decode('utf-8')) @@ -667,7 +668,7 @@ def _process_stream( ) try: - response = urllib.request.urlopen( + response = urlopen_no_redirect( request, timeout=120, context=SSL_CONTEXT ) except urllib.error.HTTPError as e: diff --git a/web/pgadmin/llm/tests/test_no_redirect.py b/web/pgadmin/llm/tests/test_no_redirect.py new file mode 100644 index 00000000000..a2d5add997c --- /dev/null +++ b/web/pgadmin/llm/tests/test_no_redirect.py @@ -0,0 +1,137 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2026, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +"""Tests that LLM API requests never follow HTTP redirects. + +validate_api_url() only checks the URL pgAdmin was configured with, so a +redirect would otherwise reach a destination that check was never applied +to. urlopen_no_redirect() must refuse every redirect status code instead. +""" + +import json +import threading +import urllib.error +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +from pgadmin.utils.route import BaseTestGenerator +from pgadmin.llm.utils import urlopen_no_redirect + + +def _make_server(redirect_code): + """ + Start a local server that redirects /redirect to /target. + + Returns (base_url, server, target_hits), where target_hits has the + request method appended to it every time /target is actually + reached, so a test can prove the redirect was not followed. + """ + target_hits = [] + + class Handler(BaseHTTPRequestHandler): + def _respond(self): + length = int(self.headers.get('Content-Length') or 0) + if length: + self.rfile.read(length) + + if self.path == '/redirect': + self.send_response(redirect_code) + self.send_header('Location', '/target') + self.send_header('Content-Length', '0') + self.end_headers() + return + + if self.path == '/target': + target_hits.append(self.command) + + body = json.dumps({'path': self.path}).encode('utf-8') + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + + do_GET = _respond + do_POST = _respond + + def log_message(self, fmt, *args): + """Keep the test output quiet.""" + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + host, port = server.server_address[:2] + return 'http://%s:%d' % (host, port), server, target_hits + + +class NoRedirectTestCase(BaseTestGenerator): + """urlopen_no_redirect() must refuse every redirect status code.""" + + scenarios = [ + ('%s %d is refused' % (method, code), dict(method=method, code=code)) + for code in (301, 302, 303, 307, 308) + for method in ('GET', 'POST') + ] + + def setUp(self): + self.base_url, self.server, self.target_hits = \ + _make_server(self.code) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + + def runTest(self): + request = urllib.request.Request( + self.base_url + '/redirect', + data=b'{}' if self.method == 'POST' else None, + method=self.method + ) + + with self.assertRaises(urllib.error.HTTPError) as ctx: + urlopen_no_redirect(request, timeout=10) + + self.assertEqual(ctx.exception.code, self.code) + self.assertIn('ALLOWED_LLM_API_URLS', str(ctx.exception)) + # The redirect target must never have been contacted. + self.assertEqual(self.target_hits, []) + + +class RedirectFixtureControlTestCase(BaseTestGenerator): + """Control cases for the refusal tests above. + + Without the first of these, NoRedirectTestCase could pass simply + because the test server never issued a redirect in the first place. + """ + + scenarios = [ + ('Default urlopen does follow the redirect', dict(follow=True)), + ('A normal 200 response is returned unchanged', dict(follow=False)), + ] + + def setUp(self): + self.base_url, self.server, self.target_hits = _make_server(302) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + + def runTest(self): + if self.follow: + request = urllib.request.Request(self.base_url + '/redirect') + with urllib.request.urlopen(request, timeout=10) as response: + body = json.loads(response.read().decode('utf-8')) + self.assertEqual(body['path'], '/target') + self.assertEqual(self.target_hits, ['GET']) + else: + request = urllib.request.Request(self.base_url + '/target') + with urlopen_no_redirect(request, timeout=10) as response: + body = json.loads(response.read().decode('utf-8')) + self.assertEqual(body['path'], '/target') + self.assertEqual(self.target_hits, ['GET']) diff --git a/web/pgadmin/llm/utils.py b/web/pgadmin/llm/utils.py index c89b6c4f8c1..e58ab86aa98 100644 --- a/web/pgadmin/llm/utils.py +++ b/web/pgadmin/llm/utils.py @@ -10,6 +10,8 @@ """Utility functions for LLM configuration access.""" import os +import urllib.error +import urllib.request from pgadmin.utils.preferences import Preferences import config @@ -203,6 +205,52 @@ def validate_api_url(url): return False +class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + """ + Refuse to follow redirects returned by an LLM API endpoint. + + validate_api_url() checks the configured URL before the request is + made, but urllib's default redirect handler would then follow a + Location header to somewhere that check was never applied to. No + legitimate LLM API replies with a redirect, so refuse them outright + rather than re-validating each hop. + """ + + def redirect_request(self, req, fp, code, msg, headers, newurl): + raise urllib.error.HTTPError( + req.full_url, code, + 'Refusing to follow the redirect to %s returned by the LLM ' + 'API endpoint; the redirect target has not been checked ' + 'against ALLOWED_LLM_API_URLS' % newurl, + headers, fp + ) + + # HTTPRedirectHandler only gained http_error_308() in Python 3.11, so on + # older interpreters a 308 is not recognised as a redirect at all: it + # bypasses redirect_request() and surfaces from http_error_default() as a + # bare 'HTTP Error 308: Permanent Redirect'. That is safe, in that the + # redirect still isn't followed, but the caller gets no explanation of + # why. Map it explicitly so every redirect status behaves identically on + # every version we support. + http_error_308 = urllib.request.HTTPRedirectHandler.http_error_301 + + +def urlopen_no_redirect(request, timeout, context=None): + """ + Open an LLM API request without following redirects. + + Mirrors urllib.request.urlopen(), including its handling of an + explicit SSL context, but installs _NoRedirectHandler so that a + redirect raises urllib.error.HTTPError instead of being followed. + """ + handlers = [_NoRedirectHandler()] + if context is not None: + handlers.append(urllib.request.HTTPSHandler(context=context)) + + opener = urllib.request.build_opener(*handlers) + return opener.open(request, timeout=timeout) + + def _read_api_key_from_file(file_path, _trusted=False): """ Read an API key from a file.