Merge pull request #67 from flavien-hugs/develop

feat: single authentication per device

.isort.cfg

@@ -1,2 +1,2 @@
 [settings]
-known_third_party = beanie,email_validator,fastapi,fastapi_cache,fastapi_jwt,fastapi_pagination,httpx,jinja2,jose,mongomock_motor,pwdlib,pydantic,pydantic_settings,pymongo,pyotp,pytest,pytest_asyncio,slugify,starlette,typer,uvicorn,yaml
+known_third_party = beanie,email_validator,fastapi,fastapi_cache,fastapi_jwt,fastapi_pagination,getmac,httpx,jinja2,jose,mongomock_motor,pwdlib,pydantic,pydantic_settings,pymongo,pyotp,pytest,pytest_asyncio,slugify,starlette,typer,uvicorn,yaml

pyproject.toml

@@ -22,6 +22,7 @@ user-agents = "^2.2.0"
 python-multipart = "^0.0.12"
 cachetools = "^5.5.0"
 python-jose = "^3.3.0"
+getmac = "0.9.5"
 
 
 [tool.poetry.group.test.dependencies]

src/services/auth/auth.py

@@ -1,9 +1,10 @@
-from datetime import datetime, timezone
+from datetime import datetime, timezone, UTC
 from typing import Optional
 
 from beanie import PydanticObjectId
 from fastapi import Request, status
 from fastapi.encoders import jsonable_encoder
+from getmac import get_mac_address
 from starlette.responses import JSONResponse
 
 from src.common.helpers.caching import delete_custom_key
@@ -62,15 +63,47 @@ async def login(request: Request, payload: LoginUser) -> JSONResponse:
         )
 
     role = await get_one_role(role_id=PydanticObjectId(user.role))
+
+    # Récupérer l'adresse IP et le device_id
+    address_ip = request.client.host
+    forwarded_for = request.headers.get("X-Forwarded-For")
+    if forwarded_for:
+        address_ip = forwarded_for.split(",")[0]
+
+    device_id = (
+        get_mac_address(ip=address_ip)
+        or get_mac_address(interface="eth0")
+        or get_mac_address(ip=address_ip, network_request=True)
+    )
+
+    # Vérifier l'authentification unique par appareil
+    if user.attributes["device_id"] and user.attributes["device_id"] != device_id:
+        raise CustomHTTPException(
+            code_error=AuthErrorCode.AUTH_ALREADY_LOGGED_IN,
+            message_error="You are already logged in on another device.",
+            status_code=status.HTTP_403_FORBIDDEN,
+        )
+
+    # Mettre à jour les informations de l'utilisateur
+    current_time = datetime.now(tz=UTC)
+    update_data = {"last_login": current_time, "address_ip": address_ip, "device_id": device_id}
+    await user.set({"attributes": {**user.attributes, **update_data}, "updated_at": current_time})
+
     user_data = user.model_dump(
         by_alias=True,
         mode="json",
-        exclude={"password", "attributes.otp_secret", "attributes.otp_created_at", "is_primary"},
+        exclude={
+            "password",
+            "attributes.otp_secret",
+            "attributes.otp_created_at",
+            "is_primary",
+        },
     )
     user_data.update(
         {"role": role.model_dump(by_alias=True, mode="json", exclude={"permissions", "created_at", "updated_at"})}
     )
 
