Плагины. Общие принципы

<mgrdata>

    <!-- Регистрация плагина. -->
    <plugin name="myplugin">
        <group>plugin</group>
        <author>Author Name</author>
    </plugin>

    <!-- Регистрация обработчиков. Количество обработчиков для одного файла не ограничено. -->
    <handler name="myplugin.py" type="xml">
        <event name="dedic.open" before="yes" base="project"/>
        <func name="myplugin.settings"/>
    </handler>

    <!-- Описание формы -->
    <metadata name="myplugin.settings" type="form" level="admin+">
        <form>
            <field name="api_key">
                <input type="text" name="api_key" required="yes"/>
            </field>
        </form>
    </metadata>

    <!-- Пункт меню -->
    <mainmenu level="29" name="Administrator">
        <node name="integration">
            <node name="myplugin.settings"/>
        </node>
    </mainmenu>

    <!-- Локализация -->
    <lang name="ru">
        <messages name="plugin">
            <msg name="desc_short_myplugin">Мой плагин</msg>
            <msg name="desc_full_myplugin">Описание плагина</msg>
        </messages>
        <messages name="myplugin.settings">
            <msg name="title">Настройки плагина</msg>
            <msg name="api_key">API-ключ</msg>
        </messages>
        <messages name="msgerror">
            <msg name="msg_error_invalid_token">Неверный токен</msg>
        </messages>
    </lang>

</mgrdata>

#!/usr/bin/env python3

import sys
sys.path.insert(0, "/usr/local/mgr5/lib/python")

import xml.etree.ElementTree as ET
import billmgr.session as session
import billmgr.exception as bill_exc
import billmgr.logger as logging

logging.init_logging("myplugin")
logger = logging.get_logger("myplugin")

def handle():
    # Читаем входной XML из stdin
    xml = session.get_input_xml(True)
    session.debug_session(xml)

    func = session.get_query_param("func")
    logger.info("func=%r", func)

    # Логика обработки
    if func == "myplugin.settings":
        ...

    # Возвращаем XML в stdout
    return ET.tostring(xml, encoding="unicode")

if __name__ == "__main__":
    try:
        print(handle())
    except bill_exc.XmlException as err:
        bill_exc.log_backtrace()
        print(err.as_xml())
    except Exception as err:
        bill_exc.log_backtrace()
        print(bill_exc.XmlException("unknown", "what", str(err)).as_xml())

<mgrdata>
    <plugin name="checker">
        <group>plugin</group>
        <author>BILLmanager team</author>
    </plugin>
    
    <handler name="checker.py" type="cgi">
        <func name="check.checker"/>
    </handler>
    
    <lang name="ru">
        <messages name="msgerror">
            <msg name="desc_too_short">Значение слишком короткое (минимум 3 символа).</msg>
            <msg name="desc_has_spaces">Значение не должно содержать пробелы.</msg>            
            <msg name="desc_empty_domain">Доменное имя не может быть пустым.</msg>
            <msg name="desc_domain_too_short">Доменное имя слишком короткое (минимум 2 символа).</msg>
            <msg name="desc_domain_too_long">Доменное имя слишком длинное (максимум 63 символа на метку).</msg>
            <msg name="desc_invalid_hyphen_position">Дефис не может быть в начале или конце метки домена.</msg>
            <msg name="desc_invalid_dot_position">Точка не может быть в начале или конце доменного имени.</msg>
            <msg name="desc_invalid_domain_format">Неверный формат доменного имени. Например: example.com</msg>
            <msg name="desc_reserved_domain">Это доменное имя зарезервировано и не может быть использовано.</msg>
            <msg name="desc_too_many_labels">Слишком много уровней в доменном имени (максимум 127).</msg>
            <msg name="desc_invalid_chars">Доменное имя содержит недопустимые символы. Разрешены: буквы, цифры, дефис и точка.</msg>
            <msg name="desc_tld_too_short">TLD (последняя часть домена) должна содержать минимум 2 буквы.</msg>
            <msg name="desc_invalid_tld">TLD может содержать только буквы.</msg>
        </messages>
    </lang>
    
    <lang name="en">
        <messages name="msgerror">
            <msg name="desc_too_short">Value is too short (minimum 3 characters).</msg>
            <msg name="desc_has_spaces">Value must not contain spaces.</msg>
            <msg name="desc_empty_domain">Domain name cannot be empty.</msg>
            <msg name="desc_domain_too_short">Domain name is too short (minimum 2 characters).</msg>
            <msg name="desc_domain_too_long">Domain name is too long (maximum 63 characters per label).</msg>
            <msg name="desc_invalid_hyphen_position">Hyphen cannot be at the beginning or end of a domain label.</msg>
            <msg name="desc_invalid_dot_position">Dot cannot be at the beginning or end of domain name.</msg>
            <msg name="desc_invalid_domain_format">Invalid domain name format. Example: example.com</msg>
            <msg name="desc_reserved_domain">This domain name is reserved and cannot be used.</msg>
            <msg name="desc_too_many_labels">Too many levels in domain name (maximum 127).</msg>
            <msg name="desc_invalid_chars">Domain name contains invalid characters. Allowed: letters, numbers, hyphen and dot.</msg>
            <msg name="desc_tld_too_short">TLD must contain at least 2 letters.</msg>
            <msg name="desc_invalid_tld">TLD can contain only letters.</msg>
        </messages>
    </lang>
