core/auth.py - master - vorobei/erp

core/auth.py
146 строк6 KB
vorobei
feat(core): интеграция модели БД, email-сервиса и админ-интерфейса
02 июл 2026, 13:38
02 июл 2026, 13:38cdf3188
# core/auth.py
from functools import wraps
from flask import session, flash, redirect, url_for, request
from core.models import User, Permission, RolePermission
from core.db import db


def login_required(f):
    """
    Декоратор: требует авторизации пользователя.
    Если пользователь не авторизован - перенаправляет на страницу входа.
    """
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not session.get('user_id'):
            flash('Необходимо войти в систему', 'warning')
            return redirect(url_for('auth.login', next=request.url))
        return f(*args, **kwargs)
    return decorated_function


def permission_required(permission_name):
    """
    Декоратор: требует наличия определённого разрешения у пользователя.
    Используется для контроля доступа к функциям.
    """
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # Проверяем авторизацию
            if not session.get('user_id'):
                flash('Необходимо войти в систему', 'warning')
                return redirect(url_for('auth.login', next=request.url))
            
            # Получаем пользователя
            user = User.query.get(session['user_id'])
            if not user:
                session.clear()
                flash('Сессия истекла, войдите заново', 'warning')
                return redirect(url_for('auth.login'))
            
            # Проверяем наличие роли
            if not user.role:
                flash('У вашей учетной записи не назначена роль. Обратитесь к администратору.', 'danger')
                return redirect(url_for('main.index'))
            
            # Проверяем разрешение
            permission = Permission.query.filter_by(name=permission_name).first()
            if not permission:
                flash(f'Системная ошибка: разрешение "{permission_name}" не найдено', 'danger')
                return redirect(url_for('main.index'))
            
            # Проверяем, есть ли у роли это разрешение
            has_permission = RolePermission.query.filter_by(
                role_id=user.role_id,
                permission_id=permission.id
            ).first()
            
            if not has_permission:
                flash('У вас нет прав для доступа к этой странице', 'danger')
                return redirect(url_for('main.index'))
            
            return f(*args, **kwargs)
        return decorated_function
    return decorator


def admin_required(f):
    """
    Декоратор: требует роль 'admin'.
    Упрощённая проверка для административных функций.
    """
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not session.get('user_id'):
            flash('Необходимо войти в систему', 'warning')
            return redirect(url_for('auth.login', next=request.url))
        
        user = User.query.get(session['user_id'])
        if not user:
            session.clear()
            flash('Сессия истекла, войдите заново', 'warning')
            return redirect(url_for('auth.login'))
        
        if not user.role or user.role.name != 'admin':
            flash('Доступ запрещён. Требуются права администратора.', 'danger')
            return redirect(url_for('main.index'))
        
        return f(*args, **kwargs)
    return decorated_function


def get_current_user():
    """
    Вспомогательная функция: возвращает текущего пользователя или None.
    """
    user_id = session.get('user_id')
    if user_id:
        return User.query.get(user_id)
    return None


# ============================================================
# ФУНКЦИИ ДЛЯ РАБОТЫ С РАЗРЕШЕНИЯМИ
# ============================================================

def get_user_permissions(user_id: int) -> set:
    """
    Возвращает множество названий разрешений пользователя.
    Если пользователь admin - возвращает все разрешения.
    """
    user = User.query.get(user_id)
    if not user:
        return set()
    
    # Администратор имеет все права
    if user.role and user.role.name == 'admin':
        # Возвращаем все существующие разрешения
        all_perms = Permission.query.all()
        return {p.name for p in all_perms}
    
    # Получаем разрешения роли пользователя
    if not user.role_id:
        return set()
    
    perms = db.session.query(Permission.name).join(
        RolePermission, RolePermission.permission_id == Permission.id
    ).filter(
        RolePermission.role_id == user.role_id
    ).all()
    
    return {p[0] for p in perms}


def has_permission(user_id: int, permission_name: str) -> bool:
    """Проверяет, есть ли у пользователя конкретное разрешение"""
    permissions = get_user_permissions(user_id)
    return permission_name in permissions


def user_has_any_permission(user_id: int, permission_names: list) -> bool:
    """Проверяет, есть ли у пользователя хотя бы одно из разрешений"""
    if not user_id:
        return False
    permissions = get_user_permissions(user_id)
    return any(p in permissions for p in permission_names)