✨ Enhance confirmation service integration: add methods for managing confirmation tokens in user workflows

Author: pcrespov

services/web/server/src/simcore_service_webserver/login/_confirmation_service.py

@@ -8,6 +8,7 @@
 
 import logging
 from datetime import UTC, datetime
+from typing import Any
 
 from models_library.users import UserID
 
@@ -85,3 +86,42 @@ async def validate_confirmation_code(self, code: str) -> Confirmation | None:
             )
             return None
         return confirmation
+
+    async def delete_confirmation(self, confirmation: Confirmation) -> None:
+        """Delete a confirmation token."""
+        await self._repository.delete_confirmation(confirmation=confirmation)
+
+    async def delete_confirmation_and_update_user(
+        self,
+        confirmation: Confirmation,
+        user_id: UserID,
+        updates: dict[str, Any],
+    ) -> None:
+        """Atomically delete confirmation and update user."""
+        await self._repository.delete_confirmation_and_update_user(
+            confirmation=confirmation,
+            user_id=user_id,
+            updates=updates,
+        )
+
+    async def get_confirmation(
+        self, filter_dict: dict[str, Any]
+    ) -> Confirmation | None:
+        """Get a confirmation by filter criteria."""
+        return await self._repository.get_confirmation(filter_dict=filter_dict)
+
+    async def create_confirmation(
+        self, user_id: UserID, action: ActionLiteralStr, data: str | None = None
+    ) -> Confirmation:
+        """Create a new confirmation token for a user action."""
+        return await self._repository.create_confirmation(
+            user_id=user_id, action=action, data=data
+        )
+
+    async def delete_confirmation_and_user(
+        self, user_id: UserID, confirmation: Confirmation
+    ) -> None:
+        """Atomically delete confirmation and user."""
+        await self._repository.delete_confirmation_and_user(
+            user_id=user_id, confirmation=confirmation
+        )

services/web/server/src/simcore_service_webserver/login/_controller/rest/change.py

@@ -15,9 +15,10 @@
 from ....utils import HOUR
 from ....utils_rate_limiting import global_rate_limit_route
 from ....web_utils import flash_response
-from ... import _auth_service, _confirmation_service, _confirmation_web
+from ... import _auth_service, _confirmation_web
+from ..._confirmation_repository import ConfirmationRepository
+from ..._confirmation_service import ConfirmationService
 from ..._emails_service import get_template_path, send_email_from_template
-from ..._login_repository_legacy import AsyncpgStorage, get_plugin_storage
 from ..._login_service import (
     ACTIVE,
     CHANGE_EMAIL,
@@ -33,12 +34,20 @@
 )
 from ...decorators import login_required
 from ...errors import WrongPasswordError
-from ...settings import LoginOptions, get_plugin_options
+from ...settings import get_plugin_options
 from .change_schemas import ChangeEmailBody, ChangePasswordBody, ResetPasswordBody
 
 _logger = logging.getLogger(__name__)
 
 
+def _get_confirmation_service(app: web.Application) -> ConfirmationService:
+    """Get confirmation service instance from app."""
+    engine = app["postgres_db_engine"]
+    repository = ConfirmationRepository(engine)
+    options = get_plugin_options(app)
+    return ConfirmationService(repository, options)
+
+
 routes = RouteTableDef()
 
 
