Re-evaluate NO_PROXY per-request so it is honored across redirects

.changes/next-release/bugfix-Endpoint-12345.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "Endpoint",
+  "description": "Re-evaluate the ``NO_PROXY`` environment variable per-request inside ``URLLib3Session`` so proxy selection follows redirected URLs (e.g. S3 cross-region ``PermanentRedirect`` responses), rather than freezing the bypass decision against the initial endpoint URL."
+}

botocore/endpoint.py

@@ -433,7 +433,7 @@ def create_endpoint(
     def _get_proxies(self, url):
         # We could also support getting proxies from a config file,
         # but for now proxy support is taken from the environment.
-        return get_environ_proxies(url)
+        return get_environ_proxies(url, no_proxy_filter=False)
 
     def _get_verify_value(self, verify):
         # This is to account for:

botocore/httpsession.py

@@ -6,6 +6,7 @@
 import warnings
 from base64 import b64encode
 from concurrent.futures import CancelledError
+from urllib.request import proxy_bypass
 
 from urllib3 import PoolManager, Timeout, proxy_from_url
 from urllib3.exceptions import (
@@ -243,12 +244,25 @@ def __init__(self, proxies=None, proxies_settings=None):
 
     def proxy_url_for(self, url):
         """Retrieves the corresponding proxy url for a given url."""
+        if self._should_bypass_proxies(url):
+            return None
         parsed_url = urlparse(url)
         proxy = self._proxies.get(parsed_url.scheme)
         if proxy:
             proxy = self._fix_proxy_url(proxy)
         return proxy
 
+    @staticmethod
+    def _should_bypass_proxies(url):
+        # Inlined to avoid a circular import with ``botocore.utils``; mirrors
+        # ``botocore.utils.should_bypass_proxies``.
+        try:
+            if proxy_bypass(urlparse(url).netloc):
+                return True
+        except (TypeError, socket.gaierror):
+            pass
+        return False
+
     def proxy_headers_for(self, proxy_url):
         """Retrieves the corresponding proxy headers for a given proxy url."""
         headers = {}

botocore/utils.py

@@ -3187,11 +3187,15 @@ def full_url(self, relative_uri):
         return f'http://{self.IP_ADDRESS}{relative_uri}'
 
 
-def get_environ_proxies(url):
-    if should_bypass_proxies(url):
+def get_environ_proxies(url, no_proxy_filter=True):
+    """Return proxy URLs configured via the environment.
+
+    When ``no_proxy_filter`` is False, ``NO_PROXY`` is not applied here; the
+    caller is responsible for re-checking ``should_bypass_proxies`` per request.
+    """
+    if no_proxy_filter and should_bypass_proxies(url):
         return {}
-    else:
-        return getproxies()
+    return getproxies()
 
 
 def should_bypass_proxies(url):

tests/functional/test_proxy_redirect.py

@@ -0,0 +1,141 @@
+"""End-to-end test for NO_PROXY re-evaluation across redirect-style flows.
+
+This test exercises the real socket path through ``URLLib3Session``: two
+sequential requests through the same session, where one host matches
+``NO_PROXY`` and the other does not. With the per-request bypass check in
+place, the first request must go direct and the second must go through the
+proxy. Without it (the legacy behaviour), the bypass decision is frozen at
+session-construction time, so the second request also bypasses the proxy.
+"""
+
+import contextlib
+import socket
+import socketserver
+import threading
+from http.server import BaseHTTPRequestHandler
+
+from botocore.awsrequest import AWSRequest
+from botocore.httpsession import URLLib3Session
+from tests import mock, unittest
+
+
+def _unused_port():
+    with contextlib.closing(socket.socket()) as sock:
+        sock.bind(('127.0.0.1', 0))
+        return sock.getsockname()[1]
+
+
+class _RecordingHandler(BaseHTTPRequestHandler):
+    """HTTP handler that records every request path and host header."""
+
+    received = []
+    ready = None  # set in setUp
+
+    def do_GET(self):
+        type(self).received.append(
+            {
+                'path': self.path,
+                'host': self.headers.get('Host', ''),
+            }
+        )
+        self.send_response(200)
+        self.send_header('Content-Length', '2')
+        self.end_headers()
+        self.wfile.write(b'ok')
+
+    def log_message(self, format, *args):
+        pass
+
+
+def _serve(handler_cls, port, request_count):
+    address = ('127.0.0.1', port)
+    server = socketserver.TCPServer(
+        address, handler_cls, bind_and_activate=False
+    )
+    server.allow_reuse_address = True
+    server.server_bind()
+    server.server_activate()
+    handler_cls.ready.set()
+    try:
+        for _ in range(request_count):
+            server.handle_request()
+    finally:
+        server.server_close()
+
+
+class TestNoProxyAcrossRedirect(unittest.TestCase):
+    def setUp(self):
+        self.backend_port = _unused_port()
+        self.proxy_port = _unused_port()
+
+        class BackendHandler(_RecordingHandler):
+            received = []
+            ready = threading.Event()
+
+        class ProxyHandler(_RecordingHandler):
+            received = []
+            ready = threading.Event()
+
+        self.BackendHandler = BackendHandler
+        self.ProxyHandler = ProxyHandler
+
+        self.backend_thread = threading.Thread(
+            target=_serve,
+            args=(BackendHandler, self.backend_port, 1),
+            daemon=True,
+        )
+        self.proxy_thread = threading.Thread(
+            target=_serve,
+            args=(ProxyHandler, self.proxy_port, 1),
+            daemon=True,
+        )
+        self.backend_thread.start()
+        self.proxy_thread.start()
+        self.assertTrue(BackendHandler.ready.wait(timeout=5))
+        self.assertTrue(ProxyHandler.ready.wait(timeout=5))
+
+    def tearDown(self):
+        self.backend_thread.join(timeout=5)
+        self.proxy_thread.join(timeout=5)
+
+    def test_no_proxy_re_evaluated_across_sequential_requests(self):
+        proxy_url = f'http://127.0.0.1:{self.proxy_port}'
+        initial_url = f'http://127.0.0.1:{self.backend_port}/initial'
+        redirect_url = f'http://localhost:{self.backend_port}/redirect'
+
+        env = {
+            'HTTP_PROXY': proxy_url,
+            'NO_PROXY': '127.0.0.1',
+        }
+        with mock.patch.dict('os.environ', env, clear=True):
+            session = URLLib3Session(proxies={'http': proxy_url})
+
+            req1 = AWSRequest(
+                method='GET', url=initial_url, headers={}, data=b''
+            )
+            resp1 = session.send(req1.prepare())
+            self.assertEqual(resp1.status_code, 200)
+
+            req2 = AWSRequest(
+                method='GET', url=redirect_url, headers={}, data=b''
+            )
+            resp2 = session.send(req2.prepare())
+            self.assertEqual(resp2.status_code, 200)
+
+            session.close()
+
+        self.assertEqual(
+            len(self.BackendHandler.received),
+            1,
+            'NO_PROXY host should have been contacted directly once',
+        )
+        self.assertEqual(self.BackendHandler.received[0]['path'], '/initial')
+
+        self.assertEqual(
+            len(self.ProxyHandler.received),
+            1,
+            'Redirect target (not in NO_PROXY) should go via the proxy',
+        )
+        proxy_req = self.ProxyHandler.received[0]
+        self.assertEqual(proxy_req['path'], redirect_url)
+        self.assertIn('localhost', proxy_req['host'])

tests/unit/test_http_session.py

@@ -1,3 +1,4 @@
+import os
 import socket
 from concurrent.futures import CancelledError
 
@@ -58,6 +59,21 @@ def test_fix_proxy_url_has_protocol_http(self):
         proxy_url = self.proxy_config.proxy_url_for(self.url)
         self.assertEqual('http://localhost:8081/', proxy_url)
 
+    def test_proxy_url_for_honors_no_proxy_per_call(self):
+        config = ProxyConfiguration(
+            proxies={'https': 'http://proxy.example.com'}
+        )
+        with mock.patch.dict(
+            os.environ, {'NO_PROXY': 'bypass.example.com'}, clear=True
+        ):
+            self.assertIsNone(
+                config.proxy_url_for('https://bypass.example.com/key')
+            )
+            self.assertEqual(
+                'http://proxy.example.com',
+                config.proxy_url_for('https://other.example.com/key'),
+            )
+
 
 class TestHttpSessionUtils(unittest.TestCase):
     def test_get_cert_path_path(self):
@@ -511,6 +527,70 @@ def test_close(self):
         session.close()
         self.pool_manager.clear.assert_called_once_with()
 
+    def test_no_proxy_env_var_bypasses_proxy_per_request(self):
+        proxies = {'https': 'http://proxy.com'}
+        non_proxy_manager = mock.Mock()
+        non_proxy_manager.connection_from_url.return_value = self.connection
+        self.pool_manager_cls.return_value = non_proxy_manager
+
+        proxy_manager = mock.Mock()
+        proxy_manager.connection_from_url.return_value = self.connection
+        self.proxy_manager_fun.return_value = proxy_manager
+
+        session = URLLib3Session(proxies=proxies)
+        bypass_url = 'https://bypass.example.com/'
+        proxied_url = 'https://other.example.com/'
+
+        with mock.patch.dict(
+            os.environ, {'NO_PROXY': 'bypass.example.com'}, clear=True
+        ):
+            req1 = AWSRequest(
+                method='GET', url=bypass_url, headers={}, data=b''
+            )
+            session.send(req1.prepare())
+            non_proxy_manager.connection_from_url.assert_called_with(
+                bypass_url
+            )
+            proxy_manager.connection_from_url.assert_not_called()
+
+            req2 = AWSRequest(
+                method='GET', url=proxied_url, headers={}, data=b''
+            )
+            session.send(req2.prepare())
+            proxy_manager.connection_from_url.assert_called_with(proxied_url)
+
+    def test_no_proxy_env_var_matches_redirect_target(self):
+        proxies = {'https': 'http://proxy.com'}
+        non_proxy_manager = mock.Mock()
+        non_proxy_manager.connection_from_url.return_value = self.connection
+        self.pool_manager_cls.return_value = non_proxy_manager
+
+        proxy_manager = mock.Mock()
+        proxy_manager.connection_from_url.return_value = self.connection
+        self.proxy_manager_fun.return_value = proxy_manager
+
+        session = URLLib3Session(proxies=proxies)
+        proxied_url = 'https://api.example.com/'
+        bypass_url = 'https://bypass.example.com/'
+
+        with mock.patch.dict(
+            os.environ, {'NO_PROXY': 'bypass.example.com'}, clear=True
+        ):
+            req1 = AWSRequest(
+                method='GET', url=proxied_url, headers={}, data=b''
+            )
+            session.send(req1.prepare())
+            proxy_manager.connection_from_url.assert_called_with(proxied_url)
+            non_proxy_manager.connection_from_url.assert_not_called()
+
+            req2 = AWSRequest(
+                method='GET', url=bypass_url, headers={}, data=b''
+            )
+            session.send(req2.prepare())
+            non_proxy_manager.connection_from_url.assert_called_with(
+                bypass_url
+            )
+
     def test_close_proxied(self):
         proxies = {'https': 'http://proxy.com', 'http': 'http://proxy2.com'}
         session = URLLib3Session(proxies=proxies)