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,11 @@
     urlsplit,
     urlunsplit,
 )
-from botocore.exceptions import NoAuthTokenError, NoCredentialsError
+from botocore.exceptions import (
+    NoAuthTokenError,
+    NoCredentialsError,
+    UnknownSignatureVersionError,
+)
 from botocore.utils import (
     is_valid_ipv6_endpoint_url,
     normalize_url_path,
@@ -1160,3 +1164,22 @@ 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',
+}
+
+
+def resolve_auth_type(auth_trait):
+    for auth_type in auth_trait:
+        if auth_type == 'smithy.api#noAuth':
+            return 'none'
+        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)

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):
@@ -955,6 +960,10 @@ def _make_api_call(self, operation_name, api_params):
             'has_streaming_input': operation_model.has_streaming_input,
             'auth_type': operation_model.auth_type,
         }
+
+        if operation_model.unsigned_payload:
+            request_context['payload_signing_enabled'] = False
+
         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: Sets override for the signing region set
+        used for sigv4a signing
+
+        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
@@ -256,6 +262,7 @@ class Config:
             ('tcp_keepalive', None),
             ('request_min_compression_size_bytes', None),
             ('disable_request_compression', None),
+            ('sigv4a_signing_region_set', None),
             ('client_context_params', 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/handlers.py

@@ -210,7 +210,8 @@ def set_operation_specific_signer(context, signing_name, **kwargs):
         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:
@@ -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,20 @@ 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):
+        if self.auth:
+            return resolve_auth_type(self.auth)
         return self._operation_model.get('authtype')
 
+    @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,76 @@
+# 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, some services may have a list of credentials requirements where one signature may fail and others may
+# succeed, for instance a service may want to use bearer by default but fall back to sigv4 if a token isn't found.
+# There currently is not a way to do this in botocore, so we added this test to make sure that we handle this gracefully
+# when the need arises.
+AUTH_TYPE_REQUIREMENTS = {
+    'aws.auth#sigv4': ['credentials'],
+    'aws.auth#sigv4a': ['credentials'],
+    'smithy.api#httpBearerAuth': ['bearer_token'],
+    'smithy.api#noAuth': [],
+}
+
+
+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):
+    if len(auth_config) > 1:
+        first_auth = auth_config.pop()
+        first_auth_reqs = AUTH_TYPE_REQUIREMENTS[first_auth]
+        assert all(
+            first_auth_reqs == AUTH_TYPE_REQUIREMENTS[req]
+            for req in auth_config
+        ), message

tests/unit/auth/test_auth_trait.py

@@ -0,0 +1,46 @@
+# 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 (
+    AUTH_TYPE_MAPS,
+    AUTH_TYPE_TO_SIGNATURE_VERSION,
+    BaseSigner,
+    resolve_auth_type,
+)
+from botocore.exceptions import UnknownSignatureVersionError
+from tests import mock, unittest
+
+
+class TestAuthTraitResolution(unittest.TestCase):
+    def test_auth_resolves_first_available(self):
+        auth = ['aws.auth#foo', 'aws.auth#bar']
+        bar_signer = mock.Mock(spec=BaseSigner)
+
+        auth_types = AUTH_TYPE_MAPS.copy()
+        auth_types['bar'] = bar_signer
+
+        auth_type_conversions = AUTH_TYPE_TO_SIGNATURE_VERSION.copy()
+        auth_type_conversions['aws.auth#foo'] = "foo"
+        auth_type_conversions['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):
+        auth = ['aws.auth#invalidAuth']
+        with self.assertRaises(UnknownSignatureVersionError):
+            resolve_auth_type(auth)

tests/unit/test_client.py

@@ -1713,6 +1713,17 @@ def test_client_internal_credential_shim(self):
         )
         self.assertEqual(service_client._get_credentials(), self.credentials)
 
+    def test_unsigned_payload_disables_signing(self):
+        creator = self.create_client_creator()
+        service_client = creator.create_client('myservice', 'us-west-2')
+        test_operation = self.service_description['operations'][
+            'TestOperation'
+        ]
+        test_operation['unsignedPayload'] = True
+        service_client.test_operation(Foo='one')
+        context = self.endpoint.make_request.call_args[0][1]['context']
+        self.assertFalse(context['payload_signing_enabled'])
+
 
 class TestClientErrors(TestAutoGeneratedClient):
     def add_error_response(self, error_response):

tests/unit/test_handlers.py

@@ -1042,7 +1042,7 @@ def test_set_operation_specific_signer_v4a_existing_signing_context(self):
             context=context, signing_name=signing_name
         )
         # region has been updated
-        self.assertEqual(context['signing']['region'], '*')
+        self.assertEqual(context['signing']['region'], 'abc')
         # signing_name has been added
         self.assertEqual(context['signing']['signing_name'], signing_name)
         # foo remained untouched
@@ -1073,6 +1073,29 @@ def test_set_operation_specific_signer_s3v4_unsigned_payload(self):
         self.assertEqual(response, 's3v4')
         self.assertEqual(context.get('payload_signing_enabled'), False)
 
+    def test_set_operation_specific_signer_defaults_to_asterisk(self):
+        signing_name = 'myservice'
+        context = {
+            'auth_type': 'v4a',
+        }
+        handlers.set_operation_specific_signer(
+            context=context, signing_name=signing_name
+        )
+        self.assertEqual(context['signing']['region'], '*')
+
+    def test_set_operation_specific_signer_prefers_client_config(self):
+        signing_name = 'myservice'
+        context = {
+            'auth_type': 'v4a',
+            'client_config': Config(
+                sigv4a_signing_region_set="region_1,region_2"
+            ),
+        }
+        handlers.set_operation_specific_signer(
+            context=context, signing_name=signing_name
+        )
+        self.assertEqual(context['signing']['region'], 'region_1,region_2')
+
 
 @pytest.mark.parametrize(
     'auth_type, expected_response', [('v4', 's3v4'), ('v4a', 's3v4a')]