Disable account

.vscode/settings.json

@@ -13,7 +13,9 @@
     "[python]": {
         "editor.defaultFormatter": "ms-python.black-formatter"
     },
-    "python.testing.pytestArgs": ["tests"],
+    "python.testing.pytestArgs": [
+        "tests"
+    ],
     "python.testing.unittestEnabled": false,
     "python.testing.pytestEnabled": true,
     "black-formatter.path": [

app/cruds/cruds_external_account.py

@@ -0,0 +1,24 @@
+from sqlalchemy import update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models import models_core
+from app.utils.types.groups_type import GroupType
+
+
+async def disable_external_accounts(db: AsyncSession):
+    try:
+        await db.execute(
+            update(models_core.CoreUser)
+            .where(
+                models_core.CoreUser.groups.any(
+                    models_core.CoreGroup.id == GroupType.external.value
+                )
+            )
+            .values(disabled=True)
+        )
+        await db.commit()
+
+    except IntegrityError as error:
+        await db.rollback()
+        raise ValueError(error)

app/cruds/cruds_users.py

@@ -53,6 +53,7 @@ async def get_users(
                     )
                     for group_id in excludedGroups
                 ],
+                not_(models_core.CoreUser.disabled),
             )
         )
     )
@@ -64,7 +65,10 @@ async def get_user_by_id(db: AsyncSession, user_id: str) -> models_core.CoreUser
 
     result = await db.execute(
         select(models_core.CoreUser)
-        .where(models_core.CoreUser.id == user_id)
+        .where(
+            models_core.CoreUser.id == user_id,
+            not_(models_core.CoreUser.disabled),
+        )
         .options(
             # The group relationship need to be loaded
             selectinload(models_core.CoreUser.groups)
@@ -79,7 +83,10 @@ async def get_user_by_email(
     """Return user with id from database as a dictionary"""
 
     result = await db.execute(
-        select(models_core.CoreUser).where(models_core.CoreUser.email == email)
+        select(models_core.CoreUser).where(
+            models_core.CoreUser.email == email,
+            not_(models_core.CoreUser.disabled),
+        )
     )
     return result.scalars().first()
 

app/dependencies.py

@@ -289,6 +289,12 @@ async def is_user_a_member_of(
         """
         A dependency that checks that user is a member of the group with the given id then returns the corresponding user.
         """
+        if user.disabled:
+            raise HTTPException(
+                status_code=403,
+                detail="Unauthorized, user is disabled",
+            )
+
         if is_user_member_of_an_allowed_group(user=user, allowed_groups=[group_id]):
             # We know the user is a member of the group, we don't need to return an error and can return the CoreUser object
             return user

app/endpoints/auth.py

@@ -68,7 +68,7 @@ async def login_for_access_token(
     settings: Settings = Depends(get_settings),
 ):
     """
-    Ask for a JWT acc   ess token using oauth password flow.
+    Ask for a JWT access token using oauth password flow.
 
     *username* and *password* must be provided
 
@@ -81,6 +81,11 @@ async def login_for_access_token(
             detail="Incorrect login or password",
             headers={"WWW-Authenticate": "Bearer"},
         )
+    if user.disabled:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Disabled account. Contact Éclair for more informations.",
+        )
     # We put the user id in the subject field of the token.
     # The subject `sub` is a JWT registered claim name, see https://datatracker.ietf.org/doc/html/rfc7519#section-4.1
     data = schemas_auth.TokenData(sub=user.id, scopes=ScopeType.API)

app/endpoints/users.py

@@ -21,7 +21,7 @@
 
 from app.core import security
 from app.core.config import Settings
-from app.cruds import cruds_groups, cruds_users
+from app.cruds import cruds_external_account, cruds_groups, cruds_users
 from app.dependencies import (
     get_db,
     get_request_id,
@@ -945,3 +945,15 @@ async def read_user_profile_picture(
         filename=str(user_id),
         default_asset="assets/images/default_profile_picture.png",
     )
+
+
+@router.get(
+    "/users/external/",
+    status_code=200,
+    tags=[Tags.external_account],
+)
+async def disable_external_users(
+    db: AsyncSession = Depends(get_db),
+    user: models_core.CoreUser = Depends(is_user_a_member_of(GroupType.admin)),
+):
+    return await cruds_external_account.disable_external_accounts(db=db)

app/models/models_core.py

@@ -2,7 +2,7 @@
 
 from datetime import date, datetime
 
-from sqlalchemy import Date, DateTime, Enum, ForeignKey, Integer, String
+from sqlalchemy import Boolean, Date, DateTime, Enum, ForeignKey, Integer, String
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database import Base
@@ -35,6 +35,7 @@ class CoreUser(Base):
     phone: Mapped[str | None] = mapped_column(String)
     floor: Mapped[FloorsType] = mapped_column(Enum(FloorsType), nullable=False)
     created_on: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    disabled: Mapped[bool] = mapped_column(Boolean, nullable=False)
 
     # We use list["CoreGroup"] with quotes as CoreGroup is only defined after this class
     # Defining CoreUser after CoreGroup would cause a similar issue

app/schemas/schemas_core.py

@@ -71,6 +71,7 @@ class CoreUser(CoreUserSimple):
     phone: str | None = None
     created_on: datetime | None = None
     groups: list[CoreGroupSimple] = []
+    disabled: bool
 
 
 class CoreUserUpdate(BaseModel):

app/utils/tools.py

@@ -60,7 +60,8 @@ def fuzzy_search_user(
     choices = []
 
     for user in users:
-        choices.append(f"{user.firstname} {user.name} {user.nickname}")
+        if not user.disabled:
+            choices.append(f"{user.firstname} {user.name} {user.nickname}")
 
     results: list[tuple[str, int | float, int]] = process.extract(
         query, choices, limit=limit

app/utils/types/groups_type.py

@@ -16,6 +16,7 @@ class GroupType(str, Enum):
     formerstudent = "ab4c7503-41b3-11ee-8177-089798f1a4a5"
     staff = "703056c4-be9d-475c-aa51-b7fc62a96aaa"
     association = "29751438-103c-42f2-b09b-33fbb20758a7"
+    external = "e9f1085f-50e5-440c-80c7-b36c9ad4d6fb"
 
     # Core groups
     admin = "0a25cb76-4b63-4fd3-b939-da6d9feabf28"
@@ -43,6 +44,7 @@ class AccountType(str, Enum):
     formerstudent = GroupType.formerstudent.value
     staff = GroupType.staff.value
     association = GroupType.association.value
+    external = GroupType.external.value
 
     def __str__(self):
         return f"{self.name}<{self.value}>"

app/utils/types/tags.py

@@ -17,3 +17,4 @@ class Tags(str, Enum):
     raffle = "Raffle"
     advert = "Advert"
     notifications = "Notifications"
+    external_account = "External_account"

tests/commons.py

@@ -141,6 +141,7 @@ async def create_user_with_groups(
         name=name or get_random_string(),
         firstname=firstname or get_random_string(),
         floor=floor,
+        disabled=False,
     )
 
     async with TestingSessionLocal() as db:

tests/test_auth.py

@@ -29,6 +29,7 @@ async def init_objects():
         birthday=date.fromisoformat("2000-01-01"),
         floor=FloorsType.Autre,
         created_on=date.fromisoformat("2000-01-01"),
+        disabled=False,
     )
     await add_object_to_db(user)
 

tests/test_external_account.py

@@ -0,0 +1,41 @@
+import pytest_asyncio
+
+from app.models import models_core
+from app.utils.types.groups_type import GroupType
+
+# We need to import event_loop for pytest-asyncio routine defined bellow
+from tests.commons import event_loop  # noqa
+from tests.commons import client, create_api_access_token, create_user_with_groups
+
+user: models_core.CoreUser
+
+token_admin: str = ""
+
+
+@pytest_asyncio.fixture(scope="module", autouse=True)
+async def init_objects():
+    global user
+    user = await create_user_with_groups([GroupType.external])
+
+    global user_admin
+    user_admin = await create_user_with_groups([GroupType.admin])
+
+    global token_admin
+    token_admin = create_api_access_token(user_admin)
+
+
+def test_disable_external_account():
+    global user
+    response = client.get(
+        "/external/",
+        follow_redirects=False,
+        headers={"Authorization": f"Bearer {token_admin}"},
+    )
+    response1 = client.get(
+        f"/users/{user.id}",
+        headers={"Authorization": f"Bearer {token_admin}"},
+    )
+    assert response.status_code == 200
+    assert response1.status_code == 200
+    data = response1.json()
+    assert data["disabled"]