</mgrdata>

#!/usr/bin/env python3
import sys
import os
import xml.etree.ElementTree as ET
from abc import abstractmethod
from typing import Callable, Dict

# Проверяем что плагин запущен внутри venv.
# Если нет — перезапускаем плагин через venv-интерпретатор.
VENV_PYTHON = "/usr/local/mgr5/src/checker/venv-checker/bin/python3"
if sys.executable != VENV_PYTHON and os.path.exists(VENV_PYTHON):
    os.execv(VENV_PYTHON, [VENV_PYTHON] + sys.argv)

sys.path.insert(0, "/usr/local/mgr5/lib/python")

import billmgr.logger as logging
import billmgr.session as session
import billmgr.exception as bill_exc

_logger = logging.get_logger("checker")

# ---------------------------------------------------------------------------
# ValueException
# ---------------------------------------------------------------------------

class ValueException(bill_exc.XmlException):
    """Ошибка валидации поля формы. Используется внутри CheckerPlugin.check().

    Эквивалент C++ mgr_err::Value(object, value, check, args).
    BILLmanager отобразит локализованное сообщение с ключом desc_{check}.

    Локализация объявляется в XML-файле плагина.

    Args:
        check: имя валидатора.
        args:  параметры валидатора.

    Пример:
        raise ValueException(check="too_short")
        raise ValueException(check="invalid_format", args=r"\\d+")
    """

    def __init__(self, err_object="", err_value="", check: str = None, args: str = None):
        super().__init__("value", err_object, err_value)
        self.messages = {}
        self.params = {}
        self.check = check
        
        if self.err_object:
            self.add_message("object", self.err_object)
        
        if self.err_value:
            self.add_param("value", self.err_value)

        if args:
            self.add_param("args", args)

        if check:
            self.add_message("desc", "desc_" + check)
        else:
            self.add_message("desc", "desk_empty")
    def add_message(self, name: str, value: str):
        self.messages[name] = value

    def as_xml(self):
        doc = ET.Element("doc")
        error = ET.SubElement(doc, "error")
        error.set("type", self.err_type)
        error.set("object", self.err_object)

        for param in self.params:
            param_node = ET.SubElement(error, "param")
            param_node.text = self.params[param]
            param_node.set("name", param)
        for msg in self.messages:
            msg_node = ET.SubElement(error, "param")
            msg_node.text = self.messages[msg]
            msg_node.set("name", msg)
            msg_node.set("type", "msg")

        return ET.tostring(doc, encoding="unicode")

# ---------------------------------------------------------------------------
# Plugin
# ---------------------------------------------------------------------------

class Plugin:
    """Универсальный базовый класс для плагинов BILLmanager.

    Один handler в BILLmanager передаёт и события (event), и функции (func) и таски(task)
    одинаково — через query-параметр "func". Поэтому Plugin обрабатывает
    все случаи через единый механизм handle().

    Наследник регистрирует обработчики в __init__ через handle():

    Сигнатура обработчика: (func: str, xml: ET.Element) -> str | None
      func:   имя текущей функции/события (из get_query_param("func"))
      xml:    входящий XML-код запроса
      return: XML-строка для ответа, или None — тогда возвращается <doc/>
    """

    #: Имя плагина — используется для инициализации логгера.
    name: str = ""

    def __init__(self) -> None:
        if not self.name:
            raise ValueError(
                f"{self.__class__.__name__} must define class attribute 'name'"
            )
        logging.init_logging(self.name)
        self._logger = logging.get_logger(self.name)
        self._handlers: Dict[str, Callable] = {}

    def handle(self, func: str, handler: Callable) -> None:
        """Зарегистрировать обработчик.

        Args:
            func:    имя функции BILLmanager.
            handler: callable с сигнатурой (func: str, xml: ET.Element) -> str | None.
        """
        self._handlers[func] = handler

    def _dispatch(self, xml: ET.Element) -> str:
        func = session.get_query_param("func", "")
        self._logger.info("func=%r", func)

        handler = self._handlers.get(func)
        if handler is None:
            self._logger.debug("no handler for %r, skipping", func)
            return '\n<doc/>'

        result = handler(func, xml)
        return result if result is not None else '\n<doc/>'

    def run(self) -> None:
        """Точка входа. Вызывается в if __name__ == '__main__'."""
        try:
            xml = session.get_input_xml(True)
            session.debug_session(xml)
            _logger.debug(f"environ: {session.envs()}")
            _logger.debug(f"input xml: {ET.tostring(xml, encoding='unicode')}")

            res = self._dispatch(xml)
            self._logger.debug(f"ans={res}")
            print(res)
        except bill_exc.XmlException as err:
            bill_exc.log_backtrace()
            print(err.as_xml())
        except Exception as err:
            bill_exc.log_backtrace()
            print(bill_exc.XmlException("unknown", "what", str(err)).as_xml())

