Implement password validation mechanism (#632)

* Implement password validation mechanism

* Add invalid password reason

* Always pass user in password validator

* Add password validation documentation

tests/conftest.py

@@ -1,6 +1,7 @@
 import asyncio
 from typing import AsyncGenerator, List, Optional
 
+import asynctest
 import httpx
 import pytest
 from asgi_lifespan import LifespanManager
@@ -15,6 +16,7 @@ from fastapi_users.authentication import Authenticator, BaseAuthentication
 from fastapi_users.db import BaseUserDatabase
 from fastapi_users.models import BaseOAuthAccount, BaseOAuthAccountMixin, BaseUserDB
 from fastapi_users.password import get_password_hash
+from fastapi_users.user import InvalidPasswordException, ValidatePasswordProtocol
 
 guinevere_password_hash = get_password_hash("guinevere")
 angharad_password_hash = get_password_hash("angharad")
@@ -400,3 +402,14 @@ def oauth_client() -> OAuth2:
         ACCESS_TOKEN_ENDPOINT,
         name="service1",
     )
+
+
+@pytest.fixture
+def validate_password() -> ValidatePasswordProtocol:
+    async def _validate_password(password: str, user: models.UD) -> None:
+        if len(password) < 3:
+            raise InvalidPasswordException(
+                reason="Password should be at least 3 characters"
+            )
+
+    return asynctest.CoroutineMock(wraps=_validate_password)

tests/test_fastapi_users.py

