ACIL FM

Nama	Ukuran	Aksi
cpapi	-	Open
lib	-	Open
public_hooks	-	Open
__pycache__	-	Open
clcagefs.py	10.99 MB	View DL Edit
clcaptain.py	1.96 MB	View DL Edit
clconfig.py	1.68 MB	View DL Edit
clconfpars.py	12.09 MB	View DL Edit
clcustomscript.py	1.16 MB	View DL Edit
cldebug.py	905 B	View DL Edit
clemail.py	1.65 MB	View DL Edit
clexception.py	1.14 MB	View DL Edit
clfunc.py	6.47 MB	View DL Edit
clhook.py	3.86 MB	View DL Edit
cllog.py	1.45 MB	View DL Edit
cloutput.py	471 B	View DL Edit
clproc.py	4.05 MB	View DL Edit
clpwd.py	7.74 MB	View DL Edit
clquota.py	1.27 MB	View DL Edit
clsec.py	657 B	View DL Edit
clwpos_lib.py	16.6 MB	View DL Edit
const.py	277 B	View DL Edit
evr_utils.py	3.55 MB	View DL Edit
features.py	5.01 MB	View DL Edit
group_info_reader.py	5.31 MB	View DL Edit
lock.py	1.02 MB	View DL Edit
mail_helper.py	4.45 MB	View DL Edit
mysql_lib.py	5.84 MB	View DL Edit
php_conf_reader.py	9.77 MB	View DL Edit
sysctl.py	7.61 MB	View DL Edit
ui_config.py	3.12 MB	View DL Edit
utils.py	33.03 MB	View DL Edit
utils_cmd.py	2.71 MB	View DL Edit
__init__.py	1.37 MB	View DL Edit

Edit file: /opt/cloudlinux/venv/lib/python3.11/site-packages/clcommon/mysql_lib.py

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
#

import os
from typing import Iterable, List
from collections import deque

import pymysql
from pymysql.cursors import DictCursor

from clcommon.clconfpars import load_fast

PLESK_CONFIG_PATH = '/etc/psa/psa.conf'
MY_CONFIG_PATH = '/etc/my.cnf'
MY_CONFIG_PATH2 = '/etc/mysql/my.cnf'

class MySQLError(Exception):
    """
    Exception related to operation with MySQL.
    """

class MySQLConnector:

def __init__(self, as_dict: bool = False, **kwargs):
        self.add_unix_socket_if_localhost(kwargs.get('host', 'localhost'), kwargs)

self._connection = None
        self._cursor_type = DictCursor if as_dict else None
        self._connect_kwargs = kwargs

def __del__(self):
        self.close()

def __enter__(self):
        return self.connect()

def __exit__(self, exc_type, exc_value, traceback):
        self.close()

def connect(self) -> 'MySQLConnector':
        if self._connection is not None:
            return self
        try:
            self._connection = pymysql.connect(**self._connect_kwargs)
            return self
        except pymysql.Error as e:
            raise MySQLError(str(e)) from e

def close(self) -> None:
        if self._connection is not None:
            self._connection.close()
            self._connection = None

def commit(self) -> None:
        if self._connection is not None:
            self._connection.commit()

def cursor(self):
        return self.connection.cursor(self._cursor_type)

@property
    def connection(self):
        if self._connection is None:
            self.connect()
        return self._connection

def execute_query(
        self,
        sql_query: str,
        args: Iterable[object] | None = None
    ) -> tuple[object, ...] | dict[str, object]:
        """
        Execute SQL query and return the result.
        """
        with self.connection.cursor(self._cursor_type) as cursor:
            cursor.execute(sql_query, args=args)
            return cursor.fetchall()

@staticmethod
    def add_unix_socket_if_localhost(host: str, kwargs: dict) -> None:
        """
        Add 'unix_socket' to kwargs if host is 'localhost'.

It seems that when the host is set to 'localhost',
        establishing a connection through TCP/IP might encounter issues
        due to some MySQL configuration options.
        To prioritize a Unix socket connection,
        we should include the 'unix_socket' query parameter.
        """
        if host == 'localhost' and 'unix_socket' not in kwargs:
            kwargs['unix_socket'] = get_unix_socket_path()

def get_rfc1738_db_uri(url):
    """
    Get a modified RFC 1738 URL for a MySQL database connection.
    """
    MySQLConnector.add_unix_socket_if_localhost(url.host, url.query)
    return url

def get_unix_socket_path(
    *,
    plesk_config_path: str = PLESK_CONFIG_PATH,
    mysql_configs: Iterable[str] = (MY_CONFIG_PATH, MY_CONFIG_PATH2)
) -> str:
    """
    Get the Unix socket path for MySQL connection.

Check Plesk and MySQL config files for the socket path.
    If found, return that path. If not, return default socket path.
    """
    if os.path.isfile(plesk_config_path):
        psa_conf = load_fast(plesk_config_path, delimiter=' ')
        psa_conf_socket = psa_conf.get('MYSQL_SOCKET')
        if psa_conf_socket is not None:
            return psa_conf_socket

mysql_configs_queue = deque(mysql_configs)
    while len(mysql_configs_queue) > 0:
        my_conf_path = mysql_configs_queue.popleft()
        if not os.path.isfile(my_conf_path):
            continue

my_conf = load_fast(my_conf_path, strip_quotes=True)
        # Use the first found socket path
        my_conf_socket = my_conf.get('socket')
        if my_conf_socket is not None:
            return my_conf_socket

# Add included config files to the queue to check them for the socket path too
        mysql_configs_queue.extend(get_cnf_paths(my_conf_path))

return '/var/lib/mysql/mysql.sock'

def get_cnf_paths(path: str) -> list[str]:
    """
    Get all included config files paths in the given config file
    marked with !include and !includedir.

Returns an empty list if there are no paths.
    """
    include_dirs: List[str] = []
    list_of_paths: List[str] = []

try:
        with open(path, "r", encoding="utf-8", errors="surrogateescape") as config_file:
            for line in config_file:
                line = line.strip()
                try:
                    if line.startswith('!includedir'):
                        included_dir_path = line.removeprefix('!includedir').strip()
                        if os.path.isdir(included_dir_path):
                            include_dirs.append(included_dir_path)
                    elif line.startswith('!include '):
                        included_file_path = line.removeprefix('!include').strip()
                        if os.path.isfile(included_file_path):
                            list_of_paths.append(included_file_path)
                except IndexError:
                    # skip broken lines
                    pass
            for dir_path in include_dirs:
                list_of_paths += get_cnf_files_in_dir(dir_path)
    except (OSError, IOError) as e:
        raise MySQLError(f"Error reading config file {path}: {e}") from e
    return list_of_paths

def get_cnf_files_in_dir(path: str) -> list[str]:
    """
    Get paths list in given dir
    Returns an empty list if there is no paths
    """
    list_of_paths = []
    if os.path.isdir(path):
        for item in os.listdir(path):
            if os.path.isfile(os.path.join(path, item)) and os.path.splitext(item)[1] == ".cnf":
                list_of_paths.append(os.path.join(path, item))
    return list_of_paths

Batal