Enforce valid email address for User email

server/mergin/auth/forms.py

@@ -1,7 +1,7 @@
 # Copyright (C) Lutra Consulting Limited
 #
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
-
+import re
 import safe
 from flask_wtf import FlaskForm
 from sqlalchemy import func
@@ -42,6 +42,26 @@ def username_validation(form, field):
             raise ValidationError(error)
 
 
+class ExtendedEmail(Email):
+    """Custom email validation extending WTForms email validation.
+    WTForms checks for
+    1. spaces,
+    2. special characters ,:;()<>[]\"
+    3, multiple @ symbols,
+    4, leading, trailing, or consecutive dots in the local part
+    5, invalid domain part - missing top level domain (user@example), consecutive dots
+    Custom check for additional invalid characters disallows |'— because they make our email sending service to fail
+    """
+
+    def __call__(self, form, field):
+        super().__call__(form, field)
+
+        if re.search(r"[|'—]", field.data):
+            raise ValidationError(
+                f"Email address '{field.data}' contains an invalid character."
+            )
+
+
 class PasswordValidator:
     def __init__(self, min_length=8):
         self.min_length = min_length
@@ -71,7 +91,7 @@ class RegisterUserForm(FlaskForm):
     )
     email = CustomStringField(
         "Email Address",
-        validators=[DataRequired(), Email()],
+        validators=[DataRequired(), ExtendedEmail()],
     )
 
     def validate(self):
@@ -94,7 +114,7 @@ def validate(self):
 class ResetPasswordForm(FlaskForm):
     email = CustomStringField(
         "Email Address",
-        [DataRequired(), Email()],
+        [DataRequired(), ExtendedEmail()],
     )
 
 
@@ -129,7 +149,7 @@ class UserProfileDataForm(UpdateForm):
     receive_notifications = BooleanField("Receive notifications", [Optional()])
     first_name = CustomStringField("First Name", [Optional()])
     last_name = CustomStringField("Last Name", [Optional()])
-    email = CustomStringField("Email", [Optional(), Email()])
+    email = CustomStringField("Email", [Optional(), ExtendedEmail()])
 
 
 class ApiLoginForm(LoginForm):

server/mergin/auth/models.py

@@ -199,7 +199,7 @@ def generate_username(cls, email: str) -> Optional[str]:
         username = email.split("@")[0].strip().lower()
         # remove forbidden chars
         username = re.sub(
-            r"[\@\#\$\%\^\&\*\(\)\{\}\[\]\?\'\"`,;\:\+\=\~\\\/\|\<\>]", "", username
+            r"[\@\#\$\%\^\&\*\(\)\{\}\[\]\?\'\"`,;\:\+\=\~\\\/\|\<\>—]", "", username
         ).ljust(4, "0")
         # additional check for reserved words
         username = f"{username}0" if is_reserved_word(username) else username

server/mergin/sync/public_api_controller.py

@@ -824,7 +824,7 @@ def project_push(namespace, project_name):
             abort(
                 400,
                 f"Unsupported file type detected: {item.path}. "
-                f"Please remove the file or try compressing it into a ZIP file before uploading",
+                f"Please remove the file or try compressing it into a ZIP file before uploading.",
             )
 
     # changes' files must be unique
@@ -1056,7 +1056,11 @@ def push_finish(transaction_id):
                 continue
         if not is_supported_type(dest_file):
             logging.info(f"Rejecting blacklisted file: {dest_file}")
-            abort(400, f"Unsupported file type detected: {f.path}")
+            abort(
+                400,
+                f"Unsupported file type detected: {f.path}. "
+                f"Please remove the file or try compressing it into a ZIP file before uploading.",
+            )
 
         if expected_size != os.path.getsize(dest_file):
             logging.error(

server/mergin/sync/schemas.py

@@ -16,7 +16,6 @@
     FileHistory,
     PushChangeType,
     ProjectRole,
-    ProjectUser,
 )
 from .workspace import WorkspaceRole
 from ..app import DateTimeWithZ, ma

server/mergin/sync/utils.py

