moving confirmation web helpers

Author: pcrespov

api/specs/web-server/_projects_groups.py

@@ -27,7 +27,7 @@
 @router.post(
     "/projects/{project_id}:share",
     response_model=Envelope[ProjectGroupGet],
-    status_code=status.HTTP_201_CREATED,
+    status_code=status.HTTP_204_NO_CONTENT,
 )
 async def share_project(
     _path: Annotated[ProjectPathParams, Depends()],

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

@@ -8,11 +8,8 @@
 
 import logging
 from datetime import datetime
-from urllib.parse import quote
 
-from aiohttp import web
 from models_library.users import UserID
-from yarl import URL
 
 from ._login_repository_legacy import (
     ActionLiteralStr,
@@ -56,19 +53,6 @@ def get_expiration_date(
     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="")
-    return app.router["auth_confirmation"].url_for(code=safe_code)
-
-
-def make_confirmation_link(
-    request: web.Request, confirmation: ConfirmationTokenDict
-) -> str:
-    link = _url_for_confirmation(request.app, code=confirmation["code"])
-    return f"{request.scheme}://{request.host}{link}"
-
-
 def is_confirmation_expired(cfg: LoginOptions, confirmation: ConfirmationTokenDict):
     age = datetime.utcnow() - confirmation["created_at"]
     lifetime = cfg.get_confirmation_lifetime(confirmation["action"])

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

@@ -0,0 +1,16 @@
+from urllib.parse import quote
+
+from aiohttp import web
+from models_library.basic_types import IDStr
+from yarl import URL
+
+
+def _url_for_confirmation(app: web.Application, code: IDStr) -> URL:
+    # NOTE: this is in a query parameter, and can contain ? for example.
+    safe_code = quote(code, safe="")
+    return app.router["auth_confirmation"].url_for(code=safe_code)
+
+
+def make_confirmation_link(request: web.Request, code: IDStr) -> str:
+    link = _url_for_confirmation(request.app, code=code)
+    return f"{request.scheme}://{request.host}{link}"

services/web/server/src/simcore_service_webserver/login/_controller/rest/change.py

@@ -18,7 +18,7 @@
 from ....users import api as users_service
 from ....utils import HOUR
 from ....utils_rate_limiting import global_rate_limit_route
-from ... import _confirmation_service
+from ... import _confirmation_service, _confirmation_web
 from ..._constants import (
     MSG_CANT_SEND_MAIL,
     MSG_CHANGE_EMAIL_REQUESTED,
@@ -187,7 +187,9 @@ def _get_error_context(
             )
 
             # Produce a link so that the front-end can hit `complete_reset_password`
-            link = _confirmation_service.make_confirmation_link(request, confirmation)
+            link = _confirmation_web.make_confirmation_link(
+                request, confirmation["code"]
+            )
 
             # primary reset email with a URL and the normal instructions.
             await send_email_from_template(
@@ -249,7 +251,7 @@ async def initiate_change_email(request: web.Request):
     confirmation = await db.create_confirmation(
         user_id=user["id"], action="CHANGE_EMAIL", data=request_body.email
     )
-    link = _confirmation_service.make_confirmation_link(request, confirmation)
+    link = _confirmation_web.make_confirmation_link(request, confirmation["code"])
     try:
         await send_email_from_template(
             request,

services/web/server/src/simcore_service_webserver/login/_controller/rest/registration.py

@@ -32,7 +32,7 @@
 from ....utils import MINUTE
 from ....utils_aiohttp import NextPage, envelope_json_response
 from ....utils_rate_limiting import global_rate_limit_route
-from ... import _auth_service, _confirmation_service, _security_service, _twofa_service
+from ... import _auth_service, _confirmation_web, _security_service, _twofa_service
 from ..._constants import (
     CODE_2FA_SMS_CODE_REQUIRED,
     MAX_2FA_CODE_RESEND,
@@ -257,8 +257,8 @@ async def register(request: web.Request):
         )
 
         try:
-            email_confirmation_url = _confirmation_service.make_confirmation_link(
-                request, _confirmation
+            email_confirmation_url = _confirmation_web.make_confirmation_link(
+                request, _confirmation["code"]
             )
             email_template_path = await get_template_path(
                 request, "registration_email.jinja2"

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

@@ -4,6 +4,7 @@
 
 import asyncpg
 from aiohttp import web
+from models_library.basic_types import IDStr
 from servicelib.utils_secrets import generate_passcode
 
 from . import _login_repository_legacy_sql
@@ -21,7 +22,7 @@
 
 
 class BaseConfirmationTokenDict(TypedDict):
-    code: str
+    code: IDStr
     action: ActionLiteralStr
 
 

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

@@ -0,0 +1,5 @@
+from ._confirmation_web import make_confirmation_link
+
+__all__: tuple[str, ...] = ("make_confirmation_link",)
+
+# nopycln: file

services/web/server/src/simcore_service_webserver/projects/_controller/groups_rest.py

@@ -2,6 +2,7 @@
 
 from aiohttp import web
 from models_library.api_schemas_webserver._base import InputSchema
+from models_library.basic_types import IDStr
 from models_library.groups import GroupID
 from models_library.projects import ProjectID
 from pydantic import BaseModel, ConfigDict, EmailStr
@@ -14,6 +15,7 @@
 
 from ..._meta import api_version_prefix as VTAG
 from ...application_settings_utils import requires_dev_feature_enabled
+from ...login import login_web
 from ...login.decorators import login_required
 from ...security.decorators import permission_required
 from ...utils_aiohttp import envelope_json_response
@@ -60,7 +62,7 @@ async def share_project(request: web.Request):
         body_params.sharee_email,
     ):
 
-        await _groups_service.share_project_by_email(
+        code: IDStr = await _groups_service.create_confirmation_action_to_share_project(
             app=request.app,
             user_id=req_ctx.user_id,
             project_id=path_params.project_id,
@@ -71,6 +73,10 @@ async def share_project(request: web.Request):
             product_name=req_ctx.product_name,
         )
 
+        confirmation_link: str = login_web.make_confirmation_link(request, code=code)
+
+        _logger.debug("Send email with confirmation link: %s", confirmation_link)
+
         return web.json_response(status=status.HTTP_204_NO_CONTENT)
 
 

services/web/server/src/simcore_service_webserver/projects/_groups_service.py

@@ -1,12 +1,16 @@
+import hashlib
 import logging
 from datetime import datetime
 
+import arrow
 from aiohttp import web
+from models_library.access_rights import AccessRights
+from models_library.basic_types import IDStr
 from models_library.groups import GroupID
 from models_library.products import ProductName
 from models_library.projects import ProjectID
 from models_library.users import UserID
-from pydantic import BaseModel, EmailStr
+from pydantic import BaseModel, EmailStr, TypeAdapter
 
 from ..users import api as users_service
 from . import _groups_repository
@@ -170,7 +174,7 @@ async def delete_project_group(
     )
 
 
-async def share_project_by_email(
+async def create_confirmation_action_to_share_project(
     app: web.Application,
     *,
     product_name: ProductName,
@@ -181,8 +185,35 @@ async def share_project_by_email(
     read: bool,
     write: bool,
     delete: bool,
-):
-    raise NotImplementedError
+) -> IDStr:
+    assert app  # nosec
+
+    _logger.debug(
+        "Checking that %s in %s has enough access rights (ownership) to %s for sharing",
+        f"{user_id=}",
+        f"{product_name=}",
+        f"{project_id=}",
+    )
+
+    sharer_user_id = user_id
+    shared_resource_type = "project"
+    shared_resource_id = project_id
+    shared_resource_access_rights = AccessRights(read=read, write=write, delete=delete)
+    shared_at = arrow.utcnow().datetime
+
+    _logger.debug(
+        "Creating confirmation token for action=SHARE with and producing a code:"
+        "\n %s," * 6,
+        sharer_user_id,
+        shared_resource_type,
+        shared_resource_id,
+        shared_resource_access_rights,
+        shared_at,
+        sharee_email,
+    )
+
+    fake_code = hashlib.sha256(sharee_email.encode()).hexdigest()
+    return TypeAdapter(IDStr).validate_python(f"fake{fake_code}")
 
 
 ### Operations without checking permissions

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

@@ -3,13 +3,7 @@
 from aiohttp.test_utils import make_mocked_request
 from aiohttp.web import Application
 from pytest_simcore.helpers.webserver_login import UserInfoDict
-from simcore_service_webserver.login._confirmation_service import (
-    get_expiration_date,
-    get_or_create_confirmation_without_data,
-    is_confirmation_expired,
-    make_confirmation_link,
-    validate_confirmation_code,
-)
+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
 
@@ -20,7 +14,7 @@ async def test_confirmation_token_workflow(
     # Step 1: Create a new confirmation token
     user_id = registered_user["id"]
     action = "RESET_PASSWORD"
-    confirmation = await get_or_create_confirmation_without_data(
+    confirmation = await _confirmation_service.get_or_create_confirmation_without_data(
         login_options, db, user_id=user_id, action=action
     )
 
@@ -29,11 +23,15 @@ async def test_confirmation_token_workflow(
     assert confirmation["action"] == action
 
     # Step 2: Check that the token is not expired
-    assert not is_confirmation_expired(login_options, confirmation)
+    assert not _confirmation_service.is_confirmation_expired(
+        login_options, confirmation
+    )
 
     # Step 3: Validate the confirmation code
     code = confirmation["code"]
-    validated_confirmation = await validate_confirmation_code(code, db, login_options)
+    validated_confirmation = await _confirmation_service.validate_confirmation_code(
+        code, db, login_options
+    )
 
     assert validated_confirmation is not None
     assert validated_confirmation["code"] == code
@@ -53,7 +51,9 @@ async def test_confirmation_token_workflow(
     )
 
     # Create confirmation link
-    confirmation_link = make_confirmation_link(request, confirmation)
+    confirmation_link = _confirmation_web.make_confirmation_link(
+        request, confirmation["code"]
+    )
 
     # Assertions
     assert confirmation_link.startswith("http://example.com/auth/confirmation/")
@@ -67,20 +67,26 @@ async def test_expired_confirmation_token(
     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
+    confirmation_1 = (
+        await _confirmation_service.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)
-    assert get_expiration_date(login_options, confirmation_1)
+    assert not _confirmation_service.is_confirmation_expired(
+        login_options, confirmation_1
+    )
+    assert _confirmation_service.get_expiration_date(login_options, confirmation_1)
 
-    confirmation_2 = await 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(
+            login_options, db, user_id=user_id, action=action
+        )
     )
 
     assert confirmation_2 == confirmation_1
@@ -89,23 +95,31 @@ async def test_expired_confirmation_token(
     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
+    confirmation_3 = (
+        await _confirmation_service.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)
+        await _confirmation_service.validate_confirmation_code(
+            confirmation_1["code"], db, login_options
+        )
         is None
     )
     assert (
-        await validate_confirmation_code(confirmation_2["code"], db, login_options)
+        await _confirmation_service.validate_confirmation_code(
+            confirmation_2["code"], db, login_options
+        )
         is None
     )
     assert (
-        await validate_confirmation_code(confirmation_3["code"], db, login_options)
+        await _confirmation_service.validate_confirmation_code(
+            confirmation_3["code"], db, login_options
+        )
         is None
     )