Fix bug with auth_scheme_preference when endpoint resolves from EP2.0 or operation-level trait (boto#3670)

The auth_scheme_preference config setting is now used to reprioritize auth schemes, if configured, if the auth scheme is resolved from Endpoints model or at the operation-level of service-2 models.

Author: aemous

.changes/next-release/bugfix-signing-59481.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "signing",
+  "description": "Fix bug so that configured auth scheme preference is used when auth scheme is resolved from endpoints rulesets, or from operation-level auth trait. Auth scheme preference can be configured using the existing ``auth_scheme_preference`` client config option, the ``auth_scheme_preference`` shared config setting, or the existing ``AWS_AUTH_SCHEME_PREFERENCE`` environment variable."
+}

botocore/args.py

@@ -143,6 +143,7 @@ def get_client_args(
         s3_disable_express_session_auth = config_kwargs[
             's3_disable_express_session_auth'
         ]
+        auth_scheme_preference = config_kwargs['auth_scheme_preference']
 
         event_emitter = copy.copy(self._event_emitter)
         signer = RequestSigner(
@@ -207,6 +208,7 @@ def get_client_args(
             credentials,
             account_id_endpoint_mode,
             s3_disable_express_session_auth,
+            auth_scheme_preference,
         )
 
         # Copy the session's user agent factory and adds client configuration.
@@ -736,6 +738,7 @@ def _build_endpoint_resolver(
         credentials,
         account_id_endpoint_mode,
         s3_disable_express_session_auth,
+        auth_scheme_preference,
     ):
         if endpoints_ruleset_data is None:
             return None
@@ -792,6 +795,7 @@ def _build_endpoint_resolver(
             event_emitter=event_emitter,
             use_ssl=is_secure,
             requested_auth_scheme=sig_version,
+            auth_scheme_preference=auth_scheme_preference,
         )
 
     def compute_endpoint_resolver_builtin_defaults(

botocore/client.py

@@ -18,7 +18,11 @@
     xform_name,
 )
 from botocore.args import ClientArgsCreator
-from botocore.auth import AUTH_TYPE_MAPS, resolve_auth_type
+from botocore.auth import (
+    AUTH_TYPE_MAPS,
+    resolve_auth_scheme_preference,
+    resolve_auth_type,
+)
 from botocore.awsrequest import prepare_request_dict
 from botocore.compress import maybe_compress_request
 from botocore.config import Config
@@ -1007,11 +1011,23 @@ def _make_api_call(self, operation_name, api_params):
             logger.debug(
                 'Warning: %s.%s() is deprecated', service_name, operation_name
             )
+        # If the operation has the `auth` property and the client has a
+        # configured auth scheme preference, use both to compute the
+        # auth type. Otherwise, fallback to auth/auth_type resolution.
+        if operation_model.auth and self.meta.config.auth_scheme_preference:
+            preferred_schemes = self.meta.config.auth_scheme_preference.split(
+                ','
+            )
+            auth_type = resolve_auth_scheme_preference(
+                preferred_schemes, operation_model.auth
+            )
+        else:
+            auth_type = operation_model.resolved_auth_type
         request_context = {
             'client_region': self.meta.region_name,
             'client_config': self.meta.config,
             'has_streaming_input': operation_model.has_streaming_input,
-            'auth_type': operation_model.resolved_auth_type,
+            'auth_type': auth_type,
             'unsigned_payload': operation_model.unsigned_payload,
             'auth_options': self._service_model.metadata.get('auth'),
         }

botocore/regions.py

@@ -25,7 +25,11 @@
 import jmespath
 
 from botocore import UNSIGNED, xform_name
-from botocore.auth import AUTH_TYPE_MAPS, HAS_CRT
+from botocore.auth import (
+    AUTH_TYPE_MAPS,
+    HAS_CRT,
+    resolve_auth_scheme_preference,
+)
 from botocore.crt import CRT_SUPPORTED_AUTH_TYPES
 from botocore.endpoint_provider import EndpointProvider
 from botocore.exceptions import (
@@ -478,6 +482,7 @@ def __init__(
         event_emitter,
         use_ssl=True,
         requested_auth_scheme=None,
+        auth_scheme_preference=None,
     ):
         self._provider = EndpointProvider(
             ruleset_data=endpoint_ruleset_data,
@@ -490,6 +495,7 @@ def __init__(
         self._event_emitter = event_emitter
         self._use_ssl = use_ssl
         self._requested_auth_scheme = requested_auth_scheme
+        self._auth_scheme_preference = auth_scheme_preference
         self._instance_cache = {}
 
     def construct_endpoint(
@@ -703,6 +709,9 @@ def auth_schemes_to_signing_ctx(self, auth_schemes):
         if self._requested_auth_scheme == UNSIGNED:
             return 'none', {}
 
+        available_ruleset_names = [
+            s['name'].split('#')[-1] for s in auth_schemes
+        ]
         auth_schemes = [
             {**scheme, 'name': self._strip_sig_prefix(scheme['name'])}
             for scheme in auth_schemes
@@ -724,6 +733,16 @@ def auth_schemes_to_signing_ctx(self, auth_schemes):
                 # exception, instead default to the logic in botocore
                 # customizations.
                 return None, {}
+        elif self._auth_scheme_preference is not None:
+            prefs = self._auth_scheme_preference.split(',')
+            auth_schemes_by_auth_type = {
+                self._strip_sig_prefix(s['name'].split('#')[-1]): s
+                for s in auth_schemes
+            }
+            name = resolve_auth_scheme_preference(
+                prefs, available_ruleset_names
+            )
+            scheme = auth_schemes_by_auth_type[name]
         else:
             try:
                 name, scheme = next(

tests/unit/test_endpoint_provider.py

@@ -82,6 +82,26 @@
         ],
     },
 }
+ENDPOINT_AUTH_SCHEMES_DICT = {
+    "url": URL_TEMPLATE,
+    "properties": {
+        "authSchemes": [
+            {
+                "disableDoubleEncoding": True,
+                "name": "foo",
+                "signingName": "s3-outposts",
+                "signingRegionSet": ["*"],
+            },
+            {
+                "disableDoubleEncoding": True,
+                "name": "bar",
+                "signingName": "s3-outposts",
+                "signingRegion": REGION_TEMPLATE,
+            },
+        ],
+    },
+    "headers": {},
+}
 
 
 @pytest.fixture(scope="module")
@@ -619,3 +639,78 @@ def test_construct_endpoint_parametrized(
         ):
             result = resolver.construct_endpoint(None, None, None)
             assert result.url == expected_url
+
+
+@pytest.mark.parametrize(
+    "auth_scheme_preference,expected_auth_scheme_name",
+    [
+        (
+            'foo,bar',
+            'foo',
+        ),
+        (
+            'bar,foo',
+            'bar',
+        ),
+        (
+            'xyz,foo,bar',
+            'foo',
+        ),
+    ],
+)
+def test_auth_scheme_preference(
+    auth_scheme_preference, expected_auth_scheme_name, monkeypatch
+):
+    conditions = (
+        [
+            PARSE_ARN_FUNC,
+            {
+                "fn": "not",
+                "argv": [STRING_EQUALS_FUNC],
+            },
+            {
+                "fn": "aws.partition",
+                "argv": [REGION_REF],
+                "assign": "PartitionResults",
+            },
+        ],
+    )
+    resolver = EndpointRulesetResolver(
+        endpoint_ruleset_data={
+            'version': '1.0',
+            'parameters': {},
+            'rules': [
+                {
+                    'conditions': conditions,
+                    'type': 'endpoint',
+                    'endpoint': ENDPOINT_AUTH_SCHEMES_DICT,
+                }
+            ],
+        },
+        partition_data={},
+        service_model=None,
+        builtins={},
+        client_context=None,
+        event_emitter=None,
+        use_ssl=True,
+        requested_auth_scheme=None,
+        auth_scheme_preference=auth_scheme_preference,
+    )
+    auth_schemes = [
+        {'name': 'foo', 'signingName': 's3', 'signingRegion': 'ap-south-1'},
+        {'name': 'bar', 'signingName': 's3', 'signingRegion': 'ap-south-2'},
+    ]
+    with (
+        patch.dict(
+            'botocore.auth.AUTH_TYPE_MAPS',
+            {'bar': None, 'foo': None},
+            clear=True,
+        ),
+        patch.dict(
+            'botocore.auth.AUTH_PREF_TO_SIGNATURE_VERSION',
+            {'bar': 'bar', 'foo': 'foo'},
+            clear=True,
+        ),
+    ):
+        name, scheme = resolver.auth_schemes_to_signing_ctx(auth_schemes)
+    assert name == expected_auth_scheme_name