# ---------------------------------------------------------------------------
# CheckerPlugin
# ---------------------------------------------------------------------------

class CheckerPlugin(Plugin):
    """Базовый класс для плагинов-чекеров. Наследник Plugin.

    Вызывается BILLmanager при валидации поля формы с атрибутом check="<name>".
    Автоматически регистрирует обработчик check.{name} и оборачивает вызов check().

    XML-регистрация:

      <handler name="checker.py" type="xml">
        <func name="check.checker"/>
      </handler>

    Наследник обязан задать name и реализовать check():

        class MyChecker(CheckerPlugin):
            name = "checker"

            def check(self, value: str, args: str) -> str:
                if len(value) < 3:
                    raise ValueException(check="too_short")
                return value.strip()

        if __name__ == "__main__":
            MyChecker().run()

    """

    def __init__(self) -> None:
        super().__init__()
        # Регистрируем обработчик check.{name} через родительский handle()
        self.handle(f"check.{self.name}", self._check_handler)

    @abstractmethod
    def check(self, object:str, value: str, args: str) -> str:
        """Логика валидации. Переопределить в наследнике.

        Args:
            value: введённое пользователем значение (уже нормализовано методом strip()).
            args:  опциональный аргумент из атрибута поля формы.

        Returns:
            Нормализованное значение подставляется обратно в поле.

        Raises:
            ValueException, если значение не прошло проверку.
        """
        raise NotImplementedError

    def _check_handler(self, func: str, xml: ET.Element) -> str:
        """Внутренний обработчик читает параметры и вызывает check()."""
        raw_value  = session.get_query_param("value", "")
        field_name = session.get_query_param("name", "value")
        args       = session.get_query_param("args", "")

        self._logger.info(
            "check.%s called, field=%r value=%r", self.name, field_name, raw_value
        )

        value = raw_value.strip()
        if not value:
            doc = ET.Element("doc")
            ET.SubElement(doc, "value")
            return ET.tostring(doc, encoding="unicode")

        try:
            normalized = self.check(field_name, value, args)
        except ValueException as err:
            self._logger.info(
                "check.%s failed: desc_%s field=%r value=%r",
                self.name, err.check, field_name, value,
            )
            return err.as_xml()

        doc = ET.Element("doc")
        ET.SubElement(doc, "value").text = normalized
        return ET.tostring(doc, encoding="unicode")

import re

