Multi-auth trait support

.changes/next-release/enhancement-signing-83434.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "signing",
+  "description": "Adds support for the new 'auth' trait on our models that will allow a priority list of auth types for a service or operation."
+}

botocore/args.py

@@ -268,6 +268,9 @@ def compute_client_args(
                     client_config.disable_request_compression
                 ),
                 client_context_params=client_config.client_context_params,
+                sigv4a_signing_region_set=(
+                    client_config.sigv4a_signing_region_set
+                ),
             )
         self._compute_retry_config(config_kwargs)
         self._compute_connect_timeout(config_kwargs)

botocore/auth.py

@@ -35,7 +35,12 @@
     urlsplit,
     urlunsplit,
 )
-from botocore.exceptions import NoAuthTokenError, NoCredentialsError
+from botocore.exceptions import (
+    NoAuthTokenError,
+    NoCredentialsError,
+    UnknownSignatureVersionError,
+    UnsupportedSignatureVersionError,
+)
 from botocore.utils import (
     is_valid_ipv6_endpoint_url,
     normalize_url_path,
@@ -1132,6 +1137,19 @@ def add_auth(self, request):
         request.headers['Authorization'] = auth_header
 
 
+def resolve_auth_type(auth_trait):
+    for auth_type in auth_trait:
+        if auth_type == 'smithy.api#noAuth':
+            return AUTH_TYPE_TO_SIGNATURE_VERSION[auth_type]
+        elif auth_type in AUTH_TYPE_TO_SIGNATURE_VERSION:
+            signature_version = AUTH_TYPE_TO_SIGNATURE_VERSION[auth_type]
+            if signature_version in AUTH_TYPE_MAPS:
+                return signature_version
+        else:
+            raise UnknownSignatureVersionError(signature_version=auth_type)
+    raise UnsupportedSignatureVersionError(signature_version=auth_trait)
+
+
 AUTH_TYPE_MAPS = {
     'v2': SigV2Auth,
     'v3': SigV3Auth,
@@ -1160,3 +1178,10 @@ def add_auth(self, request):
             's3v4-query': S3SigV4QueryAuth,
         }
     )
+
+AUTH_TYPE_TO_SIGNATURE_VERSION = {
+    'aws.auth#sigv4': 'v4',
+    'aws.auth#sigv4a': 'v4a',
+    'smithy.api#httpBearerAuth': 'bearer',
+    'smithy.api#noAuth': 'none',
+}

botocore/client.py

@@ -14,7 +14,7 @@
 
 from botocore import waiter, xform_name
 from botocore.args import ClientArgsCreator
-from botocore.auth import AUTH_TYPE_MAPS
+from botocore.auth import AUTH_TYPE_MAPS, resolve_auth_type
 from botocore.awsrequest import prepare_request_dict
 from botocore.compress import maybe_compress_request
 from botocore.config import Config
@@ -148,15 +148,20 @@ def create_client(
         region_name, client_config = self._normalize_fips_region(
             region_name, client_config
         )
+        auth = service_model.metadata.get('auth')
+        if auth:
+            service_signature_version = resolve_auth_type(auth)
+        else:
+            service_signature_version = service_model.metadata.get(
+                'signatureVersion'
+            )
         endpoint_bridge = ClientEndpointBridge(
             self._endpoint_resolver,
             scoped_config,
             client_config,
             service_signing_name=service_model.metadata.get('signingName'),
             config_store=self._config_store,
-            service_signature_version=service_model.metadata.get(
-                'signatureVersion'
-            ),
+            service_signature_version=service_signature_version,
         )
         client_args = self._get_client_args(
             service_model,
@@ -487,7 +492,7 @@ def _default_s3_presign_to_sigv2(self, signature_version, **kwargs):
             return
 
         if signature_version.startswith('v4-s3express'):
-            return f'{signature_version}'
+            return signature_version
 
         for suffix in ['-query', '-presign-post']:
             if signature_version.endswith(suffix):
@@ -953,8 +958,10 @@ def _make_api_call(self, operation_name, api_params):
             'client_region': self.meta.region_name,
             'client_config': self.meta.config,
             'has_streaming_input': operation_model.has_streaming_input,
-            'auth_type': operation_model.auth_type,
+            'auth_type': operation_model.resolved_auth_type,
+            'unsigned_payload': operation_model.unsigned_payload,
         }
+
         api_params = self._emit_api_params(
             api_params=api_params,
             operation_model=operation_model,

botocore/config.py

@@ -221,6 +221,12 @@ class Config:
 
         Defaults to None.
 
+    :type sigv4a_signing_region_set: string
+    :param sigv4a_signing_region_set: A set of AWS regions to apply the signature for
+        when using SigV4a for signing. Set to ``*`` to represent all regions.
+
+        Defaults to None.
+
     :type client_context_params: dict
     :param client_context_params: A dictionary of parameters specific to
         individual services. If available, valid parameters can be found in
@@ -257,6 +263,7 @@ class Config:
             ('request_min_compression_size_bytes', None),
             ('disable_request_compression', None),
             ('client_context_params', None),
+            ('sigv4a_signing_region_set', None),
         ]
     )
 

