[v2] Add support for configured manual auth scheme and bearer token auth (#9620)

Author: hssyoo

.changes/next-release/feature-bedrock-87718.json

@@ -0,0 +1,5 @@
+{
+  "type": "feature",
+  "category": "``bedrock``",
+  "description": "Add support for retrieving a Bearer token from environment variables to enable bearer authentication with Bedrock services."
+}

awscli/botocore/args.py

@@ -278,6 +278,7 @@ def compute_client_args(
                     client_config.response_checksum_validation
                 ),
                 account_id_endpoint_mode=client_config.account_id_endpoint_mode,
+                auth_scheme_preference=client_config.auth_scheme_preference,
             )
         self._compute_retry_config(config_kwargs)
         self._compute_request_compression_config(config_kwargs)
@@ -286,6 +287,10 @@ def compute_client_args(
         self._compute_checksum_config(config_kwargs)
         self._compute_inject_host_prefix(client_config, config_kwargs)
         self._compute_account_id_endpoint_mode_config(config_kwargs)
+        self._compute_auth_scheme_preference_config(
+            client_config, config_kwargs
+        )
+        self._compute_signature_version_config(client_config, config_kwargs)
         s3_config = self.compute_s3_config(client_config)
 
         is_s3_service = self._is_s3_service(service_name)
@@ -782,3 +787,55 @@ def _compute_account_id_endpoint_mode_config(self, config_kwargs):
             )
 
         config_kwargs[config_key] = account_id_endpoint_mode
+
+    def _compute_auth_scheme_preference_config(
+        self, client_config, config_kwargs
+    ):
+        config_key = 'auth_scheme_preference'
+        set_in_config_object = False
+
+        if client_config and client_config.auth_scheme_preference:
+            value = client_config.auth_scheme_preference
+            set_in_config_object = True
+        else:
+            value = self._config_store.get_config_variable(config_key)
+
+        if value is None:
+            config_kwargs[config_key] = None
+            return
+
+        if not isinstance(value, str):
+            raise botocore.exceptions.InvalidConfigError(
+                error_msg=(
+                    f"{config_key} must be a comma-delimited string. "
+                    f"Received {type(value)} instead: {value}."
+                )
+            )
+
+        value = ','.join(
+            item.replace(' ', '').replace('\t', '')
+            for item in value.split(',')
+            if item.strip()
+        )
+
+        if set_in_config_object:
+            value = ClientConfigString(value)
+
+        config_kwargs[config_key] = value
+
+    def _compute_signature_version_config(self, client_config, config_kwargs):
+        if client_config and client_config.signature_version:
+            value = client_config.signature_version
+            if isinstance(value, str):
+                config_kwargs['signature_version'] = ClientConfigString(value)
+
+
+class ConfigObjectWrapper:
+    """Base class to mark values set via in-code Config object."""
+
+    pass
+
+
+class ClientConfigString(str, ConfigObjectWrapper):
+    def __new__(cls, value=None):
+        return super().__new__(cls, value)

awscli/botocore/auth.py

@@ -895,6 +895,38 @@ def resolve_auth_type(auth_trait):
     raise UnsupportedSignatureVersionError(signature_version=auth_trait)
 
 
+def resolve_auth_scheme_preference(preference_list, auth_options):
+    service_supported = [scheme.split('#')[-1] for scheme in auth_options]
+
+    unsupported = [
+        scheme
+        for scheme in preference_list
+        if scheme not in AUTH_PREF_TO_SIGNATURE_VERSION
+    ]
+    if unsupported:
+        logger.debug(
+            f"Unsupported auth schemes in preference list: {', '.join(unsupported)}"
+        )
+
+    combined = preference_list + service_supported
+    prioritized_schemes = [
+        scheme
+        for scheme in dict.fromkeys(combined)
+        if scheme in service_supported
+    ]
+
+    for scheme in prioritized_schemes:
+        if scheme == 'noAuth':
+            return AUTH_PREF_TO_SIGNATURE_VERSION[scheme]
+        sig_version = AUTH_PREF_TO_SIGNATURE_VERSION.get(scheme)
+        if sig_version in AUTH_TYPE_MAPS:
+            return sig_version
+
+    raise UnsupportedSignatureVersionError(
+        signature_version=', '.join(sorted(service_supported))
+    )
+
+
 # Defined at the bottom instead of the top of the module because the Auth
 # classes weren't defined yet.
 AUTH_TYPE_MAPS = {
@@ -921,3 +953,13 @@ def resolve_auth_type(auth_trait):
     'smithy.api#httpBearerAuth': 'bearer',
     'smithy.api#noAuth': 'none',
 }
