Da

✏️ Editing: imunify360_pam.py

#!/opt/imunify360/venv/bin/python3
#
# imunify360-pam        Python script to manage imunify360 pam module
#                       enabled/diabled state.
#

import argparse
import os
import re
import shutil
import signal
import subprocess
import sys
import traceback
from collections import OrderedDict
from configparser import ConfigParser
from contextlib import closing, suppress
from distutils.version import LooseVersion
from enum import Enum
from functools import lru_cache, wraps
from pathlib import Path
from string import Template
from typing import Iterable, Tuple

import yaml

from pam_i360.internals import getLogger, logger_init, pam_imunify_config

CONFIG_DOVECOT = "/etc/dovecot/dovecot.conf"
CONFIG_DOVECOT_DATASTORE = "/var/cpanel/conf/dovecot/main"
CONFIG_DOVECOT_BASEDIR = "/var/cpanel/templates/dovecot"
CONFIG_DOVECOT_DEFAULT_SUFFIX = "2.3"
CONFIG_DOVECOT_TMPL = "main.default"
CONFIG_DOVECOT_LOCAL = "main.local"
CONFIG_PAM_DOVECOT = "/etc/pam.d/dovecot_imunify"
CONFIG_PAM_DOVECOT_DOMAINOWNER = "/etc/pam.d/dovecot_imunify_domainowner"
CONFIG_PROFTPD = "/etc/proftpd.conf"
CONFIG_PAM_PROFTPD = "/etc/pam.d/proftpd_imunify"
CONFIG_PUREFTPD = "/etc/pure-ftpd.conf"
CONFIG_TEMPLATE_PUREFTPD = "/var/cpanel/conf/pureftpd/local"
CONFIG_PAM_PUREFTPD = "/etc/pam.d/pure-ftpd"
CONFIG_IMUNIFY360 = "/etc/sysconfig/imunify360/imunify360-merged.config"

LEVELDB = "/opt/i360_pam_imunify/db/leveldb"

DOVECOT_LIB_IMUNIFY="lib_imunify360_%s.%s.%s.so"
if os.path.exists("/etc/pam.d/common-auth"):
    # Debian-like
    DOVECOT_I360_DIR = "/usr/lib/i360_pam_imunify"
    DOVECOT_AUTH_DIR = "/lib64/dovecot/auth"
else:
    # RHEL-like
    DOVECOT_I360_DIR = "/usr/lib64/i360_pam_imunify"
    DOVECOT_AUTH_DIR = "/usr/lib64/dovecot/auth"

DOVECOT_LIB_LINK = os.path.join(DOVECOT_AUTH_DIR, "lib_imunify360.so")

PAM_UNIX_REGEX = re.compile(r"auth\s+.+?\s+pam_unix\.so")

# logger late init in order to let sigterm_handler() to break
# logger_init() if needed
logger = None

class DovecotExeNotFound(Exception):
    pass

class Imunify360DovecotNotFound(Exception):
    pass

class DovecotState(Enum):
    DISABLED = 0
    PAM = 1
    NATIVE = 2

DOVECOT_STATES = {
    "disabled": DovecotState.DISABLED,
    "pam": DovecotState.PAM,
    "native": DovecotState.NATIVE,
}

class Output:
    def status_changed(self, services):
        enabled = []
        already_enabled = []
        disabled = []
        already_disabled = []

services = OrderedDict(sorted(services.items(), key=lambda x: x[0]))
        for key, value in services.items():
            enabled_prev, enabled_now = value
            if enabled_now:
                if enabled_prev:
                    already_enabled.append(key)
                else:
                    enabled.append(key)
            else:
                if not enabled_prev:
                    already_disabled.append(key)
                else:
                    disabled.append(key)

message = None
        if len(enabled) > 0:
            message = "imunify360-pam (%s) is now enabled." % ", ".join(enabled)

if len(already_enabled) > 0:
            message = "imunify360-pam (%s) is already enabled." % ", ".join(
                already_enabled
            )

if len(disabled) > 0:
            message = "imunify360-pam (%s) is now disabled." % ", ".join(disabled)

if len(already_disabled) > 0:
            message = "imunify360-pam (%s) is already disabled." % ", ".join(
                already_disabled
            )

if message:
            self._print(message)

def status(self, services):
        services = OrderedDict(sorted(services.items(), key=lambda x: x[1]))
        enabled = [key for key, value in services.items() if value]
        if len(enabled) > 0:
            self._print("status: enabled (%s)" % ", ".join(enabled))
        else:
            self._print("status: disabled")

def warning(self, *args, **kwargs):
        self._print("[WARNING]", *args, **kwargs)

def error(self, *args, **kwargs):
        self._print("[ERROR]", *args, **kwargs)

def run_and_log(self, *args, **kwargs):
        subprocess.run(*args, **kwargs)

def flush(self):
        pass

def _print(self, *args, **kwargs):
        print(*args, **kwargs)

# duplicate to pam.log
        if args[0] == "[WARNING]":
            logfun = logger.warning
            args = args[1:]
        elif args[0] == "[ERROR]":
            logfun = logger.error
            args = args[1:]
        else:
            logfun = logger.info
        logfun(" ".join(args))

class YamlOutput(Output):
    def __init__(self):
        self._buffer = {}

def flush(self):
        print(yaml.safe_dump(self._buffer, default_flow_style=False))

# duplicate to pam.log
        for k in ["status_changed", "status"]:
            if k in self._buffer:
                logger.info("%s=%r", k, self._buffer[k])

def status_changed(self, services):
        for service, value in services.items():
            enabled_prev, enabled_now = value
            self._buffer.setdefault("status_changed", {})[service] = {
                "from": "enabled" if enabled_prev else "disabled",
                "to": "enabled" if enabled_now else "disabled",
            }

