fix: honor AWS shared-config region for custom gateway hostnames

The previous check only consulted AWS_REGION* env vars before rejecting
custom hostnames, breaking deployments that configure their region via
the AWS shared config (profile). Resolve through boto3's session (env
vars + shared config) and only error when that chain yields nothing —
never sign with a silently guessed region.
This commit is contained in:
CrypticDriver 2026-07-22 07:27:10 +00:00
parent ebdad6e3ef
commit 2f342dc12d
2 changed files with 47 additions and 18 deletions

View file

@ -190,11 +190,11 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
# The signing region must match the gateway's region — derive it from
# standard gateway hostnames so callers don't have to set
# aws_region_name to a region different from their default. Custom or
# private hostnames can't be parsed: fall back to an explicitly
# configured region (param or AWS env vars), and error out rather than
# silently signing for a guessed region the gateway would reject with
# a confusing auth error.
# aws_region_name to a region different from their default. For custom
# or private hostnames, defer to BaseAWSLLM's normal region resolution
# (params, env vars, AWS shared config / profile); only error out when
# that chain yields nothing, rather than silently signing for a guessed
# region the gateway would reject with a confusing auth error.
signing_params = dict(optional_params)
if signing_params.get("aws_region_name") is None:
match = re.search(
@ -203,12 +203,21 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
)
if match:
signing_params["aws_region_name"] = match.group(1)
elif not any(get_secret_str(var) for var in ("AWS_REGION", "AWS_REGION_NAME", "AWS_DEFAULT_REGION")):
raise ValueError(
f"Cannot derive the SigV4 signing region from api_base '{api_base}'. "
"Set aws_region_name (or the AWS_REGION env var) to the gateway's "
"region when using a custom hostname."
)
else:
# boto3's session resolution covers env vars AND the AWS shared
# config (profile region) — unlike BaseAWSLLM's helper, which
# silently defaults to us-west-2 when nothing is configured.
import boto3
configured_region = boto3.Session().region_name
if configured_region:
signing_params["aws_region_name"] = configured_region
else:
raise ValueError(
f"Cannot derive the SigV4 signing region from api_base '{api_base}' "
"or the AWS configuration chain. Set aws_region_name (or AWS_REGION / "
"a profile region) to the gateway's region when using a custom hostname."
)
# api_key="" (not None, but falsy) disables BaseAWSLLM's fallback to the
# AWS_BEARER_TOKEN_BEDROCK env var: that token is a Bedrock Runtime

View file

@ -264,10 +264,12 @@ class TestAgentCoreSearch:
assert mock_base_sign.call_args.kwargs["api_key"] == ""
def test_sign_request_custom_hostname_requires_region(self):
"""Non-standard hostnames can't yield a signing region — require it explicitly."""
"""Custom hostname + empty AWS config chain → clear error, no guessed region."""
config = AgentCoreSearchConfig()
saved = {var: os.environ.pop(var, None) for var in ("AWS_REGION", "AWS_REGION_NAME", "AWS_DEFAULT_REGION")}
try:
mock_session = MagicMock()
mock_session.region_name = None # nothing configured anywhere
with patch("boto3.Session", return_value=mock_session):
with pytest.raises(ValueError, match="signing region"):
config.sign_request(
headers={},
@ -275,10 +277,28 @@ class TestAgentCoreSearch:
request_data={"jsonrpc": "2.0"},
api_base="https://gateway.internal.example.com/mcp",
)
finally:
for var, val in saved.items():
if val is not None:
os.environ[var] = val
def test_sign_request_custom_hostname_uses_shared_config_region(self):
"""Custom hostname + region from AWS shared config (profile) must be honored."""
config = AgentCoreSearchConfig()
mock_session = MagicMock()
mock_session.region_name = "eu-west-1" # e.g. from ~/.aws/config profile
with (
patch("boto3.Session", return_value=mock_session),
patch.object(
AgentCoreSearchConfig.__mro__[2], # BaseAWSLLM
"_sign_request",
return_value=({}, b"{}"),
) as mock_base_sign,
):
config.sign_request(
headers={},
optional_params={},
request_data={"jsonrpc": "2.0"},
api_base="https://gateway.internal.example.com/mcp",
)
assert mock_base_sign.call_args.kwargs["optional_params"]["aws_region_name"] == "eu-west-1"
def test_sign_request_passes_explicit_aws_credentials(self):
"""Explicit aws_* params (e.g. from a proxy search_tools entry) reach the signer."""