Path validation in upload

server/Pipfile

@@ -39,7 +39,7 @@ urllib3 = "==2.2.2"
 shapely = "==2.0.6"
 psycogreen = "==1.0.2"
 importlib-metadata = "==8.4.0"  # https://github.com/pallets/flask/issues/4502
-typing_extensions= "==4.12.2"
+typing_extensions = "==4.12.2"
 # requirements for development on windows
 colorama = "==0.4.5"
 

server/mergin/auth/forms.py

@@ -23,15 +23,15 @@ def username_validation(form, field):
     from ..sync.utils import (
         has_valid_characters,
         has_valid_first_character,
-        is_valid_filename,
+        check_filename,
         is_reserved_word,
     )
 
     errors = [
         has_valid_characters(field.data),
         has_valid_first_character(field.data),
         is_reserved_word(field.data),
-        is_valid_filename(field.data),
+        check_filename(field.data),
     ]
     for error in errors:
         if error:

server/mergin/sync/forms.py

@@ -9,7 +9,7 @@
 from mergin.sync.utils import (
     is_reserved_word,
     has_valid_characters,
-    is_valid_filename,
+    check_filename,
     has_valid_first_character,
 )
 
@@ -22,7 +22,7 @@ def project_name_validation(name: str) -> str | None:
         has_valid_characters(name),
         has_valid_first_character(name),
         is_reserved_word(name),
-        is_valid_filename(name),
+        check_filename(name),
     ]
     for error in errors:
         if error:

server/mergin/sync/public_api_controller.py

@@ -25,13 +25,12 @@
 )
 from pygeodiff import GeoDiffLibError
 from flask_login import current_user
-from sqlalchemy import and_, desc, asc, text, func, select
+from sqlalchemy import and_, desc, asc
 from sqlalchemy.exc import IntegrityError
 from binaryornot.check import is_binary
 from gevent import sleep
 import base64
 
-from sqlalchemy.orm import load_only
 from werkzeug.exceptions import HTTPException
 
 from mergin.sync.forms import project_name_validation
@@ -87,6 +86,7 @@
     is_versioned_file,
     get_project_path,
     get_device_id,
+    is_valid_path,
 )
 from .errors import StorageLimitHit
 from ..utils import format_time_delta
@@ -807,10 +807,13 @@ def project_push(namespace, project_name):
         abort(400, "Another process is running. Please try later.")
 
     upload_changes = ChangesSchema(context={"version": version + 1}).load(changes)
-    # check if same file is not already uploaded
+
     for item in upload_changes.added:
+        # check if same file is not already uploaded
         if not all(ele.path != item.path for ele in project.files):
             abort(400, f"File {item.path} has been already uploaded")
