Skip to content

Add modular credential chain for AWS identity resolution#749

Open
arandito wants to merge 1 commit into
smithy-lang:developfrom
arandito:credentials-chain
Open

Add modular credential chain for AWS identity resolution#749
arandito wants to merge 1 commit into
smithy-lang:developfrom
arandito:credentials-chain

Conversation

@arandito

Copy link
Copy Markdown
Contributor

Description

This PR adds modular credential resolution to Smithy Python via a new IdentityChain. Today users must set a single aws_credentials_identity_resolver on the client config; there is no fallback chain to automatically detect and resolve credentials from the environment, shared config files, etc. The new IdentityChain fills that gap.

IdentityChain implements the existing IdentityResolver interface, so it drops into the current config slot unchanged. A client cannot tell whether it holds a single resolver or a chain.

The chain follows a two-phase model:

Assembly (once, at construction). IdentityChain.create():

  • Discovers installed chain providers from the smithy_aws_core.identity.chain_providers entry point group. Installing a package that exposes a provider to that group is all that is needed to add a provider to the chain.
  • Validates and sorts them by standard credential-chain precedence, using Standard/Before/After ordering constraints against a fixed set of StandardProvider slots.
  • Calls each provider's setup() with a shared ChainSetup context. A provider inspects the environment and active profile, and if it detects its source, registers a resolver. Registering a terminal resolver (a source that is authoritative once detected, e.g. complete environment credentials) stops further assembly. When assembly finishes, ChainSetup is discarded and only the ordered resolvers survive.

Resolution (per request). IdentityChain.get_identity() walks the assembled resolvers in order, returning the first identity that resolves. A resolver "misses" by raising SmithyIdentityError, and the chain moves on. If every resolver misses, it raises IdentityChainError with per-provider diagnostics. When a known credential source is detected on the system but no provider claims it, the error also suggests the package to install.

This PR initially ships four standard providers (Environment, SharedConfig, ProfileSessionKeys, and ProfileStaticKeys) all of which detect their source locally with no network calls. Network-backed providers (STS, SSO, ECS, IMDS) are designed to ship as separate optional packages that register under the same entry point group.

Note

Network providers and integration into SDK client construction will be added in follow-up PRs. This PR adds standalone support for now.

Usage

Create a chain from installed providers:

chain = await IdentityChain.create(AWSCredentialsIdentity)

Construct a chain with explicit resolvers (useful for tests or an explicit, non-discovered chain). Here StaticCredentialsResolver is seeded with a fixed AWSCredentialsIdentity, so the chain falls back to those credentials if the environment resolver misses:

static_credentials = AWSCredentialsIdentity(
    access_key_id="...",
    secret_access_key="...",
)
chain = IdentityChain[AWSCredentialsIdentity](
    (
        EnvironmentCredentialsResolver(),
        StaticCredentialsResolver(static_credentials),
    )
)

Custom providers implement ChainIdentityProvider and position themselves relative to a standard slot:

class CustomProvider:
    name = "Custom"
    ordering = Before(slot=StandardProvider.EC2_INSTANCE_METADATA)

    async def setup(self, identity_type, setup: ChainSetup) -> None:
        if identity_type is AWSCredentialsIdentity:
            setup.add_resolver(CustomCredentialsResolver())

Register a custom provider so IdentityChain.create() can discover it:

# Custom package's pyproject.toml
[project.entry-points."smithy_aws_core.identity.chain_providers"]
Custom = "my_package.credentials:CustomProvider"

Testing

  • make test-py
    • Added unit tests for the new components: chain assembly (ordering, provider validation, discovery), per-request resolution and error diagnostics, and each of the four providers.
  • TODO: Implement the modular credential-chain conformance test suite. These unit tests deliberately cover the assembly and configuration-error paths that the behavioral conformance suite won't reach, to avoid overlap.
  • Exercised end to end against a real Bedrock Runtime client, resolving credentials from local AWS config/credentials files through a discovered chain:
import asyncio

from aws_sdk_bedrock_runtime.client import BedrockRuntimeClient
from aws_sdk_bedrock_runtime.config import Config
from aws_sdk_bedrock_runtime.models import ContentBlockText, ConverseInput, Message
from smithy_aws_core.identity import IdentityChain, AWSCredentialsIdentity


