Repository: Netflix/consoleme
File: consoleme/celery_tasks/celery_tasks.py (Python)

Code before:

    cloudtrail_errors = {}
    cloudtrail_errors_j = red.get(
        config.get(
            "celery.cache_cloudtrail_errors_by_arn.redis_key",
            "CLOUDTRAIL_ERRORS_BY_ARN",
        )
    )

    if cloudtrail_errors_j:
        cloudtrail_errors = json.loads(cloudtrail_errors_j)

    s3_error_topic = config.get("redis.s3_errors", "S3_ERRORS")
    all_s3_errors = red.get(s3_error_topic)
    s3_errors = {}
    if all_s3_errors:
        s3_errors = json.loads(all_s3_errors)

    allowed_tags = config.get("roles.allowed_tags")
    allowed_arns = config.get("roles.allowed_arns")
    for arn, role_details_j in all_iam_roles.items():
        role_details = ujson.loads(role_details_j)

        allowed = True

        if allowed_tags and {"consoleme":"managed"} not in role_details.get("tags"):
            allowed = False

        if allowed_arns and arn not in allowed_arns:
            allowed = False

        if not allowed:
            continue

        error_count = cloudtrail_errors.get(arn, 0)
        s3_errors_for_arn = s3_errors.get(arn, [])
        for error in s3_errors_for_arn:
            error_count += int(error.get("count"))

        account_id = arn.split(":")[4]
        account_name = accounts_d.get(str(account_id), "Unknown")
        resource_id = role_details.get("resourceId")
        items.append(
            {
                "account_id": account_id,
                "account_name": account_name,
                "arn": arn,
                "technology": "AWS::IAM::Role",
                "templated": red.hget(
                    config.get("templated_roles.redis_key", "TEMPLATED_ROLES_v2"),
                    arn.lower(),

Review comment (suggestion):
@jaydhulia This part I'm not comfortable with this since i'm not referencing a struct with defined attributes. I'm not 100% sure how to grab the tags from the `role_details` in this part

Code after:
    accounts_d = async_to_sync(get_account_id_to_name_mapping)()

    cloudtrail_errors = {}
    cloudtrail_errors_j = red.get(
        config.get(
            "celery.cache_cloudtrail_errors_by_arn.redis_key",
            "CLOUDTRAIL_ERRORS_BY_ARN",
        )
    )

    if cloudtrail_errors_j:
        cloudtrail_errors = json.loads(cloudtrail_errors_j)

    s3_error_topic = config.get("redis.s3_errors", "S3_ERRORS")
    all_s3_errors = red.get(s3_error_topic)
    s3_errors = {}
    if all_s3_errors:
        s3_errors = json.loads(all_s3_errors)

    for arn, role_details_j in all_iam_roles.items():
        role_details = ujson.loads(role_details_j)
        role_details_policy = ujson.loads(role_details.get("policy", {}))
        role_tags = role_details_policy.get("Tags", {})

        if not allowed_to_sync_role(arn, role_tags):
            continue

        error_count = cloudtrail_errors.get(arn, 0)
        s3_errors_for_arn = s3_errors.get(arn, [])
        for error in s3_errors_for_arn:
            error_count += int(error.get("count"))

        account_id = arn.split(":")[4]
        account_name = accounts_d.get(str(account_id), "Unknown")
        resource_id = role_details.get("resourceId")
        items.append(
            {
                "account_id": account_id,
                "account_name": account_name,
                "arn": arn,
                "technology": "AWS::IAM::Role",
                "templated": red.hget(
                    config.get("templated_roles.redis_key", "TEMPLATED_ROLES_v2"),
                    arn.lower(),
                ),
                "errors": error_count,
                "config_history_url": async_to_sync(
                    get_aws_config_history_url_for_resource
                )(account_id, resource_id, arn, "AWS::IAM::Role"),
            }
        )