+        if not is_valid_path(item.path):
+            abort(400, f"File {item.path} contains invalid characters.")
 
     # changes' files must be unique
     changes_files = [

server/mergin/sync/utils.py

@@ -15,7 +15,12 @@
 from flask import Request
 from typing import Optional
 from sqlalchemy import text
-from pathvalidate import validate_filename, ValidationError
+from pathvalidate import (
+    validate_filename,
+    ValidationError,
+    is_valid_filepath,
+    is_valid_filename,
+)
 
 
 def generate_checksum(file, chunk_size=4096):
@@ -284,7 +289,7 @@ def has_valid_first_character(name: str) -> str | None:
     return None
 
 
-def is_valid_filename(name: str) -> str | None:
+def check_filename(name: str) -> str | None:
     """Check if name contains only valid characters for filename"""
     error = None
     try:
@@ -352,3 +357,14 @@ def files_size():
         """
     )
     return db.session.execute(files_size).scalar()
+
+
+def is_valid_path(filepath: str) -> bool:
+    """Check filepath and filename for invalid characters, absolute path or path traversal"""
+    return (
+        not len(re.split(r"\.[/\\]", filepath)) > 1  # ./ or .\
+        and is_valid_filepath(filepath)  # invalid characters in filepath, absolute path
+        and is_valid_filename(
+            os.path.basename(filepath)
+        )  # invalid characters in filename, reserved filenames
+    )

server/mergin/tests/test_project_controller.py

@@ -36,8 +36,7 @@
     ProjectFilePath,
 )
 from ..sync.files import ChangesSchema
-from ..sync.permissions import projects_query
-from ..sync.schemas import ProjectListSchema, ProjectSchema
+from ..sync.schemas import ProjectListSchema
 from ..sync.utils import generate_checksum, is_versioned_file
 from ..auth.models import User, UserProfile
 
@@ -1487,8 +1486,7 @@ def test_whole_push_process(client):
         "テスト.txt",
         "£¥§.txt",
         "name_1_1.txt",
-        "name\\1\\1.txt",
-        "+?%@&.qgs",
+        "+%@&.qgs",
     ]
     # prepare dummy files
     os.mkdir(test_dir)
@@ -2514,3 +2512,35 @@ def test_signals(client):
     ) as push_finished_mock:
         upload_file_to_project(project, "test.txt", client)
         push_finished_mock.assert_called_once()
+
+
+def test_upload_validation(client):
+    """Test filepath and filename validation during file upload"""
+    push_start_url = url_for(
+        f"/v1.mergin_sync_public_api_controller_project_push",
+        namespace=test_workspace_name,
+        project_name=test_project,
+    )
+    filename = "image.png"
+    with open(os.path.join(TMP_DIR, filename), "w") as f:
+        f.write("Hello, Mergin!")
+    changes = {
+        "added": [file_info(TMP_DIR, filename, chunk_size=CHUNK_SIZE)],
+        "updated": [],
+        "removed": [],
+    }
+    # Manipulate the path by prepending ../../
+    manipulated_path = "../../image.png"
+    changes["added"][0]["path"] = manipulated_path
+    # Block script upload in push_start because of the invalid path
+    resp = client.post(
+        push_start_url,
+        data=json.dumps(
+            {"version": "v1", "changes": changes}, cls=DateTimeEncoder
+        ).encode("utf-8"),
+        headers=json_headers,
+    )
+    assert resp.status_code == 400
+    assert (
+        resp.json["detail"] == f"File {manipulated_path} contains invalid characters."
+    )

server/mergin/tests/test_utils.py

@@ -17,7 +17,8 @@
     is_reserved_word,
     has_valid_characters,
     has_valid_first_character,
-    is_valid_filename,
+    check_filename,
+    is_valid_path,
 )
 from ..auth.models import LoginHistory, User
 from . import json_headers
@@ -151,7 +152,7 @@ def test_is_name_allowed():
         ("-project", True),
         ("proj_ect", True),
         ("proj.ect", True),
-        # We are repmoving ! from valids
+        # We are removing ! from valids
         ("proj!ect", False),
         (" project", False),
         (".project", False),
@@ -225,8 +226,30 @@ def test_is_name_allowed():
             not (
                 has_valid_characters(name)
                 and has_valid_first_character(name)
-                and is_valid_filename(name)
+                and check_filename(name)
                 and is_reserved_word(name)
             )
             == expected
         )
+
+
+filepaths = [
+    ("/home/user/mm/project/survey.gpkg", False),
+    ("C:\Documents\Summer2018.pdf", False),
+    ("\Desktop\Summer2019.pdf", False),
+    ("../image.png", False),
+    ("./image.png", False),
+    ("assets/photos/im?age.png", False),
+    ("assets/photos/CON.png", False),
+    ("assets/photos/CONfile.png", True),
+    ("image.png", True),
+    ("images/image.png", True),
+    ("media/photos/image.png", True),
+    ("med..ia/pho.tos/ima...ge.png", True),
+    ("med/../ia/pho.tos/ima...ge.png", False),
+]
+
+
+@pytest.mark.parametrize("filepath,allow", filepaths)
+def test_is_valid_path(client, filepath, allow):
+    assert is_valid_path(filepath) == allow