def status(self, services):
        for service, enabled in services.items():
            self._buffer.setdefault("status", {})[service] = (
                "enabled" if enabled else "disabled"
            )

def warning(self, *args, **kwargs):
        self._buffer.setdefault("warnings", []).append(" ".join(args))
        logger.warning(" ".join(args))

def error(self, *args, **kwargs):
        self._buffer.setdefault("errors", []).append(" ".join(args))
        # catch message and backtrace for sentry
        logger.error(" ".join(args))

def run_and_log(self, cmd, *args, **kwargs):
        proc = subprocess.run(
            cmd,
            *args,
            **kwargs,
            stdin=subprocess.DEVNULL,
            # capture and combine both streams into one
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT
        )

if proc.returncode != 0:
            self.warning("%s exit code %d" % (" ".join(cmd), proc.returncode))

if (
            proc.returncode != 0
            or options.verbose
            or pam_imunify_config().getboolean("verbose")
        ):
            self._buffer.setdefault("subprocess_call", []).append(
                {
                    "cmd": " ".join(cmd),
                    "returncode": proc.returncode,
                    # .decode('ascii', errors='ignore') is to suppress
                    # cPanel tools output colors
                    "output": proc.stdout.decode("ascii", errors="ignore"),
                }
            )

# This function get CP name only
@lru_cache(1)
def get_cp_name():
    panel = "generic"

# cPanel check
    if os.path.isfile("/usr/local/cpanel/cpanel"):
        panel = "cpanel"

# Plesk check
    elif os.path.isfile("/usr/local/psa/version"):
        panel = "plesk"

# DirectAdmin check
    elif os.path.isfile("/usr/local/directadmin/directadmin"):
        panel = "directadmin"

return panel

@lru_cache(1)
def get_dovecot_version():
    dovecot_exe = shutil.which('dovecot')
    if dovecot_exe is None:
        raise DovecotExeNotFound("Dovecot executable not found in path")

result = subprocess.check_output([dovecot_exe, "--version"])
    version_str = result.decode().strip()
    match = re.match(r'^(\d+)\.(\d+)\.(\d+)', version_str)
    if match:
        major, minor, patch = map(int, match.groups())
        return (major, minor, patch)
    return None

def compare_dovecot_version(
        sign: str, major: int, minor: int, patch: int
    ) -> bool:

current = get_dovecot_version()
    if current == None:
        raise ValueError("Failed to get dovecot version")

target = (major, minor, patch)

if sign == ">":
        return current > target
    elif sign == "<":
        return current < target
    elif sign == "=":
        return current == target
    elif sign == ">=":
        return current >= target
    elif sign == "<=":
        return current <= target
    else:
        raise ValueError(
            f"Invalid sign: {sign}. Use one of '>', '<', '=', '>=', '<='."
        )

def is_higher_than_dovecot24() -> bool:
    try:
        return compare_dovecot_version(">=", 2, 4, 0)
    except:
        return False

def readlink_f(filename):
    """
    Pythonic way of doing /bin/readlink --canonicalize filename
    and is needed for cPanel /etc/pam.d symlinks.
    """
    try:
        result = os.readlink(filename)
    except OSError:
        # not a symlink
        return filename

if os.path.isabs(result):
        return result
    else:
        return os.path.join(os.path.dirname(filename), result)

def detect_conffiles(output=None):
    if not output:
        output = Output()

if os.path.exists("/etc/pam.d/common-auth"):
        # debian, ubuntu
        conffiles = ("/etc/pam.d/common-auth",)
    else:
        conffiles = "/etc/pam.d/password-auth", "/etc/pam.d/system-auth"

if not all(os.path.exists(conf) for conf in conffiles):
        output.error("PAM configuration file(s) not found: %s" % " ".join(conffiles))
        sys.exit(1)

return [readlink_f(fn) for fn in conffiles]

def atomic_rewrite(filename, content):
    """
    Atomically rewrites filename with given content to
    avoid possible "No space left on device"
    or unintenrional PAM module break.

Backup original file content to {filename}.i360bak
    """
    if os.path.exists(filename):
        shutil.copy(filename, filename + ".i360bak")

tmp = filename + ".i360edit"
    with open(tmp, "wb" if isinstance(content, bytes) else "w") as tf:
        tf.write(content)
        try:
            st = os.stat(filename)
        except FileNotFoundError:
            pass
        else:
            os.fchmod(tf.fileno(), st.st_mode)
            os.fchown(tf.fileno(), st.st_uid, st.st_gid)
    ext = 3
    while ext > 0:
        try:
            os.rename(tmp, filename)
        except OSError:
            ext = ext - 1
            if ext == 0:
                output.error("Trouble in renaming of %s to %s" % (tmp, filename))
                sys.exit(1)
        else:
            ext = 0

class i360RPatch:
    def __init__(self, conf_filename, output=None):
        self._conf_filename = conf_filename
        self.output = Output() if not output else output

def filename(self):
        return os.path.join(
            os.path.dirname(self._conf_filename),
            ".%s.i360patch" % os.path.basename(self._conf_filename),
        )

def create_upon(self, content):
        cmd = ["/usr/bin/diff", "--unified=1", self._conf_filename, "-"]
        proc = subprocess.Popen(
            cmd, stdin=subprocess.PIPE, stdout=open(self.filename(), "w")
        )
        proc.communicate(content.encode())

if proc.returncode != 1:
            # not a big deal: will use .i360bak as the last resort
            self.output.warning("'diff -u' error", file=sys.stderr)
            os.unlink(self.filename())

def apply(self):
        """
        :raise CalledProcessError:
        """
        cmd = ["/usr/bin/patch", "--reverse", self._conf_filename]
        subprocess.check_call(
            cmd, stdin=open(self.filename()), stdout=open("/dev/null", "w")
        )
        os.unlink(self.filename())

