Plugins. General principles

<mgrdata>

    <!-- Register the plugin. -->
    <plugin name="myplugin">
        <group>plugin</group>
        <author>Author Name</author>
    </plugin>

    <!-- Register handlers. The number of handlers for one file is unlimited. -->
    <handler name="myplugin.py" type="xml">
        <event name="dedic.open" before="yes" base="project"/>
        <func name="myplugin.settings"/>
    </handler>

    <!-- Form description -->
    <metadata name="myplugin.settings" type="form" level="admin+">
        <form>
            <field name="api_key">
                <input type="text" name="api_key" required="yes"/>
            </field>
        </form>
    </metadata>

    <!-- Menu item -->
    <mainmenu level="29" name="Administrator">
        <node name="integration">
            <node name="myplugin.settings"/>
        </node>
    </mainmenu>

    <!-- Localization -->
    <lang name="en">
        <messages name="plugin">
            <msg name="desc_short_myplugin">My plugin</msg>
            <msg name="desc_full_myplugin">Plugin description</msg>
        </messages>
        <messages name="myplugin.settings">
            <msg name="title">Plugin settings</msg>
            <msg name="api_key">API key</msg>
        </messages>
        <messages name="msgerror">
            <msg name="msg_error_invalid_token">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():
    # Read input XML from stdin
    xml = session.get_input_xml(True)
    session.debug_session(xml)

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

    # Processing logic
    if func == "myplugin.settings":
        ...

    # Return XML to 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">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 (last part of the domain) must contain at least 2 letters.</msg>
            <msg name="desc_invalid_tld">TLD can contain only letters.</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

# Check that the plugin is running inside venv.
# If not, restart the plugin through the venv interpreter.
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):
    """Validation error for a form field. Used inside CheckerPlugin.check().

    Equivalent to C++ mgr_err::Value(object, value, check, args).
    BILLmanager will display a localized message with the key desc_{check}.

    Localization is declared in the plugin XML file.

    Args:
        check: validator name.
        args:  validator parameters.

    Example:
        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:
    """Universal base class for BILLmanager plugins.

    One handler in BILLmanager passes events (event), functions (func), and tasks(task)
    in the same way — through the query parameter "func". Therefore, Plugin handles
    all cases through a single handle() mechanism.

    The descendant registers handlers in __init__ through handle():

    Handler signature: (func: str, xml: ET.Element) -> str | None
      func:   current function/event name (from get_query_param("func"))
      xml:    incoming XML request body
      return: XML string for the response, or None — then <doc/> is returned
    """

    #: Plugin name — used to initialize the logger.
    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:
        """Register a handler.

        Args:
            func:    BILLmanager function name.
            handler: callable with signature (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:
        """Entry point. Called in 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):
    """Base class for checker plugins. Inherits Plugin.

    Called by BILLmanager when validating a form field with the check="<name>" attribute.
    Automatically registers the check.{name} handler and wraps the call to check().

    XML registration:

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

    The descendant must define name and implement 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__()
        # Register the check.{name} handler through the parent handle()
        self.handle(f"check.{self.name}", self._check_handler)

    @abstractmethod
    def check(self, object:str, value: str, args: str) -> str:
        """Validation logic. Override in the descendant.

        Args:
            value: user-entered value (already normalized by the strip() method).
            args:  optional argument from the form field attribute.

        Returns:
            Normalized value is written back to the field.

        Raises:
            ValueException if the value does not pass validation.
        """
        raise NotImplementedError

    def _check_handler(self, func: str, xml: ET.Element) -> str:
        """Internal handler reads parameters and calls 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:
        """
        Domain name validation.
        Args:
            object: field name. For example, "domain".
            value: value to validate
            args: validation parameters. You can pass requirements, for example "min_length=2,max_length=63".
        """
        # Extract arguments if they are provided.
        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'

        # Check that the value is not empty.
        if not value:
            raise ValueException(check="empty_domain", err_object=object)

        # Check that the domain name length meets the requirements.
        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)

        # Check for spaces.
        if ' ' in value:
            raise ValueException(check="has_spaces", err_object=object)

        # Check for invalid characters at the beginning or end.
        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)

        # Check the domain format.
        if not self._validate_domain_format(value, allow_unicode):
            raise ValueException(check="invalid_domain_format", err_object=object)

        # Perform an additional check for reserved names:
        reserved = [
            'localhost', 'local', 'example', 'invalid', 'test',
            'xn--',  # punycode prefix is not allowed
        ]
        
        domain_lower = value.lower()
        for reserved_name in reserved:
            if domain_lower == reserved_name:
                raise ValueException(check="reserved_domain", err_object=object)

        # Check the number of dots. Maximum 127 for a full domain.
        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:
        """
        Check the domain name format.
        """
        if allow_unicode:
            # For Unicode domains (IDNA)
            # Check that the domain name can be converted to ASCII
            try:
                domain.encode('idna').decode('ascii')
            except (UnicodeError, UnicodeEncodeError):
                return False
        
        # Strict ASCII domain validation.
        # Labels: letters, numbers, hyphen (not at the beginning or end).
        # Label length: 1-63 characters.
        # TLD: only Latin letters, minimum 2 characters.
        pattern = r'^(?!(?:[0-9]+\.)+[0-9]+$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)\.(?:[A-Za-z]{2,}|[A-Za-z0-9-]{1,63}(?<!-))$'
        
        # Check a full domain name with multiple subdomains.
        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:
            # Check allowed characters for ASCII domains.
            allowed_chars = set('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-')
            if not all(c in allowed_chars for c in domain):
                return False
            
            # Check with a regular expression.
            if not re.match(full_pattern, domain):
                # Check each label separately:
                labels = domain.split('.')
                for i, label in enumerate(labels):
                    # Reject empty labels.
                    if not label:
                        return False
                    
                    # Check label length.
                    if len(label) > 63 or len(label) < 1:
                        return False
                    
                    # Check that the TLD (last label) contains only letters and is at least 2 characters long.
                    if i == len(labels) - 1:
                        if not re.match(r'^[A-Za-z]{2,}$', label):
                            return False
                    else:
                        # Check that subdomain names start and end with a letter or number.
                        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"
    
    # Create an isolated environment for the plugin dependencies.
    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
    
    # Copy source files to the target directory and run make install,
    # which will copy files to the required directories through Makefile
    mkdir -p ${DESTDIR} && cp -rfa "${CURRDIR}"/* ${DESTDIR}/
    cd ${DESTDIR} && make install || ExitError "Install failed"
    cd ${CURRDIR}
    
    Info "Installation finished successfully"
}

InstallDeps
InstallModule