Skip to content

Authentication

Authentication

With no arguments, credentials are resolved by the default chain when the first operation runs:

from capo_s3 import AsyncS3Client

async with AsyncS3Client() as s3:
    await s3.list_buckets()

The chain, in order:

  1. AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN
  2. Profile with role_arn (STS AssumeRole through source_profile or credential_source)
  3. AWS_WEB_IDENTITY_TOKEN_FILE + AWS_ROLE_ARN, or the profile's web_identity_token_file (EKS IRSA)
  4. Profile with sso_* (IAM Identity Center, after aws sso login)
  5. ~/.aws/credentials and ~/.aws/config static keys
  6. ECS / EKS Pod Identity container endpoint (AWS_CONTAINER_CREDENTIALS_*)
  7. EC2 instance metadata (IMDSv2, off with AWS_EC2_METADATA_DISABLED=true)

Steps 2, 3 and 4 need the sso extra:

uv add "capo-s3[sso]"

Static credentials:

AsyncS3Client(credentials={"access_key": "AKIA...", "secret_key": "..."})
AsyncS3Client(credentials={"access_key": "ASIA...", "secret_key": "...", "session_token": "..."})

A specific profile, either via the environment or a provider:

AWS_PROFILE=dev python app.py

from zapros import AsyncClient

from capo_s3 import (
    AssumeRoleCredentialsProvider,
    AsyncS3Client,
    ProfileCredentialsProvider,
    SsoCredentialsProvider,
)

# static keys in the profile
AsyncS3Client(credentials_provider=ProfileCredentialsProvider(profile="dev"))

# sso_* profile
AsyncS3Client(credentials_provider=SsoCredentialsProvider(AsyncClient(), profile="dev"))

# role_arn profile
AsyncS3Client(credentials_provider=AssumeRoleCredentialsProvider(AsyncClient(), profile="dev"))

Providers that call AWS take a zapros client: AsyncClient() for Async* clients, Client() for sync ones.

Your own chain:

from zapros import AsyncClient

from capo_s3 import (
    AsyncS3Client,
    CachedProvider,
    ChainedProvider,
    EnvCredentialsProvider,
    WebIdentityCredentialsProvider,
)

chain = ChainedProvider(EnvCredentialsProvider(), WebIdentityCredentialsProvider(AsyncClient()))
AsyncS3Client(credentials_provider=CachedProvider(chain))

Credentials are resolved on every request, so wrap providers that do I/O in CachedProvider. It re-resolves 60 seconds before expiration.

Your own provider:

from datetime import datetime, timedelta, timezone

from capo_s3 import AsyncS3Client, CachedProvider, Credentials, CredentialsProvider


class VaultCredentials(CredentialsProvider):
    def resolve_identity(self) -> Credentials:
        lease = vault.read("aws/creds/app")
        return {
            "access_key": lease["access_key"],
            "secret_key": lease["secret_key"],
            "session_token": lease["security_token"],
            "expiration": datetime.now(timezone.utc) + timedelta(seconds=lease["lease_duration"]),
        }

    # optional: override `aresolve_identity` to do async I/O; the default calls `resolve_identity`


AsyncS3Client(credentials_provider=CachedProvider(VaultCredentials()))

Resolve once, share across services. Every package exports the same providers and the same Credentials shape:

from zapros import AsyncClient

from capo_s3 import AsyncS3Client, SsoCredentialsProvider
from capo_sts import AsyncSTSClient

credentials = await SsoCredentialsProvider(AsyncClient()).aresolve_identity()

async with AsyncS3Client(credentials=credentials) as s3, AsyncSTSClient(credentials=credentials) as sts:
    print(await sts.get_caller_identity())
    print(await s3.list_buckets())

Per call:

await s3.get_object(
    "bucket", "key", config_overrides={"credentials_provider": ProfileCredentialsProvider(profile="other")}
)

Errors:

from capo_s3 import AssumeRoleError, IdentityNotFound, MissingDependencyError, SSOError

try:
    await s3.list_buckets()
except IdentityNotFound:
    ...  # no provider in the chain found credentials
except SSOError:
    ...  # SSO profile found but the token is unusable: run `aws sso login`
except AssumeRoleError:
    ...  # role_arn profile found but AssumeRole failed
except MissingDependencyError:
    ...  # the profile needs the `sso` extra