@@ -11,7 +11,7 @@ from tests.conftest import User, UserCreate, UserDB, UserUpdate
 @pytest.fixture
 @pytest.mark.asyncio
 async def test_app_client(
-    mock_user_db, mock_authentication, oauth_client, get_test_client
+    mock_user_db, mock_authentication, oauth_client, get_test_client, validate_password
 ) -> AsyncGenerator[httpx.AsyncClient, None]:
     fastapi_users = FastAPIUsers(
         mock_user_db,
@@ -20,6 +20,7 @@ async def test_app_client(
         UserCreate,
         UserUpdate,
         UserDB,
+        validate_password,
     )
 
     app = FastAPI()

tests/test_router_register.py

@@ -30,7 +30,7 @@ def after_register(request):
 @pytest.fixture
 @pytest.mark.asyncio
 async def test_app_client(
-    mock_user_db, after_register, get_test_client
+    mock_user_db, after_register, get_test_client, validate_password
 ) -> AsyncGenerator[httpx.AsyncClient, None]:
     create_user = get_create_user(mock_user_db, UserDB)
     register_router = get_register_router(
@@ -38,6 +38,7 @@ async def test_app_client(
         User,
         UserCreate,
         after_register,
+        validate_password,
     )
 
     app = FastAPI()
@@ -79,6 +80,22 @@ class TestRegister:
         assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
         assert after_register.called is False
 
+    async def test_invalid_password(
+        self, test_app_client: httpx.AsyncClient, after_register, validate_password
+    ):
+        json = {"email": "king.arthur@camelot.bt", "password": "g"}
+        response = await test_app_client.post("/register", json=json)
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        data = cast(Dict[str, Any], response.json())
+        assert data["detail"] == {
+            "code": ErrorCode.REGISTER_INVALID_PASSWORD,
+            "reason": "Password should be at least 3 characters",
+        }
+        validate_password.assert_called_with(
+            "g", UserCreate(email="king.arthur@camelot.bt", password="g")
+        )
+        assert after_register.called is False
+
     @pytest.mark.parametrize(
         "email", ["king.arthur@camelot.bt", "King.Arthur@camelot.bt"]
     )

tests/test_router_reset.py

@@ -56,13 +56,18 @@ def after_reset_password(request):
 @pytest.mark.asyncio
 async def test_app_client(
     mock_user_db,
-    mock_authentication,
     after_forgot_password,
     after_reset_password,
     get_test_client,
+    validate_password,
 ) -> AsyncGenerator[httpx.AsyncClient, None]:
     reset_router = get_reset_password_router(
-        mock_user_db, SECRET, LIFETIME, after_forgot_password, after_reset_password
+        mock_user_db,
+        SECRET,
+        LIFETIME,
+        after_forgot_password,
+        after_reset_password,
+        validate_password,
     )
 
     app = FastAPI()
@@ -217,6 +222,33 @@ class TestResetPassword:
         assert mock_user_db.update.called is False
         assert after_reset_password.called is False
 
+    async def test_invalid_password(
+        self,
+        mocker,
+        mock_user_db,
+        test_app_client: httpx.AsyncClient,
+        forgot_password_token,
+        user: UserDB,
+        after_reset_password,
+        validate_password,
+    ):
+        mocker.spy(mock_user_db, "update")
+
+        json = {
+            "token": forgot_password_token(user.id),
+            "password": "h",
+        }
+        response = await test_app_client.post("/reset-password", json=json)
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        data = cast(Dict[str, Any], response.json())
+        assert data["detail"] == {
+            "code": ErrorCode.RESET_PASSWORD_INVALID_PASSWORD,
+            "reason": "Password should be at least 3 characters",
+        }
+        validate_password.assert_called_with("h", user)
+        assert mock_user_db.update.called is False
+        assert after_reset_password.called is False
+
     async def test_existing_user(
         self,
         mocker,

tests/test_router_users.py

@@ -28,7 +28,7 @@ def after_update(request):
 
 
 @pytest.fixture
-def app_factory(mock_user_db, mock_authentication, after_update):
+def app_factory(mock_user_db, mock_authentication, after_update, validate_password):
     def _app_factory(requires_verification: bool) -> FastAPI:
         mock_authentication_bis = MockAuthentication(name="mock-bis")
         authenticator = Authenticator(
@@ -43,6 +43,7 @@ def app_factory(mock_user_db, mock_authentication, after_update):
             authenticator,
             after_update,
             requires_verification=requires_verification,
+            validate_password=validate_password,
         )
 
         app = FastAPI()
@@ -166,6 +167,32 @@ class TestUpdateMe:
             assert data["detail"] == ErrorCode.UPDATE_USER_EMAIL_ALREADY_EXISTS
             assert after_update.called is False
 
+    async def test_invalid_password(
+        self,
+        test_app_client: Tuple[httpx.AsyncClient, bool],
+        user: UserDB,
+        after_update,
+        validate_password,
+    ):
+        client, requires_verification = test_app_client
+        response = await client.patch(
+            "/me",
+            json={"password": "m"},
+            headers={"Authorization": f"Bearer {user.id}"},
+        )
+        if requires_verification:
+            assert response.status_code == status.HTTP_401_UNAUTHORIZED
+            assert after_update.called is False
+        else:
+            assert response.status_code == status.HTTP_400_BAD_REQUEST
+            data = cast(Dict[str, Any], response.json())
+            assert data["detail"] == {
+                "code": ErrorCode.UPDATE_USER_INVALID_PASSWORD,
+                "reason": "Password should be at least 3 characters",
+            }
+            validate_password.assert_called_with("m", user)
+            assert after_update.called is False
+
     async def test_empty_body(
         self,
         test_app_client: Tuple[httpx.AsyncClient, bool],
@@ -726,6 +753,29 @@ class TestUpdateUser:
         assert data["detail"] == ErrorCode.UPDATE_USER_EMAIL_ALREADY_EXISTS
         assert after_update.called is False
 
+    async def test_invalid_password_verified_superuser(
+        self,
+        test_app_client: Tuple[httpx.AsyncClient, bool],
+        user: UserDB,
+        verified_superuser: UserDB,
+        after_update,
+        validate_password,
+    ):
+        client, _ = test_app_client
+        response = await client.patch(
+            f"/{user.id}",
+            json={"password": "m"},
+            headers={"Authorization": f"Bearer {verified_superuser.id}"},
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        data = cast(Dict[str, Any], response.json())
+        assert data["detail"] == {
+            "code": ErrorCode.UPDATE_USER_INVALID_PASSWORD,
+            "reason": "Password should be at least 3 characters",
+        }
+        validate_password.assert_called_with("m", user)
+        assert after_update.called is False
+
     async def test_valid_body_verified_superuser(
         self,
         test_app_client: Tuple[httpx.AsyncClient, bool],