fix: validate region parameter to prevent SSRF request redirection

src/bedrock_agentcore/_utils/endpoints.py

@@ -1,16 +1,81 @@
 """Endpoint utilities for BedrockAgentCore services."""
 
 import os
+import re
+from urllib.parse import urlparse
 
 # Environment-configurable constants with fallback defaults
 DP_ENDPOINT_OVERRIDE = os.getenv("BEDROCK_AGENTCORE_DP_ENDPOINT")
 CP_ENDPOINT_OVERRIDE = os.getenv("BEDROCK_AGENTCORE_CP_ENDPOINT")
 DEFAULT_REGION = os.getenv("AWS_REGION", "us-west-2")
 
+# Regex for valid AWS region names (e.g., us-east-1, eu-west-2, cn-north-1, us-gov-west-1).
+# Uses \A and \Z anchors to prevent newline injection bypass that $ allows.
+_VALID_REGION_PATTERN = re.compile(r"\A[a-z]{2}(-[a-z]+)+-\d+\Z")
+
+
+class InvalidRegionError(ValueError):
+    """Raised when an invalid AWS region string is provided.
+
+    This prevents SSRF attacks where a crafted region value
+    (e.g., ``x@attacker.com:443/#``) could redirect SDK API calls
+    to non-AWS hosts.
+    """
+
+
+def validate_region(region: str) -> str:
+    """Validate that a region string is a well-formed AWS region name.
+
+    Args:
+        region: The region string to validate.
+
+    Returns:
+        The validated region string (unchanged).
+
+    Raises:
+        InvalidRegionError: If the region does not match the expected pattern.
+    """
+    if not isinstance(region, str) or not _VALID_REGION_PATTERN.match(region):
+        raise InvalidRegionError(
+            f"Invalid AWS region: {region!r}. Region must match pattern like 'us-east-1', 'eu-west-2', 'cn-north-1'."
+        )
+    return region
+
+
+def _validate_endpoint_url(url: str) -> str:
+    """Validate that a constructed endpoint URL resolves to an AWS host.
+
+    This is a defense-in-depth check that catches URL manipulation even if
+    the region regex is somehow bypassed.
+
+    Args:
+        url: The constructed endpoint URL.
+
+    Returns:
+        The validated URL (unchanged).
+
+    Raises:
+        InvalidRegionError: If the URL hostname does not end with an AWS domain.
+    """
+    parsed = urlparse(url)
+    hostname = parsed.hostname or ""
+    _AWS_DOMAINS = (".amazonaws.com", ".amazonaws.com.cn", ".api.aws")
+    if not any(hostname.endswith(d) for d in _AWS_DOMAINS):
+        raise InvalidRegionError(f"Constructed endpoint resolves to non-AWS host: {hostname!r}")
+    return url
+
 
 def get_data_plane_endpoint(region: str = DEFAULT_REGION) -> str:
-    return DP_ENDPOINT_OVERRIDE or f"https://bedrock-agentcore.{region}.amazonaws.com"
+    if DP_ENDPOINT_OVERRIDE:
+        return _validate_endpoint_url(DP_ENDPOINT_OVERRIDE)
+    validate_region(region)
+    url = f"https://bedrock-agentcore.{region}.amazonaws.com"
+    return _validate_endpoint_url(url)
 
 
 def get_control_plane_endpoint(region: str = DEFAULT_REGION) -> str:
-    return CP_ENDPOINT_OVERRIDE or f"https://bedrock-agentcore-control.{region}.amazonaws.com"
+    if CP_ENDPOINT_OVERRIDE:
+        return _validate_endpoint_url(CP_ENDPOINT_OVERRIDE)
+    validate_region(region)
+    url = f"https://bedrock-agentcore-control.{region}.amazonaws.com"
+    return _validate_endpoint_url(url)

src/bedrock_agentcore/memory/controlplane.py

@@ -33,12 +33,13 @@ def __init__(self, region_name: str = "us-west-2", environment: str = "prod"):
         self.region_name = region_name
         self.environment = environment
 
-        self.endpoint = os.getenv(
-            "BEDROCK_AGENTCORE_CONTROL_ENDPOINT", f"https://bedrock-agentcore-control.{region_name}.amazonaws.com"
-        )
-
         service_name = os.getenv("BEDROCK_AGENTCORE_CONTROL_SERVICE", "bedrock-agentcore-control")
-        self.client = boto3.client(service_name, region_name=self.region_name, endpoint_url=self.endpoint)
+        cp_kwargs: dict = {"region_name": self.region_name}
+        control_endpoint = os.getenv("BEDROCK_AGENTCORE_CONTROL_ENDPOINT")
+        if control_endpoint:
+            cp_kwargs["endpoint_url"] = control_endpoint
+        self.client = boto3.client(service_name, **cp_kwargs)
+        self.endpoint = self.client.meta.endpoint_url
 
         logger.info("Initialized MemoryControlPlaneClient for %s in %s", environment, region_name)
 

