1#@!#!123s

: / proc / 2 / cwd / opt / dedrads / cms_tools /
Filename : helpers.py
back
#! /usr/lib/rads/venv/bin/python3
"""Helper functions for CMS manipulation"""

# Author: Daniel K

import sys
import os
import re
import logging
import subprocess
import warnings
from datetime import datetime, timezone
import glob
import shutil
import pymysql
from cpapis import cpapi2, whmapi1


LOGGER = logging.getLogger(__name__)
# Temporary functions. Should be removed when added to py rads

# from rads import common


def common_get_string(
    prompt, string_filter=r'[a-zA-Z0-9._/-]+$', hint='regex', default=None
):
    """
    Prompt to request a string, and require it to match a regex.
    If string fails to match, give a hint, which by default is just
    the regex. If no matching string is obtained, return None.
    If empty string is entered, return default if any exists.

    Defined filters: alpha, digits, email, cpuser, database, url
    """

    # Predefined filters
    if string_filter is None:
        string_filter = '.*'
        hint = 'Sorry, that should have matched.'
    elif 'alpha' in string_filter:
        string_filter = '[a-zA-Z0-9]+$'
        if hint == 'regex':
            hint = 'Must be only alphanumeric characters.'
    elif 'digits' in string_filter:
        string_filter = '[0-9.]+'
        if hint == 'regex':
            hint = 'Must be only digits.'
    elif 'email' in string_filter:
        string_filter = (
            r'[a-z0-9._-]+@[a-z0-9._-]+'
            + r'\.([a-z]{2,15}|xn--[a-z0-9]{2,30})$'
        )
        if hint == 'regex':
            hint = 'Must be a valid email address.'
    elif 'cpuser' in string_filter:
        string_filter = '[a-z0-9]{1,14}$'
        if hint == 'regex':
            hint = (
                'Must be a valid cPanel user: '
                + 'letters and numbers, under 14 characters.'
            )
    elif 'database' in string_filter:
        # This one is not precise, but provided for convenience.
        string_filter = '[a-z0-9]{1,8}_[a-z0-9]{1,12}$'
        if hint == 'regex':
            hint = (
                'Must be a valid database user: '
                + 'letters and numbers, single underscore.'
            )
    elif 'url' in string_filter:
        string_filter = (
            r'([a-z]{3,}://)?'
            r'([a-z0-9_-]+.){1,}([a-z]{2,15}|xn--[a-z0-9]'
            r'{2,30})(:[0-9]+)?'
            r'((/[a-zA-Z0-9/.%_-]*)(\?[a-zA-Z0-9/.%=;_-]+)?)?$'
        )
        if hint == 'regex':
            hint = 'Must be a valid URL.'

    while True:

        print("%s\n" % prompt)

        try:
            choice = input()
        except KeyboardInterrupt:
            print("\nCancelled")
            return None

        if default is not None and choice == '':
            return default
        if re.match(string_filter, choice) is not None:
            return choice
        print('\nInvalid answer. ', end=' ')

        if hint == 'regex':
            print('\nString must match the patter: /%s/' % string_filter)
        elif hint is None:
            print(' ', end=' ')
        else:
            print(hint)
        print('Try again.\n')


# ==== Globals  ====

# Dictionary of dbs we have converted. This is to try to prevent
# the use of duplicate db and db user names.
DB_CONVERSIONS = {}
DB_USER_CONVERSIONS = {}

# ==== Functions  ====


def get_cp_home(cpuser):
    '''Return home directory of cPanel user'''

    if len(glob.glob("/home*/")) > 1:

        result = whmapi1('accountsummary', {'user': cpuser})

        if 0 == result['metadata']['result']:
            LOGGER.error(
                "WHM API could not find home directory for %s: %s",
                cpuser,
                result['metadata']['reason'],
            )
            return None
        partition = result['data']['acct'][0]['partition']
    else:
        if len(glob.glob("/var/cpanel/users/%s" % cpuser)) == 0:
            LOGGER.error(
                "%s does not appear to be a valid cPanel user.", cpuser
            )
            return None
        partition = "home"

    docroot = f"/{partition}/{cpuser}"

    return docroot


def find_start_path(user_path):
    '''
    Find start path from for user or path given.
    '''

    if user_path is None:
        LOGGER.error("No user or path specified")
        sys.exit(1)

    if '/' in user_path:
        requested_path = user_path
        match = re.search(r"/home[^/]*/([^/]+)", user_path)
        if match is None:
            LOGGER.error(
                "Could not find a username in path '%s'", requested_path
            )
            sys.exit(1)

        username = match.group(1)
    else:
        username = user_path
        requested_path = ''

    docroot = get_cp_home(username)
    if docroot is None:
        return None

    if requested_path == '':
        return docroot

    if re.match(docroot, requested_path) is None:
        LOGGER.error(
            "Path given (%s) is not part of %s's document root (%s)",
            requested_path,
            username,
            docroot,
        )
        return None

    if os.path.isdir(requested_path):
        return requested_path

    print("Path given does not exist: '%s'" % requested_path)
    return None


def backup_file(filename):
    '''
    Find an unused filename and make a backup of a file
    '''

    if not os.path.isfile(filename):
        LOGGER.info("File %s does not exist to backup", filename)
        return None

    date_today = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    new_file = f"{filename}.cms_tools.file.bak.{date_today}"

    if not os.path.exists(new_file):
        LOGGER.info("Copying %s -> %s", filename, new_file)
        shutil.copy2(filename, new_file)
        return new_file

    for num in range(1, 3):

        new_file = "{}.cms_tools.file.bak.{}.{}".format(
            filename, date_today, num
        )

        if not os.path.exists(new_file):
            LOGGER.info("Copying %s -> %s", filename, new_file)
            try:
                shutil.copyfile(filename, new_file)
            except OSError as error:
                LOGGER.error(
                    "File copy failed. Could not copy %s to %s: %s",
                    filename,
                    new_file,
                    error,
                )
                return False

            return new_file

    LOGGER.warning("There are already too many backup files for %s", filename)

    return False


def restore_file(source_file, destination_file):
    '''
    Replace destination file with source file, removing source file
    '''

    if not os.path.isfile(source_file):
        LOGGER.warning("File %s does not exist. Cannot restore.", source_file)
        return False

    try:
        shutil.move(source_file, destination_file)
        LOGGER.info("Restored %s from %s", destination_file, source_file)
    except OSError as error:
        LOGGER.error(
            "File restore failed. Could not restore %s to %s: %s",
            source_file,
            destination_file,
            error,
        )

    return True


def readfile(filename):
    '''
    Read data from a file or report error and return None
    '''

    LOGGER.debug("Reading from %s", filename)
    if os.path.exists(filename):
        with open(filename, encoding='utf-8') as file_handle:
            try:
                file_data = file_handle.read()
                return file_data
            except OSError:
                LOGGER.error("Error reading file")
                return None
    return None


# End readfile()


def lastmatch(regex, data):
    '''
    Return the last regex match in a set of data or None
    '''

    if None is data:
        return None
    if None is regex:
        return None

    # Create the regex as a multiline
    regex_object = re.compile(regex, re.M)

    result = regex_object.findall(data)

    if 0 == len(result):
        return None

    return result[-1]


# End lastmatch()


def strip_php_comments(data):
    '''
    Return data minus any PHP style comments
    '''

    # Remove C++ style comments
    data = re.sub(r"\s+//.*", "", data)

    # Remove C style comments
    data = re.sub(r"/\*(.*\n)*?.*?\*/", "", data)

    return data


# End strip_php_comments


def find_php_define(const_name, data):
    '''
    Find the last instance of const_name being defined in php data
    '''
    return lastmatch(
        r'define\( *["\']%s["\']\s*,\s*["\']([^"\']+)["\']' % const_name, data
    )


# End find_php_define


