✨ Refactor confirmation service and repository integration for improved token management

Author: pcrespov

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

@@ -7,74 +7,81 @@
 """
 
 import logging
-from datetime import datetime
+from datetime import UTC, datetime
 
 from models_library.users import UserID
 
-from ._login_repository_legacy import (
-    ActionLiteralStr,
-    AsyncpgStorage,
-    ConfirmationTokenDict,
-)
+from ._confirmation_repository import ConfirmationRepository
+from ._models import ActionLiteralStr, Confirmation
 from .settings import LoginOptions
 
 _logger = logging.getLogger(__name__)
 
 
-async def get_or_create_confirmation_without_data(
-    cfg: LoginOptions,
-    db: AsyncpgStorage,
-    user_id: UserID,
-    action: ActionLiteralStr,
-) -> ConfirmationTokenDict:
-
-    confirmation: ConfirmationTokenDict | None = await db.get_confirmation(
-        {"user": {"id": user_id}, "action": action}
-    )
-
-    if confirmation is not None and is_confirmation_expired(cfg, confirmation):
-        await db.delete_confirmation(confirmation)
-        _logger.warning(
-            "Used expired token [%s]. Deleted from confirmations table.",
-            confirmation,
+class ConfirmationService:
+    """Service for managing confirmation tokens and codes."""
+
+    def __init__(
+        self,
+        confirmation_repository: ConfirmationRepository,
+        login_options: LoginOptions,
+    ) -> None:
+        self._repository = confirmation_repository
+        self._options = login_options
+
+    @property
+    def options(self) -> LoginOptions:
+        """Access to login options for testing purposes."""
+        return self._options
+
+    async def get_or_create_confirmation_without_data(
+        self,
+        user_id: UserID,
+        action: ActionLiteralStr,
+    ) -> Confirmation:
+        """Get existing or create new confirmation token for user action."""
+        confirmation = await self._repository.get_confirmation(
+            filter_dict={"user_id": user_id, "action": action}
         )
-        confirmation = None
-
-    if confirmation is None:
-        confirmation = await db.create_confirmation(user_id, action=action)
-
-    return confirmation
-
-
-def get_expiration_date(
-    cfg: LoginOptions, confirmation: ConfirmationTokenDict
-) -> datetime:
-    lifetime = cfg.get_confirmation_lifetime(confirmation["action"])
-    return confirmation["created_at"] + lifetime
-
-
-def is_confirmation_expired(cfg: LoginOptions, confirmation: ConfirmationTokenDict):
-    age = datetime.utcnow() - confirmation["created_at"]
-    lifetime = cfg.get_confirmation_lifetime(confirmation["action"])
-    return age > lifetime
-
-
-async def validate_confirmation_code(
-    code: str, db: AsyncpgStorage, cfg: LoginOptions
-) -> ConfirmationTokenDict | None:
-    """
-    Returns None if validation fails
-    """
-    assert not code.startswith("***"), "forgot .get_secret_value()??"  # nosec
 
-    confirmation: ConfirmationTokenDict | None = await db.get_confirmation(
-        {"code": code}
-    )
-    if confirmation and is_confirmation_expired(cfg, confirmation):
-        await db.delete_confirmation(confirmation)
-        _logger.warning(
-            "Used expired token [%s]. Deleted from confirmations table.",
-            confirmation,
+        if confirmation is not None and self.is_confirmation_expired(confirmation):
+            await self._repository.delete_confirmation(confirmation=confirmation)
+            _logger.warning(
+                "Used expired token [%s]. Deleted from confirmations table.",
+                confirmation,
+            )
+            confirmation = None
+
+        if confirmation is None:
+            confirmation = await self._repository.create_confirmation(
+                user_id=user_id, action=action
+            )
+
+        return confirmation
+
+    def get_expiration_date(self, confirmation: Confirmation) -> datetime:
+        """Get expiration date for confirmation token."""
+        lifetime = self._options.get_confirmation_lifetime(confirmation.action)
+        return confirmation.created_at + lifetime
+
+    def is_confirmation_expired(self, confirmation: Confirmation) -> bool:
+        """Check if confirmation token has expired."""
+        age = datetime.now(tz=UTC) - confirmation.created_at
+        lifetime = self._options.get_confirmation_lifetime(confirmation.action)
+        return age > lifetime
+
+    async def validate_confirmation_code(self, code: str) -> Confirmation | None:
+        """Validate confirmation code and return confirmation if valid."""
+        assert not code.startswith("***"), "forgot .get_secret_value()??"  # nosec
+
+        confirmation = await self._repository.get_confirmation(
+            filter_dict={"code": code}
         )
-        return None
-    return confirmation
+        if confirmation and self.is_confirmation_expired(confirmation):
+            await self._repository.delete_confirmation(confirmation=confirmation)
+            _logger.warning(
+                "Used expired token [%s]. Deleted from confirmations table.",
+                confirmation,
+            )
+            return None
+        return confirmation

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

@@ -13,12 +13,10 @@
 
 
 class Confirmation(BaseModel):
-    model_config = ConfigDict(from_attributes=True)
-
     code: str
     user_id: UserID
     action: ActionLiteralStr
-    data: str | None
+    data: str | None = None
     created_at: datetime
 
 

services/web/server/tests/unit/with_dbs/03/login/conftest.py

@@ -17,6 +17,10 @@
 from pytest_simcore.helpers.webserver_users import NewUser, UserInfoDict
 from simcore_postgres_database.models.users import users
 from simcore_postgres_database.models.wallets import wallets
+from simcore_service_webserver.login._confirmation_repository import (
+    ConfirmationRepository,
+)
+from simcore_service_webserver.login._confirmation_service import ConfirmationService
 from simcore_service_webserver.login._login_repository_legacy import (
     AsyncpgStorage,
     get_plugin_storage,
@@ -84,14 +88,35 @@ def fake_weak_password(faker: Faker) -> str:
 
 
 @pytest.fixture
-def db(client: TestClient) -> AsyncpgStorage:
+def db_storage_deprecated(client: TestClient) -> AsyncpgStorage:
     """login database repository instance"""
     assert client.app
     db: AsyncpgStorage = get_plugin_storage(client.app)
     assert db
     return db
 
 
+@pytest.fixture
+def confirmation_repository(client: TestClient) -> ConfirmationRepository:
+    """Modern confirmation repository instance"""
+    assert client.app
+    # Get the async engine from the application
+    engine = client.app["postgres_db_engine"]
+    return ConfirmationRepository(engine)
+
+
+@pytest.fixture
+def confirmation_service(
+    confirmation_repository: ConfirmationRepository, login_options: LoginOptions
+) -> ConfirmationService:
+    """Confirmation service instance"""
+    from simcore_service_webserver.login._confirmation_service import (
+        ConfirmationService,
+    )
+
+    return ConfirmationService(confirmation_repository, login_options)
+
+
 @pytest.fixture
 def login_options(client: TestClient) -> LoginOptions:
     """app's login options"""

