Don't follow HTTP redirects on LLM API requests

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'))

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:

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:

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:

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:

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'])

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.