def user_has_permission()

in source/soca/cluster_web_ui/views/my_files.py [0:0]
87 lines of code
29 McCabe index (conditional complexity)

def user_has_permission(path, permission_required, type):
    logger.info("Checking " + permission_required + " for " + path + " (" + type +")")
    if type not in ["file", "folder"]:
        logger.info("Type must be file or folder")
        return False

    if permission_required not in ["write", "read"]:
        logger.info("permission_required must be write or read")
        return False

    if path.startswith("//"):
        # Remove first slash if present. Case when user click "root" label on breadcrumb and then select a folder from the top level
        path = path[1:]

    if config.Config.CHROOT_USER is True:
        if not path.lower().startswith(config.Config.USER_HOME.lower() + "/" + session["user"].lower()):
            return False

    for restricted_path in config.Config.PATH_TO_RESTRICT:
        if path.lower().startswith(restricted_path.lower()):
            return False

    min_permission_level = {"write": 6,  # Read+Write
                            "read": 5,  # Read+Execute
                            "execute": 1,  # Min permission to be able to CD into directory
                            }

    user_uid = pwd.getpwnam(session["user"]).pw_uid

    # First, make sure user can access the entire folder hierarchy
    folder_level = 1
    folder_hierarchy = path.split("/")

    if permission_required == "read":
        last_folder = folder_hierarchy[-1]
    else:
        # When we create a new folder, the last existing folder is 2 level up in the array
        last_folder = folder_hierarchy[-2]
    try:
        for folder in folder_hierarchy:
            if folder != "":
                folder_path = "/".join(folder_hierarchy[:folder_level])
                if CACHE_FOLDER_PERMISSION_PREFIX + folder_path not in cache.keys():
                    check_folder = {}
                    check_folder["folder_owner"] = os.stat(folder_path).st_uid
                    check_folder["folder_group_id"] = os.stat(folder_path).st_gid
                    try:
                        check_folder["folder_group_name"] = grp.getgrgid(check_folder["folder_group_id"]).gr_name
                    except:
                        check_folder["folder_group_name"] = "UNKNOWN"

                    check_folder["folder_permission"] = oct(os.stat(folder_path).st_mode)[-3:]
                    check_folder["group_permission"] = int(check_folder["folder_permission"][-2])
                    check_folder["other_permission"] = int(check_folder["folder_permission"][-1])
                    cache[CACHE_FOLDER_PERMISSION_PREFIX + folder_path] = check_folder
                else:
                    check_folder = cache[CACHE_FOLDER_PERMISSION_PREFIX + folder_path]

                if CACHE_GROUP_MEMBERSHIP_PREFIX + check_folder["folder_group_name"] not in cache.keys():
                    check_group_membership = get(config.Config.FLASK_ENDPOINT + "/api/ldap/group",
                                                 headers={"X-SOCA-TOKEN": config.Config.API_ROOT_KEY},
                                                 params={"group": check_folder["folder_group_name"]},
                                                 verify=False) # nosec

                    if check_group_membership.status_code == 200:
                        group_members = check_group_membership.json()["message"]["members"]
                    else:
                        logger.info("Unable to check group membership because of " + check_group_membership.text)
                        group_members = []
                    cache[CACHE_GROUP_MEMBERSHIP_PREFIX + check_folder["folder_group_name"]] = group_members
                else:
                    group_members = cache[CACHE_GROUP_MEMBERSHIP_PREFIX + check_folder["folder_group_name"]]

                if session["user"] in group_members:
                    user_belong_to_group = True
                else:
                    user_belong_to_group = False

                # Verify if user has the required permissions on the folder
                if folder == last_folder:
                    # Last folder, must have at least R or W permission
                    if check_folder["folder_owner"] != user_uid:
                        if user_belong_to_group is True:
                            if check_folder["group_permission"] < min_permission_level[permission_required]:
                                logger.info("user do not have " + permission_required + " permission for " + folder_path)
                                return False
                        else:
                            if check_folder["other_permission"] < min_permission_level[permission_required]:
                                logger.info("user do not have " + permission_required + " permission for " + folder_path)
                                return False
                else:
                    # Folder chain, must have at least Execute permission
                    if check_folder["folder_owner"] != user_uid:
                        if user_belong_to_group is True:
                            if check_folder["group_permission"] < min_permission_level[permission_required]:
                                logger.info("user do not have " + permission_required + " permission for " + folder_path)
                                return False
                        else:
                            if (check_folder["other_permission"] < min_permission_level["execute"]):
                                logger.info("user do not have EXECUTE permission for " + folder_path)
                                return False

            folder_level += 1

        logger.info("Permissions valid.")
        return True
    except FileNotFoundError:
        return False