def create_cross_account_role()

in functions/source/c1c_controltower_lifecycle.py [0:0]

• 43 lines of code
• 5 McCabe index (conditional complexity)

def create_cross_account_role(aws_account_id, c1c_connector):
    sts_session = assume_role(aws_account_id, c1cresources.ControlTowerRoleName)
    client = sts_session.client("iam")

    sts_client = sts_session.client("sts")
    sts_identity = sts_client.get_caller_identity()
    partition = sts_identity["Arn"].split(":")[1]

    logger.info(
        f"Creating role {c1cresources.IamRoleName} and policy {c1cresources.IamPolicyName} in account {aws_account_id}"
    )
    path = "/"
    try:
        logger.info("Creating role...")
        client.create_role(
            Path=path,
            RoleName=c1cresources.IamRoleName,
            AssumeRolePolicyDocument=c1cresources.get_assume_role_policy_document(
                c1c_connector
            ),
            Description="CloudOne Conformity Connector Role created by Control Tower",
        )
    except Exception as e:
        logger.error(f"Failed to create role: {e}")
        raise e
    try:
        logger.info("Creating and attaching policy parts...")
        c1c_policy_document = c1cresources.ConformityPolicyDoc()

        policy_part = 0
        for policy in c1c_policy_document.list_of_policies:
            policy_name = f"{c1cresources.IamPolicyName}{policy_part}"

            client.create_policy(
                PolicyName=policy_name,
                PolicyDocument=json.dumps(policy.get("document")),
            )
            # TODO this won't work if the partition is not aws.
            client.attach_role_policy(
                PolicyArn=f"arn:{partition}:iam::{aws_account_id}:policy/{policy_name}",
                RoleName=c1cresources.IamRoleName,
            )
            policy_part += 1
    except Exception as e:
        logger.error(f"Failed to attach policy: {e}")
        raise e
    else:
        return True