Merge pull request #5154 from scholtalbers/feature/add_permission_check_to_user_lib_import

add option to check file permissions for the user library imports

config/galaxy.ini.sample

@@ -920,6 +920,11 @@ use_interactive = True
 # create symlinks to them).
 #user_library_import_symlink_whitelist = None
 
+# In conjunction or alternatively, Galaxy can restrict user library imports to
+# those files that the user can read (by checking basic unix permissions).
+# For this to work, the username has to match the username on the filesystem.
+#user_library_import_check_permissions = False
+
 # Allow admins to paste filesystem paths during upload. For libraries this
 # adds an option to the admin library upload tool allowing admins to paste
 # filesystem paths to files and directories in a box, and these paths will be

lib/galaxy/actions/library.py

@@ -55,8 +55,9 @@ class LibraryActions(object):
             full_dir = os.path.join(import_dir, server_dir)
             unsafe = None
             if safe_relpath(server_dir):
+                username = trans.user.username if trans.app.config.user_library_import_check_permissions else None
                 if import_dir_desc == 'user_library_import_dir' and safe_contains(import_dir, full_dir, whitelist=trans.app.config.user_library_import_symlink_whitelist):
-                    for unsafe in unsafe_walk(full_dir, whitelist=[import_dir] + trans.app.config.user_library_import_symlink_whitelist):
+                    for unsafe in unsafe_walk(full_dir, whitelist=[import_dir] + trans.app.config.user_library_import_symlink_whitelist, username=username):
                         log.error('User attempted to import a path that resolves to a path outside of their import dir: %s -> %s', unsafe, os.path.realpath(unsafe))
             else:
                 log.error('User attempted to import a directory path that resolves to a path outside of their import dir: %s -> %s', server_dir, os.path.realpath(full_dir))

lib/galaxy/config.py

@@ -426,6 +426,7 @@ class Configuration(object):
         self.library_import_dir = kwargs.get('library_import_dir', None)
         self.user_library_import_dir = kwargs.get('user_library_import_dir', None)
         self.user_library_import_symlink_whitelist = listify(kwargs.get('user_library_import_symlink_whitelist', []), do_strip=True)
+        self.user_library_import_check_permissions = string_as_bool(kwargs.get('user_library_import_check_permissions', False))
         # Searching data libraries
         self.ftp_upload_dir = kwargs.get('ftp_upload_dir', None)
         self.ftp_upload_dir_identifier = kwargs.get('ftp_upload_dir_identifier', 'email')  # attribute on user - email, username, id, etc...

lib/galaxy/util/path/__init__.py

@@ -4,16 +4,20 @@ from __future__ import absolute_import
 
 import errno
 import imp
+import logging
 from functools import partial