@@ -285,7 +285,7 @@ def has_valid_characters(name: str) -> str | None:
 
 def has_valid_first_character(name: str) -> str | None:
     """Check if name contains only valid characters in first position"""
-    if re.match(r"^[\s^\.].*$", name) is not None:
+    if re.match(r"^[\s.].*$", name) is not None:
         return f"Value can not start with space or dot."
     return None
 

server/mergin/tests/test_auth.py

@@ -3,16 +3,14 @@
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
 
 from datetime import datetime, timedelta
-import os
 import time
 import pytest
 import json
 from flask import url_for
-from itsdangerous import URLSafeTimedSerializer
 from sqlalchemy import desc
 from unittest.mock import patch
 
-from mergin.tests import test_workspace
+from ..auth.forms import ResetPasswordForm
 from ..auth.app import generate_confirmation_token, confirm_token
 from ..auth.models import User, UserProfile, LoginHistory
 from ..auth.tasks import anonymize_removed_users
@@ -32,7 +30,6 @@
     login_as_admin,
     login,
     upload_file_to_project,
-    test_project_dir,
 )
 
 
@@ -118,6 +115,8 @@ def test_logout(client):
     ),  # tests with upper case, but email already exists
     (" mergin@mergin.com  ", "#pwd123", 400),  # invalid password
     ("verylonglonglonglonglonglonglongemail@example.com", "#pwd1234", 201),
+    ("us.er@mergin.com", "#pwd1234", 201),  # dot is allowed
+    ("us er@mergin.com", "#pwd1234", 400),  # space is disallowed
 ]
 
 
@@ -889,3 +888,31 @@ def test_server_usage(client):
     assert resp.json["editors"] == 1
     assert resp.json["users"] == 1
     assert resp.json["projects"] == 1
+
+
+user_data = [
+    ("user1", True),  # no problem
+    ("日人日本人", True),  # non-ascii character
+    ("usér", True),  # non-ascii character
+    ("user\\", False),  # disallowed character
+    ("user\260", True),  # non-ascii character (°)
+    ("user|", False),  # vertical bar
+    ("us er", False),  # space in the middle
+    ("us,er", False),  # comma
+    ("us—er", False),  # dash
+    ("us'er", False),  # apostrophe
+    (" user", True),  # starting with space (will be stripped)
+    ("us.er", True),  # dot in the middle
+    (".user", False),  # starting with dot
+]
+
+
+@pytest.mark.parametrize("username,is_valid", user_data)
+def test_email_format_validation(app, username, is_valid):
+    """Test email form format validation"""
+    email = username + "@email.com"
+    with app.test_request_context(method="POST"):
+        form = ResetPasswordForm(
+            data={"email": email}
+        )  # ResetPasswordForm has only email field
+        assert form.validate() == is_valid

server/mergin/tests/test_project_controller.py

@@ -2576,7 +2576,7 @@ def test_supported_file_upload(client):
     assert resp.status_code == 400
     assert (
         resp.json["detail"]
-        == f"Unsupported file type detected: {script_filename}. Please remove the file or try compressing it into a ZIP file before uploading"
+        == f"Unsupported file type detected: {script_filename}. Please remove the file or try compressing it into a ZIP file before uploading."
     )
     # Extension spoofing to trick the validator
     spoof_name = "script.gpkg"
@@ -2613,4 +2613,7 @@ def test_supported_file_upload(client):
     # Unsupported file type is revealed when reconstructed from chunks - based on the mime type - and upload is refused
     resp = client.post(f"/v1/project/push/finish/{upload.id}")
     assert resp.status_code == 400
-    assert resp.json["detail"] == f"Unsupported file type detected: {spoof_name}"
+    assert (
+        resp.json["detail"]
+        == f"Unsupported file type detected: {spoof_name}. Please remove the file or try compressing it into a ZIP file before uploading."
+    )

server/migrations/community/1ab5b02ce532_version_author_name_to_user_id.py

@@ -1,4 +1,4 @@
-"""Migrage project version author name to user.id
+"""Migrate project version author name to user.id
 
 Revision ID: 1ab5b02ce532
 Revises: 1c23e3be03a3