+
+
+# Mapping used specifically for resolving user-configured auth scheme preferences.
+# This is similar to AUTH_TYPE_TO_SIGNATURE_VERSION, but uses simplified keys by
+# stripping the auth trait prefixes ('smithy.api#httpBearerAuth' → 'httpBearerAuth').
+# These simplified keys match what customers are expected to provide in configuration.
+AUTH_PREF_TO_SIGNATURE_VERSION = {
+    auth_scheme.split('#')[-1]: sig_version
+    for auth_scheme, sig_version in AUTH_TYPE_TO_SIGNATURE_VERSION.items()
+}

awscli/botocore/client.py

@@ -79,6 +79,7 @@ def __init__(
         exceptions_factory=None,
         config_store=None,
         user_agent_creator=None,
+        auth_token_resolver=None,
     ):
         self._loader = loader
         self._endpoint_resolver = endpoint_resolver
@@ -92,6 +93,7 @@ def __init__(
         # future).
         self._config_store = config_store
         self._user_agent_creator = user_agent_creator
+        self._auth_token_resolver = auth_token_resolver
 
     def create_client(
         self,
@@ -143,6 +145,10 @@ def create_client(
             config_store=self._config_store,
             service_signature_version=service_signature_version,
         )
+        if token := self._evaluate_client_specific_token(
+            service_model.signing_name
+        ):
+            auth_token = token
         client_args = self._get_client_args(
             service_model,
             region_name,
@@ -446,6 +452,15 @@ def _api_call(self, *args, **kwargs):
         _api_call.__doc__ = docstring
         return _api_call
 
+    def _evaluate_client_specific_token(self, signing_name):
+        # Resolves an auth_token for the given signing_name.
+        # Returns None if no resolver is set or if resolution fails.
+        resolver = self._auth_token_resolver
+        if not resolver or not signing_name:
+            return None
+
+        return resolver(signing_name=signing_name)
+
 
 class ClientEndpointBridge:
     """Bridges endpoint data and client creation
@@ -829,6 +844,7 @@ def _make_api_call(self, operation_name, api_params):
             'has_streaming_input': operation_model.has_streaming_input,
             'auth_type': operation_model.resolved_auth_type,
             'unsigned_payload': operation_model.unsigned_payload,
+            'auth_options': self._service_model.metadata.get('auth'),
         }
 
         api_params = self._emit_api_params(

awscli/botocore/config.py

@@ -240,6 +240,12 @@ class Config:
 
         If a value is not provided, the client will default to ``preferred``.
 
+        Defaults to None.
+
+    :type auth_scheme_preference: str
+    :param auth_scheme_preference: A comma-delimited string of case-sensitive
+        auth scheme names used to determine the client's auth scheme preference.
+
         Defaults to None.
     """
 
@@ -270,6 +276,7 @@ class Config:
             ('request_checksum_calculation', None),
             ('response_checksum_validation', None),
             ('account_id_endpoint_mode', None),
+            ('auth_scheme_preference', None),
         ]
     )
 

awscli/botocore/configprovider.py

@@ -186,6 +186,12 @@
         None,
         utils.ensure_boolean,
     ),
+    'auth_scheme_preference': (
+        'auth_scheme_preference',
+        'AWS_AUTH_SCHEME_PREFERENCE',
+        None,
+        None,
+    ),
 }
 # A mapping for the s3 specific configuration vars. These are the configuration
 # vars that typically go in the s3 section of the config file. This mapping

awscli/botocore/handlers.py

@@ -27,6 +27,7 @@
 import botocore
 import botocore.auth
 from botocore import UNSIGNED, utils
+from botocore.args import ClientConfigString
 from botocore.compat import (
     MD5_AVAILABLE,  # noqa
     ETree,
@@ -58,10 +59,12 @@
     add_generate_presigned_post,
     add_generate_presigned_url,
 )
