auth-wip

services/auth.py

@@ -13,7 +13,7 @@ from auth.orm import Author, Role
 ALLOWED_HEADERS = ["Authorization", "Content-Type"]
 
 
-async def check_auth(req) -> Tuple[str, list[str]]:
+async def check_auth(req) -> Tuple[str, list[str], bool]:
     """
     Проверка авторизации пользователя.
 
@@ -25,11 +25,12 @@ async def check_auth(req) -> Tuple[str, list[str]]:
     Возвращает:
     - user_id: str - Идентификатор пользователя
     - user_roles: list[str] - Список ролей пользователя
+    - is_admin: bool - Флаг наличия у пользователя административных прав
     """
     # Проверяем наличие токена
     token = req.headers.get("Authorization")
     if not token:
-        return "", []
+        return "", [], False
 
     # Очищаем токен от префикса Bearer если он есть
     if token.startswith("Bearer "):
@@ -39,8 +40,39 @@ async def check_auth(req) -> Tuple[str, list[str]]:
 
     # Проверяем авторизацию внутренним механизмом
     logger.debug("Using internal authentication")
-    return await verify_internal_auth(token)
-
+    user_id, user_roles = await verify_internal_auth(token)
+    
+    # Проверяем наличие административных прав у пользователя
+    is_admin = False
+    if user_id:
+        # Быстрая проверка на админ роли в кэше
+        admin_roles = ['admin', 'super']
+        for role in user_roles:
+            if role in admin_roles:
+                is_admin = True
+                break
+                
+        # Если в ролях нет админа, но есть ID - проверяем в БД
+        if not is_admin:
+            try:
+                with local_session() as session:
+                    # Преобразуем user_id в число
+                    try:
+                        user_id_int = int(user_id.strip())
+                    except (ValueError, TypeError):
+                        logger.error(f"Невозможно преобразовать user_id {user_id} в число")
+                    else:
+                        # Проверяем наличие админских прав через БД
+                        from auth.orm import AuthorRole
+                        admin_role = session.query(AuthorRole).filter(
+                            AuthorRole.author == user_id_int,
+                            AuthorRole.role.in_(["admin", "super"])
+                        ).first()
+                        is_admin = admin_role is not None
+            except Exception as e:
+                logger.error(f"Ошибка при проверке прав администратора: {e}")
+    
+    return user_id, user_roles, is_admin
 
 async def add_user_role(user_id: str, roles: list[str] = None):
     """
@@ -84,21 +116,36 @@ async def add_user_role(user_id: str, roles: list[str] = None):
 
 
 def login_required(f):
-    """Декоратор для проверки авторизации пользователя."""
+    """Декоратор для проверки авторизации пользователя. Требуется наличие роли 'reader'."""
 
     @wraps(f)
     async def decorated_function(*args, **kwargs):
+        from graphql.error import GraphQLError
+
         info = args[1]
         req = info.context.get("request")
-        user_id, user_roles = await check_auth(req)
-        if user_id and user_roles:
-            logger.info(f" got {user_id} roles: {user_roles}")
-            info.context["user_id"] = user_id.strip()
-            info.context["roles"] = user_roles
-            author = await get_cached_author_by_user_id(user_id, get_with_stat)
-            if not author:
-                logger.error(f"author profile not found for user {user_id}")
-            info.context["author"] = author
+        user_id, user_roles, is_admin = await check_auth(req)
+        
+        if not user_id:
+            raise GraphQLError("Требуется авторизация")
+            
+        # Проверяем наличие роли reader
+        if 'reader' not in user_roles and not is_admin:
+            logger.error(f"Пользователь {user_id} не имеет роли 'reader'")
+            raise GraphQLError("У вас нет необходимых прав для доступа")
+            
+        logger.info(f"Авторизован пользователь {user_id} с ролями: {user_roles}")
+        info.context["user_id"] = user_id.strip()
+        info.context["roles"] = user_roles
+        
+        # Проверяем права администратора
+        info.context["is_admin"] = is_admin
+        
+        author = await get_cached_author_by_user_id(user_id, get_with_stat)
+        if not author:
+            logger.error(f"Профиль автора не найден для пользователя {user_id}")
+        info.context["author"] = author
+            
         return await f(*args, **kwargs)
 
     return decorated_function
@@ -113,19 +160,24 @@ def login_accepted(f):
         req = info.context.get("request")
 
         logger.debug("login_accepted: Проверка авторизации пользователя.")
-        user_id, user_roles = await check_auth(req)
+        user_id, user_roles, is_admin = await check_auth(req)
         logger.debug(f"login_accepted: user_id={user_id}, user_roles={user_roles}")
 
         if user_id and user_roles:
             logger.info(f"login_accepted: Пользователь авторизован: {user_id} с ролями {user_roles}")
             info.context["user_id"] = user_id.strip()
             info.context["roles"] = user_roles
+            
+            # Проверяем права администратора
+            info.context["is_admin"] = is_admin
 
             # Пробуем получить профиль автора
             author = await get_cached_author_by_user_id(user_id, get_with_stat)
             if author:
                 logger.debug(f"login_accepted: Найден профиль автора: {author}")
-                info.context["author"] = author.dict()
+                # Используем флаг is_admin из контекста или передаем права владельца для собственных данных
+                is_owner = True  # Пользователь всегда является владельцем собственного профиля
+                info.context["author"] = author.dict(access=is_owner or is_admin)
             else:
                 logger.error(
                     f"login_accepted: Профиль автора не найден для пользователя {user_id}. Используем базовые данные."
@@ -135,7 +187,43 @@ def login_accepted(f):
             info.context["user_id"] = None
             info.context["roles"] = None
             info.context["author"] = None
+            info.context["is_admin"] = False
 
         return await f(*args, **kwargs)
 
     return decorated_function
+
+def author_required(f):
+    """Декоратор для проверки наличия роли 'author' у пользователя."""
+
+    @wraps(f)
+    async def decorated_function(*args, **kwargs):
+        from graphql.error import GraphQLError
+
+        info = args[1]
+        req = info.context.get("request")
+        user_id, user_roles, is_admin = await check_auth(req)
+        
+        if not user_id:
+            raise GraphQLError("Требуется авторизация")
+            
+        # Проверяем наличие роли author
+        if 'author' not in user_roles and not is_admin:
+            logger.error(f"Пользователь {user_id} не имеет роли 'author'")
+            raise GraphQLError("Для выполнения этого действия необходимы права автора")
+            
+        logger.info(f"Авторизован автор {user_id} с ролями: {user_roles}")
+        info.context["user_id"] = user_id.strip()
+        info.context["roles"] = user_roles
+        
+        # Проверяем права администратора
+        info.context["is_admin"] = is_admin
+        
+        author = await get_cached_author_by_user_id(user_id, get_with_stat)
+        if not author:
+            logger.error(f"Профиль автора не найден для пользователя {user_id}")
+        info.context["author"] = author
+            
+        return await f(*args, **kwargs)
+
+    return decorated_function