+from grp import getgrgid
 from itertools import starmap
 from operator import getitem
 from os import (
     extsep,
     makedirs,
-    walk,
+    stat,
+    walk
 )
 from os.path import (
     abspath,
+    dirname,
     exists,
     isabs,
     join,
@@ -21,10 +25,13 @@ from os.path import (
     pardir,
     realpath,
     relpath,
+    sep as separator,
 )
+from pwd import getpwuid
 
 from six import iteritems, string_types
-from six.moves import filterfalse, map, zip
+from six.moves import map, zip
+log = logging.getLogger(__name__)
 
 
 def safe_contains(prefix, path, whitelist=None):
@@ -82,7 +89,7 @@ def safe_relpath(path):
     return not (isabs(path) or normpath(path).startswith(pardir))
 
 
-def unsafe_walk(path, whitelist=None):
+def unsafe_walk(path, whitelist=None, username=None):
     """Walk a path and ensure that none of its contents are symlinks outside the path.
 
     It is assumed that ``path`` itself has already been validated e.g. with :func:`safe_relpath` or
@@ -92,10 +99,72 @@ def unsafe_walk(path, whitelist=None):
     :param path:        a directory to check for unsafe contents
     :type whitelist:    list of strings
     :param whitelist:   list of additional paths under which contents may be located
-    :rtype:             iterator
-    :returns:           Iterator of "bad" files found under ``path``
+    :rtype:             list of strings
+    :returns:           A list of "bad" files found under ``path``
     """
-    return filterfalse(partial(safe_contains, path, whitelist=whitelist), __walk(abspath(path)))
+    unsafe_paths = []
+    for walked_path in __walk(abspath(path)):
+        is_safe = safe_contains(path, walked_path, whitelist=whitelist)
+        if username and is_safe:
+            is_safe = full_path_permission_for_user(path, walked_path, username=username, skip_prefix=True)
+        if not is_safe:
+            unsafe_paths.append(walked_path)
+    return unsafe_paths
+
+
+def __path_permission_for_user(path, username):
+    """
+    :type path:         string
+    :param path:        a directory or file to check
+    :type username:     string
+    :param username:    a username matching the systems username
+    """
+    group_id_of_file = stat(path).st_gid
+    file_owner = getpwuid(stat(path).st_uid)
+    group_members = getgrgid(group_id_of_file).gr_mem
+
+    oct_mode = oct(stat(path).st_mode)
+    owner_permissions = int(oct_mode[-3])
+    group_permissions = int(oct_mode[-2])
+    other_permissions = int(oct_mode[-1])
+    if other_permissions >= 4 or \
+            (file_owner == username and owner_permissions >= 4) or \
+            (username in group_members and group_permissions >= 4):
+        return True
+    return False
+
+
+def full_path_permission_for_user(prefix, path, username, skip_prefix=False):
+    """
+    Assuming username is identical to the os username, this checks that the
+    given user can read the specified path by checking the file permission
+    and each parent directory permission.
+
+    :type prefix:       string
+    :param prefix:      a directory under which ``path`` is to be checked
+    :type path:         string
+    :param path:        a filename to check
+    :type username:     string
+    :param username:    a username matching the systems username
+    :type skip_prefix:  bool
+    :param skip_prefix: skip the given prefix from being checked for permissions
+
+    """
+    full_path = realpath(join(prefix, path))
+    top_path = realpath(prefix) if skip_prefix else None
+    can_read = __path_permission_for_user(full_path, username)
+    if can_read:
+        depth = 0
+        max_depth = full_path.count(separator)
+        parent_path = dirname(full_path)
+        while can_read and depth != max_depth:
+            if parent_path in [separator, top_path]:
+                break
+            if not __path_permission_for_user(parent_path, username):
+                can_read = False
+            depth += 1
+            parent_path = dirname(parent_path)
+    return can_read
 
 
 def joinext(root, ext):

lib/galaxy/webapps/galaxy/api/library_datasets.py

@@ -25,7 +25,7 @@ from galaxy.managers import (
 )
 from galaxy.tools.actions import upload_common
 from galaxy.tools.parameters import populate_state
-from galaxy.util.path import safe_contains, safe_relpath, unsafe_walk
+from galaxy.util.path import full_path_permission_for_user, safe_contains, safe_relpath, unsafe_walk
 from galaxy.util.streamball import StreamBall
 from galaxy.web import (
     _future_expose_api as expose_api,
@@ -432,20 +432,30 @@ class LibraryDatasetsController(BaseAPIController, UsesVisualizationMixin, Libra
             path = os.path.join(import_base_dir, path)
         elif source in ['userdir_file', 'userdir_folder']:
             unsafe = None
+            username = trans.user.username if trans.app.config.user_library_import_check_permissions else None
             user_login = trans.user.email
             user_base_dir = trans.app.config.user_library_import_dir
             if user_base_dir is None:
                 raise exceptions.ConfigDoesNotAllowException('The configuration of this Galaxy instance does not allow upload from user directories.')
             full_dir = os.path.join(user_base_dir, user_login)
+
             if not safe_contains(full_dir, path, whitelist=trans.app.config.user_library_import_symlink_whitelist):
                 # the path is a symlink outside the user dir
                 path = os.path.join(full_dir, path)
                 log.error('User attempted to import a path that resolves to a path outside of their import dir: %s -> %s', path, os.path.realpath(path))
                 raise exceptions.RequestParameterInvalidException('The given path is invalid.')
+            if trans.app.config.user_library_import_check_permissions and not full_path_permission_for_user(full_dir, path, username):
+                log.error('User attempted to import a path that resolves to a path outside of their import dir: '
+                        '%s -> %s and cannot be read by them.', path, os.path.realpath(path))
+                raise exceptions.RequestParameterInvalidException('The given path is invalid.')
             path = os.path.join(full_dir, path)
-            for unsafe in unsafe_walk(path, whitelist=[full_dir] + trans.app.config.user_library_import_symlink_whitelist):
+            for unsafe in unsafe_walk(path, whitelist=[full_dir] + trans.app.config.user_library_import_symlink_whitelist, username=username):
                 # the path is a dir and contains files that symlink outside the user dir
-                log.error('User attempted to import a directory containing a path that resolves to a path outside of their import dir: %s -> %s', unsafe, os.path.realpath(unsafe))
+                error = 'User attempted to import a path that resolves to a path outside of their import dir: %s -> %s', \
+                        path, os.path.realpath(path)
+                if trans.app.config.user_library_import_check_permissions:
+                    error += ' or is not readable for them.'
+                log.error(error)
             if unsafe:
                 raise exceptions.RequestParameterInvalidException('The given path is invalid.')
             if not os.path.exists(path):