Cognito

JwtPayload module-attribute

JwtPayload = TypedDict('JwtPayload', {'sub': str, 'aud': str, 'iss': str, 'token_use': str, 'auth_time': int, 'iat': int, 'exp': int, 'email': str, 'email_verified': bool, 'cognito:username': str, 'cognito:groups': list[str], 'cognito:roles': list[str], 'cognito:preferred_role': str}, total=False)

Decoded JWT payload. (Functional syntax is needed for keys containing ':').

AWSCredentials

Bases: TypedDict

Normalized credentials exposed by manager.

CognitoCredentialsManager

CognitoCredentialsManager(identity_pool_id: str, user_pool_id: str, region: str = 'us-east-1', auth_tokens: CognitoTokens | None = None, authenticator: CognitoOAuthAuthenticator | None = None)

Manages AWS credentials using Cognito Identity Pool.

Intended long-running usage

manager = get_authenticated_cognito_manager(cogconf) session = manager.make_refreshing_session(role_suffix="SomeRole") s3 = session.client("s3")

Worker threads should use clients created from the returned session. They should not call refresh methods directly.

Parameters:

• identity_pool_id (str) –

  Cognito Identity Pool ID

• user_pool_id (str) –

  Cognito User Pool ID

• region (str, default: 'us-east-1' ) –

  AWS region

• auth_tokens (CognitoTokens | None, default: None ) –

  prefetched tokens [optional, can be fetched later]

• authenticator (CognitoOAuthAuthenticator | None, default: None ) –

  preconstructed authenticator [optional, can be provided later]

ensure_credentials

ensure_credentials(role_suffix: str | None = None, refresh_margin: timedelta = timedelta(minutes=5)) -> AWSCredentials

Ensure both layers are fresh enough:

1. Cognito User Pool ID/access tokens.
2. Cognito Identity Pool AWS credentials.

get_credentials

get_credentials(id_token: str | None = None, role_suffix: str | None = None) -> AWSCredentials

Exchange Cognito ID token for AWS credentials using Identity Pool. Updates self.creds_response and self.credentials.

Parameters:

• id_token (str | None, default: None ) –

  ID token from Cognito authentication. if not provided, uses id_token from self.auth_result if available

• role_suffix (str | None, default: None ) –

  if provided, checks available roles and requests credentials for the role with that string suffix. If there is not exactly one available role with that suffix, raises a PermissionError. If not provided, returns the first available role (if any).

Returns:

• AWSCredentials –

  self.credentials (dictionary containing AWS credentials)

make_refreshing_session

make_refreshing_session(role_suffix: str | None = None) -> boto3.Session

Create a boto3 Session backed by botocore refreshable credentials.

Prefer this method to make_session() for long-running workloads.

make_session

make_session(role_suffix: str | None = None) -> boto3.Session

Create a normal static boto3 Session. Use for short-lived operations. For longer-running workloads, prefer make_refreshable_session().

refresh_user_pool_tokens

refresh_user_pool_tokens() -> CognitoTokens

Refresh Cognito User Pool ID/access tokens using previously-retrieved refresh token.

CognitoOAuthAuthenticator

CognitoOAuthAuthenticator(cognito_domain: str, client_id: str, redirect_uri: str = 'http://localhost:3000', region: str = 'us-east-1')

Handles AWS Cognito OAuth authentication with PKCE

Initialize the Cognito OAuth authenticator

Parameters:

• cognito_domain (str) –

  Your Cognito domain (e.g., 'your-app.auth.us-east-1.amazoncognito.com')

• client_id (str) –

  The App Client ID from Cognito

• redirect_uri (str, default: 'http://localhost:3000' ) –

  The redirect URI configured in your Cognito app client

• region (str, default: 'us-east-1' ) –

  AWS region where your Cognito user pool is located

authenticate

authenticate() -> CognitoTokens

Perform the complete OAuth authentication flow

Returns:

• CognitoTokens –

  dict containing authentication tokens and user info

exchange_code_for_tokens

exchange_code_for_tokens(authorization_code: str, code_verifier: str) -> CognitoTokens

Exchange authorization code for access tokens

Parameters:

• authorization_code (str) –

  Authorization code from OAuth callback

• code_verifier (str) –

  PKCE code verifier

Returns:

• CognitoTokens –

  dict containing tokens (id_token, access_token, refresh_token)

generate_pkce_pair staticmethod

generate_pkce_pair() -> tuple[str, str]

Generate PKCE code verifier and challenge

Returns:

• tuple[str, str] –

  Tuple of (code_verifier, code_challenge)

get_authorization_url

get_authorization_url(code_challenge: str, state: str) -> str

Build the authorization URL for the OAuth flow

Parameters:

• code_challenge (str) –

  PKCE code challenge

• state (str) –

  Random state parameter for CSRF protection

Returns:

• str –

  Authorization URL

refresh_tokens

refresh_tokens(refresh_token: str) -> CognitoTokenRefreshResponse

exchange a refresh token for fresh Cognito User Pool tokens.

Usually returns new id_token/access_token. If refresh-token rotation is enabled, Cognito can also return a new refresh_token. Preserve the old one if Cognito does not return one.

start_local_server

start_local_server(port: int = 3000) -> OAuthCallbackServer

Start a local HTTP server to receive the OAuth callback

Parameters:

• port (int, default: 3000 ) –

  Port number to listen on

Returns:

• OAuthCallbackServer –

  HTTPServer instance

Raises:

• ConnectionError –

  if HTTPServer cannot connect to port

Security Notes

• Uses localhost only (no network exposure)
• Port must match the callback URL configured in Cognito
• Cannot use alternate ports without updating Cognito settings

CognitoTokenRefreshResponse

Bases: TypedDict

Raw response from Cognito refresh_token grant.

Cognito may return only new id/access tokens. If refresh-token rotation is enabled, it may also return a new refresh_token.

CognitoTokens

Bases: TypedDict

Stable token state kept in CognitoCredentialsManager.auth_result["tokens"].

This should include refresh_token after the initial authorization-code exchange. Refresh responses may omit refresh_token, but the manager should preserve the existing one.

IdentityPoolCredentials

Bases: TypedDict

Raw Credentials member from get_credentials_for_identity.

IdentityPoolCredentialsResponse

Bases: TypedDict

Raw response from Cognito Identity get_credentials_for_identity.

OAuthCallbackHandler

Bases: BaseHTTPRequestHandler

HTTP request handler for OAuth callback

RefreshableCredentialMetadata

Bases: TypedDict

Shape required by botocore.credentials.RefreshableCredentials. Yes, the key names are different from boto3 default.

decode_jwt_payload

decode_jwt_payload(jwt_token: str) -> JwtPayload

Decode a Cognito JWT token.

get_authenticated_cognito_manager

get_authenticated_cognito_manager(cogconf: CognitoConfiguration) -> CognitoCredentialsManager

Convenient entry point to this module: execute the Cognito OAuth workflow and return a credentials manager authenticated with the retrieved credentials.