def find_php_var(var_name, data):
    '''
    Find the last instance of var_name being assigned in php data
    '''
    return lastmatch(r'\$%s\s*=\s*["\']([^"\']+)["\']' % var_name, data)


# End find_php_var


def php_re_define(const_name, value, filename):
    '''
    Change all instances in filename where const_name is defined,
    and redefine it to value
    '''

    with open(filename, encoding='utf-8') as sources:
        lines = sources.readlines()
    with open(filename, "w", encoding='utf-8') as sources:
        for line in lines:
            sources.write(
                re.sub(
                    r'define\( *["\']%s["\'] *,'
                    r' *["\']([^"\']+)["\'] *\)' % const_name,
                    f"define('{const_name}','{value}')",
                    line,
                )
            )

    return True


# End php_re_define


def php_re_assign(var_name, value, filename):
    '''
    Change all instances in filename where var_name is assigned,
    and reassign it to value
    '''

    with open(filename, encoding='utf-8') as sources:
        lines = sources.readlines()
    with open(filename, "w", encoding='utf-8') as sources:
        for line in lines:
            sources.write(
                re.sub(
                    r'\$%s *= *["\']([^"\']+)["\']' % var_name,
                    f"${var_name} = '{value}'",
                    line,
                )
            )


# End php_re_define


def make_valid_db_name(cpuser, prefix, current_name, name_type="database"):
    '''
    Find a valid replacement database name. Typce can be database or user
    '''

    used_names = ()

    LOGGER.debug("Finding new name for: %s", current_name)

    # If we've already made this one, just return it
    if name_type == "database":
        if current_name in DB_CONVERSIONS:
            return DB_CONVERSIONS[current_name]
        used_names = list(DB_CONVERSIONS.values())

        result = cpapi2('MysqlFE::listdbs', user=cpuser)

        if 'result' in result['cpanelresult']['data']:
            LOGGER.error(
                "cPanel API could not list databases: %s",
                result['cpanelresult']['data']['reason'],
            )
            sys.exit(1)

        for i in result['cpanelresult']['data']:
            used_names += i['db']

    else:
        # if name_type == "user"
        if current_name in DB_USER_CONVERSIONS:
            return DB_USER_CONVERSIONS[current_name]
        used_names = list(DB_USER_CONVERSIONS.values())

        result = cpapi2('MysqlFE::listdbs', user=cpuser)
        if 'result' in result['cpanelresult']['data']:
            LOGGER.error(
                "cPanel API could not list users: %s",
                result['cpanelresult']['data']['reason'],
            )
            sys.exit(1)

        for i in result['cpanelresult']['data']:
            used_names += i['user']

    # Add cp name only so that empty additions will fail
    used_names = used_names + ["%s_" % prefix]

    is_set = False

    # For reference, this is how easily it is done with bash:
    # "${cp}_$(echo "$1"|grep -Po '^([^_]*_)?\K[a-z0-9]{1,7}')"
    # Sadly, Python can't handle variable-length lookbehinds

    # Remove unacceptable characters
    name_base = re.sub('[^a-z0-9_]', '', current_name.lower())

    # Get the group after the last _
    last_section = re.search('[^_]*$', name_base).group()

    if len(last_section) > 4:
        name_base = last_section
    else:
        first_section = re.search('^[^_]*', name_base).group()
        if len(first_section) > 4:
            name_base = first_section
        else:
            name_base = re.sub('[^a-z0-9]', '', name_base)
            name_base = name_base[:8]

    # Simply try the base name itself
    new_name = "{}_{}".format(
        prefix,
        re.search('^[a-z0-9]{1,%d}' % (15 - len(prefix)), name_base).group(0),
    )

    if new_name not in used_names:
        is_set = True

    if not is_set:
        if 14 > (len(prefix) + len(name_base)):
            for i in range(1, (10 ** (15 - len(prefix) - len(name_base)))):
                print("name base: %s" % name_base)
                new_name = "{}_{}{}".format(
                    prefix,
                    re.search(
                        '^[a-z0-9]{1,%d}' % (15 - len(prefix)), name_base
                    ).group(0),
                    i,
                )

                if new_name not in used_names:
                    is_set = True
                    break

    # If it isn't set yet, try replacing characters on the end with numbers
    if not is_set:
        for i in range(len(name_base[: (15 - len(prefix) - 1)]), 1, -1):
            tmp_base = name_base[:i]
            for i in range(1, (10 ** (15 - len(prefix) - len(tmp_base)) - 1)):
                new_name = "{}_{}{}".format(
                    prefix,
                    re.search(
                        '^[a-z0-9]{1,%d}' % (15 - len(prefix)), tmp_base
                    ).group(0),
                    i,
                )

                if new_name not in used_names:
                    is_set = True
                    break
            if is_set:
                break

    if is_set:
        # Once we have the new name, assign it to the dictionary
        LOGGER.debug("Found new name: %s -> %s", current_name, new_name)
        if name_type == "database":
            DB_CONVERSIONS[current_name] = new_name
        else:
            DB_USER_CONVERSIONS[current_name] = new_name

        return new_name

    LOGGER.critical("I give up! You find a new name for %s!", current_name)
    return None