def pam_unix_patch_around(pamconfig_lines, pam_unix_ln):
    match_offset = re.search(
        r"success=(\d)\s+default=ignore", pamconfig_lines[pam_unix_ln]
    )

patch_simple(pamconfig_lines, pam_unix_ln)
    pam_unix_ln += 1

if match_offset:
        fix_offset(pamconfig_lines, pam_unix_ln, int(match_offset.group(1)))

def dovecot_manyconfigs_basedir() -> Tuple[Iterable[str], str]:
    """
    As per DEF-18259 it comes out that dovecot config main.default and main.local
    files may reside in a directory /var/cpanel/conf/dovecot/main refers it via
    '_use_target_version: ...'

:return: dovecot multiple configs basedir list and
             warning message if any
    """
    result = set()
    warning_msg = None

def check_and_use(path_):
        # unwind '/var/cpanel/templates/dovecot -> /var/cpanel/templates/dovecot2.3'
        path_ = readlink_f(path_)

if os.path.exists(os.path.join(path_, CONFIG_DOVECOT_TMPL)):
            result.add(path_)
            return True
        return False

check_and_use(CONFIG_DOVECOT_BASEDIR)
    check_and_use(CONFIG_DOVECOT_BASEDIR + CONFIG_DOVECOT_DEFAULT_SUFFIX)

try:
        with open(CONFIG_DOVECOT_DATASTORE) as f:
            target_version = yaml.safe_load(f)["_use_target_version"]
    except (FileNotFoundError, UnicodeError, yaml.YAMLError) as e:
        warning_msg = "%s: %s" % (CONFIG_DOVECOT_DATASTORE, e)
    except KeyError:
        pass
    else:
        path_ = CONFIG_DOVECOT_BASEDIR + target_version
        if not check_and_use(path_):
            warning_msg = (
                "%s '_use_target_version: %s' refers a non existent "
                "configuration file %r"
                % (CONFIG_DOVECOT_DATASTORE, target_version, path_)
            )

return result, warning_msg

def insert_imunify_passdb(data, output=None):
    if not output:
        output = Output()

if is_higher_than_dovecot24(): # dovecot 2.4 passdb
        if not re.search(
            r"passdb\s*imunify360\s*\{", data, re.DOTALL
        ):  # passdb is already in config
            match = re.search(
                r"passdb\s+cpauthd\s*\{(?:(?:[^{}]+|\{[^{}]*\})*)\}",
                data,
                re.DOTALL,
            )
            if match:
                imunify_passdb_template = Template(
                    "passdb $passdb_name {\n"
                    "  driver = imunify360\n"
                    "  fields {\n"
                    "    key=/opt/i360_pam_imunify/key\n"
                    "    secret=/opt/i360_pam_imunify/secret\n"
                    "    socket=/opt/i360_pam_imunify/pam_imunify360.sock"
                    "$check_only"
                    "  }\n"
                    "$result_action\n"
                    "}"
                )

imunify_passdb = imunify_passdb_template.substitute(
                   passdb_name="imunify360",
                   check_only="\n",
                   result_action=f"  result_success = "
                                 f"[%IF allow_domainowner_mail_pass %]continue"
                                 f"[% ELSE %]return[% END %]"
                )

imunify_passdb_check_only = imunify_passdb_template.substitute(
                    passdb_name="imunify360_check_only",
                    check_only="\n    check_only=1\n",
                    result_action="  result_failure = return-fail\n",
                )
                # insert imunify passdb after default passdb
                data_after = (
                    data[: match.end()]
                    + "\n"
                    + imunify_passdb
                    + "\n"
                    + data[match.end() :]
                )
                # insert imunify_check_only passdb before default passdb
                return (
                    data_after[: match.start()]
                    + "\n"
                    + imunify_passdb_check_only
                    + "\n"
                    + data_after[match.start() :]
                )
            else:  # cpauthd passdb missing
                output.error(
                        "PAM configuration file parse error: cpauthd missing"
                )
                sys.exit(1)
    else:
        if not re.search(
            r"passdb\s*\{\s*driver\s*=\s*imunify360.*?}", data, re.DOTALL
        ):  # passdb is already in config
            match = re.search(
                r"passdb\s*\{\s*driver\s*=\s*dict.*?}", data, re.DOTALL
            )  # find default passdb
            if match:
                imunify_passdb_template = Template(
                    "passdb {\n"
                    "  driver = imunify360\n"
                    "  args = key=/opt/i360_pam_imunify/key \\\n"
                    "         secret=/opt/i360_pam_imunify/secret \\\n"
                    "         socket=/opt/i360_pam_imunify/pam_imunify360.sock"
                    "$check_only"
                    "$result_action\n"
                    "}"
                )

data = (
                    data[: match.end()]
                    + "\n"
                    + imunify_passdb_template.substitute(
                        check_only="\n", result_action="  result_success = continue"
                    )
                    + "\n"
                    + data[match.end() :]
                )  # insert imunify passdb after default passdb
                return (
                    data[: match.start()]
                    + "\n"
                    + imunify_passdb_template.substitute(
                        check_only=" \\\n         check_only=1\n",
                        result_action="  result_failure = return-fail\n",
                    )
                    + "\n"
                    + data[match.start() :]  # insert imunify passdb before default passdb
                )
            else:  # default passdb missing
                output.error("PAM configuration file parse error: passdb missing")
                sys.exit(1)

def ensure_dovecot_module_symlink(output):
    """
    Ensure lib_imunify360.so points to a suitable
    lib in DOVECOT_I360_DIR based on the installed dovecot version.
    """
    version = get_dovecot_version()
    if version is None:
        raise ValueError("failed to query dovecot version")

major, minor, patch = version
    exact_name = DOVECOT_LIB_IMUNIFY % (major, minor, patch)
    src = os.path.join(DOVECOT_I360_DIR, exact_name)
    if not os.path.exists(src):
        raise Imunify360DovecotNotFound(
            "no suitable imunify360 dovecot library found in %s for version %s"
            % (DOVECOT_I360_DIR, version)
        )