class MyChecker(CheckerPlugin):
    name = "checker"

    def check(self, object: str, value: str, args: str) -> str:
        """
        Проверка доменного имени.
        Args:
            object: имя поля. Например, "domain".
            value: проверяемое значение
            args: параметры проверки. Вы можете передать требования. Например "min_length=2,max_length=63".
        """
        # Извлекаем аргументы, если они переданы.
        min_length = 2
        max_length = 63
        allow_unicode = False
        
        if args:
            for arg in args.split(','):
                if '=' in arg:
                    key, val = arg.split('=')
                    if key == 'min_length':
                        min_length = int(val)
                    elif key == 'max_length':
                        max_length = int(val)
                    elif key == 'allow_unicode':
                        allow_unicode = val.lower() == 'true'

        # Проверяем, что значение не пустое.
        if not value:
            raise ValueException(check="empty_domain", err_object=object)

        # Проверяем, чтобы длина доменного имени соответствовала требованиям.
        if len(value) < min_length:
            raise ValueException(check="domain_too_short", err_object=object)
        
        if len(value) > max_length:
            raise ValueException(check="domain_too_long", err_object=object)

        # Проверяем на наличие пробелов.
        if ' ' in value:
            raise ValueException(check="has_spaces", err_object=object)

        # Проверяем на наличие недопустимых символов в начале или в конце.
        if value.startswith('-') or value.endswith('-'):
            raise ValueException(check="invalid_hyphen_position", err_object=object)

        if value.startswith('.') or value.endswith('.'):
            raise ValueException(check="invalid_dot_position", err_object=object)

        # Проверяем формат домена.
        if not self._validate_domain_format(value, allow_unicode):
            raise ValueException(check="invalid_domain_format", err_object=object)

        # Выполняем дополнительную проверку на зарезервированные имена:
        reserved = [
            'localhost', 'local', 'example', 'invalid', 'test',
            'xn--',  # punycode префикс недопустим
        ]
        
        domain_lower = value.lower()
        for reserved_name in reserved:
            if domain_lower == reserved_name:
                raise ValueException(check="reserved_domain", err_object=object)

        # Проверяем количество точек. Максимум 127 для полного домена.
        if value.count('.') > 127:
            raise ValueException(check="too_many_labels", err_object=object)

        return value.strip()

    def _validate_domain_format(self, domain: str, allow_unicode: bool = False) -> bool:
        """
        Проверяем формат доменного имени.
        """
        if allow_unicode:
            # Для Unicode доменов (IDNA)
            # Проверяем, что доменное имя можно сконвертировать в ASCII
            try:
                domain.encode('idna').decode('ascii')
            except (UnicodeError, UnicodeEncodeError):
                return False
        
        # Строгая валидация ASCII домена.
        # Метки: буквы, цифры, дефис (не в начале и не в конце).
        # Длина метки: 1-63 символа.
        # TLD: только латинские буквы, минимум 2 символа.
        pattern = r'^(?!(?:[0-9]+\.)+[0-9]+$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)\.(?:[A-Za-z]{2,}|[A-Za-z0-9-]{1,63}(?<!-))$'
        
        # Проверяем полное доменное имя с несколькими поддоменами.
        full_pattern = r'^(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
        
        if not allow_unicode:
            # Проверяем на допустимые символы для ASCII доменов.
            allowed_chars = set('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-')
            if not all(c in allowed_chars for c in domain):
                return False
            
            # Проверяем через регулярное выражение.
            if not re.match(full_pattern, domain):
                # Проверяем каждую метку отдельно:
                labels = domain.split('.')
                for i, label in enumerate(labels):
                    # Отклоняем пустые метки.
                    if not label:
                        return False
                    
                    # Проверяем длину метки.
                    if len(label) > 63 or len(label) < 1:
                        return False
                    
                    # Проверяем, что TLD (последняя метка) содержит только буквы и длина метки не менее 2 символов.
                    if i == len(labels) - 1:
                        if not re.match(r'^[A-Za-z]{2,}$', label):
                            return False
                    else:
                        # Проверяем, что имя поддоменов начинается и заканчивается буквой или цифрой.
                        if not re.match(r'^[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?$', label):
                            return False
        
        return True

if __name__ == "__main__":
    MyChecker().run()

#!/usr/bin/env bash
. /usr/local/mgr5/lib/pkgsh/core_pkg_funcs.sh

ExitError() {
    Error "${@}"
    exit 1
}

InstallDeps() {
    Info "Installing dependencies..."
    PKGS="billmanager-plugin-python-libs python3-pip make"
    
    case ${OSTYPE} in
        REDHAT) PKGS="$PKGS coremanager-devel" ;;
        DEBIAN) PKGS="$PKGS coremanager-dev python3-venv" ;;
        *)      ExitError "Unknown OS type" ;;
    esac
    
    PkgInstall "${PKGS}" || ExitError "Failed to install system packages"
    
    # Создаём изолированное окружение для зависимостей плагина.
    python3 -m venv venv-checker || ExitError "Failed to create venv"
    VENV_DIR="./venv-checker"
    "$VENV_DIR/bin/pip3" install --upgrade pip    || ExitError "pip upgrade failed"
    "$VENV_DIR/bin/pip3" install -r requirements.txt || ExitError "Failed to install Python dependencies"
}

InstallModule() {
    Info "Installing checker module..."
    CURRDIR=$(pwd)
    DESTDIR=/usr/local/mgr5/src/checker
    
    # Копируем исходники в целевую директорию и запускаем make install,
    # который скопирует файлы в нужные директории через Makefile
    mkdir -p ${DESTDIR} && cp -rfa "${CURRDIR}"/* ${DESTDIR}/
    cd ${DESTDIR} && make install || ExitError "Install failed"
    cd ${CURRDIR}
    
    Info "Installation finished successfully"
}

InstallDeps
InstallModule