There are more AWS SDK examples available in the [AWS Doc SDK Examples](https://github.com/awsdocs/aws-doc-sdk-examples) GitHub repo.
# Amazon Cognito Identity Provider examples using SDK for Python (Boto3)
The following code examples show you how to perform actions and implement common scenarios by using the AWS SDK for Python (Boto3) with Amazon Cognito Identity Provider.
*Actions* are code excerpts from larger programs and must be run in context. While actions show you how to call individual service functions, you can see actions in context in their related scenarios.
*Scenarios* are code examples that show you how to accomplish specific tasks by calling multiple functions within a service or combined with other AWS services.
Each example includes a link to the complete source code, where you can find instructions on how to set up and run the code in context.
**Topics**
+ [Get started](#get_started)
+ [Actions](#actions)
+ [Scenarios](#scenarios)
## Get started
### Hello Amazon Cognito
The following code example shows how to get started using Amazon Cognito.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
import boto3
# Create a Cognito Identity Provider client
cognitoidp = boto3.client("cognito-idp")
# Initialize a paginator for the list_user_pools operation
paginator = cognitoidp.get_paginator("list_user_pools")
# Create a PageIterator from the paginator
page_iterator = paginator.paginate(MaxResults=10)
# Initialize variables for pagination
user_pools = []
# Handle pagination
for page in page_iterator:
user_pools.extend(page.get("UserPools", []))
# Print the list of user pools
print("User Pools for the account:")
if user_pools:
for pool in user_pools:
print(f"Name: {pool['Name']}, ID: {pool['Id']}")
else:
print("No user pools found.")
```
+ For API details, see [ListUserPools](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ListUserPools) in *AWS SDK for Python (Boto3) API Reference*.
## Actions
### `AdminGetUser`
The following code example shows how to use `AdminGetUser`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def sign_up_user(self, user_name, password, user_email):
"""
Signs up a new user with Amazon Cognito. This action prompts Amazon Cognito
to send an email to the specified email address. The email contains a code that
can be used to confirm the user.
When the user already exists, the user status is checked to determine whether
the user has been confirmed.
:param user_name: The user name that identifies the new user.
:param password: The password for the new user.
:param user_email: The email address for the new user.
:return: True when the user is already confirmed with Amazon Cognito.
Otherwise, false.
"""
try:
kwargs = {
"ClientId": self.client_id,
"Username": user_name,
"Password": password,
"UserAttributes": [{"Name": "email", "Value": user_email}],
}
if self.client_secret is not None:
kwargs["SecretHash"] = self._secret_hash(user_name)
response = self.cognito_idp_client.sign_up(**kwargs)
confirmed = response["UserConfirmed"]
except ClientError as err:
if err.response["Error"]["Code"] == "UsernameExistsException":
response = self.cognito_idp_client.admin_get_user(
UserPoolId=self.user_pool_id, Username=user_name
)
logger.warning(
"User %s exists and is %s.", user_name, response["UserStatus"]
)
confirmed = response["UserStatus"] == "CONFIRMED"
else:
logger.error(
"Couldn't sign up %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
return confirmed
```
+ For API details, see [AdminGetUser](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/AdminGetUser) in *AWS SDK for Python (Boto3) API Reference*.
### `AdminInitiateAuth`
The following code example shows how to use `AdminInitiateAuth`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def start_sign_in(self, user_name, password):
"""
Starts the sign-in process for a user by using administrator credentials.
This method of signing in is appropriate for code running on a secure server.
If the user pool is configured to require MFA and this is the first sign-in
for the user, Amazon Cognito returns a challenge response to set up an
MFA application. When this occurs, this function gets an MFA secret from
Amazon Cognito and returns it to the caller.
:param user_name: The name of the user to sign in.
:param password: The user's password.
:return: The result of the sign-in attempt. When sign-in is successful, this
returns an access token that can be used to get AWS credentials. Otherwise,
Amazon Cognito returns a challenge to set up an MFA application,
or a challenge to enter an MFA code from a registered MFA application.
"""
try:
kwargs = {
"UserPoolId": self.user_pool_id,
"ClientId": self.client_id,
"AuthFlow": "ADMIN_USER_PASSWORD_AUTH",
"AuthParameters": {"USERNAME": user_name, "PASSWORD": password},
}
if self.client_secret is not None:
kwargs["AuthParameters"]["SECRET_HASH"] = self._secret_hash(user_name)
response = self.cognito_idp_client.admin_initiate_auth(**kwargs)
challenge_name = response.get("ChallengeName", None)
if challenge_name == "MFA_SETUP":
if (
"SOFTWARE_TOKEN_MFA"
in response["ChallengeParameters"]["MFAS_CAN_SETUP"]
):
response.update(self.get_mfa_secret(response["Session"]))
else:
raise RuntimeError(
"The user pool requires MFA setup, but the user pool is not "
"configured for TOTP MFA. This example requires TOTP MFA."
)
except ClientError as err:
logger.error(
"Couldn't start sign in for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
response.pop("ResponseMetadata", None)
return response
```
+ For API details, see [AdminInitiateAuth](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/AdminInitiateAuth) in *AWS SDK for Python (Boto3) API Reference*.
### `AdminRespondToAuthChallenge`
The following code example shows how to use `AdminRespondToAuthChallenge`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
Respond to an MFA challenge by providing a code generated by an associated MFA application.
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def respond_to_mfa_challenge(self, user_name, session, mfa_code):
"""
Responds to a challenge for an MFA code. This completes the second step of
a two-factor sign-in. When sign-in is successful, it returns an access token
that can be used to get AWS credentials from Amazon Cognito.
:param user_name: The name of the user who is signing in.
:param session: Session information returned from a previous call to initiate
authentication.
:param mfa_code: A code generated by the associated MFA application.
:return: The result of the authentication. When successful, this contains an
access token for the user.
"""
try:
kwargs = {
"UserPoolId": self.user_pool_id,
"ClientId": self.client_id,
"ChallengeName": "SOFTWARE_TOKEN_MFA",
"Session": session,
"ChallengeResponses": {
"USERNAME": user_name,
"SOFTWARE_TOKEN_MFA_CODE": mfa_code,
},
}
if self.client_secret is not None:
kwargs["ChallengeResponses"]["SECRET_HASH"] = self._secret_hash(
user_name
)
response = self.cognito_idp_client.admin_respond_to_auth_challenge(**kwargs)
auth_result = response["AuthenticationResult"]
except ClientError as err:
if err.response["Error"]["Code"] == "ExpiredCodeException":
logger.warning(
"Your MFA code has expired or has been used already. You might have "
"to wait a few seconds until your app shows you a new code."
)
else:
logger.error(
"Couldn't respond to mfa challenge for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return auth_result
```
+ For API details, see [AdminRespondToAuthChallenge](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/AdminRespondToAuthChallenge) in *AWS SDK for Python (Boto3) API Reference*.
### `AssociateSoftwareToken`
The following code example shows how to use `AssociateSoftwareToken`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def get_mfa_secret(self, session):
"""
Gets a token that can be used to associate an MFA application with the user.
:param session: Session information returned from a previous call to initiate
authentication.
:return: An MFA token that can be used to set up an MFA application.
"""
try:
response = self.cognito_idp_client.associate_software_token(Session=session)
except ClientError as err:
logger.error(
"Couldn't get MFA secret. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
response.pop("ResponseMetadata", None)
return response
```
+ For API details, see [AssociateSoftwareToken](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/AssociateSoftwareToken) in *AWS SDK for Python (Boto3) API Reference*.
### `ConfirmDevice`
The following code example shows how to use `ConfirmDevice`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def confirm_mfa_device(
self,
user_name,
device_key,
device_group_key,
device_password,
access_token,
aws_srp,
):
"""
Confirms an MFA device to be tracked by Amazon Cognito. When a device is
tracked, its key and password can be used to sign in without requiring a new
MFA code from the MFA application.
:param user_name: The user that is associated with the device.
:param device_key: The key of the device, returned by Amazon Cognito.
:param device_group_key: The group key of the device, returned by Amazon Cognito.
:param device_password: The password that is associated with the device.
:param access_token: The user's access token.
:param aws_srp: A class that helps with Secure Remote Password (SRP)
calculations. The scenario associated with this example uses
the warrant package.
:return: True when the user must confirm the device. Otherwise, False. When
False, the device is automatically confirmed and tracked.
"""
srp_helper = aws_srp.AWSSRP(
username=user_name,
password=device_password,
pool_id="_",
client_id=self.client_id,
client_secret=None,
client=self.cognito_idp_client,
)
device_and_pw = f"{device_group_key}{device_key}:{device_password}"
device_and_pw_hash = aws_srp.hash_sha256(device_and_pw.encode("utf-8"))
salt = aws_srp.pad_hex(aws_srp.get_random(16))
x_value = aws_srp.hex_to_long(aws_srp.hex_hash(salt + device_and_pw_hash))
verifier = aws_srp.pad_hex(pow(srp_helper.val_g, x_value, srp_helper.big_n))
device_secret_verifier_config = {
"PasswordVerifier": base64.standard_b64encode(
bytearray.fromhex(verifier)
).decode("utf-8"),
"Salt": base64.standard_b64encode(bytearray.fromhex(salt)).decode("utf-8"),
}
try:
response = self.cognito_idp_client.confirm_device(
AccessToken=access_token,
DeviceKey=device_key,
DeviceSecretVerifierConfig=device_secret_verifier_config,
)
user_confirm = response["UserConfirmationNecessary"]
except ClientError as err:
logger.error(
"Couldn't confirm mfa device %s. Here's why: %s: %s",
device_key,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return user_confirm
```
+ For API details, see [ConfirmDevice](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ConfirmDevice) in *AWS SDK for Python (Boto3) API Reference*.
### `ConfirmSignUp`
The following code example shows how to use `ConfirmSignUp`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def confirm_user_sign_up(self, user_name, confirmation_code):
"""
Confirms a previously created user. A user must be confirmed before they
can sign in to Amazon Cognito.
:param user_name: The name of the user to confirm.
:param confirmation_code: The confirmation code sent to the user's registered
email address.
:return: True when the confirmation succeeds.
"""
try:
kwargs = {
"ClientId": self.client_id,
"Username": user_name,
"ConfirmationCode": confirmation_code,
}
if self.client_secret is not None:
kwargs["SecretHash"] = self._secret_hash(user_name)
self.cognito_idp_client.confirm_sign_up(**kwargs)
except ClientError as err:
logger.error(
"Couldn't confirm sign up for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return True
```
+ For API details, see [ConfirmSignUp](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ConfirmSignUp) in *AWS SDK for Python (Boto3) API Reference*.
### `DescribeTermsByClient`
The following code example shows how to use `DescribeTermsByClient`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito/scenarios/client_token#code-examples).
```
def describe_terms_by_client(
self, user_pool_id: str, client_id: str, terms_name: str
) -> dict[str, Any]:
"""
Looks up a Terms document associated with an app client by client ID,
user pool ID, and terms name.
:param user_pool_id: The ID of the user pool that contains the Terms documents.
:param client_id: The ID of the app client associated with the Terms.
:param terms_name: The name of the terms document ('terms-of-use' or
'privacy-policy').
:return: A dictionary with the Terms document details.
"""
try:
response = self.cognito_idp_client.describe_terms_by_client(
UserPoolId=user_pool_id,
ClientId=client_id,
TermsName=terms_name,
)
terms = response.get("Terms", dict())
logger.info(
"Described terms '%s' for client %s in user pool %s.",
terms_name,
client_id,
user_pool_id,
)
return terms
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error(
"No terms documents found for client %s, user pool %s, "
"and terms name '%s'.",
client_id,
user_pool_id,
terms_name,
)
raise
```
+ For API details, see [DescribeTermsByClient](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/DescribeTermsByClient) in *AWS SDK for Python (Boto3) API Reference*.
### `DescribeUserPoolClient`
The following code example shows how to use `DescribeUserPoolClient`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito/scenarios/client_token#code-examples).
```
def describe_user_pool_client(
self, user_pool_id: str, client_id: str
) -> dict[str, Any]:
"""
Describes an app client in a user pool, including the client secret.
:param user_pool_id: The ID of the user pool that contains the app client.
:param client_id: The ID of the app client to describe.
:return: A dictionary with the app client configuration.
"""
try:
response = self.cognito_idp_client.describe_user_pool_client(
UserPoolId=user_pool_id, ClientId=client_id
)
client_info = response.get("UserPoolClient", dict())
logger.info(
"Described app client %s in user pool %s.",
client_id,
user_pool_id,
)
return client_info
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error(
"The user pool %s or client %s does not exist. "
"Verify the stack deployed successfully and the outputs are correct.",
user_pool_id,
client_id,
)
raise
```
+ For API details, see [DescribeUserPoolClient](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/DescribeUserPoolClient) in *AWS SDK for Python (Boto3) API Reference*.
### `GetClientToken`
The following code example shows how to use `GetClientToken`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito/scenarios/client_token#code-examples).
```
def get_client_token(
self,
client_id: str,
client_secret: str,
scopes: Optional[list[str]] = None,
) -> dict[str, Any]:
"""
Obtains an M2M access token for an app client using its client ID
and secret, without end-user sign-in.
:param client_id: The ID of the app client.
:param client_secret: The client secret for the app client.
:param scopes: Optional list of scopes in the format
'resource-server-identifier/scope-name'. If omitted,
all scopes configured for the app client are authorized.
:return: A dictionary containing the ClientAuthenticationResult with
AccessToken, ExpiresIn, and TokenType.
"""
try:
params = dict()
params["ClientId"] = client_id
params["Secret"] = client_secret
if scopes is not None:
params["Scopes"] = scopes
response = self.cognito_idp_client.get_client_token(**params)
auth_result = response.get("ClientAuthenticationResult", dict())
logger.info(
"Obtained M2M access token for client %s. Token type: %s, expires in: %s seconds.",
client_id,
auth_result.get("TokenType", "Unknown"),
auth_result.get("ExpiresIn", 0),
)
return auth_result
except ClientError as err:
if err.response["Error"]["Code"] == "NotAuthorizedException":
logger.error(
"Not authorized for client %s. Verify that the client ID and "
"secret are correct, and that the ALLOW_CLIENT_TOKEN_AUTH flow "
"is enabled for the app client.",
client_id,
)
raise
```
+ For API details, see [GetClientToken](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/GetClientToken) in *AWS SDK for Python (Boto3) API Reference*.
### `InitiateAuth`
The following code example shows how to use `InitiateAuth`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
This example shows you how to start authentication with a tracked device. To complete sign-in, the client must respond correctly to Secure Remote Password (SRP) challenges.
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def sign_in_with_tracked_device(
self,
user_name,
password,
device_key,
device_group_key,
device_password,
aws_srp,
):
"""
Signs in to Amazon Cognito as a user who has a tracked device. Signing in
with a tracked device lets a user sign in without entering a new MFA code.
Signing in with a tracked device requires that the client respond to the SRP
protocol. The scenario associated with this example uses the warrant package
to help with SRP calculations.
For more information on SRP, see https://en.wikipedia.org/wiki/Secure_Remote_Password_protocol.
:param user_name: The user that is associated with the device.
:param password: The user's password.
:param device_key: The key of a tracked device.
:param device_group_key: The group key of a tracked device.
:param device_password: The password that is associated with the device.
:param aws_srp: A class that helps with SRP calculations. The scenario
associated with this example uses the warrant package.
:return: The result of the authentication. When successful, this contains an
access token for the user.
"""
try:
srp_helper = aws_srp.AWSSRP(
username=user_name,
password=device_password,
pool_id="_",
client_id=self.client_id,
client_secret=None,
client=self.cognito_idp_client,
)
response_init = self.cognito_idp_client.initiate_auth(
ClientId=self.client_id,
AuthFlow="USER_PASSWORD_AUTH",
AuthParameters={
"USERNAME": user_name,
"PASSWORD": password,
"DEVICE_KEY": device_key,
},
)
if response_init["ChallengeName"] != "DEVICE_SRP_AUTH":
raise RuntimeError(
f"Expected DEVICE_SRP_AUTH challenge but got {response_init['ChallengeName']}."
)
auth_params = srp_helper.get_auth_params()
auth_params["DEVICE_KEY"] = device_key
response_auth = self.cognito_idp_client.respond_to_auth_challenge(
ClientId=self.client_id,
ChallengeName="DEVICE_SRP_AUTH",
ChallengeResponses=auth_params,
)
if response_auth["ChallengeName"] != "DEVICE_PASSWORD_VERIFIER":
raise RuntimeError(
f"Expected DEVICE_PASSWORD_VERIFIER challenge but got "
f"{response_init['ChallengeName']}."
)
challenge_params = response_auth["ChallengeParameters"]
challenge_params["USER_ID_FOR_SRP"] = device_group_key + device_key
cr = srp_helper.process_challenge(challenge_params, {"USERNAME": user_name})
cr["USERNAME"] = user_name
cr["DEVICE_KEY"] = device_key
response_verifier = self.cognito_idp_client.respond_to_auth_challenge(
ClientId=self.client_id,
ChallengeName="DEVICE_PASSWORD_VERIFIER",
ChallengeResponses=cr,
)
auth_tokens = response_verifier["AuthenticationResult"]
except ClientError as err:
logger.error(
"Couldn't start client sign in for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return auth_tokens
```
+ For API details, see [InitiateAuth](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/InitiateAuth) in *AWS SDK for Python (Boto3) API Reference*.
### `ListUsers`
The following code example shows how to use `ListUsers`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def list_users(self):
"""
Returns a list of the users in the current user pool.
:return: The list of users.
"""
try:
response = self.cognito_idp_client.list_users(UserPoolId=self.user_pool_id)
users = response["Users"]
except ClientError as err:
logger.error(
"Couldn't list users for %s. Here's why: %s: %s",
self.user_pool_id,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return users
```
+ For API details, see [ListUsers](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ListUsers) in *AWS SDK for Python (Boto3) API Reference*.
### `ResendConfirmationCode`
The following code example shows how to use `ResendConfirmationCode`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def resend_confirmation(self, user_name):
"""
Prompts Amazon Cognito to resend an email with a new confirmation code.
:param user_name: The name of the user who will receive the email.
:return: Delivery information about where the email is sent.
"""
try:
kwargs = {"ClientId": self.client_id, "Username": user_name}
if self.client_secret is not None:
kwargs["SecretHash"] = self._secret_hash(user_name)
response = self.cognito_idp_client.resend_confirmation_code(**kwargs)
delivery = response["CodeDeliveryDetails"]
except ClientError as err:
logger.error(
"Couldn't resend confirmation to %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return delivery
```
+ For API details, see [ResendConfirmationCode](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ResendConfirmationCode) in *AWS SDK for Python (Boto3) API Reference*.
### `RespondToAuthChallenge`
The following code example shows how to use `RespondToAuthChallenge`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
Sign in with a tracked device. To complete sign-in, the client must respond correctly to Secure Remote Password (SRP) challenges.
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def sign_in_with_tracked_device(
self,
user_name,
password,
device_key,
device_group_key,
device_password,
aws_srp,
):
"""
Signs in to Amazon Cognito as a user who has a tracked device. Signing in
with a tracked device lets a user sign in without entering a new MFA code.
Signing in with a tracked device requires that the client respond to the SRP
protocol. The scenario associated with this example uses the warrant package
to help with SRP calculations.
For more information on SRP, see https://en.wikipedia.org/wiki/Secure_Remote_Password_protocol.
:param user_name: The user that is associated with the device.
:param password: The user's password.
:param device_key: The key of a tracked device.
:param device_group_key: The group key of a tracked device.
:param device_password: The password that is associated with the device.
:param aws_srp: A class that helps with SRP calculations. The scenario
associated with this example uses the warrant package.
:return: The result of the authentication. When successful, this contains an
access token for the user.
"""
try:
srp_helper = aws_srp.AWSSRP(
username=user_name,
password=device_password,
pool_id="_",
client_id=self.client_id,
client_secret=None,
client=self.cognito_idp_client,
)
response_init = self.cognito_idp_client.initiate_auth(
ClientId=self.client_id,
AuthFlow="USER_PASSWORD_AUTH",
AuthParameters={
"USERNAME": user_name,
"PASSWORD": password,
"DEVICE_KEY": device_key,
},
)
if response_init["ChallengeName"] != "DEVICE_SRP_AUTH":
raise RuntimeError(
f"Expected DEVICE_SRP_AUTH challenge but got {response_init['ChallengeName']}."
)
auth_params = srp_helper.get_auth_params()
auth_params["DEVICE_KEY"] = device_key
response_auth = self.cognito_idp_client.respond_to_auth_challenge(
ClientId=self.client_id,
ChallengeName="DEVICE_SRP_AUTH",
ChallengeResponses=auth_params,
)
if response_auth["ChallengeName"] != "DEVICE_PASSWORD_VERIFIER":
raise RuntimeError(
f"Expected DEVICE_PASSWORD_VERIFIER challenge but got "
f"{response_init['ChallengeName']}."
)
challenge_params = response_auth["ChallengeParameters"]
challenge_params["USER_ID_FOR_SRP"] = device_group_key + device_key
cr = srp_helper.process_challenge(challenge_params, {"USERNAME": user_name})
cr["USERNAME"] = user_name
cr["DEVICE_KEY"] = device_key
response_verifier = self.cognito_idp_client.respond_to_auth_challenge(
ClientId=self.client_id,
ChallengeName="DEVICE_PASSWORD_VERIFIER",
ChallengeResponses=cr,
)
auth_tokens = response_verifier["AuthenticationResult"]
except ClientError as err:
logger.error(
"Couldn't start client sign in for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return auth_tokens
```
+ For API details, see [RespondToAuthChallenge](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/RespondToAuthChallenge) in *AWS SDK for Python (Boto3) API Reference*.
### `SignUp`
The following code example shows how to use `SignUp`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def sign_up_user(self, user_name, password, user_email):
"""
Signs up a new user with Amazon Cognito. This action prompts Amazon Cognito
to send an email to the specified email address. The email contains a code that
can be used to confirm the user.
When the user already exists, the user status is checked to determine whether
the user has been confirmed.
:param user_name: The user name that identifies the new user.
:param password: The password for the new user.
:param user_email: The email address for the new user.
:return: True when the user is already confirmed with Amazon Cognito.
Otherwise, false.
"""
try:
kwargs = {
"ClientId": self.client_id,
"Username": user_name,
"Password": password,
"UserAttributes": [{"Name": "email", "Value": user_email}],
}
if self.client_secret is not None:
kwargs["SecretHash"] = self._secret_hash(user_name)
response = self.cognito_idp_client.sign_up(**kwargs)
confirmed = response["UserConfirmed"]
except ClientError as err:
if err.response["Error"]["Code"] == "UsernameExistsException":
response = self.cognito_idp_client.admin_get_user(
UserPoolId=self.user_pool_id, Username=user_name
)
logger.warning(
"User %s exists and is %s.", user_name, response["UserStatus"]
)
confirmed = response["UserStatus"] == "CONFIRMED"
else:
logger.error(
"Couldn't sign up %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
return confirmed
```
+ For API details, see [SignUp](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/SignUp) in *AWS SDK for Python (Boto3) API Reference*.
### `VerifySoftwareToken`
The following code example shows how to use `VerifySoftwareToken`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def verify_mfa(self, session, user_code):
"""
Verify a new MFA application that is associated with a user.
:param session: Session information returned from a previous call to initiate
authentication.
:param user_code: A code generated by the associated MFA application.
:return: Status that indicates whether the MFA application is verified.
"""
try:
response = self.cognito_idp_client.verify_software_token(
Session=session, UserCode=user_code
)
except ClientError as err:
logger.error(
"Couldn't verify MFA. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
response.pop("ResponseMetadata", None)
return response
```
+ For API details, see [VerifySoftwareToken](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/VerifySoftwareToken) in *AWS SDK for Python (Boto3) API Reference*.
## Scenarios
### Client token and Terms discovery
The following code example shows how to use Amazon Cognito machine-to-machine (M2M) authorization and Terms discovery.
+ Deploy a CloudFormation stack with Cognito resources (user pool, resource server, app client, Terms documents).
+ Retrieve the app client secret with DescribeUserPoolClient.
+ Obtain M2M access tokens with GetClientToken.
+ Discover Terms documents with DescribeTermsByClient.
+ Clean up by deleting the CloudFormation stack.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito/scenarios/client_token#code-examples).
Run an interactive scenario at a command prompt.
```
class ClientTokenAndTermsScenario:
"""
Runs an interactive scenario that demonstrates the Amazon Cognito
GetClientToken and DescribeTermsByClient APIs.
"""
def __init__(
self,
cognito_wrapper: CognitoIdentityProviderWrapper,
cfn_client: BaseClient,
):
"""
:param cognito_wrapper: An instance of the CognitoIdentityProviderWrapper class.
:param cfn_client: A Boto3 CloudFormation client.
"""
self.cognito_wrapper = cognito_wrapper
self.cfn_client = cfn_client
self.stack_name = None
self.user_pool_id = None
self.client_id = None
self.client_secret = None
def run(self) -> None:
"""Runs the scenario from start to finish."""
print(SEPARATOR)
print("Welcome to the Amazon Cognito Client Token and Terms Scenario.")
print(SEPARATOR)
print("This example creates Cognito resources in a CloudFormation stack, then")
print("demonstrates the new GetClientToken and DescribeTermsByClient APIs.")
print()
try:
self.setup()
self.get_client_secret()
self.demonstrate_get_client_token()
self.demonstrate_describe_terms()
except Exception:
logger.exception("An error occurred during the scenario.")
finally:
self.cleanup()
print(SEPARATOR)
print("Amazon Cognito Client Token and Terms scenario completed.")
# ------------------------------------------------------------------
# Phase 1: Setup - deploy CloudFormation stack
# ------------------------------------------------------------------
def setup(self) -> None:
"""Deploys the CloudFormation prerequisite stack and retrieves outputs."""
stack_name = q.ask(
f"\nEnter a name for the CloudFormation stack [{DEFAULT_STACK_NAME}]: "
)
if not stack_name:
stack_name = DEFAULT_STACK_NAME
# Check whether the stack name already exists.
while self._stack_exists(stack_name):
print(f"A stack named '{stack_name}' already exists.")
stack_name = q.ask("Enter a different stack name: ")
self.stack_name = stack_name
template_body = self._read_template()
print(f"Deploying CloudFormation stack: {self.stack_name}")
self.cfn_client.create_stack(
StackName=self.stack_name,
TemplateBody=template_body,
Capabilities=["CAPABILITY_IAM"],
)
print(f"CloudFormation stack creation started: {self.stack_name}")
print("Waiting for CloudFormation stack creation to complete...")
waiter = self.cfn_client.get_waiter("stack_create_complete")
try:
waiter.wait(
StackName=self.stack_name,
WaiterConfig={"Delay": 10, "MaxAttempts": 60},
)
except WaiterError:
logger.error("Stack creation failed or timed out.")
self._print_stack_events()
raise
print("CloudFormation stack creation complete.")
self._retrieve_stack_outputs()
print(SEPARATOR)
def _stack_exists(self, stack_name: str) -> bool:
"""Returns True if a stack with the given name exists and is not deleted."""
if not stack_name:
return False
try:
response = self.cfn_client.describe_stacks(StackName=stack_name)
stacks = response.get("Stacks", list())
for stack in stacks:
status = stack.get("StackStatus", "")
if "DELETE_COMPLETE" not in status:
return True
return False
except ClientError as err:
if "does not exist" in str(err):
return False
raise
def _read_template(self) -> str:
"""Reads the CloudFormation template from the local file system."""
template_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)), CFN_TEMPLATE_PATH
)
if not os.path.exists(template_path):
raise FileNotFoundError(
f"CloudFormation template not found at {template_path}. "
"Ensure the resources/cfn_template.yaml file exists."
)
with open(template_path, "r") as f:
return f.read()
def _retrieve_stack_outputs(self) -> None:
"""Retrieves and displays outputs from the CloudFormation stack."""
response = self.cfn_client.describe_stacks(StackName=self.stack_name)
stacks = response.get("Stacks", list())
if not stacks:
raise RuntimeError(f"Stack '{self.stack_name}' not found.")
outputs = stacks[0].get("Outputs", list())
for output in outputs:
key = output.get("OutputKey", "")
value = output.get("OutputValue", "")
print(f"Stack output {key}: {value}")
if key == "UserPoolId":
self.user_pool_id = value
elif key == "ClientId":
self.client_id = value
if not self.user_pool_id or not self.client_id:
raise RuntimeError(
"Failed to retrieve UserPoolId or ClientId from stack outputs."
)
def _print_stack_events(self) -> None:
"""Prints recent stack events to help diagnose failures."""
try:
response = self.cfn_client.describe_stack_events(StackName=self.stack_name)
events = response.get("StackEvents", list())
for event in events[:10]:
status = event.get("ResourceStatus", "")
reason = event.get("ResourceStatusReason", "")
resource = event.get("LogicalResourceId", "")
if "FAILED" in status:
print(f" FAILED: {resource} - {reason}")
except ClientError:
logger.warning("Could not retrieve stack events.")
# ------------------------------------------------------------------
# Phase 2: Retrieve the app client secret
# ------------------------------------------------------------------
def get_client_secret(self) -> None:
"""Calls DescribeUserPoolClient to retrieve the client secret."""
print(SEPARATOR)
print("Retrieving the app client secret...")
client_info = self.cognito_wrapper.describe_user_pool_client(
self.user_pool_id, self.client_id
)
self.client_secret = client_info.get("ClientSecret", "")
if not self.client_secret:
raise RuntimeError(
"Client secret not found. Ensure the app client was created "
"with GenerateSecret: true."
)
print("Client secret retrieved successfully.")
print(SEPARATOR)
# ------------------------------------------------------------------
# Phase 3: Demonstrate GetClientToken
# ------------------------------------------------------------------
def demonstrate_get_client_token(self) -> None:
"""Demonstrates obtaining M2M access tokens with GetClientToken."""
print(SEPARATOR)
print("Obtaining an M2M access token with GetClientToken...")
print()
# 3a: Get token with all scopes (no scopes param = all configured scopes)
print("Step 1: Request a token with all configured scopes.")
auth_result = self.cognito_wrapper.get_client_token(
client_id=self.client_id,
client_secret=self.client_secret,
)
self._display_token_result(auth_result, "all scopes")
q.ask("\nPress Enter to continue...")
# 3b: Get token with specific scopes
print()
print("Step 2: Request a token with a specific scope (my-m2m-api/read).")
scoped_result = self.cognito_wrapper.get_client_token(
client_id=self.client_id,
client_secret=self.client_secret,
scopes=["my-m2m-api/read"],
)
self._display_token_result(scoped_result, "my-m2m-api/read")
print()
print("Token was successfully scoped to the requested permissions.")
print(SEPARATOR)
@staticmethod
def _display_token_result(auth_result: dict, scope_description: str) -> None:
"""Displays token metadata."""
token_type = auth_result.get("TokenType", "Unknown")
expires_in = auth_result.get("ExpiresIn", 0)
access_token = auth_result.get("AccessToken", "")
preview = access_token[:20] + "..." if len(access_token) > 20 else access_token
print(f" Scope: {scope_description}")
print(f" Token type: {token_type}")
print(f" Expires in: {expires_in} seconds")
print(f" Token preview: {preview}")
# ------------------------------------------------------------------
# Phase 4: Demonstrate DescribeTermsByClient
# ------------------------------------------------------------------
def demonstrate_describe_terms(self) -> None:
"""Demonstrates discovering Terms documents with DescribeTermsByClient."""
print(SEPARATOR)
print("Discovering Terms documents with DescribeTermsByClient...")
print()
# 4a: Look up terms-of-use
print("Looking up terms-of-use for the app client...")
terms_of_use = self.cognito_wrapper.describe_terms_by_client(
user_pool_id=self.user_pool_id,
client_id=self.client_id,
terms_name="terms-of-use",
)
self._display_terms(terms_of_use)
q.ask("\nPress Enter to continue...")
print()
# 4b: Look up privacy-policy
print("Looking up privacy-policy for the app client...")
privacy_policy = self.cognito_wrapper.describe_terms_by_client(
user_pool_id=self.user_pool_id,
client_id=self.client_id,
terms_name="privacy-policy",
)
self._display_terms(privacy_policy)
print()
print(
"DescribeTermsByClient lets you discover which Terms documents are "
"associated with a client without knowing the Terms resource ID up front."
)
print(SEPARATOR)
@staticmethod
def _display_terms(terms: dict) -> None:
"""Displays a Terms document's details."""
print(f" Terms ID: {terms.get('TermsId', 'N/A')}")
print(f" Terms name: {terms.get('TermsName', 'N/A')}")
print(f" Client ID: {terms.get('ClientId', 'N/A')}")
print(f" User pool ID: {terms.get('UserPoolId', 'N/A')}")
print(f" Enforcement: {terms.get('Enforcement', 'N/A')}")
print(f" Terms source: {terms.get('TermsSource', 'N/A')}")
links = terms.get("Links", dict())
if links:
print(" Links:")
for lang_tag, url in links.items():
print(f" {lang_tag}: {url}")
# Cognito returns CreationDate / LastModifiedDate as datetime objects
# (boto3 deserializes the timestamps), so display them directly.
creation_date = terms.get("CreationDate", None)
if creation_date is not None:
print(f" Created: {creation_date}")
modified_date = terms.get("LastModifiedDate", None)
if modified_date is not None:
print(f" Last modified: {modified_date}")
# ------------------------------------------------------------------
# Phase 5: Cleanup - delete CloudFormation stack
# ------------------------------------------------------------------
def cleanup(self) -> None:
"""Prompts the user and deletes the CloudFormation stack."""
if self.stack_name is None:
return
print(SEPARATOR)
answer = q.ask(
"Do you want to delete the CloudFormation stack and all resources? (y/n) "
)
if answer.lower() != "y":
print(
f"Resources will remain. You can delete the stack later "
f"through the AWS Console or CLI."
)
print(f"Stack name: {self.stack_name}")
return
print(
f"CloudFormation stack '{self.stack_name}' is being deleted. "
f"This may take a few minutes."
)
try:
self.cfn_client.delete_stack(StackName=self.stack_name)
waiter = self.cfn_client.get_waiter("stack_delete_complete")
print("Waiting for CloudFormation stack deletion to complete...")
waiter.wait(
StackName=self.stack_name,
WaiterConfig={"Delay": 10, "MaxAttempts": 60},
)
print(f"CloudFormation stack '{self.stack_name}' has been deleted.")
print("All Cognito resources have been removed.")
except (ClientError, WaiterError):
logger.warning(
"Stack deletion encountered an error. Attempting force delete."
)
try:
self.cfn_client.delete_stack(
StackName=self.stack_name,
DeletionMode="FORCE_DELETE_STACK",
)
print("Force delete initiated. Please verify in the console.")
except ClientError:
logger.error(
"Force delete also failed. Clean up manually via the console."
)
```
Create a class that wraps Amazon Cognito Identity Provider operations.
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito Identity Provider actions."""
def __init__(self, cognito_idp_client: BaseClient):
"""
Initializes the CognitoIdentityProviderWrapper with an Amazon Cognito
Identity Provider client.
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
"""
self.cognito_idp_client = cognito_idp_client
```
+ For API details, see the following topics in *AWS SDK for Python (Boto3) API Reference*.
+ [DescribeTermsByClient](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/DescribeTermsByClient)
+ [DescribeUserPoolClient](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/DescribeUserPoolClient)
+ [GetClientToken](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/GetClientToken)
### Sign up a user with a user pool that requires MFA
The following code example shows how to:
+ Sign up and confirm a user with a username, password, and email address.
+ Set up multi-factor authentication by associating an MFA application with the user.
+ Sign in by using a password and an MFA code.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito#code-examples).
Create a class that wraps Amazon Cognito functions used in the scenario.
```
class CognitoIdentityProviderWrapper:
"""Encapsulates Amazon Cognito actions"""
def __init__(self, cognito_idp_client, user_pool_id, client_id, client_secret=None):
"""
:param cognito_idp_client: A Boto3 Amazon Cognito Identity Provider client.
:param user_pool_id: The ID of an existing Amazon Cognito user pool.
:param client_id: The ID of a client application registered with the user pool.
:param client_secret: The client secret, if the client has a secret.
"""
self.cognito_idp_client = cognito_idp_client
self.user_pool_id = user_pool_id
self.client_id = client_id
self.client_secret = client_secret
def _secret_hash(self, user_name):
"""
Calculates a secret hash from a user name and a client secret.
:param user_name: The user name to use when calculating the hash.
:return: The secret hash.
"""
key = self.client_secret.encode()
msg = bytes(user_name + self.client_id, "utf-8")
secret_hash = base64.b64encode(
hmac.new(key, msg, digestmod=hashlib.sha256).digest()
).decode()
logger.info("Made secret hash for %s: %s.", user_name, secret_hash)
return secret_hash
def sign_up_user(self, user_name, password, user_email):
"""
Signs up a new user with Amazon Cognito. This action prompts Amazon Cognito
to send an email to the specified email address. The email contains a code that
can be used to confirm the user.
When the user already exists, the user status is checked to determine whether
the user has been confirmed.
:param user_name: The user name that identifies the new user.
:param password: The password for the new user.
:param user_email: The email address for the new user.
:return: True when the user is already confirmed with Amazon Cognito.
Otherwise, false.
"""
try:
kwargs = {
"ClientId": self.client_id,
"Username": user_name,
"Password": password,
"UserAttributes": [{"Name": "email", "Value": user_email}],
}
if self.client_secret is not None:
kwargs["SecretHash"] = self._secret_hash(user_name)
response = self.cognito_idp_client.sign_up(**kwargs)
confirmed = response["UserConfirmed"]
except ClientError as err:
if err.response["Error"]["Code"] == "UsernameExistsException":
response = self.cognito_idp_client.admin_get_user(
UserPoolId=self.user_pool_id, Username=user_name
)
logger.warning(
"User %s exists and is %s.", user_name, response["UserStatus"]
)
confirmed = response["UserStatus"] == "CONFIRMED"
else:
logger.error(
"Couldn't sign up %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
return confirmed
def resend_confirmation(self, user_name):
"""
Prompts Amazon Cognito to resend an email with a new confirmation code.
:param user_name: The name of the user who will receive the email.
:return: Delivery information about where the email is sent.
"""
try:
kwargs = {"ClientId": self.client_id, "Username": user_name}
if self.client_secret is not None:
kwargs["SecretHash"] = self._secret_hash(user_name)
response = self.cognito_idp_client.resend_confirmation_code(**kwargs)
delivery = response["CodeDeliveryDetails"]
except ClientError as err:
logger.error(
"Couldn't resend confirmation to %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return delivery
def confirm_user_sign_up(self, user_name, confirmation_code):
"""
Confirms a previously created user. A user must be confirmed before they
can sign in to Amazon Cognito.
:param user_name: The name of the user to confirm.
:param confirmation_code: The confirmation code sent to the user's registered
email address.
:return: True when the confirmation succeeds.
"""
try:
kwargs = {
"ClientId": self.client_id,
"Username": user_name,
"ConfirmationCode": confirmation_code,
}
if self.client_secret is not None:
kwargs["SecretHash"] = self._secret_hash(user_name)
self.cognito_idp_client.confirm_sign_up(**kwargs)
except ClientError as err:
logger.error(
"Couldn't confirm sign up for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return True
def list_users(self):
"""
Returns a list of the users in the current user pool.
:return: The list of users.
"""
try:
response = self.cognito_idp_client.list_users(UserPoolId=self.user_pool_id)
users = response["Users"]
except ClientError as err:
logger.error(
"Couldn't list users for %s. Here's why: %s: %s",
self.user_pool_id,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return users
def start_sign_in(self, user_name, password):
"""
Starts the sign-in process for a user by using administrator credentials.
This method of signing in is appropriate for code running on a secure server.
If the user pool is configured to require MFA and this is the first sign-in
for the user, Amazon Cognito returns a challenge response to set up an
MFA application. When this occurs, this function gets an MFA secret from
Amazon Cognito and returns it to the caller.
:param user_name: The name of the user to sign in.
:param password: The user's password.
:return: The result of the sign-in attempt. When sign-in is successful, this
returns an access token that can be used to get AWS credentials. Otherwise,
Amazon Cognito returns a challenge to set up an MFA application,
or a challenge to enter an MFA code from a registered MFA application.
"""
try:
kwargs = {
"UserPoolId": self.user_pool_id,
"ClientId": self.client_id,
"AuthFlow": "ADMIN_USER_PASSWORD_AUTH",
"AuthParameters": {"USERNAME": user_name, "PASSWORD": password},
}
if self.client_secret is not None:
kwargs["AuthParameters"]["SECRET_HASH"] = self._secret_hash(user_name)
response = self.cognito_idp_client.admin_initiate_auth(**kwargs)
challenge_name = response.get("ChallengeName", None)
if challenge_name == "MFA_SETUP":
if (
"SOFTWARE_TOKEN_MFA"
in response["ChallengeParameters"]["MFAS_CAN_SETUP"]
):
response.update(self.get_mfa_secret(response["Session"]))
else:
raise RuntimeError(
"The user pool requires MFA setup, but the user pool is not "
"configured for TOTP MFA. This example requires TOTP MFA."
)
except ClientError as err:
logger.error(
"Couldn't start sign in for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
response.pop("ResponseMetadata", None)
return response
def get_mfa_secret(self, session):
"""
Gets a token that can be used to associate an MFA application with the user.
:param session: Session information returned from a previous call to initiate
authentication.
:return: An MFA token that can be used to set up an MFA application.
"""
try:
response = self.cognito_idp_client.associate_software_token(Session=session)
except ClientError as err:
logger.error(
"Couldn't get MFA secret. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
response.pop("ResponseMetadata", None)
return response
def verify_mfa(self, session, user_code):
"""
Verify a new MFA application that is associated with a user.
:param session: Session information returned from a previous call to initiate
authentication.
:param user_code: A code generated by the associated MFA application.
:return: Status that indicates whether the MFA application is verified.
"""
try:
response = self.cognito_idp_client.verify_software_token(
Session=session, UserCode=user_code
)
except ClientError as err:
logger.error(
"Couldn't verify MFA. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
response.pop("ResponseMetadata", None)
return response
def respond_to_mfa_challenge(self, user_name, session, mfa_code):
"""
Responds to a challenge for an MFA code. This completes the second step of
a two-factor sign-in. When sign-in is successful, it returns an access token
that can be used to get AWS credentials from Amazon Cognito.
:param user_name: The name of the user who is signing in.
:param session: Session information returned from a previous call to initiate
authentication.
:param mfa_code: A code generated by the associated MFA application.
:return: The result of the authentication. When successful, this contains an
access token for the user.
"""
try:
kwargs = {
"UserPoolId": self.user_pool_id,
"ClientId": self.client_id,
"ChallengeName": "SOFTWARE_TOKEN_MFA",
"Session": session,
"ChallengeResponses": {
"USERNAME": user_name,
"SOFTWARE_TOKEN_MFA_CODE": mfa_code,
},
}
if self.client_secret is not None:
kwargs["ChallengeResponses"]["SECRET_HASH"] = self._secret_hash(
user_name
)
response = self.cognito_idp_client.admin_respond_to_auth_challenge(**kwargs)
auth_result = response["AuthenticationResult"]
except ClientError as err:
if err.response["Error"]["Code"] == "ExpiredCodeException":
logger.warning(
"Your MFA code has expired or has been used already. You might have "
"to wait a few seconds until your app shows you a new code."
)
else:
logger.error(
"Couldn't respond to mfa challenge for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return auth_result
def confirm_mfa_device(
self,
user_name,
device_key,
device_group_key,
device_password,
access_token,
aws_srp,
):
"""
Confirms an MFA device to be tracked by Amazon Cognito. When a device is
tracked, its key and password can be used to sign in without requiring a new
MFA code from the MFA application.
:param user_name: The user that is associated with the device.
:param device_key: The key of the device, returned by Amazon Cognito.
:param device_group_key: The group key of the device, returned by Amazon Cognito.
:param device_password: The password that is associated with the device.
:param access_token: The user's access token.
:param aws_srp: A class that helps with Secure Remote Password (SRP)
calculations. The scenario associated with this example uses
the warrant package.
:return: True when the user must confirm the device. Otherwise, False. When
False, the device is automatically confirmed and tracked.
"""
srp_helper = aws_srp.AWSSRP(
username=user_name,
password=device_password,
pool_id="_",
client_id=self.client_id,
client_secret=None,
client=self.cognito_idp_client,
)
device_and_pw = f"{device_group_key}{device_key}:{device_password}"
device_and_pw_hash = aws_srp.hash_sha256(device_and_pw.encode("utf-8"))
salt = aws_srp.pad_hex(aws_srp.get_random(16))
x_value = aws_srp.hex_to_long(aws_srp.hex_hash(salt + device_and_pw_hash))
verifier = aws_srp.pad_hex(pow(srp_helper.val_g, x_value, srp_helper.big_n))
device_secret_verifier_config = {
"PasswordVerifier": base64.standard_b64encode(
bytearray.fromhex(verifier)
).decode("utf-8"),
"Salt": base64.standard_b64encode(bytearray.fromhex(salt)).decode("utf-8"),
}
try:
response = self.cognito_idp_client.confirm_device(
AccessToken=access_token,
DeviceKey=device_key,
DeviceSecretVerifierConfig=device_secret_verifier_config,
)
user_confirm = response["UserConfirmationNecessary"]
except ClientError as err:
logger.error(
"Couldn't confirm mfa device %s. Here's why: %s: %s",
device_key,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return user_confirm
def sign_in_with_tracked_device(
self,
user_name,
password,
device_key,
device_group_key,
device_password,
aws_srp,
):
"""
Signs in to Amazon Cognito as a user who has a tracked device. Signing in
with a tracked device lets a user sign in without entering a new MFA code.
Signing in with a tracked device requires that the client respond to the SRP
protocol. The scenario associated with this example uses the warrant package
to help with SRP calculations.
For more information on SRP, see https://en.wikipedia.org/wiki/Secure_Remote_Password_protocol.
:param user_name: The user that is associated with the device.
:param password: The user's password.
:param device_key: The key of a tracked device.
:param device_group_key: The group key of a tracked device.
:param device_password: The password that is associated with the device.
:param aws_srp: A class that helps with SRP calculations. The scenario
associated with this example uses the warrant package.
:return: The result of the authentication. When successful, this contains an
access token for the user.
"""
try:
srp_helper = aws_srp.AWSSRP(
username=user_name,
password=device_password,
pool_id="_",
client_id=self.client_id,
client_secret=None,
client=self.cognito_idp_client,
)
response_init = self.cognito_idp_client.initiate_auth(
ClientId=self.client_id,
AuthFlow="USER_PASSWORD_AUTH",
AuthParameters={
"USERNAME": user_name,
"PASSWORD": password,
"DEVICE_KEY": device_key,
},
)
if response_init["ChallengeName"] != "DEVICE_SRP_AUTH":
raise RuntimeError(
f"Expected DEVICE_SRP_AUTH challenge but got {response_init['ChallengeName']}."
)
auth_params = srp_helper.get_auth_params()
auth_params["DEVICE_KEY"] = device_key
response_auth = self.cognito_idp_client.respond_to_auth_challenge(
ClientId=self.client_id,
ChallengeName="DEVICE_SRP_AUTH",
ChallengeResponses=auth_params,
)
if response_auth["ChallengeName"] != "DEVICE_PASSWORD_VERIFIER":
raise RuntimeError(
f"Expected DEVICE_PASSWORD_VERIFIER challenge but got "
f"{response_init['ChallengeName']}."
)
challenge_params = response_auth["ChallengeParameters"]
challenge_params["USER_ID_FOR_SRP"] = device_group_key + device_key
cr = srp_helper.process_challenge(challenge_params, {"USERNAME": user_name})
cr["USERNAME"] = user_name
cr["DEVICE_KEY"] = device_key
response_verifier = self.cognito_idp_client.respond_to_auth_challenge(
ClientId=self.client_id,
ChallengeName="DEVICE_PASSWORD_VERIFIER",
ChallengeResponses=cr,
)
auth_tokens = response_verifier["AuthenticationResult"]
except ClientError as err:
logger.error(
"Couldn't start client sign in for %s. Here's why: %s: %s",
user_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return auth_tokens
```
Create a class that runs the scenario. This example also registers an MFA device to be tracked by Amazon Cognito and shows you how to sign in by using a password and information from the tracked device. This avoids the need to enter a new MFA code.
```
def run_scenario(cognito_idp_client, user_pool_id, client_id):
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
print("-" * 88)
print("Welcome to the Amazon Cognito user signup with MFA demo.")
print("-" * 88)
cog_wrapper = CognitoIdentityProviderWrapper(
cognito_idp_client, user_pool_id, client_id
)
user_name = q.ask("Let's sign up a new user. Enter a user name: ", q.non_empty)
password = q.ask("Enter a password for the user: ", q.non_empty)
email = q.ask("Enter a valid email address that you own: ", q.non_empty)
confirmed = cog_wrapper.sign_up_user(user_name, password, email)
while not confirmed:
print(
f"User {user_name} requires confirmation. Check {email} for "
f"a verification code."
)
confirmation_code = q.ask("Enter the confirmation code from the email: ")
if not confirmation_code:
if q.ask("Do you need another confirmation code (y/n)? ", q.is_yesno):
delivery = cog_wrapper.resend_confirmation(user_name)
print(
f"Confirmation code sent by {delivery['DeliveryMedium']} "
f"to {delivery['Destination']}."
)
else:
confirmed = cog_wrapper.confirm_user_sign_up(user_name, confirmation_code)
print(f"User {user_name} is confirmed and ready to use.")
print("-" * 88)
print("Let's get a list of users in the user pool.")
q.ask("Press Enter when you're ready.")
users = cog_wrapper.list_users()
if users:
print(f"Found {len(users)} users:")
pp(users)
else:
print("No users found.")
print("-" * 88)
print("Let's sign in and get an access token.")
auth_tokens = None
challenge = "ADMIN_USER_PASSWORD_AUTH"
response = {}
while challenge is not None:
if challenge == "ADMIN_USER_PASSWORD_AUTH":
response = cog_wrapper.start_sign_in(user_name, password)
# start_sign_in may return an AuthenticationResult with no challenge
# (for example, when the user pool does not require MFA). Read the
# challenge defensively and stop if sign-in already succeeded.
challenge = response.get("ChallengeName")
if challenge is None:
auth_tokens = response.get("AuthenticationResult")
if auth_tokens is not None:
print(f"You're signed in as {user_name}.")
print("Here's your access token:")
pp(auth_tokens["AccessToken"])
break
elif response["ChallengeName"] == "MFA_SETUP":
print("First, we need to set up an MFA application.")
qr_img = qrcode.make(
f"otpauth://totp/{user_name}?secret={response['SecretCode']}"
)
qr_img.save("qr.png")
q.ask(
"Press Enter to see a QR code on your screen. Scan it into an MFA "
"application, such as Google Authenticator."
)
webbrowser.open("qr.png")
mfa_code = q.ask(
"Enter the verification code from your MFA application: ", q.non_empty
)
response = cog_wrapper.verify_mfa(response["Session"], mfa_code)
print(f"MFA device setup {response['Status']}")
print("Now that an MFA application is set up, let's sign in again.")
print(
"You might have to wait a few seconds for a new MFA code to appear in "
"your MFA application."
)
challenge = "ADMIN_USER_PASSWORD_AUTH"
elif response["ChallengeName"] == "SOFTWARE_TOKEN_MFA":
auth_tokens = None
while auth_tokens is None:
mfa_code = q.ask(
"Enter a verification code from your MFA application: ", q.non_empty
)
auth_tokens = cog_wrapper.respond_to_mfa_challenge(
user_name, response["Session"], mfa_code
)
print(f"You're signed in as {user_name}.")
print("Here's your access token:")
pp(auth_tokens["AccessToken"])
if "NewDeviceMetadata" in auth_tokens:
print("And your device information:")
pp(auth_tokens["NewDeviceMetadata"])
challenge = None
else:
raise Exception(f"Got unexpected challenge {response['ChallengeName']}")
print("-" * 88)
# If sign-in returned neither a challenge nor an AuthenticationResult, there
# are no tokens to work with; nothing more to do.
if auth_tokens is None:
return user_name
# respond_to_mfa_challenge returns NewDeviceMetadata only when the user pool
# has device tracking configured. When it's absent, skip the device-confirmation
# and tracked-device sign-in steps instead of failing on a missing key.
device_metadata = auth_tokens.get("NewDeviceMetadata")
if device_metadata is None:
print(
"This user pool doesn't track devices, so there's no device to confirm. "
"To try the tracked-device sign-in flow, enable device tracking on the "
"user pool and run this example again."
)
print("-" * 88)
else:
device_group_key = device_metadata["DeviceGroupKey"]
device_key = device_metadata["DeviceKey"]
device_password = base64.standard_b64encode(os.urandom(40)).decode("utf-8")
print(
"Let's confirm your MFA device so you don't have to re-enter MFA tokens for it."
)
q.ask("Press Enter when you're ready.")
cog_wrapper.confirm_mfa_device(
user_name,
device_key,
device_group_key,
device_password,
auth_tokens["AccessToken"],
aws_srp,
)
print(f"Your device {device_key} is confirmed.")
print("-" * 88)
print(
f"Now let's sign in as {user_name} from your confirmed device {device_key}.\n"
f"Because this device is tracked by Amazon Cognito, you won't have to re-enter an MFA code."
)
q.ask("Press Enter when ready.")
auth_tokens = cog_wrapper.sign_in_with_tracked_device(
user_name, password, device_key, device_group_key, device_password, aws_srp
)
print("You're signed in. Your access token is:")
pp(auth_tokens["AccessToken"])
print("-" * 88)
print("Don't forget to delete your user pool when you're done with this example.")
print("\nThanks for watching!")
print("-" * 88)
return user_name
def main():
parser = argparse.ArgumentParser(
description="Shows how to sign up a new user with Amazon Cognito and associate "
"the user with an MFA application for multi-factor authentication."
)
parser.add_argument(
"user_pool_id", help="The ID of the user pool to use for the example."
)
parser.add_argument(
"client_id", help="The ID of the client application to use for the example."
)
args = parser.parse_args()
client = boto3.client("cognito-idp")
created_user = None
try:
created_user = run_scenario(client, args.user_pool_id, args.client_id)
except Exception:
logging.exception("Something went wrong with the demo.")
finally:
# Clean up the user this scenario created, on success or failure, so a
# run never leaves an orphaned user in the pool. Cleanup errors are
# logged but do not mask the original outcome.
if created_user is not None:
try:
client.admin_delete_user(
UserPoolId=args.user_pool_id, Username=created_user
)
print(f"Deleted user {created_user}.")
except Exception:
logging.exception(
"Couldn't delete user %s; delete it manually.", created_user
)
if __name__ == "__main__":
main()
```
+ For API details, see the following topics in *AWS SDK for Python (Boto3) API Reference*.
+ [AdminGetUser](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/AdminGetUser)
+ [AdminInitiateAuth](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/AdminInitiateAuth)
+ [AdminRespondToAuthChallenge](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/AdminRespondToAuthChallenge)
+ [AssociateSoftwareToken](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/AssociateSoftwareToken)
+ [ConfirmDevice](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ConfirmDevice)
+ [ConfirmSignUp](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ConfirmSignUp)
+ [InitiateAuth](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/InitiateAuth)
+ [ListUsers](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ListUsers)
+ [ResendConfirmationCode](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/ResendConfirmationCode)
+ [RespondToAuthChallenge](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/RespondToAuthChallenge)
+ [SignUp](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/SignUp)
+ [VerifySoftwareToken](https://docs.aws.amazon.com/goto/boto3/cognito-idp-2016-04-18/VerifySoftwareToken)
### Use Amazon Cognito identity pools
The following code example shows how to create a web-based demo application that demonstrates identity pools authentication flows.
**SDK for Python (Boto3)**
Shows web-based demo application that demonstrates Amazon Cognito identity pools authentication flows, allowing users to interactively explore both enhanced and basic authentication flows with various identity providers.
For complete source code and instructions on how to set up and run, see the full example on [ GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cognito/scenarios/identity_pools_example_demo).
**Services used in this example**
+ Amazon Cognito Identity Provider