async def main():
    client = BedrockRuntimeClient(
        config=Config(
            endpoint_uri="https://bedrock-runtime.us-west-2.amazonaws.com",
            region="us-west-2",
            aws_credentials_identity_resolver=await IdentityChain.create(
                identity_type=AWSCredentialsIdentity
            ),
        )
    )
    response = await client.converse(
        ConverseInput(
            model_id="us.anthropic.claude-sonnet-4-6",
            messages=[
                Message(
                    role="user",
                    content=[ContentBlockText(value="Say hello in one sentence.")],
                )
            ],
        )
    )
    print(response)


asyncio.run(main())

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

@arandito
arandito requested a review from a team as a code owner July 22, 2026 23:24


@dataclass(frozen=True, kw_only=True)
class NamedResolver:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will also eventually associate an identity resolver with other provider metadata like a feature id set, not just its name. I'm open to renaming this resolver wrapper.

self._terminal = True


class ChainIdentityProvider(Protocol):

@arandito arandito Jul 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just realized that this class will eventually need to expose a feature_ids property and may need to expose new things in the future. This would break anybody that initializes their provider without creating a subclass from this protocol, even if we provide a default.

We can probably switch this to an ABC to allow us to add these properties with defaults that automatically propagate to third-party implementations. We could also implement a more specific protocol like AwsChainIdentityProvider that we control while directing third party implementations to use the base ChainIdentityProvider. Curious what reviewers think.

@jonathan343 jonathan343 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @arandito!

This PR is looking pretty good. I left a few comments, let me know if you have any questions!

self.canonical_name = canonical_name
self.module_suggestion = module_suggestion

def is_detected(self) -> bool:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit - could we use match / case here instead of multiple if's?

def is_detected(self) -> bool:
    """Return whether this slot has a cheap environment detection signal."""
    match self:
        case StandardProvider.ENVIRONMENT:
            return bool(os.getenv("AWS_ACCESS_KEY_ID"))
        case StandardProvider.WEB_IDENTITY_TOKEN_ENV:
            return bool(os.getenv("AWS_WEB_IDENTITY_TOKEN_FILE")) and bool(
                os.getenv("AWS_ROLE_ARN")
            )
        case StandardProvider.ECS_CONTAINER:
            return bool(os.getenv("AWS_CONTAINER_CREDENTIALS_FULL_URI")) or bool(
                os.getenv("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI")
            )
        case StandardProvider.SHARED_CONFIG:
            return shared_config_files_exist()
        case _:
            return False

unclaimed_sources=unclaimed_sources,
)

async def get_identity(self, *, properties: Mapping[str, Any]) -> I:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit - Can we make properties optional and default to {}? I was testing this locally and if felt weird that I was forced to pass in properties={}:

(smithy-python) $ python -m asyncio
asyncio REPL 3.12.6 (main, Sep 24 2024, 21:09:37) [Clang 16.0.0 (clang-1600.0.26.3)] on darwin
Use "await" directly instead of "asyncio.run()".
Type "help", "copyright", "credits" or "license" for more information.
>>> import asyncio
>>> from smithy_aws_core.identity.chain import IdentityChain
>>> from smithy_aws_core.identity.components import AWSCredentialsIdentity
>>> chain = await IdentityChain.create(AWSCredentialsIdentity)
>>> await chain.get_identity(properties={})
AWSCredentialsIdentity(access_key_id='...', secret_access_key='.', session_token=None, expiration=None, account_id=None)

account_id = os.getenv("AWS_ACCOUNT_ID")

if access_key_id is None or secret_access_key is None:
if not access_key_id or not secret_access_key:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why this change? Is this to catch empty strings?

return tuple(unclaimed_sources)


class IdentityChain[I: Identity](IdentityResolver[I, Mapping[str, Any]]):

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Don't we need to add an invalidate method here? Something like:

  async def invalidate(self) -> None:
      for resolver in self._resolvers:
          await resolver.invalidate()

This will also require the IdentityResolver interface to be updated. I think we can implement a no-op as default. Then resolvers that need custom behavior can implement what they need. This does force custom resolvers to implement as well, but I think that's fine.

return await self.resolver.get_identity(properties=properties)


class ChainSetup:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why no region override?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants