✨♻️ improving 2FA (OPS ⚠️)

api/specs/web-server/_auth.py

@@ -17,13 +17,13 @@
 from models_library.generics import Envelope
 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 (
     LoginBody,
     LoginNextPage,
     LoginTwoFactorAuthBody,
     LogoutBody,
 )
-from simcore_service_webserver.login.handlers_2fa import Resend2faBody
 from simcore_service_webserver.login.handlers_change import (
     ChangeEmailBody,
     ChangePasswordBody,

packages/models-library/src/models_library/authentification.py

@@ -0,0 +1,9 @@
+from enum import auto
+
+from .utils.enums import StrAutoEnum
+
+
+class TwoFAAuthentificationMethod(StrAutoEnum):
+    sms = auto()
+    email = auto()
+    disabled = auto()

services/web/server/src/simcore_service_webserver/application_settings.py

@@ -381,7 +381,10 @@ def to_client_statics(self) -> dict[str, Any]:
                 "SIMCORE_VCS_RELEASE_URL": True,
                 "SWARM_STACK_NAME": True,
                 "WEBSERVER_PROJECTS": {"PROJECTS_MAX_NUM_RUNNING_DYNAMIC_NODES"},
-                "WEBSERVER_LOGIN": {"LOGIN_ACCOUNT_DELETION_RETENTION_DAYS"},
+                "WEBSERVER_LOGIN": {
+                    "LOGIN_ACCOUNT_DELETION_RETENTION_DAYS",
+                    "LOGIN_2FA_REQUIRED",
+                },
                 "WEBSERVER_SESSION": {"SESSION_COOKIE_MAX_AGE"},
             },
             exclude_none=True,

services/web/server/src/simcore_service_webserver/login/_2fa.py → services/web/server/src/simcore_service_webserver/login/_2fa_api.py

@@ -11,12 +11,15 @@
 import logging
 
 from aiohttp import web
+from models_library.users import UserID
 from pydantic import BaseModel, Field
-from servicelib.logging_utils import log_decorator
+from servicelib.error_codes import create_error_code
+from servicelib.logging_utils import LogExtra, get_log_record_extra, log_decorator
 from servicelib.utils_secrets import generate_passcode
 from settings_library.twilio import TwilioSettings
 from twilio.rest import Client
 
+from ..login.errors import SendingVerificationEmailError, SendingVerificationSmsError
 from ..products.api import Product
 from ..redis import get_redis_validation_code_client
 from .utils_email import get_template_path, send_email_from_template
@@ -95,33 +98,46 @@ async def send_sms_code(
     twilio_messaging_sid: str,
     twilio_alpha_numeric_sender: str,
     first_name: str,
+    user_id: UserID | None = None,
 ):