+    # Générer les tokens
     response_data = {
         "access_token": CustomAccessBearer.access_token(data=jsonable_encoder(user_data), user_id=str(user.id)),
         "referesh_token": CustomAccessBearer.refresh_token(data=jsonable_encoder(user_data), user_id=str(user.id)),
@@ -82,8 +115,14 @@ async def login(request: Request, payload: LoginUser) -> JSONResponse:
 
 async def logout(request: Request) -> JSONResponse:
     authorization = request.headers.get("Authorization")
-    token = authorization.split()[1]
-    await blacklist_token.add_blacklist_token(token)
+    token = authorization.split()[1] if authorization else None
+
+    decode_token = CustomAccessBearer.decode_access_token(token=token)
+    user_id = decode_token.get("subject", {}).get("_id")
+    user = await get_one_user(user_id=PydanticObjectId(user_id))
+    await user.set({"attributes": {**user.attributes, "device_id": None}})
+
+    await blacklist_token.add_blacklist_token(token=token)
 
     await delete_custom_key(settings.APP_NAME + "access")
     await delete_custom_key(settings.APP_NAME + "validate")

src/shared/error_codes.py

@@ -14,6 +14,7 @@ class AuthErrorCode(StrEnum):
     AUTH_INSUFFICIENT_PERMISSION = "auth/insufficient-permission"
     AUTH_OTP_NOT_VALID = "auth/otp-not-valid"
     AUTH_OTP_EXPIRED = "auth/otp-code-expired"
+    AUTH_ALREADY_LOGGED_IN = "auth/already-logged-in-another-device"
 
 
 class UserErrorCode(StrEnum):

tests/services/test_auth.py

@@ -1,5 +1,6 @@
 import json
 from unittest import mock
+from datetime import datetime
 
 import pytest
 from starlette import status
@@ -14,6 +15,8 @@
 
 
 @pytest.mark.asyncio
+@mock.patch("src.services.auth.auth.get_mac_address")
+@mock.patch("src.services.auth.auth.User.set", new_callable=mock.AsyncMock)
 @mock.patch("src.services.auth.auth.User.find_one", new_callable=mock.AsyncMock)
 @mock.patch("src.services.auth.auth.verify_password", return_value=True)
 @mock.patch("src.services.auth.auth.get_one_role", new_callable=mock.AsyncMock)
@@ -25,10 +28,15 @@ async def test_login_success(
     mock_get_one_role,
     mock_verify_password,
     mock_find_one,
+    mock_user_set,
+    mock_get_mac_address,
     fixture_models,
     mock_task,
     mock_request,
 ):
+    # setup device_id mock
+    mock_get_mac_address.return_value = "test_device_id"
+
     for register_with_email in [True, False]:
         settings.REGISTER_WITH_EMAIL = register_with_email
 
@@ -40,6 +48,11 @@ async def test_login_success(
                 password="hashedpassword",
                 role=PydanticObjectId("66e85363aa07cb1e95d3e3d0"),
                 is_active=True,
+                attributes={
+                    "device_id": None,
+                    "address_ip": None,
+                    "last_login": None
+                }
             )
             payload = LoginUser(email="test@example.com", password="testpassword")
         else:
@@ -51,21 +64,38 @@ async def test_login_success(
                 password="hashedpassword",
                 role=PydanticObjectId("66e85363aa07cb1e95d3e3d0"),
                 is_active=True,
+                attributes={
+                    "device_id": None,
+                    "address_ip": None,
+                    "last_login": None
+                }
             )
             payload = LoginUser(phonenumber="+2250151571396", password="testpassword")
 
+        # setup mocks
         mock_find_one.return_value = fake_user
         mock_get_one_role.return_value.model_dump = mock.Mock(
             return_value={"_id": "66e85363aa07cb1e95d3e3d0", "name": "admin"}
         )
+        mock_user_set.return_value = fake_user
+
+        # mock request headers for X-Forwarded-For
+        mock_request.headers = {"X-Forwarded-For": "192.168.1.1"}
+        mock_request.client.host = "127.0.0.1"
 
+        # execute login
         response = await auth.login(request=mock_request, payload=payload)
+
+        # assertions
         assert isinstance(response, JSONResponse)
         assert response.status_code == status.HTTP_200_OK
 
         response_data = json.loads(response.body.decode())
+
+        # token assertions
         assert response_data["access_token"] == "access_token"
-        assert response_data["referesh_token"] == "refresh_token"
+
+        # user data assertions
         assert response_data["user"][identifier] == payload.email if register_with_email else payload.phonenumber
         assert response_data["user"]["role"]["name"] == "admin"