test coverage

Author: pcrespov

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

@@ -24,27 +24,38 @@
 _logger = logging.getLogger(__name__)
 
 
-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
+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(
-        {"code": code}
+        {"user": {"id": user_id}, "action": action}
     )
-    if confirmation and is_confirmation_expired(cfg, confirmation):
+
+    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,
         )
-        return None
+        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 _url_for_confirmation(app: web.Application, code: str) -> URL:
     # NOTE: this is in a query parameter, and can contain ? for example.
     safe_code = quote(code, safe="")
@@ -58,39 +69,28 @@ def make_confirmation_link(
     return f"{request.scheme}://{request.host}{link}"
 
 
-def get_expiration_date(
-    cfg: LoginOptions, confirmation: ConfirmationTokenDict
-) -> datetime:
+def is_confirmation_expired(cfg: LoginOptions, confirmation: ConfirmationTokenDict):
+    age = datetime.utcnow() - confirmation["created_at"]
     lifetime = cfg.get_confirmation_lifetime(confirmation["action"])
-    return confirmation["created_at"] + lifetime
+    return age > lifetime
 
 
-async def get_or_create_confirmation_without_data(
-    cfg: LoginOptions,
-    db: AsyncpgStorage,
-    user_id: UserID,
-    action: ActionLiteralStr,
-) -> ConfirmationTokenDict:
+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(
-        {"user": {"id": user_id}, "action": action}
+        {"code": code}
     )
-
-    if confirmation is not None and is_confirmation_expired(cfg, confirmation):
+    if confirmation and is_confirmation_expired(cfg, confirmation):
         await db.delete_confirmation(confirmation)
         _logger.warning(
             "Used expired token [%s]. Deleted from confirmations table.",
             confirmation,
         )
-        confirmation = None
-
-    if confirmation is None:
-        confirmation = await db.create_confirmation(user_id, action=action)
-
+        return None
     return confirmation
-
-
-def is_confirmation_expired(cfg: LoginOptions, confirmation: ConfirmationTokenDict):
-    age = datetime.utcnow() - confirmation["created_at"]
-    lifetime = cfg.get_confirmation_lifetime(confirmation["action"])
-    return age > lifetime

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

@@ -1,6 +1,8 @@
+from datetime import timedelta
+
 from aiohttp.test_utils import make_mocked_request
 from aiohttp.web import Application
-from models_library.users import UserID
+from pytest_simcore.helpers.webserver_login import UserInfoDict
 from simcore_service_webserver.login._confirmation_service import (
     get_or_create_confirmation_without_data,
     is_confirmation_expired,
@@ -12,9 +14,10 @@
 
 
 async def test_confirmation_token_workflow(
-    db: AsyncpgStorage, login_options: LoginOptions, user_id: UserID
+    db: AsyncpgStorage, login_options: LoginOptions, registered_user: UserInfoDict
 ):
     # Step 1: Create a new confirmation token
+    user_id = registered_user["id"]
     action = "RESET_PASSWORD"
     confirmation = await get_or_create_confirmation_without_data(
         login_options, db, user_id=user_id, action=action
@@ -42,13 +45,65 @@ async def test_confirmation_token_workflow(
         "/auth/confirmation/{code}", lambda request: None, name="auth_confirmation"
     )
     request = make_mocked_request(
-        "GET", "/auth/confirmation/{code}", app=app, headers={"Host": "example.com"}
+        "GET",
+        "http://example.com/auth/confirmation/{code}",
+        app=app,
+        headers={"Host": "example.com"},
     )
-    request.scheme = "http"
 
     # Create confirmation link
     confirmation_link = make_confirmation_link(request, confirmation)
 
     # Assertions
     assert confirmation_link.startswith("http://example.com/auth/confirmation/")
     assert confirmation["code"] in confirmation_link
+
+
+async def test_expired_confirmation_token(
+    db: AsyncpgStorage, login_options: LoginOptions, registered_user: UserInfoDict
+):
+    user_id = registered_user["id"]
+    action = "CHANGE_EMAIL"
+
+    # Create a brand new confirmation token
+    confirmation_1 = await get_or_create_confirmation_without_data(
+        login_options, db, user_id=user_id, action=action
+    )
+
+    assert confirmation_1 is not None
+    assert confirmation_1["user_id"] == user_id
+    assert confirmation_1["action"] == action
+
+    # Check that the token is not expired
+    assert not is_confirmation_expired(login_options, confirmation_1)
+
+    confirmation_2 = await get_or_create_confirmation_without_data(
+        login_options, db, 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_3 = await get_or_create_confirmation_without_data(
+        login_options, db, user_id=user_id, action=action
+    )
+
+    # when expired, it gets renewed
+    assert confirmation_3 != confirmation_1
+
+    # now all have expired
+    assert (
+        await validate_confirmation_code(confirmation_1["code"], db, login_options)
+        is None
+    )
+    assert (
+        await validate_confirmation_code(confirmation_2["code"], db, login_options)
+        is None
+    )
+    assert (
+        await validate_confirmation_code(confirmation_3["code"], db, login_options)
+        is None
+    )