Merge pull request #23 from dataware-tools/feature/avoid-directory-tr…

…aversal

Avoid Directory Traversal

api/server.py

@@ -17,7 +17,7 @@
 from dataware_tools_api_helper import get_forward_headers, get_jwt_payload_from_request
 
 from api.settings import META_STORE_SERVICE, UPLOADED_FILE_PATH_PREFIX
-from api.utils import get_valid_filename, is_file_in_directory
+from api.utils import get_valid_filename, is_file_in_directory, is_valid_path
 
 # Metadata
 description = "An API for downloading files."
@@ -108,7 +108,7 @@ async def on_post(self, req, resp):
         if all([database_id is not None, record_id is not None]):
             payload['content_type'] = _get_content_type(req, database_id, record_id, path)
 
-        if not os.path.isfile(path):
+        if not is_valid_path(path, check_existence=True):
             resp.status_code = 404
             resp.media = {'reason': 'No such file'}
             return
@@ -178,7 +178,7 @@ async def on_get(self, req, resp, *, token):
             )
 
         # Check file
-        if not os.path.isfile(payload.get('path')):
+        if not is_valid_path(payload.get('path'), check_existence=True):
             resp.status_code = 404
             resp.media = {'reason': 'No such file: {}'.format(payload.get('path'))}
             return
@@ -216,6 +216,12 @@ def save_file(save_file_path, file):
             f'record_{get_valid_filename(record_id)}',
             file['filename'],
         )
+        if not is_valid_path(save_file_path, check_existence=False):
+            resp.status_code = 403
+            resp.media = {
+                'reason': f'Invalid path: {save_file_path}',
+            }
+            return
         if os.path.exists(save_file_path):
             resp.status_code = 409
             resp.media = {
@@ -226,7 +232,8 @@ def save_file(save_file_path, file):
             save_file(save_file_path, file)
 
         # Add metadata to meta-store
-        fetch_success, fetch_res = _update_metastore(req, database_id, record_id, save_file_path, file_metadata)
+        fetch_success, fetch_res = _update_metastore(req, database_id, record_id, save_file_path,
+                                                     file_metadata)
 
         if fetch_success and fetch_res is not None:
             resp.status_code = fetch_res.status_code if fetch_res.status_code != 200 else 201
@@ -255,6 +262,14 @@ async def on_delete(self, req: responder.Request, resp: responder.Response):
         """
         file_path = req.params.get('path', '')
 
+        # Validation
+        if not is_valid_path(file_path, check_existence=False):
+            resp.status_code = 403
+            resp.media = {
+                'reason': f'Deleting ({file_path}) is forbbiden.',
+            }
+            return
+
         # Check if path is in the UPLOADED_FILE_PATH_PREFIX directory
         if not is_file_in_directory(file_path, UPLOADED_FILE_PATH_PREFIX):
             resp.status_code = 403
@@ -307,7 +322,7 @@ async def get_file(req, resp):
     elif all([database_id is not None, record_id is not None]):
         resp.headers['Content-Type'] = _get_content_type(req, database_id, record_id, path)
 
-    if not os.path.isfile(path):
+    if not is_valid_path(path, check_existence=True):
         resp.status_code = 404
         resp.media = {'reason': 'No such file'}
         return

api/utils.py

@@ -37,3 +37,33 @@ def is_file_in_directory(file: str, directory: str) -> bool:
     # return true, if the common prefix of both is equal to directory
     # e.g. /a/b/c/d.rst and directory is /a/b, the common prefix is /a/b
     return os.path.commonprefix([file, directory]) == directory
+
+
+def is_valid_path(path: str, check_existence=False) -> bool:
+    """Checks if the path is valid.
+
+    Args:
+        path (str): File path
+        check_existence (bool): If True, this function also checks the existence of the file
+
+    Returns:
+        (bool): True if the path is valid, otherwise False
+
+    """
+    # Avoid Directory Traversal
+    abspath = os.path.abspath(os.path.realpath(path))
+    abspath_splitted = abspath.split(os.sep)
+    if len(abspath_splitted) < 2:
+        return False
+    else:
+        if abspath_splitted[1] in ['bin', 'boot', 'dev', 'etc', 'home', 'lib', 'lib64', 'media',
+                                   'proc', 'root', 'run', 'sbin', 'srv', 'sys', 'tmp', 'usr',
+                                   'var']:
+            return False
+
+    # Check the existence of the file
+    if check_existence:
+        if not os.path.isfile(path):
+            return False
+
+    return True

test/test_api.py

@@ -81,6 +81,11 @@ def assert_file_get_200(api, file_path, content_type=None):
     ('/opt/app/test/files/records/sample/data/records.bag', 'application/rosbag'),
 ]
 
+file_pathes_invalid = [
+    ('/proc/self/environ', 'text/plain'),
+    ('/etc/passwd', 'text/plain'),
+]
+
 
 @pytest.mark.parametrize("file_path, content_type", file_pathes)
 def test_file_get_200(api, file_path, content_type):
@@ -333,3 +338,12 @@ def test_download_403(api, file_path, content_type):
     # Get file with the token
     r = api.requests.get(url=api.url_for(server.Download, token=token))
     assert r.status_code == 403
+
+
+@pytest.mark.parametrize("file_path, content_type", file_pathes_invalid)
+def test_invalid_path(api, file_path, content_type):
+    params = {'path': file_path}
+    if content_type is not None:
+        params.update({'content_type': content_type})
+    r = api.requests.post(url=api.url_for(server.Downloads), data=params)
+    assert r.status_code == 404