Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

CognitoIDP: update_group, global_sign_out, update_user_attributes (#5086)

Browse Source
This commit is contained in:
Bert Blommers 2022-05-01 21:04:57 +00:00 committed by GitHub
parent 3b68be55d3
commit 1d87b90e75
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 178 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
moto/cognitoidp/models.py
View File

@ -578,6 +578,14 @@ class CognitoIdpUserPool(BaseModel):
extra_data.update({attribute[0]["Name"]: attribute[0]["Value"]})
return extra_data
def sign_out(self, username):
for token, token_tuple in list(self.refresh_tokens.items()):
if token_tuple is None:
continue
_, logged_in_user = token_tuple
if username == logged_in_user:
self.refresh_tokens[token] = None
class CognitoIdpUserPoolDomain(BaseModel):
def __init__(self, user_pool_id, domain, custom_domain_config=None):
@ -679,6 +687,15 @@ class CognitoIdpGroup(BaseModel):
# Note that these links are bidirectional.
self.users = set()
def update(self, description, role_arn, precedence):
if description is not None:
self.description = description
if role_arn is not None:
self.role_arn = role_arn
if precedence is not None:
self.precedence = precedence
self.last_modified_date = datetime.datetime.now()
def to_json(self):
return {
"GroupName": self.group_name,
@ -1014,6 +1031,13 @@ class CognitoIdpBackend(BaseBackend):
del user_pool.groups[group_name]
def update_group(self, user_pool_id, group_name, description, role_arn, precedence):
group = self.get_group(user_pool_id, group_name)
group.update(description, role_arn, precedence)
return group
def admin_add_user_to_group(self, user_pool_id, group_name, username):
group = self.get_group(user_pool_id, group_name)
user = self.admin_get_user(user_pool_id, username)
@ -1441,12 +1465,16 @@ class CognitoIdpBackend(BaseBackend):
user_pool = self.describe_user_pool(user_pool_id)
self.admin_get_user(user_pool_id, username)
for token, token_tuple in list(user_pool.refresh_tokens.items()):
if token_tuple is None:
continue
_, username = token_tuple
if username == username:
user_pool.refresh_tokens[token] = None
user_pool.sign_out(username)
def global_sign_out(self, access_token):
for user_pool in self.user_pools.values():
if access_token in user_pool.access_tokens:
_, username = user_pool.access_tokens[access_token]
user_pool.sign_out(username)
return
raise NotAuthorizedError(access_token)
def create_resource_server(self, user_pool_id, identifier, name, scopes):
user_pool = self.describe_user_pool(user_pool_id)
@ -1752,6 +1780,20 @@ class CognitoIdpBackend(BaseBackend):
user_pool = self.describe_user_pool(user_pool_id)
user_pool.add_custom_attributes(custom_attributes)
def update_user_attributes(self, access_token, attributes):
"""
The parameter ClientMetadata has not yet been implemented. No CodeDeliveryDetails are returned.
"""
for user_pool in self.user_pools.values():
if access_token in user_pool.access_tokens:
_, username = user_pool.access_tokens[access_token]
user = self.admin_get_user(user_pool.id, username)
user.update_attributes(attributes)
return
raise NotAuthorizedError(access_token)
class GlobalCognitoIdpBackend(CognitoIdpBackend):
# Some operations are unauthenticated

26
moto/cognitoidp/responses.py
View File

@ -256,6 +256,19 @@ class CognitoIdpResponse(BaseResponse):
cognitoidp_backends[self.region].delete_group(user_pool_id, group_name)
return ""
def update_group(self):
group_name = self._get_param("GroupName")
user_pool_id = self._get_param("UserPoolId")
description = self._get_param("Description")
role_arn = self._get_param("RoleArn")
precedence = self._get_param("Precedence")
group = cognitoidp_backends[self.region].update_group(
user_pool_id, group_name, description, role_arn, precedence
)
return json.dumps({"Group": group.to_json()})
def admin_add_user_to_group(self):
user_pool_id = self._get_param("UserPoolId")
username = self._get_param("Username")
@ -501,6 +514,11 @@ class CognitoIdpResponse(BaseResponse):
)
return ""
def global_sign_out(self):
access_token = self._get_param("AccessToken")
cognitoidp_backends[self.region].global_sign_out(access_token)
return ""
# Resource Server
def create_resource_server(self):
user_pool_id = self._get_param("UserPoolId")
@ -595,6 +613,14 @@ class CognitoIdpResponse(BaseResponse):
)
return ""
def update_user_attributes(self):
access_token = self._get_param("AccessToken")
attributes = self._get_param("UserAttributes")
cognitoidp_backends[self.region].update_user_attributes(
access_token, attributes
)
return json.dumps({})
class CognitoIdpJsonWebKeyResponse(BaseResponse):
def __init__(self):

