Adds support for the new 'auth' trait to allow a priority list of aut…

…h types.

awscli/botocore/client.py

@@ -299,6 +299,31 @@ def _register_s3_control_events(self, client):
             return
         S3ControlArnParamHandlerv2().register(client.meta.events)
 
+    def _default_s3_presign_to_sigv2(self, signature_version, **kwargs):
+        """
+        Returns the 's3' (sigv2) signer if presigning an s3 request. This is
+        intended to be used to set the default signature version for the signer
+        to sigv2. Situations where an asymmetric signature is required are the
+        exception, for example MRAP needs v4a.
+
+        :type signature_version: str
+        :param signature_version: The current client signature version.
+
+        :type signing_name: str
+        :param signing_name: The signing name of the service.
+
+        :return: 's3' if the request is an s3 presign request, None otherwise
+        """
+        if signature_version.startswith('v4a'):
+            return
+
+        if signature_version.startswith('v4-s3express'):
+            return signature_version
+
+        for suffix in ['-query', '-presign-post']:
+            if signature_version.endswith(suffix):
+                return f's3{suffix}'
+
     def _get_client_args(
             self,
             service_model,
@@ -682,7 +707,8 @@ 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(

awscli/botocore/config.py

@@ -189,6 +189,11 @@ class Config(object):
     set to True.
 
         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.
     """
     OPTION_DEFAULTS = OrderedDict([
         ('region_name', None),
@@ -212,6 +217,7 @@ class Config(object):
         ('ignore_configured_endpoint_urls', None),
         ('request_min_compression_size_bytes', None),
         ('disable_request_compression', None),
+        ('sigv4a_signing_region_set', None),
     ])
 
     def __init__(self, *args, **kwargs):

awscli/botocore/configprovider.py

@@ -131,6 +131,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

awscli/botocore/exceptions.py

@@ -457,7 +457,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):

awscli/botocore/handlers.py

@@ -183,14 +183,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:
@@ -212,6 +218,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:

awscli/botocore/model.py

@@ -14,6 +14,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,
@@ -587,10 +588,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", [])

awscli/botocore/regions.py

@@ -664,7 +664,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/botocore/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/botocore/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([])

tests/unit/botocore/test_handlers.py

@@ -979,13 +979,11 @@ def test_set_operation_specific_signer_v4a_existing_signing_context(self):
         signing_name = 'myservice'
         context = {
             'auth_type': 'v4a',
-            'signing': {'foo': 'bar', 'region': 'abc'},
+            'signing': {'foo': 'bar'},
         }
         handlers.set_operation_specific_signer(
             context=context, signing_name=signing_name
         )
-        # region has been updated
-        self.assertEqual(context['signing']['region'], '*')
         # signing_name has been added
         self.assertEqual(context['signing']['signing_name'], signing_name)
         # foo remained untouched
@@ -1014,6 +1012,61 @@ 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"
+            ),
+            'signing': {
+                'region': 'abc',
+            },
+        }
+        handlers.set_operation_specific_signer(
+            context=context, signing_name=signing_name
+        )
+        self.assertEqual(context['signing']['region'], 'region_1,region_2')
+
+    def test_payload_signing_disabled_sets_proper_key(self):
+        signing_name = 'myservice'
+        context = {
+            'auth_type': 'v4',
+            'signing': {
+                'foo': 'bar',
+                'region': 'abc',
+            },
+            'unsigned_payload': True,
+        }
+        handlers.set_operation_specific_signer(
+            context=context, signing_name=signing_name
+        )
+        self.assertEqual(context.get('payload_signing_enabled'), False)
+
+    def test_no_payload_signing_disabled_does_not_set_key(self):
+        signing_name = 'myservice'
+        context = {
+            'auth_type': 'v4',
+            'signing': {
+                'foo': 'bar',
+                'region': 'abc',
+            },
+        }
+        handlers.set_operation_specific_signer(
+            context=context, signing_name=signing_name
+        )
+        self.assertNotIn('payload_signing_enabled', context)
+
 
 @pytest.mark.parametrize(
     'auth_type, expected_response', [('v4', 's3v4'), ('v4a', 's3v4a')]

tests/unit/botocore/test_model.py

@@ -185,7 +185,8 @@ def setUp(self):
                     },
                     'errors': [{'shape': 'NoSuchResourceException'}],
                     'documentation': 'Docs for OperationName',
-                    'authtype': 'v4'
+                    'authtype': 'v4',
+                    'auth': ['aws.auth#sigv4'],
                 },
                 'OperationTwo': {
                     'http': {
@@ -407,6 +408,22 @@ def test_error_shapes(self):
         self.assertEqual(
             operation.error_shapes[0].name, 'NoSuchResourceException')
 
+    def test_has_auth(self):
+        operation = self.service_model.operation_model('OperationName')
+        self.assertEqual(operation.auth, ["aws.auth#sigv4"])
+
+    def test_auth_not_set(self):
+        operation = self.service_model.operation_model('OperationTwo')
+        self.assertIsNone(operation.auth)
+
+    def test_has_resolved_auth_type(self):
+        operation = self.service_model.operation_model('OperationName')
+        self.assertEqual(operation.resolved_auth_type, 'v4')
+
+    def test_resolved_auth_type_not_set(self):
+        operation = self.service_model.operation_model('OperationTwo')
+        self.assertIsNone(operation.resolved_auth_type)
+
     def test_has_auth_type(self):
         operation = self.service_model.operation_model('OperationName')
         self.assertEqual(operation.auth_type, 'v4')