services/web/server/tests/unit/with_dbs/03/login/test_login_confirmation_service.py

@@ -1,47 +1,47 @@
 from datetime import timedelta
 
 from aiohttp.test_utils import make_mocked_request
-from aiohttp.web import Application
+from aiohttp.web import Application, Response
 from pytest_simcore.helpers.webserver_users import UserInfoDict
-from simcore_service_webserver.login import _confirmation_service, _confirmation_web
-from simcore_service_webserver.login._login_repository_legacy import AsyncpgStorage
-from simcore_service_webserver.login.settings import LoginOptions
+from simcore_service_webserver.login import _confirmation_web
+from simcore_service_webserver.login._confirmation_service import ConfirmationService
 
 
 async def test_confirmation_token_workflow(
-    db: AsyncpgStorage, login_options: LoginOptions, registered_user: UserInfoDict
+    confirmation_service: ConfirmationService,
+    registered_user: UserInfoDict,
 ):
     # Step 1: Create a new confirmation token
     user_id = registered_user["id"]
     action = "RESET_PASSWORD"
-    confirmation = await _confirmation_service.get_or_create_confirmation_without_data(
-        login_options, db, user_id=user_id, action=action
+    confirmation = await confirmation_service.get_or_create_confirmation_without_data(
+        user_id=user_id, action=action
     )
 
     assert confirmation is not None
-    assert confirmation["user_id"] == user_id
-    assert confirmation["action"] == action
+    assert confirmation.user_id == user_id
+    assert confirmation.action == action
 
     # Step 2: Check that the token is not expired
-    assert not _confirmation_service.is_confirmation_expired(
-        login_options, confirmation
-    )
+    assert not confirmation_service.is_confirmation_expired(confirmation)
 
     # Step 3: Validate the confirmation code
-    code = confirmation["code"]
-    validated_confirmation = await _confirmation_service.validate_confirmation_code(
-        code, db, login_options
-    )
+    code = confirmation.code
+    validated_confirmation = await confirmation_service.validate_confirmation_code(code)
 
     assert validated_confirmation is not None
-    assert validated_confirmation["code"] == code
-    assert validated_confirmation["user_id"] == user_id
-    assert validated_confirmation["action"] == action
+    assert validated_confirmation.code == code
+    assert validated_confirmation.user_id == user_id
+    assert validated_confirmation.action == action
 
     # Step 4: Create confirmation link
     app = Application()
+
+    async def mock_handler(request):
+        return Response()
+
     app.router.add_get(
-        "/auth/confirmation/{code}", lambda request: None, name="auth_confirmation"
+        "/auth/confirmation/{code}", mock_handler, name="auth_confirmation"
     )
     request = make_mocked_request(
         "GET",
@@ -52,74 +52,63 @@ async def test_confirmation_token_workflow(
 
     # Create confirmation link
     confirmation_link = _confirmation_web.make_confirmation_link(
-        request, confirmation["code"]
+        request, confirmation.code
     )
 
     # Assertions
     assert confirmation_link.startswith("https://example.com/auth/confirmation/")
-    assert confirmation["code"] in confirmation_link
+    assert confirmation.code in confirmation_link
 
 
 async def test_expired_confirmation_token(
-    db: AsyncpgStorage, login_options: LoginOptions, registered_user: UserInfoDict
+    confirmation_service: ConfirmationService,
+    registered_user: UserInfoDict,
 ):
     user_id = registered_user["id"]
     action = "CHANGE_EMAIL"
 
     # Create a brand new confirmation token
-    confirmation_1 = (
-        await _confirmation_service.get_or_create_confirmation_without_data(
-            login_options, db, user_id=user_id, action=action
-        )
+    confirmation_1 = await confirmation_service.get_or_create_confirmation_without_data(
+        user_id=user_id, action=action
     )
 
     assert confirmation_1 is not None
-    assert confirmation_1["user_id"] == user_id
-    assert confirmation_1["action"] == action
+    assert confirmation_1.user_id == user_id
+    assert confirmation_1.action == action
 
     # Check that the token is not expired
-    assert not _confirmation_service.is_confirmation_expired(
-        login_options, confirmation_1
-    )
-    assert _confirmation_service.get_expiration_date(login_options, confirmation_1)
+    assert not confirmation_service.is_confirmation_expired(confirmation_1)
+    assert confirmation_service.get_expiration_date(confirmation_1)
 
-    confirmation_2 = (
-        await _confirmation_service.get_or_create_confirmation_without_data(
-            login_options, db, user_id=user_id, action=action
-        )
+    confirmation_2 = await confirmation_service.get_or_create_confirmation_without_data(
+        user_id=user_id, action=action
     )
 
     assert confirmation_2 == confirmation_1
 
     # Enforce ALL EXPIRED
-    login_options.CHANGE_EMAIL_CONFIRMATION_LIFETIME = 0
-    assert login_options.get_confirmation_lifetime(action) == timedelta(seconds=0)
+    confirmation_service.options.CHANGE_EMAIL_CONFIRMATION_LIFETIME = 0
+    assert confirmation_service.options.get_confirmation_lifetime(action) == timedelta(
+        seconds=0
+    )
 
-    confirmation_3 = (
-        await _confirmation_service.get_or_create_confirmation_without_data(
-            login_options, db, user_id=user_id, action=action
-        )
+    confirmation_3 = await confirmation_service.get_or_create_confirmation_without_data(
+        user_id=user_id, action=action
     )
 
     # when expired, it gets renewed
     assert confirmation_3 != confirmation_1
 
     # now all have expired
     assert (
-        await _confirmation_service.validate_confirmation_code(
-            confirmation_1["code"], db, login_options
-        )
+        await confirmation_service.validate_confirmation_code(confirmation_1.code)
         is None
     )
     assert (
-        await _confirmation_service.validate_confirmation_code(
-            confirmation_2["code"], db, login_options
-        )
+        await confirmation_service.validate_confirmation_code(confirmation_2.code)
         is None
     )
     assert (
-        await _confirmation_service.validate_confirmation_code(
-            confirmation_3["code"], db, login_options
-        )
+        await confirmation_service.validate_confirmation_code(confirmation_3.code)
         is None
     )