1
tests/terraformtests/terraform-tests.success.txt
View File

@ -24,6 +24,7 @@ cloudwatch:
- TestAccCloudWatchLogsGroupDataSource
cognitoidp:
- TestAccCognitoIDPIdentityProvider
- TestAccCognitoIDPUserGroup_
- TestAccCognitoIDPUserPool_
- TestAccCognitoUser_
- TestAccCognitoUserInGroup_

103
tests/test_cognitoidp/test_cognitoidp.py
View File

@ -1316,6 +1316,44 @@ def test_create_group():
result["Group"]["CreationDate"].should.be.a("datetime.datetime")
@mock_cognitoidp
def test_update_group():
conn = boto3.client("cognito-idp", "us-west-2")
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
group_name = str(uuid.uuid4())
description = str(uuid.uuid4())
description2 = str(uuid.uuid4())
role_arn = "arn:aws:iam:::role/my-iam-role"
role_arn2 = "arn:aws:iam:::role/my-iam-role2"
precedence = random.randint(0, 100000)
precedence2 = random.randint(0, 100000)
conn.create_group(
GroupName=group_name,
UserPoolId=user_pool_id,
Description=description,
RoleArn=role_arn,
Precedence=precedence,
)
result = conn.update_group(
GroupName=group_name,
UserPoolId=user_pool_id,
Description=description2,
RoleArn=role_arn2,
Precedence=precedence2,
)
result["Group"]["GroupName"].should.equal(group_name)
result["Group"]["UserPoolId"].should.equal(user_pool_id)
result["Group"]["Description"].should.equal(description2)
result["Group"]["RoleArn"].should.equal(role_arn2)
result["Group"]["Precedence"].should.equal(precedence2)
result["Group"]["LastModifiedDate"].should.be.a("datetime.datetime")
result["Group"]["CreationDate"].should.be.a("datetime.datetime")
@mock_cognitoidp
def test_group_in_access_token():
conn = boto3.client("cognito-idp", "us-west-2")
@ -2990,6 +3028,36 @@ def test_admin_user_global_sign_out_unknown_user():
err["Message"].should.equal("User does not exist.")
@mock_cognitoidp
def test_global_sign_out():
conn = boto3.client("cognito-idp", "us-west-2")
result = user_authentication_flow(conn)
conn.global_sign_out(AccessToken=result["access_token"])
with pytest.raises(ClientError) as ex:
conn.initiate_auth(
ClientId=result["client_id"],
AuthFlow="REFRESH_TOKEN",
AuthParameters={
"REFRESH_TOKEN": result["refresh_token"],
"SECRET_HASH": result["secret_hash"],
},
)
err = ex.value.response["Error"]
err["Code"].should.equal("NotAuthorizedException")
err["Message"].should.equal("Refresh Token has been revoked")
@mock_cognitoidp
def test_global_sign_out_unknown_accesstoken():
conn = boto3.client("cognito-idp", "us-east-2")
with pytest.raises(ClientError) as ex:
conn.global_sign_out(AccessToken="n/a")
err = ex.value.response["Error"]
err["Code"].should.equal("NotAuthorizedException")
@mock_cognitoidp
def test_admin_update_user_attributes():
conn = boto3.client("cognito-idp", "us-west-2")
@ -3146,6 +3214,41 @@ def test_admin_delete_user_attributes_non_existing_pool():
err["Message"].should.equal(f"User pool {user_pool_id} does not exist.")
@mock_cognitoidp
def test_update_user_attributes():
conn = boto3.client("cognito-idp", "us-west-2")
result = authentication_flow(conn, auth_flow="ADMIN_USER_PASSWORD_AUTH")
access_token = result["access_token"]
username = result["username"]
user_pool_id = result["user_pool_id"]
conn.update_user_attributes(
AccessToken=access_token,
UserAttributes=[
{"Name": "family_name", "Value": "Doe"},
{"Name": "given_name", "Value": "Jane"},
],
)
user = conn.admin_get_user(UserPoolId=user_pool_id, Username=username)
attributes = user["UserAttributes"]
attributes.should.contain({"Name": "family_name", "Value": "Doe"})
attributes.should.contain({"Name": "given_name", "Value": "Jane"})
@mock_cognitoidp
def test_update_user_attributes_unknown_accesstoken():
conn = boto3.client("cognito-idp", "us-east-2")
with pytest.raises(ClientError) as ex:
conn.update_user_attributes(
AccessToken="n/a", UserAttributes=[{"Name": "a", "Value": "b"}]
)
err = ex.value.response["Error"]
err["Code"].should.equal("NotAuthorizedException")
@mock_cognitoidp
def test_resource_server():