CognitoIDP: get_user (#4038)

* cognito-idp get_user support

* add carium expected attributes

* CognitoIDP#get_user - Add negative tests

Co-authored-by: Lalitha Kolla <lalitha.kolla@carium.com>
This commit is contained in:
Bert Blommers 2021-06-27 15:48:31 +01:00 committed by GitHub
commit 2590bf0e80
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 89 additions and 13 deletions

View file

@ -136,6 +136,7 @@ class CognitoIdpUserPool(BaseModel):
"token_use": token_use,
"auth_time": now,
"exp": now + expires_in,
"email": self.users[username].username,
}
payload.update(extra_data)
headers = {"kid": "dummy"} # KID as present in jwks-public.json
@ -657,6 +658,10 @@ class CognitoIdpBackend(BaseBackend):
UserStatus["FORCE_CHANGE_PASSWORD"],
attributes,
)
user.attributes.append({"Name": "sub", "Value": user.id})
user.attributes.append({"Name": "email_verified", "Value": True})
user.attributes.append({"Name": "name", "Value": ""})
user.attributes.append({"Name": "family_name", "Value": ""})
user_pool.users[user.username] = user
return user
@ -670,6 +675,20 @@ class CognitoIdpBackend(BaseBackend):
return user_pool.users[username]
def get_user(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 = user_pool.users.get(username)
if (
not user
or not user.enabled
or user.status != UserStatus["CONFIRMED"]
):
raise NotAuthorizedError("username")
return user
raise NotAuthorizedError("Invalid token")
@paginate(60, "pagination_token", "limit")
def list_users(self, user_pool_id, pagination_token=None, limit=None):
user_pool = self.user_pools.get(user_pool_id)

View file

@ -313,6 +313,11 @@ class CognitoIdpResponse(BaseResponse):
user = cognitoidp_backends[self.region].admin_get_user(user_pool_id, username)
return json.dumps(user.to_json(extended=True, attributes_key="UserAttributes"))
def get_user(self):
access_token = self._get_param("AccessToken")
user = cognitoidp_backends[self.region].get_user(access_token=access_token)
return json.dumps(user.to_json(extended=True, attributes_key="UserAttributes"))
def list_users(self):
user_pool_id = self._get_param("UserPoolId")
limit = self._get_param("Limit")