OwlCyberSecurity - MANAGER

Edit File: 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
#
from __future__ import annotations

import os
from typing import Iterable

# TODO: Once the full migration from Python 3.7 to 3.11 is completed,
# MySQLdb import below can be safely removed
try:
    import pymysql
except ImportError:
    try:
        import MySQLdb as pymysql
    except ImportError:
        pymysql = None
    else:
        from MySQLdb.cursors import DictCursor
else:
    from pymysql.cursors import DictCursor

from clcommon.clconfpars import load_fast
from clcommon.cpapi.cpapiexceptions import NoPackage

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

class MySQLConnector:

def __init__(self, as_dict: bool = False, **kwargs):
        if pymysql is None:
            raise NoPackage(
                'Can not connect to database; MySQL-client libraries are not installed.'
            )

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 __enter__(self):
        return self.connect()

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

def connect(self) -> MySQLConnector:
        self._connection = pymysql.connect(**self._connect_kwargs)
        return self

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

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

for my_conf_path in mysql_configs:
        if not os.path.isfile(my_conf_path):
            continue

my_conf = load_fast(my_conf_path, strip_quotes=True)
        my_conf_socket = my_conf.get('socket')
        if my_conf_socket is not None:
            return my_conf_socket

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