[v2] Fix bug with auth_scheme_preference when endpoint resolves from EP2.0 or operation-level trait (#10169)

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-24412.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`` shared config setting, or the existing ``AWS_AUTH_SCHEME_PREFERENCE`` environment variable."
+}

awscli/botocore/args.py

@@ -119,6 +119,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(
@@ -169,6 +170,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.
@@ -589,6 +591,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
@@ -645,6 +648,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(

awscli/botocore/client.py

@@ -19,7 +19,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
 
@@ -838,11 +842,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'),
         }

awscli/botocore/regions.py

@@ -24,7 +24,7 @@
 
 import jmespath
 from botocore import UNSIGNED, xform_name
-from botocore.auth import AUTH_TYPE_MAPS
+from botocore.auth import AUTH_TYPE_MAPS, resolve_auth_scheme_preference
 from botocore.endpoint_provider import EndpointProvider
 from botocore.exceptions import (
     EndpointProviderError,
@@ -471,6 +471,7 @@ def __init__(
         event_emitter,
         use_ssl=True,
         requested_auth_scheme=None,
+        auth_scheme_preference=None,
     ):
         self._provider = EndpointProvider(
             ruleset_data=endpoint_ruleset_data,
@@ -483,6 +484,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(
@@ -698,6 +700,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
@@ -719,6 +724,14 @@ 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/botocore/test_endpoint_provider.py

@@ -14,6 +14,7 @@
 import json
 import logging
 import os
+from unittest import mock
 
 import pytest
 
@@ -80,6 +81,28 @@
         ],
     },
 }
+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")
@@ -562,4 +585,79 @@ def test_construct_endpoint_parametrized(
             resolver, '_get_provider_params', return_value=provider_params
         ):
             result = resolver.construct_endpoint(None, None, None)
-            assert result.url == expected_url
+            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 (
+        mock.patch.dict(
+            'botocore.auth.AUTH_TYPE_MAPS',
+            {'bar': None, 'foo': None},
+            clear=True
+        ),
+        mock.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