Validate usernames

src/dstack/_internal/server/services/users.py

@@ -1,5 +1,6 @@
 import hashlib
 import os
+import re
 import uuid
 from datetime import timezone
 from typing import Awaitable, Callable, List, Optional, Tuple
@@ -8,7 +9,7 @@
 from sqlalchemy import func as safunc
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from dstack._internal.core.errors import ResourceExistsError
+from dstack._internal.core.errors import ResourceExistsError, ServerClientError
 from dstack._internal.core.models.users import (
     GlobalRole,
     User,
@@ -78,6 +79,7 @@ async def create_user(
     active: bool = True,
     token: Optional[str] = None,
 ) -> UserModel:
+    validate_username(username)
     user_model = await get_user_model_by_name(session=session, username=username, ignore_case=True)
     if user_model is not None:
         raise ResourceExistsError()
@@ -225,6 +227,15 @@ def get_user_permissions(user_model: UserModel) -> UserPermissions:
     )
 
 
+def validate_username(username: str):
+    if not is_valid_username(username):
+        raise ServerClientError("Username should match regex '^[a-zA-Z0-9-_]{1,60}$'")
+
+
+def is_valid_username(username: str) -> bool:
+    return re.match("^[a-zA-Z0-9-_]{1,60}$", username) is not None
+
+
 _CREATE_USER_HOOKS = []
 
 

src/tests/_internal/server/routers/test_users.py

@@ -236,6 +236,29 @@ async def test_return_400_if_username_taken(
         )
         assert len(res.scalars().all()) == 1
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    @freeze_time(datetime(2023, 1, 2, 3, 4, tzinfo=timezone.utc))
+    async def test_returns_400_if_username_invalid(
+        self,
+        test_db,
+        session: AsyncSession,
+        client: AsyncClient,
+    ):
+        user = await create_user(
+            name="admin",
+            session=session,
+        )
+        response = await client.post(
+            "/api/users/create",
+            headers=get_auth_headers(user.token),
+            json={
+                "username": "Invalid#$username",
+                "global_role": GlobalRole.USER,
+            },
+        )
+        assert response.status_code == 400
+
 
 class TestDeleteUsers:
     @pytest.mark.asyncio

src/tests/_internal/server/services/test_users.py

@@ -0,0 +1,29 @@
+import pytest
+
+from dstack._internal.server.services.users import is_valid_username
+
+
+class TestIsValidUsername:
+    @pytest.mark.parametrize(
+        "username",
+        [
+            "special#$symbols",
+            "A,B",
+            "",
+            "a" * 61,
+        ],
+    )
+    def test_valid(self, username: str):
+        assert not is_valid_username(username)
+
+    @pytest.mark.parametrize(
+        "username",
+        [
+            "regularusername",
+            "CaseUsername",
+            "username_with_underscores-and-dashes1234",
+            "a" * 60,
+        ],
+    )
+    def test_invalid(self, username: str):
+        assert is_valid_username(username)