Add self delete option

auth_backend/routes/user.py

@@ -1,9 +1,10 @@
 import logging
 from typing import Any, Literal
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi_sqlalchemy import db
 from sqlalchemy.orm import Session
+from starlette.status import HTTP_403_FORBIDDEN
 
 from auth_backend.auth_method import AuthPluginMeta
 from auth_backend.auth_plugins.email import Email
@@ -141,26 +142,32 @@ async def patch_user(
 @user.delete("/{user_id}", response_model=None)
 async def delete_user(
     user_id: int,
-    current_user: UserSession = Depends(UnionAuth(scopes=["auth.user.delete"], allow_none=False, auto_error=True)),
+    current_user: UserSession = Depends(UnionAuth(scopes=[], allow_none=False, auto_error=True)),
 ) -> None:
     """
-    Scopes: `["auth.user.delete"]`
+    Scopes: `["auth.user.delete"]` or `["auth.user.selfdelete"]` for self delete
     """
-    logger.debug(f'User id={current_user.id} triggered delete_user')
-    old_user = {"user_id": current_user.id}
-    user: User = User.get(user_id, session=db.session)
+    session_scopes = set([scope.name.lower() for scope in current_user.scopes])
+    if "auth.user.delete" in session_scopes or (
+        "auth.user.selfdelete" in session_scopes and user_id == current_user.user_id
+    ):
+        logger.debug(f'User id={current_user.id} triggered delete_user')
+        old_user = {"user_id": current_user.id}
+        user: User = User.get(user_id, session=db.session)
 
-    for method in user._auth_methods:
-        if method.is_deleted:
-            continue
-        # Сохраняем старое состояние пользователя
-        if method.auth_method not in old_user:
-            old_user[method.auth_method] = {}
-        old_user[method.auth_method][method.param] = method.value
-        # Удаляем AuthMethod
-        AuthMethod.delete(method.id, session=db.session)
-        logger.info(f'{method=} for {user.id=} deleted')
-    User.delete(user_id, session=db.session)
-    db.session.commit()
-    await AuthPluginMeta.user_updated(None, old_user)
-    logger.info(f'{user=} deleted')
+        for method in user._auth_methods:
+            if method.is_deleted:
+                continue
+            # Сохраняем старое состояние пользователя
+            if method.auth_method not in old_user:
+                old_user[method.auth_method] = {}
+            old_user[method.auth_method][method.param] = method.value
+            # Удаляем AuthMethod
+            AuthMethod.delete(method.id, session=db.session)
+            logger.info(f'{method=} for {user.id=} deleted')
+        User.delete(user_id, session=db.session)
+        db.session.commit()
+        await AuthPluginMeta.user_updated(None, old_user)
+        logger.info(f'{user=} deleted')
+    else:
+        raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="Not authorized")

tests/test_routes/test_user.py

@@ -25,16 +25,18 @@ def test_user_email(client: TestClient, dbsession: Session, user_factory):
     dbsession.commit()
 
 
-def test_delete_user(client: TestClient, dbsession: Session, user_factory):
-    user1 = user_factory(client)
+def test_delete_user(client_auth: TestClient, dbsession: Session, user_factory, user_scopes):
+    token = user_scopes[0]
+    header = {"Authorization": token}
+    user1 = user_factory(client_auth)
     time1 = datetime.utcnow()
     email_user = AuthMethod(user_id=user1, param="email", auth_method="email", value="testemailx@x.xy")
     dbsession.add(email_user)
     dbsession.commit()
     body = {"name": f"group{time1}", "parent_id": None, "scopes": []}
-    group = client.post(url="/group", json=body).json()["id"]
-    client.patch(f"/user/{user1}", json={"groups": [group]})
-    resp = client.delete(f"user/{user1}")
+    group = client_auth.post(url="/group", json=body, headers=header).json()["id"]
+    client_auth.patch(f"/user/{user1}", json={"groups": [group]}, headers=header)
+    resp = client_auth.delete(f"user/{user1}", headers=header)
     assert resp.status_code == 200
     user = dbsession.query(User).filter(User.id == user1).one_or_none()
     assert user.is_deleted