fix: always read fresh credentials from disk on every request

Author: anasstahr

mcp_proxy_for_aws/sigv4_helper.py

@@ -112,34 +112,20 @@ def create_aws_session(profile: Optional[str] = None) -> boto3.Session:
 
 
 class SessionHolder:
-    """Holds a boto3 session that can be refreshed on credential errors.
+    """Provides fresh boto3 sessions so every request reads current credentials from disk.
 
-    Wraps a boto3.Session so the signing hook always uses the current session,
-    and can create a fresh session when the previous request got an auth error.
+    Instead of caching a session and refreshing reactively on auth errors,
+    this always creates a new session so account switches and credential
+    refreshes take effect immediately.
     """
 
-    def __init__(self, session: boto3.Session, profile: Optional[str] = None) -> None:
-        """Initialize SessionHolder with the given session and optional profile."""
-        self.session = session
+    def __init__(self, profile: Optional[str] = None) -> None:
+        """Initialize SessionHolder with the given profile."""
         self._profile = profile
-        self._needs_refresh = False
 
-    def mark_needs_refresh(self) -> None:
-        """Mark that the next request should use a fresh session."""
-        self._needs_refresh = True
-
-    def refresh_if_needed(self) -> None:
-        """Create a fresh session if a previous request got an auth error."""
-        if not self._needs_refresh:
-            return
-        logger.info('Refreshing AWS session to pick up new credentials')
-        try:
-            self.session = create_aws_session(self._profile)
-            self._needs_refresh = False
-        except Exception:
-            logger.warning(
-                'Failed to create fresh AWS session, keeping current session', exc_info=True
-            )
+    def get_session(self) -> boto3.Session:
+        """Create and return a fresh boto3 session reading current credentials from disk."""
+        return create_aws_session(self._profile)
 
 
 def create_sigv4_client(
@@ -219,17 +205,13 @@ async def _handle_error_response(session_holder: SessionHolder, response: httpx.
     and provide more detailed error information when requests fail.
 
     Args:
-        session_holder: SessionHolder to refresh on credential errors
+        session_holder: SessionHolder (unused, kept for hook signature compatibility)
         response: The HTTP response object
 
     Raises:
         No raises. let the mcp http client handle the errors.
     """
     if response.is_error:
-        # Mark session for refresh so the next request picks up new credentials
-        if response.status_code in (401, 403):
-            session_holder.mark_needs_refresh()
-
         # warning only because the SDK logs error
         log_level = logging.WARNING
         if (
@@ -290,13 +272,10 @@ async def _sign_request_hook(
     # Set Content-Length for signing
     request.headers['Content-Length'] = str(len(request.content))
 
-    # Refresh session if a previous request got an auth error.
-    # Done here (at signing time) so the new session reads credentials
-    # that the user may have refreshed since the error occurred.
-    session_holder.refresh_if_needed()
-
-    # Get AWS credentials from the session
-    credentials = session_holder.session.get_credentials()
+    # Always read fresh credentials from disk so account switches
+    # and credential refreshes take effect immediately.
+    session = session_holder.get_session()
+    credentials = session.get_credentials()
 
     if credentials is None:
         if skip_auth:

mcp_proxy_for_aws/utils.py

@@ -19,7 +19,7 @@
 import logging
 import os
 from fastmcp.client.transports import StreamableHttpTransport
-from mcp_proxy_for_aws.sigv4_helper import SessionHolder, create_aws_session, create_sigv4_client
+from mcp_proxy_for_aws.sigv4_helper import SessionHolder, create_sigv4_client
 from typing import Any, Dict, Optional, Tuple
 from urllib.parse import urlparse
 
@@ -91,10 +91,9 @@ def create_transport_with_sigv4(
     Returns:
         StreamableHttpTransport instance with SigV4 authentication
     """
-    # Create AWS session with a holder that can refresh on credential errors
-    logger.debug('Creating AWS session with profile: %s', profile)
-    session = create_aws_session(profile)
-    session_holder = SessionHolder(session, profile)
+    # Create session holder that reads fresh credentials on every request
+    logger.debug('Creating session holder with profile: %s', profile)
+    session_holder = SessionHolder(profile=profile)
 
     def client_factory(
         headers: Optional[Dict[str, str]] = None,

tests/unit/test_hooks.py

@@ -56,16 +56,16 @@ def create_mock_session():
 def create_mock_session_holder():
     """Helper to create a mocked SessionHolder."""
     holder = MagicMock(spec=SessionHolder)
-    holder.session = create_mock_session()
+    holder.get_session.return_value = create_mock_session()
     return holder
 
 
 class TestHandleErrorResponse:
     """Test cases for the _handle_error_response function."""
 
     @pytest.mark.asyncio
-    async def test_handle_error_response_marks_refresh_on_401(self):
-        """401 response marks the session holder for refresh."""
+    async def test_handle_error_response_logs_401(self):
+        """401 response is handled without error."""
         request = httpx.Request('POST', 'https://example.com/mcp')
         response = httpx.Response(
             status_code=401,
@@ -77,11 +77,9 @@ async def test_handle_error_response_marks_refresh_on_401(self):
 
         await _handle_error_response(holder, response)
 
-        holder.mark_needs_refresh.assert_called_once()
-
     @pytest.mark.asyncio
-    async def test_handle_error_response_marks_refresh_on_403(self):
-        """403 response marks the session holder for refresh."""
+    async def test_handle_error_response_logs_403(self):
+        """403 response is handled without error."""
         request = httpx.Request('POST', 'https://example.com/mcp')
         response = httpx.Response(
             status_code=403,
@@ -93,25 +91,6 @@ async def test_handle_error_response_marks_refresh_on_403(self):
 
         await _handle_error_response(holder, response)
 
-        holder.mark_needs_refresh.assert_called_once()
-
-    @pytest.mark.asyncio
-    async def test_handle_error_response_does_not_mark_refresh_on_other_errors(self):
-        """Non-auth error codes (400, 404, 500) do not mark refresh."""
-        for status_code in (400, 404, 500):
-            request = httpx.Request('POST', 'https://example.com/mcp')
-            response = httpx.Response(
-                status_code=status_code,
-                headers={'content-type': 'text/plain'},
-                content=b'Error',
-                request=request,
-            )
-            holder = create_mock_session_holder()
-
-            await _handle_error_response(holder, response)
-
-            holder.mark_needs_refresh.assert_not_called()
-
     @pytest.mark.asyncio
     async def test_handle_error_response_with_json_error(self):
         """Test error handling with JSON error response."""
@@ -401,15 +380,15 @@ class TestSignRequestHook:
     """Test cases for sign_request_hook function."""
 
     @pytest.mark.asyncio
-    async def test_sign_request_hook_calls_refresh_if_needed(self):
-        """Signing hook calls refresh_if_needed before signing."""
+    async def test_sign_request_hook_calls_get_session(self):
+        """Signing hook calls get_session to read fresh credentials."""
         holder = create_mock_session_holder()
         request_body = b'{"test": "data"}'
         request = httpx.Request('POST', 'https://example.com/mcp', content=request_body)
 
         await _sign_request_hook('us-east-1', 'execute-api', holder, False, request)
 
-        holder.refresh_if_needed.assert_called_once()
+        holder.get_session.assert_called_once()
 
     @pytest.mark.asyncio
     async def test_sign_request_hook_signs_request(self):
@@ -482,7 +461,7 @@ async def test_sign_request_hook_with_partial_application(self):
     async def test_sign_request_hook_skips_signing_when_skip_auth(self):
         """Request is sent unsigned when credentials are unavailable and skip_auth is True."""
         holder = create_mock_session_holder()
-        holder.session.get_credentials.return_value = None
+        holder.get_session.return_value.get_credentials.return_value = None
 
         request_body = b'{"test": "data"}'
         request = httpx.Request('POST', 'https://example.com/mcp', content=request_body)
@@ -497,7 +476,7 @@ async def test_sign_request_hook_skips_signing_when_skip_auth(self):
     async def test_sign_request_hook_raises_when_no_credentials_and_no_skip_auth(self):
         """ValueError is raised when credentials are unavailable and skip_auth is False."""
         holder = create_mock_session_holder()
-        holder.session.get_credentials.return_value = None
+        holder.get_session.return_value.get_credentials.return_value = None
 
         request_body = b'{"test": "data"}'
         request = httpx.Request('POST', 'https://example.com/mcp', content=request_body)
@@ -506,25 +485,25 @@ async def test_sign_request_hook_raises_when_no_credentials_and_no_skip_auth(sel
             await _sign_request_hook('us-east-1', 'execute-api', holder, False, request)
 
     @pytest.mark.asyncio
-    async def test_sign_request_hook_no_credentials_still_refreshes(self):
-        """refresh_if_needed is called even when credentials end up None."""
+    async def test_sign_request_hook_no_credentials_still_calls_get_session(self):
+        """get_session is called even when credentials end up None."""
         holder = create_mock_session_holder()
-        holder.session.get_credentials.return_value = None
+        holder.get_session.return_value.get_credentials.return_value = None
 
         request_body = b'test'
         request = httpx.Request('POST', 'https://example.com/mcp', content=request_body)
 
         with pytest.raises(ValueError):
             await _sign_request_hook('us-east-1', 'execute-api', holder, False, request)
 
-        holder.refresh_if_needed.assert_called_once()
+        holder.get_session.assert_called_once()
 
     @pytest.mark.asyncio
     @patch('mcp_proxy_for_aws.sigv4_helper.SigV4HTTPXAuth')
     async def test_sign_request_hook_no_credentials_does_not_create_auth(self, mock_auth_class):
         """SigV4HTTPXAuth is never instantiated when credentials are None and skip_auth is True."""
         holder = create_mock_session_holder()
-        holder.session.get_credentials.return_value = None
+        holder.get_session.return_value.get_credentials.return_value = None
 
         request_body = b'test'
         request = httpx.Request('POST', 'https://example.com/mcp', content=request_body)

tests/unit/test_server.py

@@ -406,9 +406,7 @@ def test_validate_service_name_service_parsing(self):
     @patch('mcp_proxy_for_aws.sigv4_helper.httpx.AsyncClient')
     def test_create_sigv4_client(self, mock_async_client):
         """Test creating SigV4 authenticated client with request hooks."""
-        mock_session = Mock()
-        mock_session.get_credentials.return_value = Mock(access_key='test-key')
-        session_holder = SessionHolder(mock_session, profile='test-profile')
+        session_holder = SessionHolder(profile='test-profile')
 
         create_sigv4_client(
             service='test-service', region='us-west-2', session_holder=session_holder
@@ -423,8 +421,7 @@ def test_create_sigv4_client(self, mock_async_client):
 
     def test_create_sigv4_client_no_credentials(self):
         """Test that credential check happens in sign_request_hook, not during client creation."""
-        mock_session = Mock()
-        session_holder = SessionHolder(mock_session)
+        session_holder = SessionHolder()
 
         client = create_sigv4_client(
             service='test-service', region='test-region', session_holder=session_holder