@@ -82,8 +91,6 @@ async def initiate_reset_password(request: web.Request):
      - 4. Who requested the reset?
     """
 
-    db: AsyncpgStorage = get_plugin_storage(request.app)
-    cfg: LoginOptions = get_plugin_options(request.app)
     product: Product = products_web.get_current_product(request)
 
     request_body = await parse_request_body_as(ResetPasswordBody, request)
@@ -177,16 +184,15 @@ def _get_error_context(
         try:
             # Confirmation token that includes code to `complete_reset_password`.
             # Recreated if non-existent or expired  (Guideline #2)
+            confirmation_service = _get_confirmation_service(request.app)
             confirmation = (
-                await _confirmation_service.get_or_create_confirmation_without_data(
-                    cfg, db, user_id=user["id"], action="RESET_PASSWORD"
+                await confirmation_service.get_or_create_confirmation_without_data(
+                    user_id=user["id"], action="RESET_PASSWORD"
                 )
             )
 
             # Produce a link so that the front-end can hit `complete_reset_password`
-            link = _confirmation_web.make_confirmation_link(
-                request, confirmation["code"]
-            )
+            link = _confirmation_web.make_confirmation_link(request, confirmation.code)
 
             # primary reset email with a URL and the normal instructions.
             await send_email_from_template(
@@ -220,8 +226,8 @@ def _get_error_context(
 
 async def initiate_change_email(request: web.Request):
     # NOTE: This code have been intentially disabled in https://github.com/ITISFoundation/osparc-simcore/pull/5472
-    db: AsyncpgStorage = get_plugin_storage(request.app)
     product: Product = products_web.get_current_product(request)
+    confirmation_service = _get_confirmation_service(request.app)
 
     request_body = await parse_request_body_as(ChangeEmailBody, request)
 
@@ -238,15 +244,17 @@ async def initiate_change_email(request: web.Request):
         raise web.HTTPUnprocessableEntity(text="This email cannot be used")
 
     # Reset if previously requested
-    confirmation = await db.get_confirmation({"user": user, "action": CHANGE_EMAIL})
-    if confirmation:
-        await db.delete_confirmation(confirmation)
+    existing_confirmation = await confirmation_service.get_confirmation(
+        filter_dict={"user_id": user["id"], "action": CHANGE_EMAIL}
+    )
+    if existing_confirmation:
+        await confirmation_service.delete_confirmation(existing_confirmation)
 
     # create new confirmation to ensure email is actually valid
-    confirmation = await db.create_confirmation(
+    confirmation = await confirmation_service.create_confirmation(
         user_id=user["id"], action="CHANGE_EMAIL", data=request_body.email
     )
-    link = _confirmation_web.make_confirmation_link(request, confirmation["code"])
+    link = _confirmation_web.make_confirmation_link(request, confirmation.code)
     try:
         await send_email_from_template(
             request,
@@ -261,7 +269,7 @@ async def initiate_change_email(request: web.Request):
         )
     except Exception as err:  # pylint: disable=broad-except
         _logger.exception("Can not send change_email_email")
-        await db.delete_confirmation(confirmation)
+        await confirmation_service.delete_confirmation(confirmation)
         raise web.HTTPServiceUnavailable(text=MSG_CANT_SEND_MAIL) from err
 
     return flash_response(MSG_CHANGE_EMAIL_REQUESTED)

services/web/server/src/simcore_service_webserver/login/_controller/rest/confirmation.py

@@ -26,15 +26,14 @@
 from ....web_utils import flash_response
 from ... import (
     _auth_service,
-    _confirmation_service,
     _registration_service,
     _security_service,
     _twofa_service,
 )
+from ..._confirmation_repository import ConfirmationRepository
+from ..._confirmation_service import ConfirmationService
 from ..._login_repository_legacy import (
-    AsyncpgStorage,
     ConfirmationTokenDict,
-    get_plugin_storage,
 )
 from ..._login_service import (
     ACTIVE,
@@ -43,6 +42,7 @@
     RESET_PASSWORD,
     notify_user_confirmation,
 )
+from ..._models import Confirmation
 from ...constants import (
     MSG_PASSWORD_CHANGE_NOT_ALLOWED,
     MSG_PASSWORD_CHANGED,
@@ -64,47 +64,66 @@
 _logger = logging.getLogger(__name__)
 
 
+def _get_confirmation_service(app: web.Application) -> ConfirmationService:
+    """Get confirmation service instance from app."""
+    engine = app["postgres_db_engine"]
+    repository = ConfirmationRepository(engine)
+    options = get_plugin_options(app)
+    return ConfirmationService(repository, options)
+
+
+def _confirmation_to_legacy_dict(confirmation: Confirmation) -> ConfirmationTokenDict:
+    """Convert new Confirmation model to legacy ConfirmationTokenDict format."""
+    return {
+        "code": confirmation.code,
+        "user_id": confirmation.user_id,
+        "action": confirmation.action,
+        "data": confirmation.data,
+        "created_at": confirmation.created_at,
+    }
+
+
 routes = RouteTableDef()
 
 
 async def _handle_confirm_registration(
     app: web.Application,
     product_name: ProductName,
-    confirmation: ConfirmationTokenDict,
+    confirmation: Confirmation,
 ):
-    db: AsyncpgStorage = get_plugin_storage(app)
-    user_id = confirmation["user_id"]
+    confirmation_service = _get_confirmation_service(app)
+    user_id = confirmation.user_id
 
     # activate user and consume confirmation token
-    await db.delete_confirmation_and_update_user(
+    await confirmation_service.delete_confirmation_and_update_user(
+        confirmation=confirmation,
         user_id=user_id,
         updates={"status": ACTIVE},
-        confirmation=confirmation,
     )
 
     await notify_user_confirmation(
         app,
         user_id=user_id,
         product_name=product_name,
-        extra_credits_in_usd=parse_extra_credits_in_usd_or_none(confirmation),
+        extra_credits_in_usd=parse_extra_credits_in_usd_or_none(
+            _confirmation_to_legacy_dict(confirmation)
+        ),
     )
 
 
 async def _handle_confirm_change_email(
-    app: web.Application, confirmation: ConfirmationTokenDict
+    app: web.Application, confirmation: Confirmation
 ):
-    db: AsyncpgStorage = get_plugin_storage(app)
-    user_id = confirmation["user_id"]
+    confirmation_service = _get_confirmation_service(app)
+    user_id = confirmation.user_id
 
     # update and consume confirmation token
-    await db.delete_confirmation_and_update_user(
+    await confirmation_service.delete_confirmation_and_update_user(
+        confirmation=confirmation,
         user_id=user_id,
         updates={
-            "email": TypeAdapter(LowerCaseEmailStr).validate_python(
-                confirmation["data"]
-            )
+            "email": TypeAdapter(LowerCaseEmailStr).validate_python(confirmation.data)
         },
-        confirmation=confirmation,
     )
 
 
@@ -125,22 +144,20 @@ async def validate_confirmation_and_redirect(request: web.Request):
             - show the reset-password page
             - use the token to submit a POST /v0/auth/confirmation/{code} and finalize reset action
     """
