py.lib.aw_web_tools

     1.1 --- /dev/null	Thu Jan 01 00:00:00 1970 +0000
     1.2 +++ b/src/aw_web_tools/authelia_helper.py	Sat Nov 09 16:45:06 2024 +0300
     1.3 @@ -0,0 +1,96 @@
     1.4 +# coding: utf-8
     1.5 +"""\
     1.6 +Поддержка работы с системой авторизации [authelia](https://www.authelia.com/)
     1.7 +"""
     1.8 +import bottle
     1.9 +from typing import Optional, List
    1.10 +from collections import namedtuple
    1.11 +
    1.12 +from . import Error
    1.13 +
    1.14 +
    1.15 +# Константы
    1.16 +HEADER_USER = 'Remote-User'      # Заголовок по умолчанию, в котором система аутентификации передаёт имя УЗ пользователя
    1.17 +HEADER_GROUPS = 'Remote-Groups'  # Заголовок по умолчанию, в котором система аутентификации передаёт группы пользовтеля
    1.18 +HEADER_EMAIL = 'Remote-Email'    # Заголовок по умолчанию, в котором система аутентификации передаёт e-mail пользователя
    1.19 +HEADER_NAME = 'Remote-Name'      # Заголовок по умолчанию, в котором система аутентификации передаёт имя пользователя
    1.20 +
    1.21 +
    1.22 +class AHError(Error):
    1.23 +    """\
    1.24 +    Базовый класс исключений модуля адаптера Authelia
    1.25 +    """
    1.26 +
    1.27 +
    1.28 +class AHAuthError(AHError):
    1.29 +    """\
    1.30 +    Проблемы с авторизацией пользователя
    1.31 +    """
    1.32 +    def __init__(self):
    1.33 +        super().__init__('Пользователь не авторизован')
    1.34 +
    1.35 +
    1.36 +AHUser = namedtuple('AHUser', ['uname', 'groups', 'email', 'name'])
    1.37 +
    1.38 +
    1.39 +class AutheliaHelper(object):
    1.40 +    def __init__(self,
    1.41 +                 group_filter: Optional[List[str]] = None,
    1.42 +                 header_user: str = HEADER_USER,
    1.43 +                 header_groups: str = HEADER_GROUPS,
    1.44 +                 header_email: str = HEADER_EMAIL,
    1.45 +                 header_name: str = HEADER_NAME
    1.46 +                 ):
    1.47 +        """\
    1.48 +        :param group_filter: Фильтр передаваемый системой аутентификации групп, которые интересны нам. Ели ``None``
    1.49 +                             группы, передаваемые в заголовке не фильтруются
    1.50 +        :param header_user: Имя заголовка с именем УЗ пользователя.
    1.51 +        :param header_groups: Имя заголовка с группами пользователя
    1.52 +        :param header_name: Имя заголовка с именем пользователя
    1.53 +        """
    1.54 +        self.header_user = header_user
    1.55 +        self.header_groups = header_groups
    1.56 +        self.header_email = header_email
    1.57 +        self.header_name = header_name
    1.58 +
    1.59 +        self.group_filter = None
    1.60 +        self.group_filter_set = None
    1.61 +        if group_filter is not None:
    1.62 +            self.group_filter = {}
    1.63 +            for i in group_filter:
    1.64 +                self.group_filter[i.lower()] = i
    1.65 +
    1.66 +            self.group_filter_set = set(self.group_filter.keys())
    1.67 +
    1.68 +    def __call__(self, request: bottle.BaseRequest = bottle.request) -> AHUser:
    1.69 +        """\
    1.70 +        Обработка запроса. Если заголовки системы авторизации присутствуют, информацией из них заполняется
    1.71 +        объект ``AHUser``. Если данные не найдены, то будет возбуждено исключение.
    1.72 +
    1.73 +        :param request: Обрабатываемый запрос
    1.74 +        :returns: Объект ``AHUser`` с данным авторизации.
    1.75 +        """
    1.76 +
    1.77 +        uname = request.get_header(self.header_user)
    1.78 +        email = request.get_header(self.header_email)
    1.79 +        name = request.get_header(self.header_name)
    1.80 +        groups = request.get_header(self.header_groups)
    1.81 +
    1.82 +        if uname is None:
    1.83 +            raise AHAuthError()
    1.84 +
    1.85 +        if groups is not None:
    1.86 +            groups = tuple(map(lambda x: x.stip(), groups.split(',')))
    1.87 +            if self.group_filter is not None:
    1.88 +                _groups = []
    1.89 +                _buf = set(map(lambda x: x.lower(), groups))
    1.90 +
    1.91 +                for grp in _buf & self.group_filter_set:
    1.92 +                    try:
    1.93 +                        _groups.append(self.group_filter[grp])
    1.94 +
    1.95 +                    except KeyError:
    1.96 +                        AHError(f'Ошибка в фильтре групп. Для ключа "{grp}" не нашлось значения, хотя оно присутствует '
    1.97 +                                f'в множестве возможных значений.')
    1.98 +
    1.99 +        return AHUser(uname=uname, groups=groups, name=name, email=email)