cleanup and fixes static analysis

Author: pcrespov

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 ._login_repository_legacy import AsyncpgStorage, get_plugin_storage
-from ._login_service import validate_user_status
 
 
 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_service.py

@@ -21,7 +21,7 @@
 )
 from .settings import LoginOptions
 
-log = logging.getLogger(__name__)
+_logger = logging.getLogger(__name__)
 
 
 async def validate_confirmation_code(
@@ -37,7 +37,7 @@ async def validate_confirmation_code(
     )
     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,
         )
@@ -78,7 +78,7 @@ async def get_or_create_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,
         )

services/web/server/src/simcore_service_webserver/login/_controller/auth_rest.py

@@ -22,7 +22,7 @@
 )
 from ...users import preferences_api as user_preferences_api
 from ...utils_aiohttp import NextPage
-from .. import _auth_service, _security_service, _twofa_service
+from .. import _auth_service, _login_service, _security_service, _twofa_service
 from .._constants import (
     CODE_2FA_EMAIL_CODE_REQUIRED,
     CODE_2FA_SMS_CODE_REQUIRED,
@@ -37,8 +37,6 @@
     MSG_WRONG_2FA_CODE__EXPIRED,
     MSG_WRONG_2FA_CODE__INVALID,
 )
-from .._login_repository_legacy import AsyncpgStorage, get_plugin_storage
-from .._login_service import envelope_response, flash_response, notify_user_logout
 from .._models import InputSchema
 from ..decorators import login_required
 from ..errors import handle_login_exceptions
@@ -136,7 +134,7 @@ async def login(request: web.Request):
         user_2fa_authentification_method == TwoFactorAuthentificationMethod.SMS
         and not user["phone"]
     ):
-        return envelope_response(
+        return _login_service.envelope_response(
             # LoginNextPage
             {
                 "name": CODE_PHONE_NUMBER_REQUIRED,
@@ -171,7 +169,7 @@ async def login(request: web.Request):
             user_id=user["id"],
         )
 
-        return envelope_response(
+        return _login_service.envelope_response(
             # LoginNextPage
             {
                 "name": CODE_2FA_SMS_CODE_REQUIRED,
@@ -198,7 +196,7 @@ async def login(request: web.Request):
         product=product,
         user_id=user["id"],
     )
-    return envelope_response(
+    return _login_service.envelope_response(
         {
             "name": CODE_2FA_EMAIL_CODE_REQUIRED,
             "parameters": {
@@ -226,8 +224,6 @@ async def login_2fa(request: web.Request):
     settings: LoginSettingsForProduct = get_plugin_settings(
         request.app, product_name=product.name
     )
-    db: AsyncpgStorage = get_plugin_storage(request.app)
-
     if not settings.LOGIN_2FA_REQUIRED:
         raise web.HTTPServiceUnavailable(
             reason="2FA login is not available",
@@ -284,8 +280,10 @@ async def logout(request: web.Request) -> web.Response:
         f"{logout_.client_session_id=}",
         extra=get_log_record_extra(user_id=user_id),
     ):
-        response = flash_response(MSG_LOGGED_OUT, "INFO")
-        await notify_user_logout(request.app, user_id, logout_.client_session_id)
+        response = _login_service.flash_response(MSG_LOGGED_OUT, "INFO")
+        await _login_service.notify_user_logout(
+            request.app, user_id, logout_.client_session_id
+        )
         await security_service.forget_identity(request, response)
 
         return response

services/web/server/src/simcore_service_webserver/login/_controller/registration_rest.py

@@ -405,7 +405,7 @@ async def register_phone(request: web.Request):
                     "expiration_2fa": settings.LOGIN_2FA_CODE_EXPIRATION_SEC,
                 },
                 "message": MSG_2FA_CODE_SENT.format(
-                    phone_number=mask_phone_number(registration.phone)
+                    phone_number=_twofa_service.mask_phone_number(registration.phone)
                 ),
                 "level": "INFO",
                 "logger": "user",

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

@@ -1,5 +1,5 @@
 import sys
-from datetime import datetime
+from datetime import UTC, datetime
 
 import typer
 from servicelib.utils_secrets import generate_password
@@ -46,7 +46,7 @@ def invitations(
         fg=typer.colors.BLUE,
     )
 
-    utcnow = datetime.utcnow()
+    utcnow = datetime.now(tz=UTC)
     today: datetime = utcnow.today()
     print("code,user_id,action,data,created_at", file=sys.stdout)
     for n, code in enumerate(codes, start=1):