def get_mysql_err(err: pymysql.Error):
    try:
        errno, errmsg = err.args
    except ValueError:
        errno, errmsg = None, 'UNKNOWN'
    return errno, errmsg


def restore_db(dbuser, password, dbname, dump_file):
    '''
    Restore database from dump file, removing dump file
    '''

    if not os.path.isfile(dump_file):
        LOGGER.warning(
            "File %s does not exist. Cannot restore %s.", dump_file, dbname
        )
        return False

    print("Attempting to remove tables")

    try:
        with pymysql.connect(
            host='localhost', user=dbuser, password=password, database=dbname
        ) as conn:
            with conn.cursor() as cursor:
                warnings.filterwarnings("ignore", "Unknown table.*")
                LOGGER.debug(
                    "Reading from %s to import into %s", dump_file, dbname
                )
                cursor.execute("SHOW TABLES")
                for (table_name,) in cursor.fetchall():
                    cursor.execute(
                        "DROP TABLE `%s`" % table_name.replace('`', '``')
                    )

    except pymysql.Error as err:
        errno, sterror = get_mysql_err(err)
        LOGGER.warning(
            "Error removing tables from %s. %s: %s", dbname, errno, sterror
        )
        return False

    if not import_db(dbuser, password, dbname, dump_file):
        LOGGER.critical("Could not restore %s", dbname)
        return False

    os.remove(dump_file)
    return True


def import_db(dbuser: str, password: str, dbname: str, dump_file):
    '''
    Use database credentials to import data from dump_file
    '''
    if not os.path.exists(dump_file):
        logging.warning("File '%s' does not exist", dump_file)
        return False
    with open(dump_file, 'rb') as file:
        try:
            subprocess.run(
                ['/usr/bin/mysql', '--user', dbuser, dbname],
                env={'MYSQL_PWD': password},
                stdin=file,
                encoding='utf-8',
                stdout=subprocess.DEVNULL,
                stderr=subprocess.PIPE,
                check=True,
            )
        except subprocess.CalledProcessError as exc:
            LOGGER.error(str(exc.stderr, 'utf-8', 'replace').strip())
            return False
    LOGGER.debug("Database import completed for %s from %s", dbname, dump_file)
    return True


def find_dump_file(dbname, dump_path, add_to_name=''):
    '''
    Find a non-existent file into which to dump the a db
    '''

    if add_to_name != '':
        add_to_name = '.' + add_to_name

    if not os.path.isdir(dump_path):
        LOGGER.warning("Path %s does not exist to dump %s", dump_path, dbname)
        return False

    date_today = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    # That's just pretenious
    # for num in [''] + ['.' + str(i) for i in range(1,4)]
    for num in ['', '.1', '.2', '.3']:
        new_file = os.path.join(
            dump_path,
            f"{dbname}{add_to_name}.{date_today}{num}.sql",
        )

        if not os.path.exists(new_file):
            return new_file

    LOGGER.warning("There are already too many backup files for %s", dbname)
    return False


