More secure path mangling

src/actinia_core/core/utils.py

@@ -28,15 +28,16 @@
 import os
 from actinia_core.core.common.process_object import Process
 from actinia_core.core.common.exceptions import SecurityError
+from actinia_core.core.common.config import global_config
 
 __license__ = "GPLv3"
 __author__ = "Sören Gebbert, Anika Weinmann"
 __copyright__ = "Copyright 2016-2021, Sören Gebbert and mundialis GmbH & Co. KG"
 __maintainer__ = "mundialis"
 
 
-def os_path_normpath(path_parts):
-    """The function returns a joined norm path or raises an error if it is not
+def ensure_valid_path(path_parts):
+    """The function returns a joined and normalized path or raises an error if it is not
     as expected.
 
     Args:
@@ -49,11 +50,22 @@ def os_path_normpath(path_parts):
         raises a SecurityError if the path is not as expected
     """
     path = os.path.normpath(os.path.join(*path_parts))
-    if not path.startswith(path_parts[0]):
+    if not path.startswith(path_parts[0]) or not path_is_in_allowed_paths(path):
         raise SecurityError(f"Used path '{path}' is insecure")
     return path
 
 
+def path_is_in_allowed_paths(path):
+    # TODO we should double check whether they are all needed, better would be one
+    # single location
+    allowed_starts = [
+        global_config.GRASS_USER_DATABASE,
+        global_config.GRASS_DATABASE,
+        global_config.GRASS_RESOURCE_DIR
+    ]
+    return any(path.startswith(allowed) for allowed in allowed_starts)
+
+
 def get_wget_process(source, url):
     """The function returns a wget Process for the given source and url
 

src/actinia_core/rest/location_management.py

@@ -49,7 +49,7 @@
 from actinia_core.rest.mapset_management import PersistentGetProjectionRegionInfo
 from actinia_core.core.common.redis_interface import enqueue_job
 from actinia_core.core.common.exceptions import AsyncProcessError
-from actinia_core.core.utils import os_path_normpath
+from actinia_core.core.utils import ensure_valid_path
 
 __license__ = "GPLv3"
 __author__ = "Sören Gebbert, Carmen Tawalika"
@@ -214,10 +214,10 @@ def delete(self, location_name):
         """Delete an existing location and everything inside from the user database.
         """
         # Delete only locations from the user database
-        location = os_path_normpath(
+        location = ensure_valid_path(
             [self.grass_user_data_base, self.user_group, location_name])
-        permanent_mapset = os_path_normpath([location, "PERMANENT"])
-        wind_file = os_path_normpath([permanent_mapset, "WIND"])
+        permanent_mapset = ensure_valid_path([location, "PERMANENT"])
+        wind_file = ensure_valid_path([permanent_mapset, "WIND"])
         # Check the location path, only "valid" locations can be deleted
         if os.path.isdir(location):
             if os.path.isdir(permanent_mapset) and os.path.isfile(wind_file):
@@ -272,7 +272,7 @@ def post(self, location_name):
         """Create a new location based on EPSG code in the user database.
         """
         # Create only new locations if they did not exist in the global database
-        location = os_path_normpath([self.grass_data_base, location_name])
+        location = ensure_valid_path([self.grass_data_base, location_name])
 
         # Check the location path
         if os.path.isdir(location):
@@ -281,7 +281,7 @@ def post(self, location_name):
                         "Location <%s> exists in global database." % location_name)
 
         # Check also for the user database
-        location = os_path_normpath(
+        location = ensure_valid_path(
             [self.grass_user_data_base, self.user_group, location_name])
         # Check the location path
         if os.path.isdir(location):

src/actinia_core/rest/resource_streamer.py

@@ -31,7 +31,7 @@
 from actinia_core.core.common.config import global_config
 from actinia_core.core.common.app import auth
 from actinia_core.core.common.api_logger import log_api_call
-from actinia_core.core.utils import os_path_normpath
+from actinia_core.core.utils import ensure_valid_path
 
 
 __license__ = "GPLv3"
@@ -84,9 +84,9 @@ def get(self, user_id, resource_id, file_name):
         """
 
         resource_dir = global_config.GRASS_RESOURCE_DIR
-        user_export_path = os_path_normpath([resource_dir, user_id])
-        resource_export_path = os_path_normpath([user_export_path, resource_id])
-        resource_export_file_path = os_path_normpath([resource_export_path,
+        user_export_path = ensure_valid_path([resource_dir, user_id])
+        resource_export_path = ensure_valid_path([user_export_path, resource_id])
+        resource_export_file_path = ensure_valid_path([resource_export_path,
                                                       file_name])
 
         if (os.path.exists(resource_export_file_path) is True