CognitoIDP: ID-token is different when username=email (#7927)

moto/cognitoidp/models.py

@@ -538,8 +538,22 @@ def create_jwt(
             "token_use": token_use,
             "auth_time": now,
             "exp": now + expires_in,
-            "username" if token_use == "access" else "cognito:username": username,
         }
+        username_is_email = "email" in self.extended_config.get(
+            "UsernameAttributes", []
+        )
+        if token_use == "access":
+            if username_is_email:
+                payload["username"] = payload["sub"]
+            else:
+                payload["username"] = username
+        if token_use == "id":
+            if username_is_email:
+                payload["cognito:username"] = payload["sub"]
+                payload["email"] = username
+            else:
+                payload["cognito:username"] = username
+
         payload.update(extra_data or {})
         headers = {"kid": "dummy", "alg": "RS256"}  # KID as present in jwks-public.json
 

requirements-tests.txt

@@ -1,4 +1,5 @@
 coverage
+pycognito
 pytest
 pytest-cov
 pytest-order

tests/test_cognitoidp/__init__.py

@@ -1 +1,75 @@
-# This file is intentionally left blank.
+import os
+from functools import wraps
+from uuid import uuid4
+
+import boto3
+
+from moto import mock_aws
+
+
+def cognitoidp_aws_verified(
+    username_attributes=None,
+    read_attributes=None,
+    explicit_auth_flows=None,
+    generate_secret=False,
+):
+    """
+    Function that is verified to work against AWS.
+    Can be run against AWS at any time by setting:
+      MOTO_TEST_ALLOW_AWS_REQUEST=true
+
+    If this environment variable is not set, the function runs in a `mock_aws` context.
+
+    This decorator will:
+      - Create a UserPool
+      - Run the test and pass the name as an argument
+      - Delete the UserPool
+    """
+
+    def inner(func):
+        @wraps(func)
+        def pagination_wrapper(*args, **kwargs):
+            def create_user_pool_and_test():
+                conn = boto3.client("cognito-idp", "us-west-2")
+                pool_name = f"TestUserPool_{str(uuid4())[0:6]}"
+                userpool_args = {}
+                if username_attributes:
+                    userpool_args["UsernameAttributes"] = username_attributes
+                user_pool = conn.create_user_pool(PoolName=pool_name, **userpool_args)
+                pool_id = user_pool["UserPool"]["Id"]
+                pool_client_kwargs = {}
+                if read_attributes:
+                    pool_client_kwargs["ReadAttributes"] = read_attributes
+                if explicit_auth_flows:
+                    pool_client_kwargs["ExplicitAuthFlows"] = explicit_auth_flows
+                if generate_secret:
+                    pool_client_kwargs["GenerateSecret"] = True
+
+                try:
+                    user_pool_client = conn.create_user_pool_client(
+                        UserPoolId=pool_id,
+                        ClientName="TestAppClient",
+                        **pool_client_kwargs,
+                    )
+
+                    kwargs["user_pool"] = user_pool
+                    kwargs["user_pool_client"] = user_pool_client
+                    resp = func(*args, **kwargs)
+                finally:
+                    conn.delete_user_pool(UserPoolId=pool_id)
+
+                return resp
+
+            allow_aws_request = (
+                os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
+            )
+
+            if allow_aws_request:
+                return create_user_pool_and_test()
+            else:
+                with mock_aws():
+                    return create_user_pool_and_test()
+
+        return pagination_wrapper
+
+    return inner