botocore/configprovider.py

@@ -168,6 +168,12 @@
         False,
         utils.ensure_boolean,
     ),
+    'sigv4a_signing_region_set': (
+        'sigv4a_signing_region_set',
+        'AWS_SIGV4A_SIGNING_REGION_SET',
+        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

botocore/exceptions.py

@@ -514,7 +514,7 @@ class UnknownClientMethodError(BotoCoreError):
 class UnsupportedSignatureVersionError(BotoCoreError):
     """Error when trying to use an unsupported Signature Version."""
 
-    fmt = 'Signature version is not supported: {signature_version}'
+    fmt = 'Signature version(s) are not supported: {signature_version}'
 
 
 class ClientError(Exception):

botocore/handlers.py

@@ -203,14 +203,20 @@ def set_operation_specific_signer(context, signing_name, **kwargs):
     if auth_type == 'bearer':
         return 'bearer'
 
+    # If the operation needs an unsigned body, we set additional context
+    # allowing the signer to be aware of this.
+    if context.get('unsigned_payload') or auth_type == 'v4-unsigned-body':
+        context['payload_signing_enabled'] = False
+
     if auth_type.startswith('v4'):
         if auth_type == 'v4-s3express':
             return auth_type
 
         if auth_type == 'v4a':
             # If sigv4a is chosen, we must add additional signing config for
             # global signature.
-            signing = {'region': '*', 'signing_name': signing_name}
+            region = _resolve_sigv4a_region(context)
+            signing = {'region': region, 'signing_name': signing_name}
             if 'signing' in context:
                 context['signing'].update(signing)
             else:
@@ -219,11 +225,6 @@ def set_operation_specific_signer(context, signing_name, **kwargs):
         else:
             signature_version = 'v4'
 
-        # If the operation needs an unsigned body, we set additional context
-        # allowing the signer to be aware of this.
-        if auth_type == 'v4-unsigned-body':
-            context['payload_signing_enabled'] = False
-
         # Signing names used by s3 and s3-control use customized signers "s3v4"
         # and "s3v4a".
         if signing_name in S3_SIGNING_NAMES:
@@ -232,6 +233,15 @@ def set_operation_specific_signer(context, signing_name, **kwargs):
         return signature_version
 
 
+def _resolve_sigv4a_region(context):
+    region = None
+    if 'client_config' in context:
+        region = context['client_config'].sigv4a_signing_region_set
+    if not region and context.get('signing', {}).get('region'):
+        region = context['signing']['region']
+    return region or '*'
+
+
 def decode_console_output(parsed, **kwargs):
     if 'Output' in parsed:
         try:

botocore/model.py

@@ -15,6 +15,7 @@
 from collections import defaultdict
 from typing import NamedTuple, Union
 
+from botocore.auth import resolve_auth_type
 from botocore.compat import OrderedDict
 from botocore.exceptions import (
     MissingServiceIdError,
@@ -623,10 +624,24 @@ def context_parameters(self):
     def request_compression(self):
         return self._operation_model.get('requestcompression')
 
+    @CachedProperty
+    def auth(self):
+        return self._operation_model.get('auth')
+
     @CachedProperty
     def auth_type(self):
         return self._operation_model.get('authtype')
 
+    @CachedProperty
+    def resolved_auth_type(self):
+        if self.auth:
+            return resolve_auth_type(self.auth)
+        return self.auth_type
+
+    @CachedProperty
+    def unsigned_payload(self):
+        return self._operation_model.get('unsignedPayload')
+
     @CachedProperty
     def error_shapes(self):
         shapes = self._operation_model.get("errors", [])

botocore/regions.py

@@ -722,7 +722,9 @@ def auth_schemes_to_signing_ctx(self, auth_schemes):
             signing_context['region'] = scheme['signingRegion']
         elif 'signingRegionSet' in scheme:
             if len(scheme['signingRegionSet']) > 0:
-                signing_context['region'] = scheme['signingRegionSet'][0]
+                signing_context['region'] = ','.join(
+                    scheme['signingRegionSet']
+                )
         if 'signingName' in scheme:
             signing_context.update(signing_name=scheme['signingName'])
         if 'disableDoubleEncoding' in scheme:

tests/functional/test_auth_config.py

@@ -0,0 +1,77 @@
+# Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+import pytest
+
+from botocore.session import get_session
+
+# In the future, a service may have a list of credentials requirements where one
+# signature may fail and others may succeed. e.g. a service may want to use bearer
+# auth but fall back to sigv4 if a token isn't available. There's currently no way to do
+# this in botocore, so this test ensures we handle this gracefully when the need arises.
+
+
+# The dictionary's value here needs to be hashable to be added to the set below; any
+# new auth types with multiple requirements should be added in a comma-separated list
+AUTH_TYPE_REQUIREMENTS = {
+    'aws.auth#sigv4': 'credentials',
+    'aws.auth#sigv4a': 'credentials',
+    'smithy.api#httpBearerAuth': 'bearer_token',
+    'smithy.api#noAuth': 'none',
+}
+
+
+def _all_test_cases():
+    session = get_session()
+    loader = session.get_component('data_loader')
+
+    services = loader.list_available_services('service-2')
+    auth_services = []
+    auth_operations = []
+
+    for service in services:
+        service_model = session.get_service_model(service)
+        auth_config = service_model.metadata.get('auth', {})
+        if auth_config:
+            auth_services.append([service, auth_config])
+        for operation in service_model.operation_names:
+            operation_model = service_model.operation_model(operation)
+            if operation_model.auth:
+                auth_operations.append([service, operation_model])
+    return auth_services, auth_operations
+
+
+AUTH_SERVICES, AUTH_OPERATIONS = _all_test_cases()
+
+
+@pytest.mark.validates_models
+@pytest.mark.parametrize("auth_service, auth_config", AUTH_SERVICES)
+def test_all_requirements_match_for_service(auth_service, auth_config):
+    # Validates that all service-level signature types have the same requirements
+    message = f'Found mixed signer requirements for service: {auth_service}'
+    assert_all_requirements_match(auth_config, message)
+
+
+@pytest.mark.validates_models
+@pytest.mark.parametrize("auth_service, operation_model", AUTH_OPERATIONS)
+def test_all_requirements_match_for_operation(auth_service, operation_model):
+    # Validates that all operation-level signature types have the same requirements
+    message = f'Found mixed signer requirements for operation: {auth_service}.{operation_model.name}'
+    auth_config = operation_model.auth
+    assert_all_requirements_match(auth_config, message)
+
+
+def assert_all_requirements_match(auth_config, message):
+    auth_requirements = set(
+        AUTH_TYPE_REQUIREMENTS[auth_type] for auth_type in auth_config
+    )
+    assert len(auth_requirements) == 1

tests/unit/auth/test_auth_trait.py

@@ -0,0 +1,42 @@
+# Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+from botocore.auth import BaseSigner, resolve_auth_type
+from botocore.exceptions import (
+    UnknownSignatureVersionError,
+    UnsupportedSignatureVersionError,
+)
+from tests import mock, unittest
+
+
+class TestAuthTraitResolution(unittest.TestCase):
+    def test_auth_resolves_first_available(self):
+        auth = ['aws.auth#foo', 'aws.auth#bar']
+        # Don't declare a signer for "foo"
+        auth_types = {'bar': mock.Mock(spec=BaseSigner)}
+        auth_type_conversions = {'aws.auth#foo': 'foo', 'aws.auth#bar': 'bar'}
+
+        with mock.patch('botocore.auth.AUTH_TYPE_MAPS', auth_types):
+            with mock.patch(
+                'botocore.auth.AUTH_TYPE_TO_SIGNATURE_VERSION',
+                auth_type_conversions,
+            ):
+                assert resolve_auth_type(auth) == 'bar'
+
+    def test_invalid_auth_type_error(self):
+        with self.assertRaises(UnknownSignatureVersionError):
+            resolve_auth_type(['aws.auth#invalidAuth'])
+
+    def test_no_known_auth_type(self):
+        with self.assertRaises(UnsupportedSignatureVersionError):
+            resolve_auth_type([])