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

Code before:
    )
    items = []
    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)

    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)
        role_details_policy = ujson.loads(role_details.get("policy"), {})

        allowed = True

        actual_tags = {
            k: v for d in role_details_policy.get("Tags") for k, v in d.items()
        }  # Convert List[Dicts] to 1 Dict
        if allowed_tags and not allowed_tags.items() <= actual_tags.items():
            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(
            {

Review comment (security):
Ahhh, I don't seem to have permission to push to this branch... this line should be:
`ujson.loads(role_details.get("policy", {}))` instead of `ujson.loads(role_details.get("policy"), {})`
and that should fix the unit tests.

Code after:
        default={},
    )
    items = []
    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