src/bedrock_agentcore/runtime/a2a.py

@@ -75,6 +75,8 @@ def build_runtime_url(agent_arn: str, region: Optional[str] = None) -> str:
     """
     from urllib.parse import quote
 
+    from .._utils.endpoints import validate_region
+
     if region is None:
         # ARN format: arn:aws:bedrock-agentcore:<region>:<account>:runtime/<id>
         parts = agent_arn.split(":")
@@ -83,6 +85,7 @@ def build_runtime_url(agent_arn: str, region: Optional[str] = None) -> str:
         else:
             raise ValueError(f"Cannot extract region from ARN: {agent_arn}")
 
+    validate_region(region)
     encoded_arn = quote(agent_arn, safe="")
     return f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{encoded_arn}/invocations"
 

src/bedrock_agentcore/runtime/agent_core_runtime_client.py

@@ -43,6 +43,9 @@ def __init__(self, region: str, session: Optional[boto3.Session] = None) -> None
             session (Optional[boto3.Session]): Optional boto3 session. If not provided,
                 a new session will be created using default credentials.
         """
+        from .._utils.endpoints import validate_region
+
+        validate_region(region)
         self.region = region
         self.logger = logging.getLogger(__name__)
 

src/bedrock_agentcore/services/identity.py

@@ -10,7 +10,7 @@
 import boto3
 from pydantic import BaseModel
 
-from bedrock_agentcore._utils.endpoints import get_control_plane_endpoint, get_data_plane_endpoint
+from bedrock_agentcore._utils.endpoints import CP_ENDPOINT_OVERRIDE, DP_ENDPOINT_OVERRIDE
 
 
 class TokenPoller(ABC):
@@ -75,12 +75,14 @@ class IdentityClient:
     def __init__(self, region: str):
         """Initialize the identity client with the specified region."""
         self.region = region
-        self.cp_client = boto3.client(
-            "bedrock-agentcore-control", region_name=region, endpoint_url=get_control_plane_endpoint(region)
-        )
-        self.dp_client = boto3.client(
-            "bedrock-agentcore", region_name=region, endpoint_url=get_data_plane_endpoint(region)
-        )
+        cp_kwargs: dict = {"region_name": region}
+        if CP_ENDPOINT_OVERRIDE:
+            cp_kwargs["endpoint_url"] = CP_ENDPOINT_OVERRIDE
+        self.cp_client = boto3.client("bedrock-agentcore-control", **cp_kwargs)
+        dp_kwargs: dict = {"region_name": region}
+        if DP_ENDPOINT_OVERRIDE:
+            dp_kwargs["endpoint_url"] = DP_ENDPOINT_OVERRIDE
+        self.dp_client = boto3.client("bedrock-agentcore", **dp_kwargs)
         self.logger = logging.getLogger("bedrock_agentcore.identity_client")
 
     def create_oauth2_credential_provider(self, req):

src/bedrock_agentcore/services/resource_policy.py

@@ -6,7 +6,7 @@
 
 import boto3
 
-from bedrock_agentcore._utils.endpoints import get_control_plane_endpoint
+from bedrock_agentcore._utils.endpoints import CP_ENDPOINT_OVERRIDE
 
 
 class ResourcePolicyClient:
@@ -19,11 +19,10 @@ class ResourcePolicyClient:
     def __init__(self, region: str):
         """Initialize the client for the specified region."""
         self.region = region
-        self.client = boto3.client(
-            "bedrock-agentcore-control",
-            region_name=region,
-            endpoint_url=get_control_plane_endpoint(region),
-        )
+        cp_kwargs: dict = {"region_name": region}
+        if CP_ENDPOINT_OVERRIDE:
+            cp_kwargs["endpoint_url"] = CP_ENDPOINT_OVERRIDE
+        self.client = boto3.client("bedrock-agentcore-control", **cp_kwargs)
         self.logger = logging.getLogger("bedrock_agentcore.resource_policy_client")
 
     def put_resource_policy(self, resource_arn: str, policy: Union[str, dict]) -> dict:

src/bedrock_agentcore/tools/browser_client.py

@@ -21,7 +21,7 @@
 
 from bedrock_agentcore._utils.user_agent import build_user_agent_suffix
 