+from botocore.useragent import register_feature_id
 from botocore.utils import (
     SAFE_CHARS,
     SERVICE_NAME_ALIASES,
     ArnParser,
+    get_token_from_environment,
     hyphenize_service_id,  # noqa
     is_global_accesspoint,  # noqa
     percent_encode,
@@ -985,12 +988,14 @@ def remove_qbusiness_chat(class_attributes, **kwargs):
         del class_attributes['chat']
 
 
-def remove_bedrock_runtime_invoke_model_with_bidirectional_stream(class_attributes, **kwargs):
+def remove_bedrock_runtime_invoke_model_with_bidirectional_stream(
+    class_attributes, **kwargs
+):
     """Operation requires h2 which is currently unsupported in Python"""
     if 'invoke_model_with_bidirectional_stream' in class_attributes:
         del class_attributes['invoke_model_with_bidirectional_stream']
- 
- 
+
+
 def remove_bucket_from_url_paths_from_model(params, model, context, **kwargs):
     """Strips leading `{Bucket}/` from any operations that have it.
 
@@ -1201,6 +1206,100 @@ def _handle_request_validation_mode_member(params, model, **kwargs):
         params.setdefault(mode_member, "ENABLED")
 
 
+def _set_auth_scheme_preference_signer(context, signing_name, **kwargs):
+    """
+    Determines the appropriate signer to use based on the client configuration,
+    authentication scheme preferences, and the availability of a bearer token.
+    """
+    client_config = context.get('client_config')
+    if client_config is None:
+        return
+
+    signature_version = client_config.signature_version
+    auth_scheme_preference = client_config.auth_scheme_preference
+    auth_options = context.get('auth_options')
+
+    signature_version_set_in_code = (
+        isinstance(signature_version, ClientConfigString)
+        or signature_version is botocore.UNSIGNED
+    )
+    auth_preference_set_in_code = isinstance(
+        auth_scheme_preference, ClientConfigString
+    )
+    has_in_code_configuration = (
+        signature_version_set_in_code or auth_preference_set_in_code
+    )
+
+    resolved_signature_version = signature_version
+
+    # If signature version was not set in code, but an auth scheme preference
+    # is available, resolve it based on the preferred schemes and supported auth
+    # options for this service.
+    if (
+        not signature_version_set_in_code
+        and auth_scheme_preference
+        and auth_options
+    ):
+        preferred_schemes = auth_scheme_preference.split(',')
+        resolved = botocore.auth.resolve_auth_scheme_preference(
+            preferred_schemes, auth_options
+        )
+        resolved_signature_version = (
+            botocore.UNSIGNED if resolved == 'none' else resolved
+        )
+
+    # Prefer 'bearer' signature version if a bearer token is available, and it
+    # is allowed for this service. This can override earlier resolution if the
+    # config object didn't explicitly set a signature version.
+    if _should_prefer_bearer_auth(
+        has_in_code_configuration,
+        signing_name,
+        resolved_signature_version,
+        auth_options,
+    ):
+        register_feature_id('BEARER_SERVICE_ENV_VARS')
+        resolved_signature_version = 'bearer'
+
+    if resolved_signature_version == signature_version:
+        return None
+    return resolved_signature_version
+
+
+def _should_prefer_bearer_auth(
+    has_in_code_configuration,
+    signing_name,
+    resolved_signature_version,
+    auth_options,
+):
+    if signing_name not in get_bearer_auth_supported_services():
+        return False
+
+    if not auth_options or 'smithy.api#httpBearerAuth' not in auth_options:
+        return False
+
+    has_token = get_token_from_environment(signing_name) is not None
+
+    # Prefer 'bearer' if a bearer token is available, and either:
+    #   Bearer was already resolved, or
+    #   No auth-related values were explicitly set in code
+    return has_token and (
+        resolved_signature_version == 'bearer' or not has_in_code_configuration
+    )
+
+
+def get_bearer_auth_supported_services():
+    """
+    Returns a set of services that support bearer token authentication.
+    These values correspond to the service's `signingName` property as defined
+    in model.py, falling back to `endpointPrefix` if `signingName` is not set.
+
+    Warning: This is a private interface and is subject to abrupt breaking changes,
+    including removal, in any botocore release. It is not intended for external use,
+    and its usage outside of botocore is not advised or supported.
+    """
+    return {'bedrock'}
+
+
 def _set_extra_headers_for_unsigned_request(
     request, signature_version, **kwargs
 ):
@@ -1242,7 +1341,10 @@ def _set_extra_headers_for_unsigned_request(
     ),
     ('creating-client-class.lex-runtime-v2', remove_lex_v2_start_conversation),
     ('creating-client-class.qbusiness', remove_qbusiness_chat),
-    ('creating-client-class.bedrock-runtime', remove_bedrock_runtime_invoke_model_with_bidirectional_stream),
+    (
+        'creating-client-class.bedrock-runtime',
+        remove_bedrock_runtime_invoke_model_with_bidirectional_stream,
+    ),
     ('after-call.iam', json_decode_policies),
     ('after-call.ec2.GetConsoleOutput', decode_console_output),
     ('after-call.cloudformation.GetTemplate', json_decode_template_body),
@@ -1299,6 +1401,7 @@ def _set_extra_headers_for_unsigned_request(
     ('choose-signer.sts.AssumeRoleWithSAML', disable_signing),
     ('choose-signer.sts.AssumeRoleWithWebIdentity', disable_signing),
     ('choose-signer', set_operation_specific_signer),
+    ('choose-signer', _set_auth_scheme_preference_signer),
     ('before-parameter-build.s3.HeadObject', sse_md5),
     ('before-parameter-build.s3.GetObject', sse_md5),
     ('before-parameter-build.s3.PutObject', sse_md5),