dst_dir = DOVECOT_AUTH_DIR
    os.makedirs(dst_dir, exist_ok=True)
    dst = DOVECOT_LIB_LINK

# If symlink already points to the exact version, skip updating
    if os.path.islink(dst):
        try:
            current_target = os.readlink(dst)
            if os.path.basename(current_target) == os.path.basename(src):
                logger.info(
                    "symlink skipped: dovecot module points to the right version"
                )
                return
        except OSError:
            pass

with suppress(FileNotFoundError):
        os.unlink(dst)

try:
        os.symlink(src, dst)
        logger.info("symlinked dovecot module: %s -> %s", dst, src)
    except OSError as e:
        output.error("failed to symlink %s -> %s: %s" % (dst, src, e))
        raise e

def patch_dovecot_config_template(dovecot_state: str, config_basedir: str):
    config_template = os.path.join(config_basedir, CONFIG_DOVECOT_TMPL)
    config_local = os.path.join(config_basedir, CONFIG_DOVECOT_LOCAL)
    if is_higher_than_dovecot24():
        passdb_regex = re.compile(
                    r"^\s*passdb\s*cpauthd(?:\s+\w+)?\s*\{(?:[^{}]|\{[^{}]*\})*\}\s*$",
                    re.DOTALL | re.MULTILINE
                )
    else:
        passdb_regex = re.compile(r"^\s*passdb\s*\{.*?\}\s*$", re.DOTALL | re.MULTILINE)

if dovecot_state == DovecotState.PAM or dovecot_state == DovecotState.NATIVE:

def passdb_replace(match):
            repl = None
            if dovecot_state == DovecotState.PAM:
                if is_higher_than_dovecot24():
                    repl = re.sub(
                        r"passdb\s*cpauthd\s*\{",
                        "passdb pam {",
                        match.group(0)
                    )
                    repl = re.sub(
                        r"^\s*driver\s*=.*$",
                        r"  service_name = "
                        r"[% IF allow_domainowner_mail_pass %]"
                        r"dovecot_imunify_domainowner"
                        r"[% ELSE %]dovecot_imunify[% END %]",
                        repl,
                        flags=re.MULTILINE
                    )
                    repl = re.sub(r"^\s*lua_file\s*=.*\n", "",
                                  repl,
                                  flags=re.MULTILINE
                    )
                    repl = re.sub(
                        r"^\s*lua_settings\s*\{[^}]*\}\s*\n",
                        "",
                        repl,
                        flags=re.MULTILINE | re.DOTALL
                    )
                else:
                    repl = re.sub(r"driver\s*=.*", "driver = pam", match.group(0))
                    repl = re.sub(
                        r"args\s*=.*",
                        r"args = "
                        r"[% IF allow_domainowner_mail_pass %]"
                        r"dovecot_imunify_domainowner"
                        r"[% ELSE %]dovecot_imunify[% END %]",
                        repl,
                    )

if dovecot_state == DovecotState.NATIVE:
                repl = re.sub(
                    r"result_internalfail\s*=.*",
                    "result_success = continue-ok",
                    match.group(0),
                )
                repl = re.sub(
                    r"result_failure\s*=.*", "result_failure = continue-fail", repl
                )
            return repl

data = Path(config_template).read_text()
        data = re.sub(passdb_regex, passdb_replace, data)
        if dovecot_state == DovecotState.NATIVE:
            data = insert_imunify_passdb(data)

if not options.dry:
            with open(config_local, "w") as f:
                f.write(data)
        else:
            return
    elif dovecot_state == DovecotState.DISABLED:
        with suppress(FileNotFoundError):
            os.unlink(config_local)

def change_dovecot_state(dovecot_state, output=None):
    """
    Enable or disable pam_imunify support for Dovecot
    """
    if not output:
        output = Output()

manyconfigs, warn = dovecot_manyconfigs_basedir()
    if warn:
        output.warning(warn)
    if len(manyconfigs) == 0:
        output.error("Dovecot config template file not found. Aborting.")
        sys.exit(1)
    for config_basedir in manyconfigs:
        patch_dovecot_config_template(dovecot_state, config_basedir)

if not options.norestart:
        if os.path.isfile("/scripts/builddovecotconf"):
            output.run_and_log(["/scripts/builddovecotconf"])
        if os.path.isfile("/scripts/restartsrv_dovecot"):
            output.run_and_log(["/scripts/restartsrv_dovecot"])

def service_incompatibility_panic(msg):
    """
    So far we decided to report service incompatibility error as warning,
    with the only exception for --dry-run option.

Otherwise we break agent PAM subsystem loop with the error when
    a client copied imunify360-merged.config from one server to another server
    in a case when the first server is compatible with that PAM integration feature
    but the second server is not capable with.
    """
    if options.dry:
        output.error(msg)
        sys.exit(1)
    else:
        output.warning(msg)
        sys.exit(0)

def cpanel_only_feature(service):
    def decorator(fun):
        @wraps(fun)
        def wrapper(*args, **kwargs):
            if get_cp_name() != "cpanel":
                service_incompatibility_panic(
                    "%s is not supported for %s."
                    % (service, get_cp_name().capitalize())
                )
            else:
                return fun(*args, **kwargs)

return wrapper

return decorator

def toggle_proftpd_support(enable=True, output=None):
    """
    Enable or disable pam_imunify support for ProFTPd
    """
    conf = CONFIG_PROFTPD
    if not os.path.isfile(conf):
        output.error("ProFTPD config file not found. Aborting.")
        sys.exit(1)