-    db: AsyncpgStorage = get_plugin_storage(request.app)
     cfg: LoginOptions = get_plugin_options(request.app)
     product: Product = products_web.get_current_product(request)
+    confirmation_service = _get_confirmation_service(request.app)
 
     path_params = parse_request_path_parameters_as(CodePathParam, request)
 
-    confirmation: ConfirmationTokenDict | None = (
-        await _confirmation_service.validate_confirmation_code(
-            path_params.code.get_secret_value(),
-            db=db,
-            cfg=cfg,
+    confirmation: Confirmation | None = (
+        await confirmation_service.validate_confirmation_code(
+            path_params.code.get_secret_value()
         )
     )
 
     redirect_to_login_url = URL(cfg.LOGIN_REDIRECT)
-    if confirmation and (action := confirmation["action"]):
+    if confirmation and (action := confirmation.action):
         try:
             if action == REGISTRATION:
                 await _handle_confirm_registration(
@@ -249,20 +266,19 @@ async def complete_reset_password(request: web.Request):
     - Changes password using a token code without login
     - Code is provided via email by calling first initiate_reset_password
     """
-    db: AsyncpgStorage = get_plugin_storage(request.app)
-    cfg: LoginOptions = get_plugin_options(request.app)
     product: Product = products_web.get_current_product(request)
+    confirmation_service = _get_confirmation_service(request.app)
 
     path_params = parse_request_path_parameters_as(CodePathParam, request)
     request_body = await parse_request_body_as(ResetPasswordConfirmation, request)
 
-    confirmation = await _confirmation_service.validate_confirmation_code(
-        code=path_params.code.get_secret_value(), db=db, cfg=cfg
+    confirmation = await confirmation_service.validate_confirmation_code(
+        code=path_params.code.get_secret_value()
     )
 
     if confirmation:
         user = await _auth_service.get_user_or_none(
-            request.app, user_id=confirmation["user_id"]
+            request.app, user_id=confirmation.user_id
         )
         assert user  # nosec
 
@@ -274,7 +290,7 @@ async def complete_reset_password(request: web.Request):
             verify_current_password=False,  # confirmed by code
         )
 
-        await db.delete_confirmation(confirmation)
+        await confirmation_service.delete_confirmation(confirmation)
 
         return flash_response(MSG_PASSWORD_CHANGED)
 

services/web/server/src/simcore_service_webserver/login/_controller/rest/registration.py

@@ -30,6 +30,8 @@
     _security_service,
     _twofa_service,
 )
+from ..._confirmation_repository import ConfirmationRepository
+from ..._confirmation_service import ConfirmationService
 from ..._emails_service import get_template_path, send_email_from_template
 from ..._invitations_service import (
     ConfirmedInvitationData,
@@ -39,12 +41,12 @@
 )
 from ..._login_repository_legacy import (
     AsyncpgStorage,
-    ConfirmationTokenDict,
     get_plugin_storage,
 )
 from ..._login_service import (
     notify_user_confirmation,
 )
+from ..._models import Confirmation
 from ...constants import (
     CODE_2FA_SMS_CODE_REQUIRED,
     MAX_2FA_CODE_RESEND,
@@ -70,6 +72,14 @@
 _logger = logging.getLogger(__name__)
 
 
+def _get_confirmation_service(app: web.Application) -> ConfirmationService:
+    """Get confirmation service instance from app."""
+    engine = app["postgres_db_engine"]
+    repository = ConfirmationRepository(engine)
+    options = get_plugin_options(app)
+    return ConfirmationService(repository, options)
+
+
 routes = RouteTableDef()
 
 
@@ -224,15 +234,16 @@ async def register(request: web.Request):
 
     if settings.LOGIN_REGISTRATION_CONFIRMATION_REQUIRED:
         # Confirmation required: send confirmation email
-        _confirmation: ConfirmationTokenDict = await db.create_confirmation(
+        confirmation_service = _get_confirmation_service(request.app)
+        _confirmation: Confirmation = await confirmation_service.create_confirmation(
             user_id=user["id"],
             action="REGISTRATION",
             data=invitation.model_dump_json() if invitation else None,
         )
 
         try:
             email_confirmation_url = _confirmation_web.make_confirmation_link(
-                request, _confirmation["code"]
+                request, _confirmation.code
             )
             email_template_path = await get_template_path(
                 request, "registration_email.jinja2"
@@ -270,7 +281,9 @@ async def register(request: web.Request):
                 )
             )
 
-            await db.delete_confirmation_and_user(user["id"], _confirmation)
+            await confirmation_service.delete_confirmation_and_user(
+                user_id=user["id"], confirmation=_confirmation
+            )
 
             raise web.HTTPServiceUnavailable(text=user_error_msg) from err