Gecko [ tanishkshah.com ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2026-09-04 11:33
__init__.py	381 B	-rw-r--r--	2026-08-06 17:01
base.py	65.16 KB	-rw-r--r--	2026-08-06 17:01
cpanel.py	23.74 KB	-rw-r--r--	2026-08-06 17:01
da.py	17.31 KB	-rw-r--r--	2026-08-06 17:01
ispmanager.py	3.19 KB	-rw-r--r--	2026-08-06 17:01
iworx.py	1.04 KB	-rw-r--r--	2026-08-06 17:01
plesk.py	1.24 KB	-rw-r--r--	2026-08-06 17:01
storage.py	6.69 KB	-rw-r--r--	2026-08-06 17:01

Rename

#coding:utf-8

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
#
"""
This module contains class for managing governor on DirectAdmin server
"""
import os
import pwd
import shutil
import subprocess
from glob import glob

from utilities import (
    check_file,
    exec_command,
    exec_command_out,
    grep,
    read_file,
    remove_packages,
    write_file,
)

from .base import InstallManager

class DirectAdminManager(InstallManager):
    """
    Implementation for DA panel
    """
    CONF_FILE_MYSQL = "/usr/local/directadmin/conf/mysql.conf"

def update_user_map_file(self):
        """
        Update user mapping file for cPanel
        """
        self._script("dbgovernor_map.py")

def _delete(self, installed_packages):
        """
        Remove installed packages
        """
        check_file("/usr/local/directadmin/custombuild/build")

print("Removing mysql for db_governor start")

self._mysqlservice("stop")
        # remove governor package
        exec_command_out("rpm -e governor-mysql")
        # delete installed packages
        remove_packages(installed_packages)

param = "mysql"
        if os.path.exists("/usr/share/lve/dbgovernor/da.tp.old"):
            param = read_file("/usr/share/lve/dbgovernor/da.tp.old")

# allowlist gates ONLY the set mysql_inst call; an empty/invalid value
        # skips that one call but must NOT skip the restore below.
        param = param.strip()
        if param in ("mysql", "mariadb", "no"):
            # no-shell argv vector: param can't be parsed as shell syntax
            subprocess.run(
                ["/usr/local/directadmin/custombuild/build", "set", "mysql_inst", param],
                check=False)

exec_command_out("/usr/local/directadmin/custombuild/build mysql update")

print("Removing mysql for db_governor completed")

def _before_install_new_packages(self):
        """
        Specific actions before new packages installation
        """
        print("The installation of MySQL for db_governor has started")

check_file("/usr/local/directadmin/custombuild/build")
        check_file("/usr/local/directadmin/custombuild/options.conf")

# MYSQL_DA_TYPE=`cat /usr/local/directadmin/custombuild/options.conf | grep mysql_inst= | cut -d= -f2`
        try:
            MYSQL_DA_TYPE = grep("/usr/local/directadmin/custombuild/options.conf", "mysql_inst=")[0].split("=")[1]
        except IndexError:
            MYSQL_DA_TYPE = ""

if os.path.exists("/usr/share/lve/dbgovernor/da.tp.old"):
            if MYSQL_DA_TYPE == "no":
                MYSQL_DA_TYPE = read_file("/usr/share/lve/dbgovernor/da.tp.old")
            else:
                write_file("/usr/share/lve/dbgovernor/da.tp.old", MYSQL_DA_TYPE)
        else:
            write_file("/usr/share/lve/dbgovernor/da.tp.old", MYSQL_DA_TYPE)

exec_command_out("/usr/local/directadmin/custombuild/build set mysql_inst no")

self._mysqlservice("stop")

def get_mysql_user(self):
        """
        Retrieve MySQL user name and password and save it into self attributes
        """
        if not os.path.exists(self.CONF_FILE_MYSQL):
            return None
        try:
            self.MYSQLUSER = grep(self.CONF_FILE_MYSQL, "user=")[0].split("=")[1]
            self.MYSQLPASSWORD = grep(self.CONF_FILE_MYSQL, "passwd=")[0].split("=")[1]
        except IndexError:
            pass

def _after_install_new_packages(self):
        """
        Specific actions after new packages installation
        """
        # call parent after_install
        InstallManager._after_install_new_packages(self)
        print("Rebuild php please... /usr/local/directadmin/custombuild/build php")

def _get_custombuild_option(self, option_name):
        """
        Get an option from the DirectAdmin custombuild options.conf file
        """
        CUSTOMBUILD_OPTIONS = "/usr/local/directadmin/custombuild/options.conf"
        option_regex = "{}=".format(option_name)
        try:
            option_grep = grep(CUSTOMBUILD_OPTIONS, option_regex)
            if not option_grep:
                return None
            option_value = option_grep[0].split("=")[1].strip()
            return option_value
        except IndexError:
            return None

def _detect_version_if_auto(self):
        """
        Detect version of MySQL if mysql.type is auto
        """
        print("Detecting MySQL version for AUTO")

try:
            # We can reach this section before calling _check_mysql_version in
            # InstallManager.install by calling manager.unsupported_db_version
            # from install\mysqlgovernor.py
            # self.prev_version won't be assigned then, try it now
            if not self.prev_version:
                self.prev_version = self._check_mysql_version()
            MYSQL_DA_VER = self.prev_version['full']
            print(f'Detected successfully from installed mysql binary: {MYSQL_DA_VER}')
        except (KeyError, AttributeError):
            print('Failed to detect from mysql binary, trying to detect from custombuild options')
            check_file("/usr/local/directadmin/custombuild/build")
            check_file("/usr/local/directadmin/custombuild/options.conf")

# MYSQL_DA_TYPE=`cat /usr/local/directadmin/custombuild/options.conf | grep mysql_inst= | cut -d= -f2`
            # This parameter is used to indicate what type of DB should be installed.
            # Typical values are 'mysql', 'mariadb' or 'no'.
            MYSQL_DA_TYPE = self._get_custombuild_option("mysql_inst")

# On newer versions of DirectAdmin, the config parameter used to define MariaDB version is mariadb.
            # On older ones, both MySQL and MariaDB versions are defined by mysql parameter.
            if MYSQL_DA_TYPE == "mariadb":
                MARIADB_DA_VER = self._get_custombuild_option("mariadb")

MYSQL_DA_VER = self._get_custombuild_option("mysql")

# If we have a specified MariaDB version, we should use it.
            # Otherwise, fall back to the older approach and use the mysql parameter for MariaDB versions too.
            if MYSQL_DA_TYPE == "mariadb" and MARIADB_DA_VER:
                MYSQL_DA_VER = MARIADB_DA_VER

if MYSQL_DA_TYPE == "no":
                if os.path.exists("/usr/share/lve/dbgovernor/da.tp.old"):
                    MYSQL_DA_TYPE = read_file("/usr/share/lve/dbgovernor/da.tp.old")
                elif os.path.exists("/usr/bin/mysql"):
                    result = exec_command("/usr/bin/mysql -V | grep -c 'MariaDB' -i || true", True)
                    if result == "0":
                        MYSQL_DA_TYPE = "mysql"
                    else:
                        MYSQL_DA_TYPE = "mariadb"

print("I got %s and %s" % (MYSQL_DA_VER, MYSQL_DA_TYPE))

mysql_version_map = {
                "5.0": "mysql50",
                "5.1": "mysql51",
                "5.5": "mysql55",
                "5.6": "mysql56",
                "5.7": "mysql57",
                "8.0": "mysql80",
                "8.4": "mysql84",
                "10.0.0": "mariadb100",
                "10.1.1": "mariadb101"
            }
            mariadb_version_map = {
                "11.8": "mariadb1108",
                "11.08": "mariadb1108",
                "11.4": "mariadb1104",
                "11.04": "mariadb1104",
                "10.11": "mariadb1011",
                "10.6": "mariadb106",
                "10.5": "mariadb105",
                "10.4": "mariadb104",
                "10.3": "mariadb103",
                "10.2": "mariadb102",
                "10.1": "mariadb101",
                "10.0": "mariadb100",
                "5.6": "mariadb100",
                "5.5": "mariadb100",
                "10.0.0": "mariadb100",
                "10.1.1": "mariadb101"
            }

# Double-check that we actually have a valid version and type
            try:
                # Did we actually detect a version from the mapping?
                if not MYSQL_DA_TYPE or not MYSQL_DA_VER:
                    # It's OK not to reraise the exception here, the scenario being handled is different
                    # pylint: disable=raise-missing-from
                    raise AttributeError("MySQL/MariaDB version could not be detected")
                if MYSQL_DA_TYPE == "mysql":
                    MYSQL_DA_VER = mysql_version_map[MYSQL_DA_VER]
                elif MYSQL_DA_TYPE == "mariadb":
                    MYSQL_DA_VER = mariadb_version_map[MYSQL_DA_VER]
            # In case we have a version that is not in the mapping
            except KeyError as e:
                raise RuntimeError(f"Unsupported MySQL version: {MYSQL_DA_VER} ({MYSQL_DA_TYPE})") from e

return MYSQL_DA_VER

def _is_trusted_cache_dir(self, path):
        """
        Refuse to trust an rpm found under a group/world-writable directory,
        or one not owned by root or the DirectAdmin service account -- an
        unprivileged local user could have planted or replaced it there, or
        could own the directory outright even when its mode bits look safe
        (owner-write is not covered by the group/world-write mask above).
        """
        try:
            dir_stat = os.stat(os.path.dirname(path))
        except OSError:
            return False
        if dir_stat.st_mode & (0o020 | 0o002):
            return False
        trusted_uids = {0}
        try:
            trusted_uids.add(pwd.getpwnam("diradmin").pw_uid)
        except KeyError:
            pass
        return dir_stat.st_uid in trusted_uids

def _custom_rpm_locator(self, package_name):
        """
        Where we should download installed MySQL package from.

For the special argument value "package_name"=="+"
        returning "yes" means that this _custom_rpm_locator() implementation is fully functional,
        and returning any other value means that it's not supported.

There could be a lot of packages in /usr/local/directadmin/custombuild/mysql,
        not all of them relevant (installed) or fine ones (not corrupted)
        """
        if package_name == "+":   # supported?
            return "yes"

bad_pkg = False
        list_of_rpm = glob("/usr/local/directadmin/custombuild/mysql/*.rpm") + glob(
            "/usr/local/directadmin/scripts/packages/*.rpm")
        for found_package in list_of_rpm:
            try:
                # no-shell argv vector: found_package is glob()-sourced and
                # attacker-influenceable, so it must never reach a shell
                qp = subprocess.run(
                    ["/bin/rpm", "-qp", found_package],
                    check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            except OSError as e:
                print(f"Failed to query package {found_package}: {e}\n")
                bad_pkg = True
                continue
            if qp.returncode != 0:
                print(f"Failed to query package {found_package}: "
                      f"{qp.stderr.decode(errors='replace')}\n")
                bad_pkg = True
                continue
            result = qp.stdout.decode().strip()
            # exact NEVRA match only: substring containment let a
            # differently-named rpm masquerade as the requested package
            if package_name == result:
                pkg_name_real = found_package
                if pkg_name_real != "" and os.path.exists(pkg_name_real) \
                        and self._is_trusted_cache_dir(pkg_name_real):
                    return f"file:{pkg_name_real}"
                # name matched but the cache dir is untrusted (writable by
                # a non-owner principal): refuse it, don't force-install
                bad_pkg = True

if bad_pkg:
            return f"bad_file:{package_name}"
        else:
            return ""

def _verify_rpm_signature(self, path):
        """
        Refuse to trust an rpm whose GPG signature does not verify against
        the system's already-imported keyring -- the same mechanism yum/dnf
        rely on whenever --nogpgcheck is NOT passed. `rpm -K` alone is not
        enough: a completely unsigned rpm still reports header/payload
        digests "OK", so we additionally require the terse output to
        affirmatively confirm a verified signature; a missing signature, an
        unimported/untrusted key ("nokey"), or a failed check ("not ok")
        are all treated as untrusted.

rpm's terse checksig wording differs by version: rpm >= 4.14 (CL8+)
        summarizes as "... digests signatures OK"; rpm 4.11 (CL7) instead
        names each individual check that passed, e.g. "rsa sha1 (md5) pgp
        md5 OK", with no "signatures" word at all. Requiring the literal
        "signatures ok" substring refuses every legitimately signed package
        on CL7. Accept both eras: the output must end in "OK" AND name an
        actual signature check -- either the word "signatures" (new wording)
        or a signature-algorithm token (rsa/dsa/pgp/gpg, old wording) -- a
        digest-only pass ("... digests OK") satisfies neither and stays
        refused.
        """
        try:
            result = subprocess.run(
                ["/bin/rpm", "-K", path],
                check=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        except OSError:
            return False
        # surrogateescape (not "replace") to match how `path` itself was
        # decoded from raw filesystem bytes upstream (glob()/os.fsdecode);
        # otherwise a filename byte that isn't valid UTF-8 decodes
        # differently in each string and the startswith() prefix-strip
        # below silently fails to fire.
        output = result.stdout.decode(errors="surrogateescape")
        # rpm -K always echoes the literal path argument as the first token
        # of its own terse output ("<path>: <result>"); strip it before any
        # token/substring search below so a crafted filename (e.g. containing
        # "rsa"/"pgp"/"ok") can never spoof the acceptance check -- only
        # rpm's own verdict text must ever be inspected.
        if output.startswith(path):
            output = output[len(path):]
        output = output.lower()
        if result.returncode != 0 or "not ok" in output or "nokey" in output \
                or "missing keys" in output:
            return False
        if not output.rstrip().endswith("ok"):
            return False
        return any(token in output for token in
                   ("signatures", "pgp", "gpg", "rsa", "dsa"))

def _custom_rpm_installer(self, package_name, indicator=False):
        """
        Specific package installer
        :param package_name:
        :param indicator:
        :return: for indicator=False, True if the install was attempted,
                 False if refused by the signature gate. Callers (see
                 install_packages()/install_rollback() in base.py) must
                 check this so a systematic refusal is never reported as a
                 successful rollback.
        """
        if not indicator:
            if not self._verify_rpm_signature(package_name):
                # Loud and actionable on purpose: a keyring-only refusal
                # (no matching key imported) is plausible for a legitimate
                # vendor package (e.g. Percona, DirectAdmin's own custombuild
                # output) that was never signed with a locally-known key, not
                # only for an attacker-planted one -- the admin needs to see
                # WHY the rollback stopped and what to do about it, not just
                # a bare "refusing" line.
                print(f"Refusing to install {package_name}: GPG signature "
                      f"verification failed or no matching key is imported "
                      f"locally. This can be a legitimate vendor package "
                      f"whose signing key was never imported (not "
                      f"necessarily tampering) -- import the vendor's GPG "
                      f"key with 'rpm --import' or restore this package "
                      f"manually if you trust its origin.\n")
                return False
            # no-shell argv vector: package_name can't be parsed as shell syntax
            subprocess.run(
                ["/bin/rpm", "-ihv", "--force", "--nodeps", package_name],
                check=False)
            return True
        else:
            return "yes"

def fix_mysqld_service(self):
        """
        Restore mysqld.service
        """
        src = self._rel("scripts/mysqld.service")
        dst = '/usr/local/directadmin/custombuild/configure/systemd/mysqld.service'
        try:
            # refuse group/world-writable parent + O_NOFOLLOW dest
            parent = os.path.dirname(dst)
            pst = os.stat(parent)
            if pst.st_mode & (0o020 | 0o002):
                print('ERROR occurred while attempting to restore mysqld.service!')
                return
            with open(src, 'rb') as fsrc:
                data = fsrc.read()
            # O_NOFOLLOW: refuse symlinked dest
            fd = os.open(dst, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o644)
            try:
                os.write(fd, data)
            finally:
                os.close(fd)
            print('mysqld.service restored!')
        except Exception:
            print('ERROR occurred while attempting to restore mysqld.service!')