fix: update env provider to skip terminal resolver registration if asume role creds are configured

packages/smithy-aws-core/.changes/next-release/smithy-aws-core-bugfix-0391d2827ce9402b8a77fd4ba5df090f.json

@@ -0,0 +1,4 @@
+{
+  "type": "bugfix",
+  "description": "Updated the Environment credentials provider to skip registering a terminal resolver when a configured profile assumes a role using `credential_source = Environment`."
+}

packages/smithy-aws-core/src/smithy_aws_core/identity/chain/providers/environment.py

@@ -4,13 +4,16 @@
 
 from smithy_core.interfaces.identity import Identity
 
+from ....config import load_config
 from ...components import AWSCredentialsIdentity
 from ...environment import EnvironmentCredentialsResolver
 from ..ordering import Standard, StandardProvider
 from ..provider import ChainSetup
 
 _ACCESS_KEY_ID = "AWS_ACCESS_KEY_ID"
 _SECRET_ACCESS_KEY = "AWS_SECRET_ACCESS_KEY"  # noqa: S105
+_ROLE_ARN = "role_arn"
+_CREDENTIAL_SOURCE = "credential_source"
 
 
 class EnvironmentCredentialsProvider:
@@ -38,4 +41,15 @@ async def setup(
         if not os.getenv(_ACCESS_KEY_ID) or not os.getenv(_SECRET_ACCESS_KEY):
             return
 
+        # Skip environment provider if a profile is explicitly provided and
+        # that profile configures assume role credentials with 'Environment' as the
+        # credential source
+        profile_name = setup.profile_name
+        if profile_name:
+            config = setup.config_file or await load_config()
+            role_arn = config.get(profile_name, _ROLE_ARN)
+            credential_source = config.get(profile_name, _CREDENTIAL_SOURCE)
+            if role_arn is not None and credential_source == "Environment":
+                return
+
         setup.add_terminal_resolver(EnvironmentCredentialsResolver())

packages/smithy-aws-core/tests/unit/identity/chain/providers/test_environment.py

@@ -1,8 +1,9 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Mapping
 
 import pytest
+from smithy_aws_core.config.merged_config import MergedConfig
 from smithy_aws_core.identity.chain.provider import ChainSetup
 from smithy_aws_core.identity.chain.providers.environment import (
     EnvironmentCredentialsProvider,
@@ -64,3 +65,82 @@ async def test_registers_terminal_resolver(
     assert len(setup.resolvers) == 1
     assert setup.resolvers[0].provider_name == "Environment"
     assert isinstance(setup.resolvers[0].resolver, EnvironmentCredentialsResolver)
+
+
+async def test_skips_when_profile_uses_environment_credential_source(
+    setup_provider: Callable[..., Awaitable[ChainSetup]],
+    merged_config: Callable[..., MergedConfig],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "akid")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "secret")
+
+    config = merged_config(
+        {
+            "assume": {
+                "role_arn": "arn:aws:iam::123456789012:role/example",
+                "credential_source": "Environment",
+            }
+        }
+    )
+
+    setup = await setup_provider(
+        EnvironmentCredentialsProvider(),
+        config_file=config,
+        profile_name="assume",
+    )
+
+    assert setup.resolvers == ()
+    assert not setup.terminal
+
+
+@pytest.mark.parametrize(
+    "profile_properties",
+    [
+        {
+            "role_arn": "arn:aws:iam::123456789012:role/example",
+            "credential_source": "Ec2InstanceMetadata",
+        },
+        {"credential_source": "Environment"},
+        {"role_arn": "arn:aws:iam::123456789012:role/example"},
+    ],
+)
+async def test_registers_resolver_when_not_assuming_role_from_environment(
+    profile_properties: Mapping[str, str],
+    setup_provider: Callable[..., Awaitable[ChainSetup]],
+    merged_config: Callable[..., MergedConfig],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "akid")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "secret")
+
+    config = merged_config({"assume": profile_properties})
+
+    setup = await setup_provider(
+        EnvironmentCredentialsProvider(),
+        config_file=config,
+        profile_name="assume",
+    )
+
+    assert setup.terminal
+    assert len(setup.resolvers) == 1
+    assert isinstance(setup.resolvers[0].resolver, EnvironmentCredentialsResolver)
+
+
+async def test_registers_resolver_when_profile_missing_from_config(
+    setup_provider: Callable[..., Awaitable[ChainSetup]],
+    merged_config: Callable[..., MergedConfig],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "akid")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "secret")
+
+    setup = await setup_provider(
+        EnvironmentCredentialsProvider(),
+        config_file=merged_config({}),
+        profile_name="does-not-exist",
+    )
+
+    assert setup.terminal
+    assert len(setup.resolvers) == 1
+    assert isinstance(setup.resolvers[0].resolver, EnvironmentCredentialsResolver)