if enable:
        version_output = subprocess.check_output(
            [
                "in.proftpd" if get_cp_name() == "plesk" else "proftpd",
                "--version-status",
            ],
            stderr=subprocess.DEVNULL,
        ).decode(sys.stdout.encoding)
        if "mod_auth_pam" not in version_output:
            service_incompatibility_panic(
                "ProFTPD built without PAM support. "
                "pam_imunify for FTP is NOT enabled."
            )

version_regex = re.compile(r"ProFTPD Version:\s([0-9a-z\.]+)")
        version_found = version_regex.search(version_output)
        if version_found:
            version = LooseVersion(version_found.group(1))
            if version < LooseVersion("1.3.6c") or version.vstring.startswith(
                "1.3.6rc"
            ):
                if get_cp_name() == "cpanel":
                    service_incompatibility_panic(
                        "ProFTPD needs to be upgraded to "
                        "cPanel version 88 or higher. "
                        "pam_imunify for FTP is NOT enabled."
                    )
                else:
                    service_incompatibility_panic(
                        "ProFTPD needs to be upgraded. "
                        "pam_imunify for FTP is NOT enabled."
                    )

if not output:
        output = Output()

authpam_imunify = (
        "AuthOrder mod_auth_pam.c* mod_auth_file.c\n"
        "AuthPAM on\n"
        "AuthPAMConfig proftpd_imunify\n"
    )

authpam_regex = re.compile(r"(^AuthPAM.*\n?)+", re.MULTILINE)

data = open(conf).read()
    authpam_found = authpam_regex.search(data)

if enable:
        if authpam_found:
            authpam_span = authpam_found.span()
            data = data[: authpam_span[0]] + authpam_imunify + data[authpam_span[1] :]
        else:
            data = authpam_imunify + "\n" + data

if not options.dry:
            atomic_rewrite(conf, data)
        else:
            return
    else:
        conf_bak = conf + ".i360bak"
        if os.path.isfile(conf_bak):
            os.rename(conf_bak, conf)
        else:
            output.error(
                "Failed to disable proftpd integration: %s not found" % conf_bak
            )
            sys.exit(1)
    if os.path.isfile("/scripts/restartsrv_ftpd"):
        output.run_and_log(["/scripts/restartsrv_ftpd"])

def file_patchline(config: str, pattern, repl: bytes, reverse: bool):
    """
    Patch config file line inplace and backup config to '%s.i360bak' % config
    :param config: file path
    :param pattern: re.compile(b'...') result type
    :param reverse: revert back the previos operation on the same config file
    """
    if reverse:
        # lookup replacement in config_i360bak
        try:
            with open("%s.i360bak" % config, "rb") as f:
                repl = next(ln for ln in f if re.match(pattern, ln))
        except (FileNotFoundError, StopIteration):
            # there were no such entry before us
            repl = b""

if os.path.exists(config):
        with open(config, "rb") as f:
            conf_before = f.read()
            conf_after = re.sub(pattern, repl, conf_before, count=1)
            if conf_after == conf_before and not repl in conf_after:
                conf_after = (
                    conf_before + (b"" if conf_before.endswith(b"\n") else b"\n") + repl
                )
        if conf_after != conf_before:
            atomic_rewrite(config, conf_after)
    else:
        atomic_rewrite(config, repl)

def is_pureftpd_supported():
    # Pure-FTPd writes output to stdin, so we have to use
    # pipes to read from stdin afterwards...
    pipe_r, pipe_w = map(os.fdopen, os.pipe())

def finalize():
        pipe_w.close()
        pipe_r.close()

try:
        subprocess.check_output(
            ["pure-ftpd", "-l", "pam"],
            stdin=pipe_w,
            stderr=subprocess.STDOUT,
            timeout=1,
        )
    except subprocess.CalledProcessError:
        # after pipe_w.close() we can do pipe_r.read()
        pipe_w.close()
        with closing(pipe_r):
            # 421 Unknown authentication method: pam
            if pipe_r.read().startswith("421 "):
                return False
    except subprocess.TimeoutExpired:
        # This could happen if pam is supported
        # and pure-ftpd has started
        finalize()
    else:
        finalize()

return True

def is_pureftpd_enabled():
    """
    Check if pure-ftpd.conf contains /var/run/ftpd.imunify360.sock
    """
    if not os.path.isfile(CONFIG_PUREFTPD):
        return False

imunify360_regex = re.compile(
        rb"^(?!#).*\/var\/run\/ftpd.imunify360.sock",
        re.MULTILINE
    )

return imunify360_regex.search(
               open(CONFIG_PUREFTPD, "rb").read()) is not None

def toggle_pureftpd_conf_support(enable, output):
    extauth_regex = re.compile(rb"^\s*ExtAuth\s.*$", re.MULTILINE)
    extauth_imunify = b"ExtAuth /var/run/ftpd.imunify360.sock"

file_patchline(CONFIG_PUREFTPD, extauth_regex, extauth_imunify, reverse=enable is False)

if not options.norestart_pureftpd and \
            os.path.isfile("/scripts/restartsrv_ftpd"):
        output.run_and_log(["/scripts/restartsrv_ftpd"])

def toggle_pureftpd_cpanel_support(enable, output):
    extauth_regex = re.compile(rb"^\s*ExtAuth:\s.*$", re.MULTILINE)
    extauth_imunify = b"ExtAuth: /var/run/ftpd.imunify360.sock"

file_patchline(CONFIG_TEMPLATE_PUREFTPD, extauth_regex, extauth_imunify, reverse=enable is False)

if os.path.isfile("/scripts/setupftpserver"):
        output.run_and_log(["/usr/local/cpanel/scripts/setupftpserver", "--force", "pure-ftpd"])

def disable_ftp_protection():
    if not os.path.exists(CONFIG_IMUNIFY360):
        return

ftp_protection_pattern = re.compile(
        rb"^(?!#)([^\S\r\n]*ftp_protection:[^\S\r\n])*true",
        re.MULTILINE
    )
    def replacement(match):
        return match.group(1) + b'false'