def dump_db(dbuser, password, dbname, dump_path, add_to_name=''):
    '''
    Export MySQL database as a backup.
    '''

    dump_file = find_dump_file(dbname, dump_path, add_to_name)
    if not dump_file:
        return False

    with open(dump_file, 'w+', encoding='utf-8') as fhandle:
        LOGGER.info("Attempting dump to %s ", dump_file)
        cmd = [
            "mysqldump",
            "--add-drop-table",
            "--routines",
            "-u",
            dbuser,
            dbname,
        ]
        try:
            stderr = subprocess.run(
                cmd,
                env={'MYSQL_PWD': password},
                stdout=fhandle,
                stderr=subprocess.PIPE,
                encoding='utf-8',
                errors='replace',
                check=True,
            ).stderr.strip()
            if not stderr:
                LOGGER.info("Dumped %s to %s", dbname, dump_file)
                return dump_file
            LOGGER.error(
                "Unable to dump %s to %s: %s", dbname, dump_file, stderr
            )
            return False

        except subprocess.CalledProcessError as error:
            LOGGER.error("Error attempting dump of %s: %s", dbname, error)
            return False


# -- CP funcitons --


def db_exists(cpuser, dbname):
    '''
    Check for existance of dbname
    '''

    result = cpapi2('MysqlFE::listdbs', user=cpuser)

    if 'result' in result['cpanelresult']['data']:
        LOGGER.error(
            "cPanel API could not list databases: %s",
            result['cpanelresult']['data']['reason'],
        )
        sys.exit(1)

    for i in result['cpanelresult']['data']:
        if dbname == i['db']:
            return True

    return False


def db_user_exists(cpuser, dbname):
    '''
    Check for existance of dbname
    '''

    result = cpapi2('MysqlFE::listdbs', user=cpuser)
    if 'result' in result['cpanelresult']['data']:
        LOGGER.error(
            "cPanel API could not list database users: %s",
            result['cpanelresult']['data']['reason'],
        )
        sys.exit(1)

    for i in result['cpanelresult']['data']:
        if dbname == i['user']:
            return True

    return False


def create_db(cpuser, dbname):
    '''
    Create database dbname for cpuser
    '''

    result = cpapi2('MysqlFE::createdb', user=cpuser, args={"db": dbname})

    if 'result' in result['cpanelresult']['data']:
        LOGGER.error(
            "cPanel API could not create database: %s",
            result['cpanelresult']['data']['reason'],
        )
        sys.exit(1)


def create_db_user(cpuser, dbuser, password):
    '''
    Create database dbname for cpuser
    '''

    result = cpapi2(
        'MysqlFE::createdbuser',
        user=cpuser,
        args={"dbuser": dbuser, "password": password},
    )

    if 'result' in result['cpanelresult']['data']:
        LOGGER.error(
            "cPanel API could not create database user: %s",
            result['cpanelresult']['data']['reason'],
        )
        sys.exit(1)


def change_db_pass(cpuser, dbuser, password):
    '''
    Create database dbname for cpuser
    '''

    result = cpapi2(
        'MysqlFE::changedbuserpassword',
        user=cpuser,
        args={'dbuser': dbuser, 'password': password},
    )

    if 'result' in result['cpanelresult']['data']:
        LOGGER.error(
            "cPanel API could not change database user password: %s",
            result['cpanelresult']['data']['reason'],
        )
        return False

    return True


def associate_db_user(cpuser, dbname, dbuser):
    '''
    Create database dbname for cpuser
    '''

    result = cpapi2(
        'MysqlFE::setdbuserprivileges',
        user=cpuser,
        args={'privileges': 'ALL PRIVILEGES', 'db': dbname, 'dbuser': dbuser},
    )

    if 'result' in result['cpanelresult']['data']:
        LOGGER.error(
            "cPanel API could not change database user password: %s",
            result['cpanelresult']['data']['reason'],
        )
        sys.exit(1)