Implement a configurable credentials resolver chain

codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsAuthIntegration.java

@@ -57,6 +57,10 @@ public List<RuntimeClientPlugin> getClientPlugins(GenerationContext context) {
                                         .build())
                                 // TODO: Initialize with the provider chain?
                                 .nullable(true)
+                                .initialize(writer -> {
+                                    writer.addImport("smithy_aws_core.credentials_resolvers", "CredentialsResolverChain");
+                                    writer.write("self.aws_credentials_identity_resolver = aws_credentials_identity_resolver or CredentialsResolverChain()");
+                                })
                                 .build())
                         .addConfigProperty(REGION)
                         .authScheme(new Sigv4AuthScheme())

packages/smithy-aws-core/src/smithy_aws_core/credentials_resolvers/__init__.py

@@ -1,6 +1,12 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
+
 from .environment import EnvironmentCredentialsResolver
 from .static import StaticCredentialsResolver
+from .credentials_resolver_chain import CredentialsResolverChain
 
-__all__ = ("EnvironmentCredentialsResolver", "StaticCredentialsResolver")
+__all__ = (
+    "EnvironmentCredentialsResolver",
+    "StaticCredentialsResolver",
+    "CredentialsResolverChain",
+)

packages/smithy-aws-core/src/smithy_aws_core/credentials_resolvers/credentials_resolver_chain.py

@@ -0,0 +1,54 @@
+from typing import Callable, List
+
+from smithy_aws_core.credentials_resolvers import EnvironmentCredentialsResolver
+from smithy_aws_core.identity import AWSCredentialsIdentity, AWSCredentialsResolver
+from smithy_core.aio.interfaces.identity import IdentityResolver
+from smithy_core.exceptions import SmithyIdentityException
+from smithy_core.interfaces.identity import IdentityProperties
+
+import os
+
+
+def _env_creds_available() -> bool:
+    return bool(os.getenv("AWS_ACCESS_KEY_ID")) and bool(
+        os.getenv("AWS_SECRET_ACCESS_KEY")
+    )
+
+
+def _build_env_creds() -> AWSCredentialsResolver:
+    return EnvironmentCredentialsResolver()
+
+
+type CredentialSource = tuple[Callable[[], bool], Callable[[], AWSCredentialsResolver]]
+_DEFAULT_SOURCES: list[CredentialSource] = [(_env_creds_available, _build_env_creds)]
+
+
+class CredentialsResolverChain(
+    IdentityResolver[AWSCredentialsIdentity, IdentityProperties]
+):
+    """Resolves AWS Credentials from system environment variables."""
+
+    def __init__(self, *, sources: List[CredentialSource] | None = None):
+        if sources is None:
+            sources = _DEFAULT_SOURCES
+        self._sources: List[CredentialSource] = sources
+        self._credentials_resolver: AWSCredentialsResolver | None = None
+
+    async def get_identity(
+        self, *, identity_properties: IdentityProperties
+    ) -> AWSCredentialsIdentity:
+        if self._credentials_resolver is not None:
+            return await self._credentials_resolver.get_identity(
+                identity_properties=identity_properties
+            )
+
+        for source in self._sources:
+            if source[0]():
+                self._credentials_resolver = source[1]()
+                return await self._credentials_resolver.get_identity(
+                    identity_properties=identity_properties
+                )
+
+        raise SmithyIdentityException(
+            "None of the configured credentials sources were able to resolve credentials."
+        )

packages/smithy-aws-core/tests/unit/credentials_resolvers/test_credentials_resolver_chain.py

@@ -0,0 +1,62 @@
+import pytest
+
+from smithy_aws_core.credentials_resolvers import (
+    CredentialsResolverChain,
+    StaticCredentialsResolver,
+)
+from smithy_aws_core.identity import AWSCredentialsIdentity
+from smithy_core.exceptions import SmithyIdentityException
+from smithy_core.interfaces.identity import IdentityProperties
+
+
+async def test_no_sources_resolve():
+    resolver_chain = CredentialsResolverChain(sources=[])
+    with pytest.raises(SmithyIdentityException):
+        await resolver_chain.get_identity(identity_properties=IdentityProperties())
+
+
+async def test_env_credentials_resolver_not_set(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.delenv("AWS_ACCESS_KEY_ID", raising=False)
+    monkeypatch.delenv("AWS_SECRET_ACCESS_KEY", raising=False)
+    resolver_chain = CredentialsResolverChain()
+
+    with pytest.raises(SmithyIdentityException):
+        await resolver_chain.get_identity(identity_properties=IdentityProperties())
+
+
+async def test_env_credentials_resolver_partial(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "akid")
+    monkeypatch.delenv("AWS_SECRET_ACCESS_KEY", raising=False)
+    resolver_chain = CredentialsResolverChain()
+
+    with pytest.raises(SmithyIdentityException):
+        await resolver_chain.get_identity(identity_properties=IdentityProperties())
+
+
+async def test_env_credentials_resolver_success(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "akid")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "secret")
+    resolver_chain = CredentialsResolverChain()
+
+    credentials = await resolver_chain.get_identity(
+        identity_properties=IdentityProperties()
+    )
+    assert credentials.access_key_id == "akid"
+    assert credentials.secret_access_key == "secret"
+
+
+async def test_custom_sources_with_static_credentials():
+    static_credentials = AWSCredentialsIdentity(
+        access_key_id="static_akid",
+        secret_access_key="static_secret",
+    )
+    static_resolver = StaticCredentialsResolver(credentials=static_credentials)
+    resolver_chain = CredentialsResolverChain(
+        sources=[(lambda: False, lambda: None), (lambda: True, lambda: static_resolver)]  # type: ignore
+    )
+
+    credentials = await resolver_chain.get_identity(
+        identity_properties=IdentityProperties()
+    )
+    assert credentials.access_key_id == "static_akid"
+    assert credentials.secret_access_key == "static_secret"