with open(CONFIG_IMUNIFY360, "rb") as f:
        contents = f.read()
        if ftp_protection_pattern.search(contents):
            disabled_ftp_content = re.sub(
                ftp_protection_pattern,
                replacement,
                contents,
                count=1
            )
            atomic_rewrite(CONFIG_IMUNIFY360, disabled_ftp_content)

def toggle_pureftpd_support(enable=True, output=None):
    """
    Enable or disable pam_imunify support for PureFTPd
    """
    conf = CONFIG_PUREFTPD
    if not os.path.isfile(conf):
        output.error("Pure-FTPd config file not found. Aborting.")
        sys.exit(1)
    if enable:
        if not is_pureftpd_supported():
            service_incompatibility_panic(
                "Pure-FTPd built without PAM support. "
                "pam_imunify for FTP is NOT enabled."
            )

# ensure that ftp_protection is disabled in imunify360-merged.config
    if enable is False:
        disable_ftp_protection()

if not output:
        output = Output()
    if options.dry:
        return

toggle_pureftpd_conf_support(enable, output)

def toggle_sshd_support(conffiles, enable=True, output=None):
    """
    Enable or disable pam_imunify module for sshd authentication
    """
    if not output:
        output = Output()

if enable:
        for conf in conffiles:
            lines = open(conf).readlines()

try:
                pam_unix_ln = next(
                    ln for ln, line in enumerate(lines) if PAM_UNIX_REGEX.search(line)
                )
            except StopIteration:
                output.error("PAM configuration file %s parse error" % conf)
                sys.exit(1)

pam_unix_patch_around(lines, pam_unix_ln)

content = "".join(lines)
            i360RPatch(conf, output).create_upon(content)
            if not options.dry:
                atomic_rewrite(conf, content)
    else:
        for conf in conffiles:
            rpatch = i360RPatch(conf, output)
            if os.path.exists(rpatch.filename()):
                try:
                    rpatch.apply()
                    continue
                except subprocess.CalledProcessError as e:
                    output.warning(
                        "'patch -R' was not successful: %s" % e, file=sys.stderr
                    )
            else:
                output.warning(
                    "File not found: %s" % rpatch.filename(), file=sys.stderr
                )

atomic_rewrite(conf, open(conf + ".i360bak").read())

def set_panel_integration(confs, output=None):
    imunify_regex = re.compile(r"auth\s+sufficient\s+pam_imunify\.so")
    assert get_cp_name() == "cpanel", "The only supported integration so far."

if not output:
        output = Output()

for conf in confs:
        lines = open(conf).readlines()

try:
            imunify_ln = next(
                ln for ln, line in enumerate(lines) if imunify_regex.search(line)
            )
        except StopIteration:
            output.error("PAM configuration file %s parse error" % conf)
            sys.exit(1)

match_count = 0

def panel_replace(match):
            nonlocal match_count
            if match.group() in ["cpanel", "plesk", "directadmin"]:
                match_count += 1
                return get_cp_name()
            return match.group()

lines[imunify_ln] = re.sub(r"\b([^\s]+)\b", panel_replace, lines[imunify_ln])

if match_count == 0:
            lines[imunify_ln] = "%s %s\n" % (lines[imunify_ln].strip(), get_cp_name())

content = "".join(lines)
        atomic_rewrite(conf, content)
        os.unlink(conf + ".i360bak")

def patch_simple(pamconfig_lines, pam_unix_ln):
    pamconfig_lines.insert(pam_unix_ln + 1, "auth\trequired\tpam_imunify.so\n")
    pamconfig_lines.insert(pam_unix_ln, "auth\trequired\tpam_imunify.so\tcheck_only\n")

def fix_offset(pamconfig_lines, pam_unix_ln, pam_unix_success_offset):
    bump_to = pam_unix_success_offset + 1
    pamconfig_lines[pam_unix_ln] = re.sub(
        r"success=\d", "success=%d" % bump_to, pamconfig_lines[pam_unix_ln]
    )

def config_dovecot_state():
    """
    Read dovecot state from imunify360-merged.config

:return: DovecotState based on exim_dovecot_native
        and exim_dovecot_protection:
             - NATIVE if exim_dovecot_native is true
             - PAM if exim_dovecot_protection is true
             - DISABLED if both are false
             Returns None if config file doesn't exist or can't be parsed
    """
    if not os.path.exists(CONFIG_IMUNIFY360):
        return None

try:
        with open(CONFIG_IMUNIFY360, "r") as f:
            config = yaml.safe_load(f)

if not config:
            return None

pam_config = config.get("PAM", {})
        exim_dovecot_native = pam_config.get("exim_dovecot_native", False)
        exim_dovecot_protection = pam_config.get(
                "exim_dovecot_protection",
                False
            )

if exim_dovecot_native and exim_dovecot_protection:
            return DovecotState.NATIVE

if exim_dovecot_protection:
            return DovecotState.PAM

return DovecotState.DISABLED
    except (yaml.YAMLError, IOError, KeyError) as e:
        logger.warning(
                "Failed to read dovecot stats from %s: %s",
                CONFIG_IMUNIFY360,
                e
        )
        return None

def dovecot_state():
    pam_dovecot_enabled = (
        os.path.isfile(CONFIG_DOVECOT)
        and "dovecot_imunify" in open(CONFIG_DOVECOT).read()
    )
    native_dovecot_enabled = (
        os.path.isfile(CONFIG_DOVECOT) and "imunify360" in open(CONFIG_DOVECOT).read()
    )

if pam_dovecot_enabled:
        return DovecotState.PAM
    elif native_dovecot_enabled:
        return DovecotState.NATIVE
    else:
        return DovecotState.DISABLED

class Cmd:
    @classmethod
    def enable(cls, conffiles, output=None):
        if not output:
            output = Output()

