fix(security): add URL scheme validation to prevent credential interception

AWS credentials must be transmitted over HTTPS to prevent interception
via man-in-the-middle attacks. This adds validation to reject HTTP
endpoints for remote hosts while allowing HTTP for localhost during
local development.

- Add validate_endpoint_url() function to utils.py
- Integrate validation into get_service_name_and_region_from_endpoint()
- Integrate validation into aws_iam_streamablehttp_client()
- Add comprehensive tests for URL scheme validation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: DennisTraub

mcp_proxy_for_aws/client.py

@@ -22,6 +22,7 @@
 from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client
 from mcp.shared.message import SessionMessage
 from mcp_proxy_for_aws.sigv4_helper import SigV4HTTPXAuth
+from mcp_proxy_for_aws.utils import validate_endpoint_url
 from typing import Optional
 
 
@@ -53,7 +54,8 @@ def aws_iam_streamablehttp_client(
     authentication via SigV4 signing. Use with 'async with' to manage the connection lifecycle.
 
     Args:
-        endpoint: The URL of the MCP server to connect to. Must be a valid HTTP/HTTPS URL.
+        endpoint: The URL of the MCP server to connect to. Must use HTTPS for remote endpoints
+            (HTTP is allowed for localhost during development).
         aws_service: The name of the AWS service the MCP server is hosted on, e.g. "bedrock-agentcore".
         aws_region: The AWS region name of the MCP server, e.g. "us-west-2".
         aws_profile: The AWS profile to use for authentication.
@@ -81,6 +83,9 @@ def aws_iam_streamablehttp_client(
     """
     logger.debug('Preparing AWS IAM MCP client for endpoint: %s', endpoint)
 
+    # Validate URL scheme for security - AWS credentials must be transmitted over HTTPS
+    validate_endpoint_url(endpoint)
+
     if credentials is not None:
         creds = credentials
         region = aws_region

mcp_proxy_for_aws/utils.py

@@ -27,6 +27,45 @@
 logger = logging.getLogger(__name__)
 
 
+def validate_endpoint_url(endpoint: str, allow_localhost_http: bool = True) -> None:
+    """Validate endpoint URL scheme for secure credential transmission.
+
+    AWS credentials are sent via SigV4 signing headers, so endpoints must use HTTPS
+    to prevent credential interception via man-in-the-middle attacks.
+
+    Args:
+        endpoint: The endpoint URL to validate
+        allow_localhost_http: Whether to allow HTTP for localhost (default: True)
+
+    Raises:
+        ValueError: If URL scheme is not HTTPS (or HTTP for allowed localhost)
+    """
+    parsed = urlparse(endpoint)
+
+    if not parsed.scheme:
+        raise ValueError(
+            f"Invalid endpoint URL '{endpoint}': missing URL scheme. "
+            "Use https:// prefix for secure connections."
+        )
+
+    if parsed.scheme == 'https':
+        return  # Valid
+
+    if parsed.scheme == 'http':
+        if allow_localhost_http and parsed.hostname in ('localhost', '127.0.0.1', '::1'):
+            return  # Allow HTTP for local development
+        raise ValueError(
+            f"Invalid endpoint URL '{endpoint}': HTTP is not allowed for remote endpoints. "
+            "AWS credentials must be transmitted over HTTPS to prevent interception. "
+            "Use https:// instead."
+        )
+
+    raise ValueError(
+        f"Invalid endpoint URL '{endpoint}': unsupported scheme '{parsed.scheme}'. "
+        "Only HTTPS is supported for secure credential transmission."
+    )
+
+
 def create_transport_with_sigv4(
     url: str,
     service: str,
@@ -78,16 +117,23 @@ def get_service_name_and_region_from_endpoint(endpoint: str) -> Tuple[str, str]:
     """Extract service name and region from an endpoint URL.
 
     Args:
-        endpoint: The endpoint URL to parse
+        endpoint: The endpoint URL to parse (must use HTTPS for remote endpoints)
 
     Returns:
         Tuple of (service_name, region). Either value may be empty string if not found.
 
+    Raises:
+        ValueError: If endpoint URL does not use HTTPS (except localhost for development)
+
     Notes:
+        - Validates URL scheme is HTTPS before parsing
         - Matches bedrock-agentcore endpoints (gateway and runtime)
         - Matches AWS API Gateway endpoints (service.region.api.aws)
         - Falls back to extracting first hostname segment as service name
     """
+    # Validate URL scheme for security
+    validate_endpoint_url(endpoint)
+
     # Parse AWS service from endpoint URL
     parsed = urlparse(endpoint)
     hostname = parsed.hostname or ''

tests/unit/test_client.py

@@ -279,3 +279,39 @@ async def test_credentials_parameter_bypasses_boto3_session(mock_streams):
                     pass
 
                 mock_boto.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_http_endpoint_raises_security_error():
+    """Test that HTTP endpoints raise ValueError for security.
+
+    AWS credentials must be transmitted over HTTPS to prevent interception.
+    This test verifies the client rejects HTTP endpoints for remote hosts.
+    """
+    with pytest.raises(ValueError, match='HTTP is not allowed'):
+        async with aws_iam_streamablehttp_client(
+            endpoint='http://example.com/mcp',
+            aws_service='bedrock-agentcore',
+            aws_region='us-west-2',
+        ):
+            pass
+
+
+@pytest.mark.asyncio
+async def test_http_localhost_endpoint_allowed(mock_session, mock_streams):
+    """Test that HTTP localhost endpoints are allowed for local development."""
+    mock_read, mock_write, mock_get_session = mock_streams
+
+    with patch('boto3.Session', return_value=mock_session):
+        with patch('mcp_proxy_for_aws.client.streamablehttp_client') as mock_stream_client:
+            mock_stream_client.return_value.__aenter__ = AsyncMock(
+                return_value=(mock_read, mock_write, mock_get_session)
+            )
+            mock_stream_client.return_value.__aexit__ = AsyncMock(return_value=None)
+
+            # Should not raise - localhost HTTP is allowed for development
+            async with aws_iam_streamablehttp_client(
+                endpoint='http://localhost:8080/mcp',
+                aws_service='bedrock-agentcore',
+            ):
+                pass

tests/unit/test_utils.py

@@ -20,10 +20,98 @@
     create_transport_with_sigv4,
     determine_aws_region,
     determine_service_name,
+    validate_endpoint_url,
 )
 from unittest.mock import MagicMock, patch
 
 
+class TestValidateEndpointUrl:
+    """Test cases for validate_endpoint_url function."""
+
+    def test_https_url_passes(self):
+        """HTTPS URLs should pass validation."""
+        validate_endpoint_url('https://example.com/mcp')
+        # No exception = pass
+
+    def test_https_with_port_passes(self):
+        """HTTPS URLs with explicit port should pass."""
+        validate_endpoint_url('https://example.com:443/mcp')
+
+    def test_https_aws_endpoints_pass(self):
+        """HTTPS URLs for AWS endpoints should pass."""
+        aws_endpoints = [
+            'https://test-service.us-west-2.api.aws/mcp',
+            'https://bedrock-agentcore.us-east-1.amazonaws.com',
+            'https://gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp',
+        ]
+        for endpoint in aws_endpoints:
+            validate_endpoint_url(endpoint)  # No exception = pass
+
+    def test_http_remote_raises_error(self):
+        """HTTP URLs for remote hosts should raise ValueError."""
+        with pytest.raises(ValueError) as exc_info:
+            validate_endpoint_url('http://example.com/mcp')
+        assert 'HTTP is not allowed' in str(exc_info.value)
+        assert 'HTTPS' in str(exc_info.value)
+
+    def test_http_remote_error_includes_url(self):
+        """Error message should include the invalid URL."""
+        with pytest.raises(ValueError) as exc_info:
+            validate_endpoint_url('http://remote-server.example.com/mcp')
+        assert 'remote-server.example.com' in str(exc_info.value)
+
+    def test_http_localhost_allowed(self):
+        """HTTP URLs for localhost should pass by default."""
+        validate_endpoint_url('http://localhost:8080/mcp')
+        validate_endpoint_url('http://127.0.0.1:8080/mcp')
+        validate_endpoint_url('http://[::1]:8080/mcp')
+
+    def test_http_localhost_without_port_allowed(self):
+        """HTTP URLs for localhost without port should pass."""
+        validate_endpoint_url('http://localhost/mcp')
+        validate_endpoint_url('http://127.0.0.1/mcp')
+
+    def test_http_localhost_disallowed_when_strict(self):
+        """HTTP localhost should fail when allow_localhost_http=False."""
+        with pytest.raises(ValueError) as exc_info:
+            validate_endpoint_url('http://localhost:8080/mcp', allow_localhost_http=False)
+        assert 'HTTP is not allowed' in str(exc_info.value)
+
+    def test_missing_scheme_raises_error(self):
+        """URLs without scheme should raise ValueError."""
+        with pytest.raises(ValueError) as exc_info:
+            validate_endpoint_url('example.com/mcp')
+        assert 'missing URL scheme' in str(exc_info.value)
+
+    def test_unsupported_scheme_ftp_raises_error(self):
+        """FTP scheme should raise ValueError."""
+        with pytest.raises(ValueError) as exc_info:
+            validate_endpoint_url('ftp://example.com/mcp')
+        assert 'unsupported scheme' in str(exc_info.value)
+        assert 'ftp' in str(exc_info.value)
+
+    def test_unsupported_scheme_file_raises_error(self):
+        """File scheme should raise ValueError."""
+        with pytest.raises(ValueError) as exc_info:
+            validate_endpoint_url('file:///path/to/file')
+        assert 'unsupported scheme' in str(exc_info.value)
+        assert 'file' in str(exc_info.value)
+
+    def test_unsupported_scheme_ws_raises_error(self):
+        """WebSocket scheme should raise ValueError."""
+        with pytest.raises(ValueError) as exc_info:
+            validate_endpoint_url('ws://example.com/mcp')
+        assert 'unsupported scheme' in str(exc_info.value)
+        assert 'ws' in str(exc_info.value)
+
+    def test_unsupported_scheme_wss_raises_error(self):
+        """Secure WebSocket scheme should raise ValueError."""
+        with pytest.raises(ValueError) as exc_info:
+            validate_endpoint_url('wss://example.com/mcp')
+        assert 'unsupported scheme' in str(exc_info.value)
+        assert 'wss' in str(exc_info.value)
+
+
 class TestCreateTransportWithSigv4:
     """Test cases for create_transport_with_sigv4 function (line 129)."""
 
@@ -183,15 +271,15 @@ def test_validate_service_name_without_service_failure(self):
         assert '--service argument' in str(exc_info.value)
 
     def test_validate_service_name_invalid_url_failure(self):
-        """Test validation with invalid URL."""
+        """Test validation with invalid URL raises scheme validation error."""
         endpoint = 'not-a-url'
 
         with pytest.raises(ValueError) as exc_info:
             determine_service_name(endpoint)
 
-        assert 'Could not determine AWS service name' in str(exc_info.value)
+        # URL validation now catches this first with a scheme error
+        assert 'missing URL scheme' in str(exc_info.value)
         assert endpoint in str(exc_info.value)
-        assert '--service argument' in str(exc_info.value)
 
 
 class TestDetermineRegion: