Source code for ecs_composex.cognito_userpool.cognito_aws
# -*- coding: utf-8 -*-
# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2021 John Mille <john@compose-x.io>
"""
Module to find the Cognito UserPools from tagging API
"""
import re
from compose_x_common.compose_x_common import keyisset
from ecs_composex.cognito_userpool.cognito_params import (
USERPOOL_ARN,
USERPOOL_DOMAIN,
USERPOOL_ID,
)
from ecs_composex.common import LOG
from ecs_composex.common.aws import (
define_lookup_role_from_info,
find_aws_resource_arn_from_tags_api,
)
[docs]def get_userpool_config(userpool_arn, session):
"""
:param str userpool_arn:
:param boto3.session.Session session:
:return:
"""
userpool_parts = re.compile(
r"(?:^arn:aws(?:-[a-z]+)?:cognito-idp:[\S]+:[0-9]{12}:userpool/)([\S]+)$"
)
userpool_id = userpool_parts.match(userpool_arn).groups()[0]
userpool_config = {
USERPOOL_ARN.title: userpool_arn,
USERPOOL_ID.title: userpool_id,
}
client = session.client("cognito-idp")
try:
userpool_r = client.describe_user_pool(UserPoolId=userpool_id)
if keyisset("CustomDomain", userpool_r["UserPool"]):
userpool_config[USERPOOL_DOMAIN.title] = userpool_r["UserPool"][
"CustomDomain"
]
elif keyisset("Domain", userpool_r["UserPool"]):
userpool_config[USERPOOL_DOMAIN.title] = userpool_r["UserPool"]["Domain"]
LOG.debug(f"Pool domain is {userpool_config[USERPOOL_DOMAIN.title]}")
except client.exceptions:
LOG.error("Failed to retrieve the Pool Domain. Moving on.")
return userpool_config
[docs]def lookup_userpool_config(lookup, session):
"""
Function to find the DB in AWS account
:param dict lookup: The Lookup definition for DB
:param boto3.session.Session session: Boto3 session for clients
:return:
"""
lookup_types = {
"cognito-idp": {
"regexp": r"(?:^arn:aws(?:-[a-z]+)?:cognito-idp:[\S]+:[0-9]{12}:userpool/)([\S]+)$"
},
}
lookup_session = define_lookup_role_from_info(lookup, session)
userpool_arn = find_aws_resource_arn_from_tags_api(
lookup,
lookup_session,
"cognito-idp",
types=lookup_types,
)
if not userpool_arn:
return None
config = get_userpool_config(userpool_arn, lookup_session)
LOG.debug(config)
return config