♻️ webserver: Refactor login domain for CSR compliance and future confirmation logic (#7417)

Author: pcrespov

api/specs/web-server/_auth.py

@@ -15,29 +15,29 @@
 from models_library.rest_error import EnvelopedError, Log
 from pydantic import BaseModel, Field, confloat
 from simcore_service_webserver._meta import API_VTAG
-from simcore_service_webserver.login._2fa_handlers import Resend2faBody
-from simcore_service_webserver.login._auth_handlers import (
+from simcore_service_webserver.login._controller.rest.auth import (
     LoginBody,
     LoginNextPage,
     LoginTwoFactorAuthBody,
     LogoutBody,
 )
-from simcore_service_webserver.login.handlers_change import (
+from simcore_service_webserver.login._controller.rest.change import (
     ChangeEmailBody,
     ChangePasswordBody,
     ResetPasswordBody,
 )
-from simcore_service_webserver.login.handlers_confirmation import (
+from simcore_service_webserver.login._controller.rest.confirmation import (
     PhoneConfirmationBody,
     ResetPasswordConfirmation,
 )
-from simcore_service_webserver.login.handlers_registration import (
+from simcore_service_webserver.login._controller.rest.registration import (
     InvitationCheck,
     InvitationInfo,
     RegisterBody,
     RegisterPhoneBody,
     RegisterPhoneNextPage,
 )
+from simcore_service_webserver.login._controller.rest.twofa import Resend2faBody
 
 router = APIRouter(prefix=f"/{API_VTAG}", tags=["auth"])
 

packages/pytest-simcore/src/pytest_simcore/helpers/webserver_login.py

@@ -11,8 +11,11 @@
 from simcore_service_webserver.db.models import UserRole, UserStatus
 from simcore_service_webserver.groups.api import auto_add_user_to_product_group
 from simcore_service_webserver.login._constants import MSG_LOGGED_IN
-from simcore_service_webserver.login._registration import create_invitation_token
-from simcore_service_webserver.login.storage import AsyncpgStorage, get_plugin_storage
+from simcore_service_webserver.login._invitations_service import create_invitation_token
+from simcore_service_webserver.login._login_repository_legacy import (
+    AsyncpgStorage,
+    get_plugin_storage,
+)
 from simcore_service_webserver.products.products_service import list_products
 from simcore_service_webserver.security.api import clean_auth_policy_cache
 from yarl import URL

services/web/server/src/simcore_service_webserver/garbage_collector/_tasks_users.py

@@ -1,5 +1,5 @@
 """
-    Scheduled tasks addressing users
+Scheduled tasks addressing users
 
 """
 
@@ -14,7 +14,7 @@
 from tenacity.before_sleep import before_sleep_log
 from tenacity.wait import wait_exponential
 
-from ..login.utils import notify_user_logout
+from ..login import login_service
 from ..security.api import clean_auth_policy_cache
 from ..users.api import update_expired_users
 
@@ -39,7 +39,7 @@ async def notify_user_logout_all_sessions(
         get_log_record_extra(user_id=user_id),
     ):
         try:
-            await notify_user_logout(app, user_id, client_session_id=None)
+            await login_service.notify_user_logout(app, user_id, client_session_id=None)
         except Exception:  # pylint: disable=broad-except
             _logger.warning(
                 "Ignored error while notifying logout for %s",

services/web/server/src/simcore_service_webserver/login/_auth_api.py renamed to services/web/server/src/simcore_service_webserver/login/_auth_service.py

@@ -10,9 +10,9 @@
 from ..groups.api import is_user_by_email_in_group
 from ..products.models import Product
 from ..security.api import check_password, encrypt_password
+from . import _login_service
 from ._constants import MSG_UNKNOWN_EMAIL, MSG_WRONG_PASSWORD
-from .storage import AsyncpgStorage, get_plugin_storage
-from .utils import validate_user_status
+from ._login_repository_legacy import AsyncpgStorage, get_plugin_storage
 
 
 async def get_user_by_email(app: web.Application, *, email: str) -> dict[str, Any]:
@@ -55,7 +55,7 @@ async def check_authorized_user_credentials_or_raise(
             reason=MSG_UNKNOWN_EMAIL, content_type=MIMETYPE_APPLICATION_JSON
         )
 
-    validate_user_status(user=user, support_email=product.support_email)
+    _login_service.validate_user_status(user=user, support_email=product.support_email)
 
     if not check_password(password, user["password_hash"]):
         raise web.HTTPUnauthorized(

services/web/server/src/simcore_service_webserver/login/_confirmation.py renamed to services/web/server/src/simcore_service_webserver/login/_confirmation_service.py

@@ -14,33 +14,48 @@
 from models_library.users import UserID
 from yarl import URL
 
+from ._login_repository_legacy import (
+    ActionLiteralStr,
+    AsyncpgStorage,
+    ConfirmationTokenDict,
+)
 from .settings import LoginOptions
-from .storage import ActionLiteralStr, AsyncpgStorage, ConfirmationTokenDict
 
-log = logging.getLogger(__name__)
+_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)
-        log.warning(
+        _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="")
@@ -54,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(
-    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)
-        log.warning(
+        _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