if any("pam_imunify.so" in open(conf).read() for conf in conffiles):
            cls._cphulk_check()
            output.status_changed({"sshd": (True, True)})
            return

toggle_sshd_support(conffiles, True, output)

if not options.dry:
            cls._cphulk_check(output)
            output.status_changed({"sshd": (False, True)})

@staticmethod
    @cpanel_only_feature("Dovecot")
    def set_dovecot(conffiles, output=None):
        if not output:
            output = Output()

target_dovecot_state = DOVECOT_STATES[options.dovecot_state]
        if target_dovecot_state:
            prev_dovecot_state = dovecot_state()
            if prev_dovecot_state != target_dovecot_state:
                set_panel_integration(
                    [CONFIG_PAM_DOVECOT, CONFIG_PAM_DOVECOT_DOMAINOWNER], output
                )

if not options.dry \
                    and target_dovecot_state == DovecotState.NATIVE:
                    try:
                        ensure_dovecot_module_symlink(output)
                    except Imunify360DovecotNotFound as e:
                        output.error("Missing imunify360 dovecot-native:%s" % e)
                        sys.exit(1)

change_dovecot_state(target_dovecot_state, output)
            if target_dovecot_state in [DovecotState.PAM, DovecotState.NATIVE]:
                output.status_changed(
                    {
                        "dovecot-{}".format(options.dovecot_state): (
                            prev_dovecot_state == target_dovecot_state,
                            True,
                        )
                    }
                )
            else:
                output.status_changed(
                    {"dovecot": (prev_dovecot_state != target_dovecot_state, False)}
                )

else:
            output.error("Unexpected dovecot state {}".format(options.dovecot_state))

@staticmethod
    @cpanel_only_feature("Dovecot")
    def dovecot_native(output=None):
        if not output:
            output = Output()
        if not options.dry:
            ensure_dovecot_module_symlink(output)

@staticmethod
    @cpanel_only_feature("Dovecot")
    def dovecot_reset(conffiles, output=None):
        if not output:
            output = Output()

logger.info("Resetting dovecot state")

current_config_state = config_dovecot_state()
        imunify360DovecotLinkMissing = False
        try:
            ensure_dovecot_module_symlink(output)
        except Imunify360DovecotNotFound as e:
            imunify360DovecotLinkMissing = True
            # Unlink dovecot module
            with suppress(FileNotFoundError):
                logger.info("Unlinking dovecot %s", DOVECOT_LIB_LINK)
                os.unlink(DOVECOT_LIB_LINK)

# Disable dovecot native
            if current_config_state == DovecotState.NATIVE \
                    and dovecot_state() == DovecotState.NATIVE:
                change_dovecot_state(DovecotState.DISABLED, output)
                output.status_changed({f"dovecot-native": (True, False)})

if current_config_state == DovecotState.PAM:
            change_dovecot_state(DovecotState.DISABLED, output)
            change_dovecot_state(DovecotState.PAM, output)
            output.status_changed({f"dovecot-pam": (False, True)})
        elif current_config_state == DovecotState.NATIVE \
                and imunify360DovecotLinkMissing is False:
            change_dovecot_state(DovecotState.DISABLED, output)
            change_dovecot_state(DovecotState.NATIVE, output)
            output.status_changed({f"dovecot-native": (False, True)})
        return

@staticmethod
    @cpanel_only_feature("ProFTPd")
    def enable_proftpd(conffiles, output=None):
        if not output:
            output = Output()

set_panel_integration([CONFIG_PAM_PROFTPD], output)

proftpd_enabled = (
            os.path.isfile(CONFIG_PROFTPD)
            and "proftpd_imunify" in open(CONFIG_PROFTPD).read()
        )

if not proftpd_enabled:
            toggle_proftpd_support(True, output)

if not options.dry:
            output.status_changed({"ftp": (proftpd_enabled, True)})

@staticmethod
    @cpanel_only_feature("Pure-FTPd")
    def enable_pureftpd(conffiles, output=None):
        if not output:
            output = Output()

panel = get_cp_name()

pureftpd_enabled = is_pureftpd_enabled()

if not pureftpd_enabled:
            toggle_pureftpd_support(True, output)

if not options.dry:
            output.status_changed({"ftp": (pureftpd_enabled, True)})

@staticmethod
    def disable_all(conffiles, output=None):
        if not output:
            output = Output()

dovecot_enabled = dovecot_state() != DovecotState.DISABLED
        proftpd_enabled = (
            os.path.isfile(CONFIG_PROFTPD)
            and "proftpd_imunify" in open(CONFIG_PROFTPD).read()
        )
        pureftpd_enabled = is_pureftpd_enabled()
        sshd_enabled = any("pam_imunify.so" in open(conf).read() for conf in conffiles)

if dovecot_enabled:
            change_dovecot_state(DovecotState.DISABLED, output)

if proftpd_enabled:
            toggle_proftpd_support(False, output)

if pureftpd_enabled:
            toggle_pureftpd_support(False, output)

if sshd_enabled:
            toggle_sshd_support(conffiles, False, output)

output.status_changed(
            {
                "sshd": (sshd_enabled, False),
                "dovecot": (dovecot_enabled, False),
                "ftp": (proftpd_enabled or pureftpd_enabled, False),
            }
        )

@staticmethod
    def disable(conffiles, output=None):
        if not output:
            output = Output()

sshd_enabled = any("pam_imunify.so" in open(conf).read() for conf in conffiles)

if sshd_enabled:
            toggle_sshd_support(conffiles, False, output)

output.status_changed({"sshd": (sshd_enabled, False)})

@staticmethod
    def disable_proftpd(conffiles, output=None):
        if not output:
            output = Output()

proftpd_enabled = (
            os.path.isfile(CONFIG_PROFTPD)
            and "proftpd_imunify" in open(CONFIG_PROFTPD).read()
        )
        if proftpd_enabled:
            toggle_proftpd_support(False, output)

