diff --git a/.github/workflows/tests_servermode.yml b/.github/workflows/tests_servermode.yml index c69590d2189b..63a9ced1e1d1 100644 --- a/.github/workflows/tests_servermode.yml +++ b/.github/workflows/tests_servermode.yml @@ -21,7 +21,7 @@ jobs: run: | pip install build python -m build - docker run --rm -t --name motoserver -e TEST_SERVER_MODE=true -e MOTO_EC2_LOAD_DEFAULT_AMIS=false -e AWS_SECRET_ACCESS_KEY=server_secret -e AWS_ACCESS_KEY_ID=server_key -v `pwd`:/moto -p 5000:5000 -v /var/run/docker.sock:/var/run/docker.sock python:${{ matrix.python-version }}-slim /moto/scripts/ci_moto_server.sh & + docker run --rm -t --name motoserver -e TEST_SERVER_MODE=true -e MOTO_EC2_LOAD_DEFAULT_AMIS=false -e MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP=true -e AWS_SECRET_ACCESS_KEY=server_secret -e AWS_ACCESS_KEY_ID=server_key -v `pwd`:/moto -p 5000:5000 -v /var/run/docker.sock:/var/run/docker.sock python:${{ matrix.python-version }}-slim /moto/scripts/ci_moto_server.sh & python scripts/ci_wait_for_server.py - name: Get pip cache dir id: pip-cache diff --git a/moto/cognitoidp/exceptions.py b/moto/cognitoidp/exceptions.py index b0bdac427b28..a4f09dad1729 100644 --- a/moto/cognitoidp/exceptions.py +++ b/moto/cognitoidp/exceptions.py @@ -59,3 +59,8 @@ def __init__(self) -> None: error_type="InvalidPasswordException", message="The provided password does not confirm to the configured password policy", ) + + +class CodeMismatchException(JsonRESTError): + def __init__(self, message: str): + super().__init__(error_type="CodeMismatchException", message=message) diff --git a/moto/cognitoidp/models.py b/moto/cognitoidp/models.py index 7bf26f742d2b..e1c7bba6c9e4 100644 --- a/moto/cognitoidp/models.py +++ b/moto/cognitoidp/models.py @@ -2,8 +2,9 @@ import re import time from collections import OrderedDict -from typing import Any, Optional +from typing import Any, Final, Optional +from cryptography.hazmat.primitives.twofactor import InvalidToken from joserfc import jwk, jwt from moto.core.base_backend import BackendDict, BaseBackend @@ -15,10 +16,12 @@ from ..settings import ( get_cognito_idp_user_pool_client_id_strategy, + get_cognito_idp_user_pool_enable_totp, get_cognito_idp_user_pool_id_strategy, ) from .exceptions import ( AliasExistsException, + CodeMismatchException, ExpiredCodeException, GroupExistsException, InvalidParameterException, @@ -32,12 +35,16 @@ from .utils import ( PAGINATION_MODEL, check_secret_hash, + cognito_totp, expand_attrs, flatten_attrs, generate_id, validate_username_format, ) +# FIXME: Should be per user and stored in the user's profile +COGNITO_TOTP_MFA_SECRET: Final[str] = "asdfasdfasdf" + class UserStatus(str, enum.Enum): FORCE_CHANGE_PASSWORD = "FORCE_CHANGE_PASSWORD" @@ -970,8 +977,12 @@ class CognitoIdpBackend(BaseBackend): In some cases, you need to have reproducible IDs for the user pool. For example, a single initialization before the start of integration tests. - This behavior can be enabled by passing the environment variable: MOTO_COGNITO_IDP_USER_POOL_ID_STRATEGY=HASH. - Passing MOTO_COGNITO_IDP_USER_POOL_CLIENT_ID_STRATEGY=HASH enables the same logic for user pool clients. + This behavior can be enabled by passing the environment variable: `MOTO_COGNITO_IDP_USER_POOL_ID_STRATEGY=HASH`. + Passing `MOTO_COGNITO_IDP_USER_POOL_CLIENT_ID_STRATEGY=HASH` enables the same logic for user pool clients. + + Support for MFA TOTP can be enabled by setting `MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP=true`. + Moto will validate the TOTP MFA provided by the user when registering MFA or subsequently authenticating. + At this time, Moto uses a single fixed secret across all users. """ def __init__(self, region_name: str, account_id: str): @@ -1720,6 +1731,16 @@ def respond_to_auth_challenge( ): raise NotAuthorizedError(secret_hash) + if ( + challenge_name == "SOFTWARE_TOKEN_MFA" + and get_cognito_idp_user_pool_enable_totp() + ): + totp = cognito_totp(COGNITO_TOTP_MFA_SECRET) + try: + totp.verify(mfa_code.encode("utf-8"), int(time.time())) + except InvalidToken: + raise CodeMismatchException("MFA Code Mismatch") + del self.sessions[session] return self._log_user_in(user_pool, client, username) @@ -2173,7 +2194,7 @@ def initiate_auth( def associate_software_token( self, access_token: str, session: str ) -> dict[str, str]: - secret_code = "asdfasdfasdf" + secret_code = COGNITO_TOTP_MFA_SECRET if session: if session in self.sessions: return {"SecretCode": secret_code, "Session": session} @@ -2188,16 +2209,25 @@ def associate_software_token( raise NotAuthorizedError(access_token) - def verify_software_token(self, access_token: str, session: str) -> dict[str, str]: - """ - The parameter UserCode has not yet been implemented - """ + def verify_software_token( + self, access_token: str, session: str, user_code: str, friendly_device_name: str + ) -> dict[str, str]: + totp = cognito_totp(COGNITO_TOTP_MFA_SECRET) if session: if session not in self.sessions: raise ResourceNotFoundError(session) username, user_pool = self.sessions[session] user = self.admin_get_user(user_pool.id, username) + + if get_cognito_idp_user_pool_enable_totp(): + try: + totp.verify(user_code.encode("utf-8"), int(time.time())) + except InvalidToken: + raise CodeMismatchException( + f"Code mismatch ({friendly_device_name})" + ) + user.token_verified = True session = str(random.uuid4()) @@ -2209,6 +2239,14 @@ def verify_software_token(self, access_token: str, session: str) -> dict[str, st _, username = user_pool.access_tokens[access_token] user = self.admin_get_user(user_pool.id, username) + if get_cognito_idp_user_pool_enable_totp(): + try: + totp.verify(user_code.encode("utf-8"), int(time.time())) + except InvalidToken: + raise CodeMismatchException( + f"Code mismatch ({friendly_device_name})" + ) + user.token_verified = True session = str(random.uuid4()) @@ -2454,9 +2492,17 @@ def associate_software_token( backend = self._find_backend_by_access_token_or_session(access_token, session) return backend.associate_software_token(access_token, session) - def verify_software_token(self, access_token: str, session: str) -> dict[str, str]: + def verify_software_token( + self, + access_token: str, + session: str, + user_code: str, + friendly_device_name: str, + ) -> dict[str, str]: backend = self._find_backend_by_access_token_or_session(access_token, session) - return backend.verify_software_token(access_token, session) + return backend.verify_software_token( + access_token, session, user_code, friendly_device_name + ) def set_user_mfa_preference( self, diff --git a/moto/cognitoidp/responses.py b/moto/cognitoidp/responses.py index 80f91c88f0c7..f089515c0fc4 100644 --- a/moto/cognitoidp/responses.py +++ b/moto/cognitoidp/responses.py @@ -585,8 +585,10 @@ def associate_software_token(self) -> ActionResult: def verify_software_token(self) -> ActionResult: access_token = self._get_param("AccessToken") session = self._get_param("Session") + user_code = self._get_param("UserCode") + friendly_device_name = self._get_param("FriendlyDeviceName") result = self._get_region_agnostic_backend().verify_software_token( - access_token, session + access_token, session, user_code, friendly_device_name ) return ActionResult(result) diff --git a/moto/cognitoidp/utils.py b/moto/cognitoidp/utils.py index 0165ee4dd0d0..bc768eac5cd9 100644 --- a/moto/cognitoidp/utils.py +++ b/moto/cognitoidp/utils.py @@ -5,6 +5,9 @@ import string from typing import Any, Optional +from cryptography.hazmat.primitives.hashes import SHA1 +from cryptography.hazmat.primitives.twofactor.totp import TOTP + from moto.moto_api._internal import mock_random as random FORMATS = { @@ -120,3 +123,19 @@ def _generate_id_hash(args: Any) -> str: hasher.update(str(arg).encode()) return hasher.hexdigest() + + +def cognito_totp(key: str) -> TOTP: + key_padded = key + # Pad the secret if required before converting it to bytes + padding = len(key) % 8 + if padding != 0: + key_padded += "=" * (8 - padding) + # https://docs.aws.amazon.com/cognito/latest/developerguide/user-pool-settings-mfa-totp.html + return TOTP( + key=base64.b32decode(key_padded, casefold=True), + length=6, + algorithm=SHA1(), + time_step=30, + enforce_key_length=False, + ) diff --git a/moto/settings.py b/moto/settings.py index 6ac3b2fb51d5..8bbd3ce49fa1 100644 --- a/moto/settings.py +++ b/moto/settings.py @@ -190,6 +190,13 @@ def get_cognito_idp_user_pool_client_id_strategy() -> Optional[str]: return os.environ.get("MOTO_COGNITO_IDP_USER_POOL_CLIENT_ID_STRATEGY") +def get_cognito_idp_user_pool_enable_totp() -> bool: + return ( + os.environ.get("MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP", "false").lower() + == "true" + ) + + def enable_iso_regions() -> bool: return os.environ.get("MOTO_ENABLE_ISO_REGIONS", "false").lower() == "true" diff --git a/tests/test_cognitoidp/test_cognitoidp.py b/tests/test_cognitoidp/test_cognitoidp.py index 6c114a27b8b4..e98ffc9ef25b 100644 --- a/tests/test_cognitoidp/test_cognitoidp.py +++ b/tests/test_cognitoidp/test_cognitoidp.py @@ -536,7 +536,7 @@ def test_create_user_pool_default_id_strategy(): @mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ID_STRATEGY": "HASH"}) def test_create_user_pool_hash_id_strategy_with_equal_pool_name(): if settings.TEST_SERVER_MODE: - raise SkipTest("Cannot set environemnt variables in ServerMode") + raise SkipTest("Cannot set environment variables in ServerMode") conn = boto3.client("cognito-idp", "us-west-2") @@ -550,7 +550,7 @@ def test_create_user_pool_hash_id_strategy_with_equal_pool_name(): @mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ID_STRATEGY": "HASH"}) def test_create_user_pool_hash_id_strategy_with_different_pool_name(): if settings.TEST_SERVER_MODE: - raise SkipTest("Cannot set environemnt variables in ServerMode") + raise SkipTest("Cannot set environment variables in ServerMode") conn = boto3.client("cognito-idp", "us-west-2") @@ -564,7 +564,7 @@ def test_create_user_pool_hash_id_strategy_with_different_pool_name(): @mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ID_STRATEGY": "HASH"}) def test_create_user_pool_hash_id_strategy_with_different_attributes(): if settings.TEST_SERVER_MODE: - raise SkipTest("Cannot set environemnt variables in ServerMode") + raise SkipTest("Cannot set environment variables in ServerMode") conn = boto3.client("cognito-idp", "us-west-2") @@ -1002,7 +1002,7 @@ def test_create_user_pool_client_default_id_strategy(): @mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_CLIENT_ID_STRATEGY": "HASH"}) def test_create_user_pool_client_hash_id_strategy_with_equal_args(): if settings.TEST_SERVER_MODE: - raise SkipTest("Cannot set environemnt variables in ServerMode") + raise SkipTest("Cannot set environment variables in ServerMode") conn = boto3.client("cognito-idp", "us-west-2") @@ -1024,7 +1024,7 @@ def test_create_user_pool_client_hash_id_strategy_with_equal_args(): @mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_CLIENT_ID_STRATEGY": "HASH"}) def test_create_user_pool_client_hash_id_strategy_with_different_args(): if settings.TEST_SERVER_MODE: - raise SkipTest("Cannot set environemnt variables in ServerMode") + raise SkipTest("Cannot set environment variables in ServerMode") conn = boto3.client("cognito-idp", "us-west-2") @@ -3210,6 +3210,7 @@ def user_authentication_flow( refresh_token = result["AuthenticationResult"]["RefreshToken"] # add mfa token + secret_code = None if with_mfa: resp = conn.associate_software_token( AccessToken=result["AuthenticationResult"]["AccessToken"] @@ -3284,6 +3285,7 @@ def user_authentication_flow( "client_id": client_id, "client_secret": client_secret, "secret_hash": secret_hash, + "secret_code": secret_code, "id_token": result["AuthenticationResult"]["IdToken"], "access_token": result["AuthenticationResult"]["AccessToken"], "refresh_token": refresh_token, @@ -3294,6 +3296,7 @@ def user_authentication_flow( @cognitoidp_aws_verified(generate_secret=True, with_mfa="ON") +@mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP": "true"}) @pytest.mark.aws_verified def test_user_authentication_flow_mfa_on(user_pool=None, user_pool_client=None): conn = boto3.client("cognito-idp", "us-west-2") @@ -3397,6 +3400,7 @@ def test_user_authentication_flow_mfa_on(user_pool=None, user_pool_client=None): @cognitoidp_aws_verified(generate_secret=True, with_mfa="OPTIONAL") +@mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP": "true"}) @pytest.mark.aws_verified def test_user_authentication_flow_mfa_optional(user_pool=None, user_pool_client=None): conn = boto3.client("cognito-idp", "us-west-2") @@ -4840,6 +4844,7 @@ def test_initiate_auth_USER_PASSWORD_AUTH_when_software_token_mfa_enabled(): password = result["password"] client_id = result["client_id"] secret_hash = result["secret_hash"] + secret_code = result["secret_code"] result = conn.admin_get_user(UserPoolId=user_pool_id, Username=username) assert result["PreferredMfaSetting"] == "SOFTWARE_TOKEN_MFA" @@ -4854,12 +4859,15 @@ def test_initiate_auth_USER_PASSWORD_AUTH_when_software_token_mfa_enabled(): assert result["ChallengeParameters"] == {} assert result["Session"] is not None + totp = pyotp.TOTP(secret_code) + user_code = totp.now() + result = conn.respond_to_auth_challenge( ClientId=client_id, ChallengeName="SOFTWARE_TOKEN_MFA", Session=result["Session"], ChallengeResponses={ - "SOFTWARE_TOKEN_MFA_CODE": "123456", + "SOFTWARE_TOKEN_MFA_CODE": user_code, "USERNAME": username, "SECRET_HASH": secret_hash, }, @@ -5019,6 +5027,7 @@ def test_initiate_auth_USER_PASSWORD_AUTH_with_FORCE_CHANGE_PASSWORD_status(): @cognitoidp_aws_verified(explicit_auth_flows=["USER_PASSWORD_AUTH"], with_mfa="ON") +@mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP": "true"}) @pytest.mark.aws_verified def test_initiate_mfa_auth_USER_PASSWORD_AUTH_with_FORCE_CHANGE_PASSWORD_status( user_pool=None, user_pool_client=None @@ -5241,6 +5250,7 @@ def test_initiate_auth_with_invalid_secret_hash(): @mock_aws +@mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP": "true"}) def test_setting_mfa(): conn = boto3.client("cognito-idp", "us-west-2") @@ -5248,9 +5258,13 @@ def test_setting_mfa(): result = authentication_flow(conn, auth_flow) # Set MFA method - conn.associate_software_token(AccessToken=result["access_token"]) + resp = conn.associate_software_token(AccessToken=result["access_token"]) + secret_code = resp["SecretCode"] + totp = pyotp.TOTP(secret_code) + user_code = totp.now() + conn.verify_software_token( - AccessToken=result["access_token"], UserCode="123456" + AccessToken=result["access_token"], UserCode=user_code ) conn.set_user_mfa_preference( AccessToken=result["access_token"], @@ -5325,6 +5339,7 @@ def test_admin_setting_single_mfa(): @mock_aws +@mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP": "true"}) def test_admin_setting_mfa_totp_and_sms(): conn = boto3.client("cognito-idp", "us-west-2") @@ -5332,8 +5347,12 @@ def test_admin_setting_mfa_totp_and_sms(): access_token = result["access_token"] user_pool_id = result["user_pool_id"] username = result["username"] - conn.associate_software_token(AccessToken=access_token) - conn.verify_software_token(AccessToken=access_token, UserCode="123456") + resp = conn.associate_software_token(AccessToken=access_token) + secret_code = resp["SecretCode"] + totp = pyotp.TOTP(secret_code) + user_code = totp.now() + + conn.verify_software_token(AccessToken=access_token, UserCode=user_code) # Set MFA TOTP and SMS methods conn.admin_set_user_mfa_preference( @@ -5359,6 +5378,7 @@ def test_admin_setting_mfa_totp_and_sms(): @mock_aws +@mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP": "true"}) def test_admin_initiate_auth_when_token_totp_enabled(): conn = boto3.client("cognito-idp", "us-west-2") @@ -5368,8 +5388,73 @@ def test_admin_initiate_auth_when_token_totp_enabled(): username = result["username"] client_id = result["client_id"] password = result["password"] - conn.associate_software_token(AccessToken=access_token) - conn.verify_software_token(AccessToken=access_token, UserCode="123456") + resp = conn.associate_software_token(AccessToken=access_token) + secret_code = resp["SecretCode"] + totp = pyotp.TOTP(secret_code) + user_code = totp.now() + conn.verify_software_token(AccessToken=access_token, UserCode=user_code) + + # Set MFA TOTP and SMS methods + conn.admin_set_user_mfa_preference( + Username=username, + UserPoolId=user_pool_id, + SoftwareTokenMfaSettings={"Enabled": True, "PreferredMfa": True}, + SMSMfaSettings={"Enabled": True, "PreferredMfa": False}, + ) + result = conn.admin_get_user(UserPoolId=user_pool_id, Username=username) + assert len(result["UserMFASettingList"]) == 2 + assert result["PreferredMfaSetting"] == "SOFTWARE_TOKEN_MFA" + + # Initiate auth with TOTP + result = conn.admin_initiate_auth( + UserPoolId=user_pool_id, + ClientId=client_id, + AuthFlow="ADMIN_NO_SRP_AUTH", + AuthParameters={ + "USERNAME": username, + "PASSWORD": password, + }, + ) + + assert result["ChallengeName"] == "SOFTWARE_TOKEN_MFA" + assert result["Session"] != "" + + # Respond to challenge with TOTP + result = conn.admin_respond_to_auth_challenge( + UserPoolId=user_pool_id, + ClientId=client_id, + ChallengeName="SOFTWARE_TOKEN_MFA", + Session=result["Session"], + ChallengeResponses={ + "SOFTWARE_TOKEN_MFA_CODE": totp.now(), + "USERNAME": username, + }, + ) + + assert result["AuthenticationResult"]["IdToken"] != "" + assert result["AuthenticationResult"]["AccessToken"] != "" + assert result["AuthenticationResult"]["RefreshToken"] != "" + assert result["AuthenticationResult"]["TokenType"] == "Bearer" + + +@mock_aws +def test_admin_initiate_auth_when_token_totp_disabled(): + if settings.TEST_SERVER_MODE: + raise SkipTest("TOTP is enabled in server mode") + + conn = boto3.client("cognito-idp", "us-west-2") + + result = authentication_flow(conn, "ADMIN_NO_SRP_AUTH") + access_token = result["access_token"] + user_pool_id = result["user_pool_id"] + username = result["username"] + client_id = result["client_id"] + password = result["password"] + resp = conn.associate_software_token(AccessToken=access_token) + secret_code = resp["SecretCode"] + totp = pyotp.TOTP(secret_code) + user_code = totp.now() + conn.verify_software_token(AccessToken=access_token, UserCode=user_code) # Set MFA TOTP and SMS methods conn.admin_set_user_mfa_preference( @@ -5414,6 +5499,63 @@ def test_admin_initiate_auth_when_token_totp_enabled(): assert result["AuthenticationResult"]["TokenType"] == "Bearer" +@mock_aws +@mock.patch.dict(os.environ, {"MOTO_COGNITO_IDP_USER_POOL_ENABLE_TOTP": "true"}) +def test_admin_initiate_auth_when_token_totp_enabled_invalid(): + conn = boto3.client("cognito-idp", "us-west-2") + + result = authentication_flow(conn, "ADMIN_NO_SRP_AUTH") + access_token = result["access_token"] + user_pool_id = result["user_pool_id"] + username = result["username"] + client_id = result["client_id"] + password = result["password"] + resp = conn.associate_software_token(AccessToken=access_token) + secret_code = resp["SecretCode"] + totp = pyotp.TOTP(secret_code) + user_code = totp.now() + conn.verify_software_token(AccessToken=access_token, UserCode=user_code) + + # Set MFA TOTP and SMS methods + conn.admin_set_user_mfa_preference( + Username=username, + UserPoolId=user_pool_id, + SoftwareTokenMfaSettings={"Enabled": True, "PreferredMfa": True}, + SMSMfaSettings={"Enabled": True, "PreferredMfa": False}, + ) + result = conn.admin_get_user(UserPoolId=user_pool_id, Username=username) + assert len(result["UserMFASettingList"]) == 2 + assert result["PreferredMfaSetting"] == "SOFTWARE_TOKEN_MFA" + + # Initiate auth with TOTP + result = conn.admin_initiate_auth( + UserPoolId=user_pool_id, + ClientId=client_id, + AuthFlow="ADMIN_NO_SRP_AUTH", + AuthParameters={ + "USERNAME": username, + "PASSWORD": password, + }, + ) + + assert result["ChallengeName"] == "SOFTWARE_TOKEN_MFA" + assert result["Session"] != "" + + with pytest.raises(ClientError) as exc: + result = conn.admin_respond_to_auth_challenge( + UserPoolId=user_pool_id, + ClientId=client_id, + ChallengeName="SOFTWARE_TOKEN_MFA", + Session=result["Session"], + ChallengeResponses={ + "SOFTWARE_TOKEN_MFA_CODE": "123456", + "USERNAME": username, + }, + ) + err = exc.value.response["Error"] + assert err["Code"] == "CodeMismatchException" + + @mock_aws def test_admin_initiate_auth_when_sms_mfa_enabled(): conn = boto3.client("cognito-idp", "us-west-2") diff --git a/tests/test_cognitoidp/test_cognitoidp_utils.py b/tests/test_cognitoidp/test_cognitoidp_utils.py new file mode 100644 index 000000000000..67f141c00c6e --- /dev/null +++ b/tests/test_cognitoidp/test_cognitoidp_utils.py @@ -0,0 +1,16 @@ +import time + +import pyotp + +from moto.cognitoidp.models import COGNITO_TOTP_MFA_SECRET +from moto.cognitoidp.utils import cognito_totp + + +def test_cognito_totp(): + client_totp = pyotp.TOTP(s=COGNITO_TOTP_MFA_SECRET) + internal_totp = cognito_totp(COGNITO_TOTP_MFA_SECRET) + + client_code = client_totp.now() + internal_code = internal_totp.generate(int(time.time())).decode("utf-8") + + assert internal_code == client_code