Add modular credential chain for AWS identity resolution

packages/smithy-aws-core/.changes/next-release/smithy-aws-core-feature-b59653bc7c5e42dcb0e64bf9a2d066aa.json

@@ -0,0 +1,4 @@
+{
+  "type": "feature",
+  "description": "Added a modular credential chain for AWS identity resolution. `IdentityChain.create()` discovers installed chain providers via entry points, orders them by precedence, and assembles a resolver chain. Initially ships with the Environment, SharedConfig, ProfileSessionKeys, and ProfileStaticKeys providers."
+}

packages/smithy-aws-core/pyproject.toml

@@ -37,6 +37,12 @@ dependencies = [
 "Code" = "https://github.com/smithy-lang/smithy-python/blob/develop/packages/smithy-aws-core/"
 "Issue tracker" = "https://github.com/smithy-lang/smithy-python/issues"
 
+[project.entry-points."smithy_aws_core.identity.chain_providers"]
+Environment = "smithy_aws_core.identity.chain.providers.environment:EnvironmentCredentialsProvider"
+SharedConfig = "smithy_aws_core.identity.chain.providers.shared_config:SharedConfigProvider"
+ProfileSessionKeys = "smithy_aws_core.identity.chain.providers.profile:ProfileSessionCredentialsProvider"
+ProfileStaticKeys = "smithy_aws_core.identity.chain.providers.profile:ProfileStaticCredentialsProvider"
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

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

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from .aws_config import AsyncAwsConfig
-from .context import SharedConfigContext, load_config
+from .context import SharedConfigContext, load_config, shared_config_files_exist
 from .exceptions import (
     ConfigError,
     ConfigParseError,
@@ -25,4 +25,5 @@
     "ProfileNotFoundError",
     "SharedConfigContext",
     "load_config",
+    "shared_config_files_exist",
 ]

packages/smithy-aws-core/src/smithy_aws_core/config/context.py

@@ -86,6 +86,27 @@ async def load_config(
     return MergedConfig(std_config, std_credentials)
 
 
+def shared_config_files_exist(
+    config_file_path: Path | None = None,
+    credentials_file_path: Path | None = None,
+) -> bool:
+    """Return whether either the shared config or credentials file exists.
+
+    A cheap filesystem check (no parsing) used to detect whether the shared
+    config credential source appears configured.
+
+    :param config_file_path: Override path for config file.
+        Defaults to AWS_CONFIG_FILE env var or ~/.aws/config.
+    :param credentials_file_path: Override path for credentials file.
+        Defaults to AWS_SHARED_CREDENTIALS_FILE env var or ~/.aws/credentials.
+    :returns: True if either file exists on disk.
+    """
+    config_path, credentials_path = _resolve_config_paths(
+        config_file_path, credentials_file_path
+    )
+    return config_path.is_file() or credentials_path.is_file()
+
+
 class SharedConfigContext:
     """Resolution context shared across resolvers during config construction.
 

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

@@ -2,6 +2,13 @@
 #  SPDX-License-Identifier: Apache-2.0
 from smithy_core.types import PropertyKey
 
+from .chain import IdentityChain, IdentityChainError, UnclaimedSource
+from .chain.providers.environment import EnvironmentCredentialsProvider
+from .chain.providers.profile import (
+    ProfileSessionCredentialsProvider,
+    ProfileStaticCredentialsProvider,
+)
+from .chain.providers.shared_config import SharedConfigProvider
 from .components import (
     AWSCredentialsIdentity,
     AWSCredentialsResolver,
@@ -18,9 +25,16 @@
     "AWSCredentialsResolver",
     "AWSIdentityProperties",
     "ContainerCredentialsResolver",
+    "EnvironmentCredentialsProvider",
     "EnvironmentCredentialsResolver",
     "IMDSCredentialsResolver",
+    "IdentityChain",
+    "IdentityChainError",
+    "ProfileSessionCredentialsProvider",
+    "ProfileStaticCredentialsProvider",
+    "SharedConfigProvider",
     "StaticCredentialsResolver",
+    "UnclaimedSource",
 )
 
 AWS_IDENTITY_CONFIG = PropertyKey(key="config", value_type=AWSIdentityConfig)

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

@@ -0,0 +1,249 @@
+#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  SPDX-License-Identifier: Apache-2.0
+import logging
+from collections.abc import Callable, Mapping, Sequence
+from importlib import metadata
+from typing import Any, cast
+
+from smithy_core.aio.interfaces.identity import IdentityResolver
+from smithy_core.exceptions import SmithyIdentityError
+from smithy_core.interfaces.identity import Identity
+from smithy_http.aio.interfaces import HTTPClient
+
+from ...config.merged_config import MergedConfig
+from .exceptions import (
+    IdentityChainConfigurationError,
+    IdentityChainError,
+    IdentityResolverFailure,
+    UnclaimedSource,
+)
+from .ordering import (
+    After,
+    Before,
+    OrderingConstraint,
+    Standard,
+    StandardProvider,
+)
+from .provider import (
+    ChainIdentityProvider,
+    ChainSetup,
+    NamedResolver,
+)
+
+__all__ = (
+    "After",
+    "Before",
+    "ChainIdentityProvider",
+    "ChainSetup",
+    "IdentityChain",
+    "IdentityChainConfigurationError",
+    "IdentityChainError",
+    "IdentityResolverFailure",
+    "OrderingConstraint",
+    "Standard",
+    "StandardProvider",
+    "UnclaimedSource",
+)
+
+_CHAIN_PROVIDER_ENTRY_POINT_GROUP = "smithy_aws_core.identity.chain_providers"
+logger = logging.getLogger(__name__)
+
+
+def _discover_chain_identity_providers() -> tuple[ChainIdentityProvider, ...]:
+    discovered: list[ChainIdentityProvider] = []
+    for entry_point in metadata.entry_points(group=_CHAIN_PROVIDER_ENTRY_POINT_GROUP):
+        provider_factory = cast(
+            Callable[[], ChainIdentityProvider],
+            entry_point.load(),
+        )
+        discovered.append(provider_factory())
+    return tuple(discovered)
+
+
+def _sort_by_ordering(
+    providers: Sequence[ChainIdentityProvider],
+) -> tuple[ChainIdentityProvider, ...]:
+    if not providers:
+        return ()
+    slot_indexes = {slot: index for index, slot in enumerate(StandardProvider)}
+
+    def sort_key(
+        indexed_provider: tuple[int, ChainIdentityProvider],
+    ) -> tuple[int, int, int]:
+        discovery_index, provider = indexed_provider
+        ordering = provider.ordering
+
+        match ordering:
+            case Before():
+                constraint_precedence = 0
+            case Standard():
+                constraint_precedence = 1
+            case After():
+                constraint_precedence = 2
+            case _:
+                raise IdentityChainConfigurationError(
+                    f"Provider {type(provider).__name__} returned an unsupported "
+                    f"ordering constraint: {ordering!r}."
+                )
+
+        # Slot precedence, constraint precedence, then discovery order
+        return (slot_indexes[ordering.slot], constraint_precedence, discovery_index)
+
+    indexed_providers = enumerate(providers)
+    ordered_providers = sorted(indexed_providers, key=sort_key)
+    return tuple(provider for _, provider in ordered_providers)
+
+
+def _validate_providers(providers: Sequence[ChainIdentityProvider]) -> None:
+    discovered_names: dict[str, ChainIdentityProvider] = {}
+    discovered_standard_slots: dict[StandardProvider, ChainIdentityProvider] = {}
+
+    for provider in providers:
+        if (previous := discovered_names.get(provider.name)) is not None:
+            raise IdentityChainConfigurationError(
+                f"Credential providers {type(previous).__name__} and "
+                f"{type(provider).__name__} use the same name: {provider.name}."
+            )
+        discovered_names[provider.name] = provider
+
+        ordering = provider.ordering
+        if isinstance(ordering, Standard):
+            if (previous := discovered_standard_slots.get(ordering.slot)) is not None:
+                raise IdentityChainConfigurationError(
+                    f"Credential providers {type(previous).__name__} and "
+                    f"{type(provider).__name__} both claim standard slot: "
+                    f"{ordering.slot.name}."
+                )
+            discovered_standard_slots[ordering.slot] = provider
+
+
+def _find_unclaimed_sources(
+    providers: Sequence[ChainIdentityProvider],
+) -> tuple[UnclaimedSource, ...]:
+    claimed_slots = {
+        provider.ordering.slot
+        for provider in providers
+        if isinstance(provider.ordering, Standard)
+    }
+    unclaimed_sources: list[UnclaimedSource] = []
+
+    for slot in StandardProvider:
+        if slot in claimed_slots or not slot.is_detected():
+            continue
+        package = slot.module_suggestion
+        if package:
+            unclaimed_sources.append(
+                UnclaimedSource(source_name=slot.canonical_name, package=package)
+            )
+
+    return tuple(unclaimed_sources)
+
+
+class IdentityChain[I: Identity](IdentityResolver[I, Mapping[str, Any]]):
+    """Resolves identities from an assembled sequence of resolvers."""
+
+    _resolvers: tuple[IdentityResolver[I, Any], ...]
+    _identity_type: type[I] | None
+    _unclaimed_sources: tuple[UnclaimedSource, ...]
+
+    def __init__(
+        self,
+        resolvers: Sequence[IdentityResolver[I, Any]],
+        *,
+        identity_type: type[I] | None = None,
+        unclaimed_sources: Sequence[UnclaimedSource] = (),
+    ) -> None:
+        """Initialize the chain with resolvers in precedence order.
+
+        :param resolvers: Identity resolvers to iterate in precedence order.
+        :param identity_type: The identity type this chain resolves, or None when
+            constructing a chain directly without declaring one.
+        :param unclaimed_sources: Detected-but-unclaimed sources discovered during
+            chain assembly. This parameter is used by :meth:`create`; omit it when
+            constructing a chain directly.
+        """
+        self._resolvers = tuple(resolvers)
+        self._identity_type = identity_type
+        self._unclaimed_sources = tuple(unclaimed_sources)
+
+    @property
+    def identity_type(self) -> type[I] | None:
+        """The identity type this chain resolves, or None if not declared."""
+        return self._identity_type
+
+    @staticmethod
+    async def create[ChainIdentity: Identity](
+        identity_type: type[ChainIdentity],
+        *,
+        profile_file: MergedConfig | None = None,
+        profile_name_override: str | None = None,
+        region_override: str | None = None,
+        http_client: HTTPClient | None = None,
+    ) -> "IdentityChain[ChainIdentity]":
+        """Create an identity chain from discovered providers.
+
+        :param identity_type: The identity type to resolve.
+        :param profile_file: Parsed config/credentials file. Loaded from disk
+            when not set.
+        :param profile_name_override: Profile name to use, taking precedence over
+            ``AWS_PROFILE``.
+        :param region_override: Region to use for providers whose resolvers
+            fetch credentials through a service call.
+        :param http_client: HTTP client to use for providers whose resolvers make
+            network calls.
+        """
+        discovered_providers = _discover_chain_identity_providers()
+        _validate_providers(discovered_providers)
+        providers = _sort_by_ordering(discovered_providers)
+        setup = ChainSetup(
+            profile_file=profile_file,
+            profile_name_override=profile_name_override,
+            region_override=region_override,
+            http_client=http_client,
+        )
+        unclaimed_sources = _find_unclaimed_sources(discovered_providers)
+
+        for provider in providers:
+            setup.set_current_provider(provider)
+            await provider.setup(identity_type, setup)
+            if setup.terminal:
+                break
+
+        for source in unclaimed_sources:
+            logger.warning(str(source))
+        return IdentityChain(
+            setup.resolvers,
+            identity_type=identity_type,
+            unclaimed_sources=unclaimed_sources,
+        )
+
+    async def get_identity(self, *, properties: Mapping[str, Any]) -> I:
+        """Return the first identity resolved by the chain."""
+        failures: list[IdentityResolverFailure] = []
+        for resolver in self._resolvers:
+            try:
+                return await resolver.get_identity(properties=properties)
+            except SmithyIdentityError as error:
+                if isinstance(resolver, NamedResolver):
+                    provider_name = resolver.provider_name
+                    failed_resolver = resolver.resolver
+                else:
+                    provider_name = type(resolver).__name__
+                    failed_resolver = resolver
+                failures.append(
+                    IdentityResolverFailure(
+                        provider_name=provider_name,
+                        resolver=failed_resolver,
+                        error=error,
+                    )
+                )
+
+        raise IdentityChainError(
+            failures=tuple(failures),
+            unclaimed_sources=self._unclaimed_sources,
+        )
+
+    async def invalidate(self) -> None:
+        """Invalidate every resolver in the chain."""
+        for resolver in self._resolvers:
+            await resolver.invalidate()