output.status_changed({"ftp": (proftpd_enabled, False)})

@staticmethod
    def disable_pureftpd(conffiles, output=None):
        if not output:
            output = Output()

pureftpd_enabled = is_pureftpd_enabled()
        if pureftpd_enabled:
            toggle_pureftpd_support(False, output)

output.status_changed({"ftp": (pureftpd_enabled, False)})

@staticmethod
    @cpanel_only_feature("FTP service")
    def enable_ftp(conffiles, output=None):
        if not output:
            output = Output()

with open("/var/cpanel/cpanel.config", "r") as cpcfg:
            data = cpcfg.read()
            if "ftpserver=proftpd" in data:
                Cmd.enable_proftpd(conffiles, output)
            elif "ftpserver=pure-ftpd" in data:
                Cmd.enable_pureftpd(conffiles, output)
            else:
                service_incompatibility_panic("No supported FTP found.")

@staticmethod
    def disable_ftp(conffiles, output=None):
        if not output:
            output = Output()

proftpd_enabled = (
            os.path.isfile(CONFIG_PROFTPD)
            and "proftpd_imunify" in open(CONFIG_PROFTPD).read()
        )
        pureftpd_enabled = is_pureftpd_enabled()

if proftpd_enabled:
            toggle_proftpd_support(False, output)

if pureftpd_enabled:
            toggle_pureftpd_support(False, output)

output.status_changed({"ftp": (proftpd_enabled or pureftpd_enabled, False)})

@classmethod
    def status(cls, conffiles, output=None):
        if not output:
            output = Output()

dovecot_enabled = dovecot_state() != DovecotState.DISABLED
        sshd_enabled = any("pam_imunify.so" in open(conf).read() for conf in conffiles)
        proftpd_enabled = (
            os.path.isfile(CONFIG_PROFTPD)
            and "proftpd_imunify" in open(CONFIG_PROFTPD).read()
        )
        pureftpd_enabled = is_pureftpd_enabled()

if dovecot_enabled or sshd_enabled or proftpd_enabled or pureftpd_enabled:
            cls._cphulk_check(output)

output.status(
            {
                "sshd": sshd_enabled,
                "dovecot-pam": dovecot_state() == DovecotState.PAM,
                "dovecot-native": dovecot_state() == DovecotState.NATIVE,
                "ftp": proftpd_enabled or pureftpd_enabled,
            }
        )

@staticmethod
    def state_reset(*_):
        subprocess.check_call(["service", "imunify360-pam", "stop"])
        shutil.rmtree(LEVELDB)
        logger.info("rm -rf %s", LEVELDB)
        subprocess.check_call(["service", "imunify360-pam", "start"])

@staticmethod
    def _cphulk_check(output=None):
        if not os.path.isfile("/usr/sbin/whmapi1"):
            return
        if not options.verbose and not pam_imunify_config().getboolean("verbose"):
            return

if not output:
            output = Output()

proc = subprocess.run(
            ["whmapi1", "servicestatus", "service=cphulkd"],
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
        )
        if proc.returncode != 0:
            # we expect err dump is printed to stderr
            return

try:
            status = yaml.safe_load(proc.stdout)
            if status["data"]["service"][0]["enabled"]:
                output.warning("cPHulk is enabled", file=sys.stderr)
        except (yaml.YAMLError, IndexError, KeyError) as e:
            output.warning("whmapi error:", e, file=sys.stderr)

def sigterm_handler(signum, frame):
    """
    generate backtrace on SIGTERM
    """
    traceback.print_stack(frame, file=sys.stderr)
    print("caught SIGTERM.", file=sys.stderr)
    if logger is not None:
        logger.fatal("caught SIGTERM.")
    sys.exit(15)

if __name__ == "__main__":
    def add_opt_args(parser):
        parser.add_argument(
            "-r",
            "--dry-run",
            dest="dry",
            action="store_true",
            help="Dry run the command, whithout changing of state",
        )
        parser.add_argument(
            "-n",
            "--no-restart",
            dest="norestart",
            action="store_true",
            help="Don't restart dovecot and don't rebuild, just patch local dovecot template (cPanel only)",
        )
        parser.add_argument(
            "--no-restart-pureftpd",
            dest="norestart_pureftpd",
            action="store_true",
            help="Don't restart pureftpd",
        )
        parser.add_argument(
            "--yaml", dest="yaml", action="store_true", help="for YAML output"
        )
        parser.add_argument("-v", "--verbose", dest="verbose", action="store_true")

signal.signal(signal.SIGTERM, sigterm_handler)
    logger = logger_init(console_stream=None)
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="cmd")
    subparsers.required = True

for command in sorted(
        (cmd.replace("_", "-") for cmd in dir(Cmd) if not cmd.startswith("_")),
        reverse=True,
    ):
        if command == "set-dovecot":
            command_parser = subparsers.add_parser(command)
            command_parser.add_argument(
                "dovecot_state", type=str, choices=["pam", "native", "disabled"]
            )
            add_opt_args(command_parser)
        elif command == "dovecot-native":
            command_parser = subparsers.add_parser(command)
            command_parser.add_argument("action", choices=["symlink"])
            add_opt_args(command_parser)
        else:
            command_parser = subparsers.add_parser(command)
            add_opt_args(command_parser)

options = parser.parse_args()
    output = YamlOutput() if options.yaml else Output()
    try:
        cmd = getattr(Cmd, options.cmd.replace("-", "_"))
        if options.cmd == "dovecot-native":
            cmd(output)
        else:
            cmd(detect_conffiles(output), output)
    except Exception as e:
        logger.exception("unexpected error: %s", e)
    finally:
        # even if we caught an exception there is still a possibility
        # that stdout is still usable (at least partially)
        output.flush()

📁 Sid Gifari File Manager