-    create_kwargs = {
-        "messaging_service_sid": twilio_messaging_sid,
-        "to": phone_number,
-        "body": f"Dear {first_name}, your verification code is {code}",
-    }
-    if twilo_auth.is_alphanumeric_supported(phone_number):
-        create_kwargs["from_"] = twilio_alpha_numeric_sender
-
-    def _sender():
-        log.info(
-            "Sending sms code to %s from product %s",
-            f"{phone_number=}",
-            twilio_alpha_numeric_sender,
+    try:
+        create_kwargs = {
+            "messaging_service_sid": twilio_messaging_sid,
+            "to": phone_number,
+            "body": f"Dear {first_name}, your verification code is {code}",
+        }
+        if twilo_auth.is_alphanumeric_supported(phone_number):
+            create_kwargs["from_"] = twilio_alpha_numeric_sender
+
+        def _sender():
+            log.info(
+                "Sending sms code to %s from product %s",
+                f"{phone_number=}",
+                twilio_alpha_numeric_sender,
+            )
+            #
+            # SEE https://www.twilio.com/docs/sms/quickstart/python
+            #
+            client = Client(twilo_auth.TWILIO_ACCOUNT_SID, twilo_auth.TWILIO_AUTH_TOKEN)
+            message = client.messages.create(**create_kwargs)
+
+            log.debug(
+                "Got twilio client %s",
+                f"{message=}",
+            )
+
+        await asyncio.get_event_loop().run_in_executor(executor=None, func=_sender)
+
+    except Exception as exc:
+        error_code = create_error_code(exc)
+        more_extra: LogExtra = get_log_record_extra(user_id=user_id) or {}
+        log.exception(
+            "Failed while setting up 2FA code and sending SMS to %s [%s]",
+            mask_phone_number(phone_number),
+            f"{error_code}",
+            extra={"error_code": error_code, **more_extra},
         )
-        #
-        # SEE https://www.twilio.com/docs/sms/quickstart/python
-        #
-        client = Client(twilo_auth.TWILIO_ACCOUNT_SID, twilo_auth.TWILIO_AUTH_TOKEN)
-        message = client.messages.create(**create_kwargs)
-
-        log.debug(
-            "Got twilio client %s",
-            f"{message=}",
-        )
-
-    await asyncio.get_event_loop().run_in_executor(executor=None, func=_sender)
+        raise SendingVerificationSmsError(reason=exc) from exc
 
 
 #
@@ -141,21 +157,33 @@ async def send_email_code(
     code: str,
     first_name: str,
     product: Product,
+    user_id: UserID | None = None,
 ):
-    email_template_path = await get_template_path(request, "new_2fa_code.jinja2")
-    await send_email_from_template(
-        request,
-        from_=support_email,
-        to=user_email,
-        template=email_template_path,
-        context={
-            "host": request.host,
-            "code": code,
-            "name": first_name,
-            "support_email": support_email,
-            "product": product,
-        },
-    )
+    try:
+        email_template_path = await get_template_path(request, "new_2fa_code.jinja2")
+        await send_email_from_template(
+            request,
+            from_=support_email,
+            to=user_email,
+            template=email_template_path,
+            context={
+                "host": request.host,
+                "code": code,
+                "name": first_name,
+                "support_email": support_email,
+                "product": product,
+            },
+        )
+    except Exception as exc:
+        error_code = create_error_code(exc)
+        more_extra: LogExtra = get_log_record_extra(user_id=user_id) or {}
+        log.exception(
+            "Failed while setting up 2FA code and sending Email to %s [%s]",
+            user_email,
+            f"{error_code}",
+            extra={"error_code": error_code, **more_extra},
+        )
+        raise SendingVerificationEmailError(reason=exc) from exc
 
 
 #

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

@@ -0,0 +1,145 @@
+import logging
+from typing import Literal
+
+from aiohttp import web
+from aiohttp.web import RouteTableDef
+from models_library.emails import LowerCaseEmailStr
+from pydantic import Field
+from servicelib.aiohttp import status
+from servicelib.aiohttp.requests_validation import parse_request_body_as
+from servicelib.mimetype_constants import MIMETYPE_APPLICATION_JSON
+
+from ..products.api import Product, get_current_product
+from ..session.access_policies import session_access_required
+from ._2fa_api import (
+    create_2fa_code,
+    delete_2fa_code,
+    get_2fa_code,
+    mask_phone_number,
+    send_email_code,
+    send_sms_code,
+)
+from ._constants import (
+    CODE_2FA_EMAIL_CODE_REQUIRED,
+    CODE_2FA_SMS_CODE_REQUIRED,
+    MSG_2FA_CODE_SENT,
+    MSG_EMAIL_SENT,
+    MSG_UNKNOWN_EMAIL,
+)
+from ._models import InputSchema
+from .errors import handle_login_exceptions
+from .settings import LoginSettingsForProduct, get_plugin_settings
+from .storage import AsyncpgStorage, get_plugin_storage
+from .utils import envelope_response
+
+_logger = logging.getLogger(__name__)
+
+
+routes = RouteTableDef()
+
+
+class Resend2faBody(InputSchema):
+    email: LowerCaseEmailStr = Field(..., description="User email (identifier)")
+    via: Literal["SMS", "Email"] = "SMS"
+
+
+@routes.post("/v0/auth/two_factor:resend", name="auth_resend_2fa_code")
+@session_access_required(
+    name="auth_resend_2fa_code",
+    one_time_access=False,
+)
+@handle_login_exceptions
+async def resend_2fa_code(request: web.Request):
+    """Resends 2FA code via SMS/Email"""
+    product: Product = get_current_product(request)
+    settings: LoginSettingsForProduct = get_plugin_settings(
+        request.app, product_name=product.name
+    )
+    db: AsyncpgStorage = get_plugin_storage(request.app)
+    resend_2fa_ = await parse_request_body_as(Resend2faBody, request)
+
+    user = await db.get_user({"email": resend_2fa_.email})
+    if not user:
+        raise web.HTTPUnauthorized(
+            reason=MSG_UNKNOWN_EMAIL, content_type=MIMETYPE_APPLICATION_JSON
+        )
+
+    if not settings.LOGIN_2FA_REQUIRED:
+        raise web.HTTPServiceUnavailable(
+            reason="2FA login is not available",
+            content_type=MIMETYPE_APPLICATION_JSON,
+        )
+
+    # Already a code?
+    previous_code = await get_2fa_code(request.app, user_email=resend_2fa_.email)
+    if previous_code is not None:
+        await delete_2fa_code(request.app, user_email=resend_2fa_.email)
+
+    # guaranteed by LoginSettingsForProduct
+    assert settings.LOGIN_2FA_REQUIRED  # nosec
+    assert settings.LOGIN_TWILIO  # nosec
+    assert product.twilio_messaging_sid  # nosec
+
+    # creates and stores code
+    code = await create_2fa_code(
+        request.app,
+        user_email=user["email"],
+        expiration_in_seconds=settings.LOGIN_2FA_CODE_EXPIRATION_SEC,
+    )
+
+    # sends via SMS
+    if resend_2fa_.via == "SMS":
+        await send_sms_code(
+            phone_number=user["phone"],
+            code=code,
+            twilo_auth=settings.LOGIN_TWILIO,
+            twilio_messaging_sid=product.twilio_messaging_sid,
+            twilio_alpha_numeric_sender=product.twilio_alpha_numeric_sender_id,
+            first_name=user["first_name"] or user["name"],
+            user_id=user["id"],
+        )
+
+        response = envelope_response(
+            {
+                "name": CODE_2FA_SMS_CODE_REQUIRED,
+                "parameters": {
+                    "message": MSG_2FA_CODE_SENT.format(
+                        phone_number=mask_phone_number(user["phone"])
+                    ),
+                    "retry_2fa_after": settings.LOGIN_2FA_CODE_EXPIRATION_SEC,
+                },
+                # NOTE: REMOVE when frontend is refactored
+                "reason": MSG_2FA_CODE_SENT.format(
+                    phone_number=mask_phone_number(user["phone"])
+                ),
+            },
+            status=status.HTTP_200_OK,
+        )
+
+    # sends via Email
+    else:
+        assert resend_2fa_.via == "Email"  # nosec
+        await send_email_code(
+            request,
+            user_email=user["email"],
+            support_email=product.support_email,
+            code=code,
+            first_name=user["first_name"] or user["name"],
+            product=product,
+            user_id=user["id"],
+        )
+
+        response = envelope_response(
+            {
+                "name": CODE_2FA_EMAIL_CODE_REQUIRED,
+                "parameters": {
+                    "message": MSG_EMAIL_SENT.format(email=user["email"]),
+                    "retry_2fa_after": settings.LOGIN_2FA_CODE_EXPIRATION_SEC,
+                },
+                # NOTE: REMOVE when frontend is refactored
+                "reason": MSG_EMAIL_SENT.format(email=user["email"]),
+            },
+            status=status.HTTP_200_OK,
+        )
+
+    return response