-from .._utils.endpoints import get_control_plane_endpoint, get_data_plane_endpoint
+from .._utils.endpoints import get_data_plane_endpoint
 from .config import (
     BrowserExtension,
     Certificate,
@@ -68,6 +68,9 @@ def __init__(self, region: str, integration_source: Optional[str] = None) -> Non
                 for telemetry (e.g., 'langchain', 'crewai'). Used to track
                 customer acquisition from different integrations.
         """
+        from bedrock_agentcore._utils.endpoints import CP_ENDPOINT_OVERRIDE, DP_ENDPOINT_OVERRIDE, validate_region
+
+        validate_region(region)
         self.region = region
         self.logger = logging.getLogger(__name__)
         self.integration_source = integration_source
@@ -76,21 +79,17 @@ def __init__(self, region: str, integration_source: Optional[str] = None) -> Non
         user_agent_extra = build_user_agent_suffix(integration_source)
         client_config = Config(user_agent_extra=user_agent_extra)
 
-        # Control plane client for browser management
-        self.control_plane_client = boto3.client(
-            "bedrock-agentcore-control",
-            region_name=region,
-            endpoint_url=get_control_plane_endpoint(region),
-            config=client_config,
-        )
-
-        # Data plane client for session operations
-        self.data_plane_client = boto3.client(
-            "bedrock-agentcore",
-            region_name=region,
-            endpoint_url=get_data_plane_endpoint(region),
-            config=client_config,
-        )
+        # Control plane client — let boto3 resolve endpoint natively.
+        cp_kwargs: dict = {"region_name": region, "config": client_config}
+        if CP_ENDPOINT_OVERRIDE:
+            cp_kwargs["endpoint_url"] = CP_ENDPOINT_OVERRIDE
+        self.control_plane_client = boto3.client("bedrock-agentcore-control", **cp_kwargs)
+
+        # Data plane client — same pattern.
+        dp_kwargs: dict = {"region_name": region, "config": client_config}
+        if DP_ENDPOINT_OVERRIDE:
+            dp_kwargs["endpoint_url"] = DP_ENDPOINT_OVERRIDE
+        self.data_plane_client = boto3.client("bedrock-agentcore", **dp_kwargs)
 
         self._identifier = None
         self._session_id = None

src/bedrock_agentcore/tools/code_interpreter_client.py

@@ -14,7 +14,7 @@
 import boto3
 from botocore.config import Config
 
-from bedrock_agentcore._utils.endpoints import get_control_plane_endpoint, get_data_plane_endpoint
+from bedrock_agentcore._utils.endpoints import CP_ENDPOINT_OVERRIDE, DP_ENDPOINT_OVERRIDE
 from bedrock_agentcore._utils.user_agent import build_user_agent_suffix
 
 from .config import Certificate
@@ -102,21 +102,18 @@ def __init__(
         # Data plane config (preserve existing read_timeout)
         data_config = Config(read_timeout=300, user_agent_extra=user_agent_extra)
 
-        # Control plane client for interpreter management
-        self.control_plane_client = session.client(
-            "bedrock-agentcore-control",
-            region_name=region,
-            endpoint_url=get_control_plane_endpoint(region),
-            config=control_config,
-        )
-
-        # Data plane client for session operations
-        self.data_plane_client = session.client(
-            "bedrock-agentcore",
-            region_name=region,
-            endpoint_url=get_data_plane_endpoint(region),
-            config=data_config,
-        )
+        # Control plane client — let boto3 resolve endpoint natively (includes region validation).
+        # Only pass endpoint_url when an environment override is set.
+        cp_kwargs: dict = {"region_name": region, "config": control_config}
+        if CP_ENDPOINT_OVERRIDE:
+            cp_kwargs["endpoint_url"] = CP_ENDPOINT_OVERRIDE
+        self.control_plane_client = session.client("bedrock-agentcore-control", **cp_kwargs)
+
+        # Data plane client — same pattern.
+        dp_kwargs: dict = {"region_name": region, "config": data_config}
+        if DP_ENDPOINT_OVERRIDE:
+            dp_kwargs["endpoint_url"] = DP_ENDPOINT_OVERRIDE
+        self.data_plane_client = session.client("bedrock-agentcore", **dp_kwargs)
 
         self._identifier = None
         self._session_id = None

tests/bedrock_agentcore/services/test_identity.py

@@ -29,7 +29,6 @@ def test_initialization(self):
             mock_boto_client.assert_called_with(
                 "bedrock-agentcore",
                 region_name=region,
-                endpoint_url="https://bedrock-agentcore.us-east-1.amazonaws.com",
             )
 
     def test_create_oauth2_credential_provider(self):

tests/bedrock_agentcore/services/test_resource_policy.py

@@ -17,7 +17,6 @@ def test_initialization(self):
             mock_boto.assert_called_once_with(
                 "bedrock-agentcore-control",
                 region_name=TEST_REGION,
-                endpoint_url=f"https://bedrock-agentcore-control.{TEST_REGION}.amazonaws.com",
             )
 
     def test_put_serializes_dict_to_json(self):