Require current password when updating password (#621)

fixbackend/auth/schemas.py

@@ -35,7 +35,7 @@ class UserCreate(schemas.BaseUserCreate):
 
 
 class UserUpdate(schemas.BaseUserUpdate):
-    pass
+    current_password: Optional[str] = None
 
 
 class OAuthProviderAuthUrl(BaseModel):

fixbackend/auth/user_manager.py

@@ -29,7 +29,7 @@
 from starlette.responses import Response
 
 from fixbackend.auth.models import User
-from fixbackend.auth.schemas import OTPConfig, UserCreate
+from fixbackend.auth.schemas import OTPConfig, UserCreate, UserUpdate
 from fixbackend.auth.user_repository import UserRepository
 from fixbackend.auth.user_verifier import AuthEmailSender
 from fixbackend.config import Config
@@ -299,6 +299,25 @@ async def validate_password(self, password: str, user: Union[UserCreate, User])
         if not re.search(r"[0-9]", password):
             raise fastapi_users.InvalidPasswordException(reason="Password must contain at least one digit.")
 
+    async def update(
+        self,
+        user_update: UserUpdate,  # type: ignore
+        user: User,
+        safe: bool = False,
+        request: Optional[Request] = None,
+    ) -> User:
+        if user_update.password:
+            if not user_update.current_password:
+                raise exceptions.InvalidPasswordException(reason="Current password is required to update password.")
+
+            db_pwd_hash = user.hashed_password
+            user_pwd_hash = self.password_helper.hash(user_update.current_password)
+
+            if not secrets.compare_digest(db_pwd_hash, user_pwd_hash):
+                raise exceptions.InvalidPasswordException(reason="Current password is incorrect.")
+
+        return await super().update(user_update, user, safe)
+
 
 def get_password_helper(deps: FixDependency) -> PasswordHelperProtocol | None:
     return deps.service(ServiceNames.password_helper, PasswordHelper)

tests/fixbackend/auth/router_test.py

@@ -234,6 +234,20 @@ async def test_registration_flow(
     assert isinstance(event1, WorkspaceCreated)
     assert str(event1.workspace_id) == workspace_json["id"]
 
+    # password can be reset only with providing a current one
+    response = await api_client.patch(
+        "/api/users/me", json={"password": "foobar@foo.com"}, cookies={session_cookie_name: auth_cookie}
+    )
+    assert response.status_code == 400
+
+    # password can be reset with providing a current one
+    response = await api_client.patch(
+        "/api/users/me",
+        json={"password": "FooBar123456789123456789", "current_password": registration_json["password"]},
+        cookies={session_cookie_name: auth_cookie},
+    )
+    assert response.status_code == 200
+
 
 @pytest.mark.asyncio
 async def test_mfa_flow(