1#@!#!123s

: / proc / self / root / opt / dedrads / cms_tools /
Filename : cms.py
back
#! /usr/lib/rads/venv/bin/python3
"""CMS class and associated classes.
Used to find CMS and present methods for their manipulation."""
# Author: Daniel K

import os
import re
import logging

from cms_tools.helpers import (
    backup_file,
    restore_file,
    restore_db,
    readfile,
    strip_php_comments,
    find_php_define,
    find_php_var,
    php_re_define,
    php_re_assign,
)


LOGGER = logging.getLogger(__name__)

# ==== Classes  ====


class CMSError(Exception):
    '''Error class for CMS'''

    def __init__(self, arg):
        self.args = (arg,)
        super().__init__(self, arg)


class VariableData:
    '''
    Class to define where a variable is set and manipulate it
    '''

    type = 'const'
    file = None
    value = ''

    def __init__(self, variable_type='const', value='', filename=None):
        '''
        Allow for initialization with or without values
        '''
        self.type = variable_type
        self.file = filename
        self.value = value

    def get_value(self):
        '''
        Return the value of the indicated by the data
        '''
        if "const" == self.type:
            return self.value

        file_data = readfile(self.file)
        if None is file_data:
            LOGGER.warning("Could not get value")
            return None

        if "php_define" == self.type:
            file_data = strip_php_comments(file_data)
            new_value = find_php_define(self.value, file_data)
            LOGGER.debug("Got php define %s = %s", self.value, new_value)
            return new_value
        if "php_variable" == self.type:
            file_data = strip_php_comments(file_data)
            return find_php_var(self.value, file_data)
        LOGGER.warning("Variable type '%s' not recognized", self.type)
        return None

    # End get_value

    def set_value(self, new_value):
        '''
        Set a new value and update appropriate file
        '''
        if "const" == self.type:
            self.value = new_value
            return True
        if "php_define" == self.type:
            return php_re_define(self.value, new_value, self.file)
        if "php_variable" == self.type:
            return php_re_assign(self.value, new_value, self.file)
        LOGGER.warning("Variable type '%s' not recognized", self.type)
        return False

    # End set_value


# End VariableData


# class CMSStatus(Enum):
class CMSStatus:
    '''Status of CMS. Negative values are errors.'''

    db_has_matching_tables = 5
    db_has_tables = 4
    db_is_connecting = 3
    db_is_set = 2
    initiated = 1
    uninitiated = 0
    warning = -1
    error = -2
    critical = -3


class CMS:
    '''
    Base class for CMS installations
    '''

    db_file_search_path = ()

    ilevel = 0

    status = CMSStatus.uninitiated
    previous_status = CMSStatus.uninitiated
    reason = 'uninitiated'

    # Location of installation
    directory_root = None

    # CMS software: WordPress, Joomla, etc.
    type = None

    # Config file locaton
    config = ''

    # cPanel username
    cpuser = None

    dbprefix = None

    # Database settings
    is_db_set = False
    is_db_connecting = False
    db_name_data = None
    db_user_data = None
    db_pass_data = None
    db_pref_data = None
    db_host_data = None

    db_name = None
    db_user = None
    db_pass = None
    db_pref = None
    db_host = None

    orig_db_name = None
    orig_db_user = None
    orig_db_pass = None
    orig_db_pref = None
    orig_db_host = None

    # URL information
    servername = None
    siteurl = 'Unknown'
    version = 'Unknown'
    tempurl = None

    modified_files = {}
    modified_dbs = {}

    cms_directories = []

    def setup(self):
        '''Virtual function to allow CMS to set the necessary variables'''

        self.db_name_data = VariableData('const', '')
        self.db_user_data = VariableData('const', '')
        self.db_pass_data = VariableData('const', '')
        self.db_pref_data = VariableData('const', '')

        return True

    def __init__(self, the_directory, ilevel=0):
        '''
        Constructor which should be run for all derived classes
        '''

        self.ilevel = ilevel

        self.modified_files = {}
        self.modified_dbs = {}

        self.servername = os.getenv('HOSTNAME')
        self.directory_root = the_directory

        find_cpuser = re.findall(r"^/home[0-9]*/([^/]+)", the_directory)
        if len(find_cpuser) == 0:
            LOGGER.warning(
                "Could no find cPanel user from path '%s'", the_directory
            )
            return
        self.cpuser = find_cpuser[-1]

        self.dbprefix = self.cpuser

        self.db_file_search_path = (
            '/home/XFER/mysqldump',
            '/home/XFER',
            '/home/%s' % self.cpuser,
            '/home',
            os.getcwd(),
            self.directory_root,
            '/home/%s/cpmove_failed*/' % self.cpuser,
        )

        if len(self.cpuser) > 8:
            self.dbprefix = self.cpuser[:8]

        if re.search(
            "database_prefix *= *0", readfile("/var/cpanel/cpanel.config")
        ):
            raise CMSError("Database prefixing is disabled. Aborting")

        self.db_host_data = VariableData('const', 'localhost')

        if not self.cpuser:
            # Maybe we could use some other heuristic later
            self.cpuser = None
            LOGGER.warning(
                "Could not init CMS class. Not able to find cPanel user"
            )
            return

        self.tempurl = "http://{}/~{}/{}".format(
            self.servername,
            self.cpuser,
            re.sub(
                r"/home[0-9]*/%s/public_html/?" % self.cpuser,
                "",
                self.directory_root,
            ),
        )

        if not self.setup():
            LOGGER.error(
                "Could not setup %s instance at %s",
                self.type,
                self.directory_root,
            )
            return

        self.set_status(
            CMSStatus.initiated,
            f"{self.type} instance created for {self.directory_root}",
        )

        self.set_db_creds()

    # End __init__

    def set_status(self, new_status, reason):
        '''Set new status for CMS'''

        if self.status < 0 < new_status:
            LOGGER.critical(
                "Attempting to increase status when error exists. "
                "Attempting to change from %d to %d. Last status: %s",
                self.status,
                new_status,
                self.reason,
            )
            raise CMSError("Unhandled error")

        if new_status < 0 <= self.status:
            self.previous_status = self.status

        self.status = new_status
        self.reason = reason

        if new_status == CMSStatus.error:
            LOGGER.error("%s: %s", self.directory_root, reason)
        elif new_status == CMSStatus.warning:
            LOGGER.warning("%s: %s", self.directory_root, reason)
        elif new_status == CMSStatus.critical:
            LOGGER.critical("%s: %s", self.directory_root, reason)

        LOGGER.debug(
            "Set status of %s to %d: %s",
            self.directory_root,
            self.status,
            self.reason,
        )

        if new_status is CMSStatus.db_has_matching_tables:
            self.post_tables_hook()

    def revert(self):
        '''
        Revert changes attempted on this run.
        '''

        LOGGER.info("Reverting %s", self.directory_root)

        if not len(self.modified_files) == 0:
            for orig_file, new_file in self.modified_files.items():
                if not restore_file(new_file, orig_file):
                    LOGGER.critical("Could not restore %s!", orig_file)
                    raise CMSError("Could not restore %s!" % orig_file)

        if not len(self.modified_dbs) == 0:
            for database, dump_path in self.modified_dbs.items():
                if not restore_db(
                    self.db_user,
                    self.db_pass,
                    self.db_name,
                    dump_path,
                ):
                    LOGGER.critical("Could not restore %s!", database)
                    raise CMSError("Could not restore %s!" % database)

        self.modified_files = {}
        self.modified_dbs = {}

        return True

    def set_variable(self, cms_variable, value):
        '''
        Safely set a cms variable, keeping track of modified files
        '''

        variable_data = None

        if cms_variable == "db_name":
            variable_data = self.db_name_data
        elif cms_variable == "db_user":
            variable_data = self.db_user_data
        elif cms_variable == "db_pass":
            variable_data = self.db_pass_data
        elif cms_variable == "db_pref":
            variable_data = self.db_pref_data
        elif cms_variable == "db_host":
            variable_data = self.db_host_data
        else:
            self.set_status(
                CMSStatus.warning, "Uknown variable type '%s'." % cms_variable
            )
            return False

        if variable_data is None:
            self.set_status(
                CMSStatus.warning, "Variable '%s' was not set." % cms_variable
            )
            return False

        config_file = variable_data.file

        if config_file not in self.modified_files:
            new_file = backup_file(config_file)
            if not new_file:
                self.set_status(
                    CMSStatus.error,
                    "Could not backup {}. Cannot modify {}".format(
                        config_file, cms_variable
                    ),
                )
                return False

            LOGGER.info("Modifying %s", config_file)
            self.modified_files[config_file] = new_file

        variable_data.set_value(value)
        return True

    def get_variable(self, cms_variable):
        '''
        Get a cms variable. Just a wrapper for consistency
        '''

        variable_data = None

        if cms_variable == "db_name":
            variable_data = self.db_name_data
        elif cms_variable == "db_user":
            variable_data = self.db_user_data
        elif cms_variable == "db_pass":
            variable_data = self.db_pass_data
        elif cms_variable == "db_pref":
            variable_data = self.db_pref_data
        elif cms_variable == "db_host":
            variable_data = self.db_host_data
        else:
            LOGGER.warning("Uknown variable type '%s'.", cms_variable)
            return False

        if variable_data is None:
            LOGGER.warning("Variable '%s' was not set.", cms_variable)
            return False

        return variable_data.get_value()

    def set_db_creds(self):
        '''
        Get database credentials
        '''

        if None in (
            self.db_name_data,
            self.db_user_data,
            self.db_pass_data,
            self.db_pref_data,
            self.db_host_data,
        ):
            LOGGER.error(
                "Database information for %s was not set! "
                "This should have already been done.",
                self.directory_root,
            )
            return False

        self.db_name = self.get_variable('db_name')
        self.db_user = self.get_variable('db_user')
        self.db_pass = self.get_variable('db_pass')
        self.db_pref = self.get_variable('db_pref')
        self.db_host = self.get_variable('db_host')

        self.orig_db_name = self.db_name
        self.orig_db_user = self.db_user
        self.orig_db_pass = self.db_pass
        self.orig_db_pref = self.db_pref
        self.orig_db_host = self.db_host

        if None in (
            self.db_name,
            self.db_user,
            self.db_pass,
            self.db_pref,
            self.db_host,
        ):
            self.set_status(CMSStatus.warning, "Database credentials not set")
            return False

        self.set_status(CMSStatus.db_is_set, "Database credentials set")
        return True

    # End set_db_creds

    def set_db_search_path(self, dumppath):
        '''
        Replace Database search path with specified directory if it exists.
        Return True if reset or False otherwise.
        '''

        if not dumppath:
            return False

        if not os.path.isdir(dumppath):
            LOGGER.warning(
                "Database search path not reset for %s. "
                "'%s' is not a directory",
                self.tempurl,
                dumppath,
            )
            return False

        self.db_file_search_path = (dumppath,)
        LOGGER.info(
            "Database search path for %s set to %s", self.tempurl, dumppath
        )
        return True

    def show_creds(self):
        '''
        Display the database credentials
        '''

        print("Databse Name: %s" % self.db_name)
        print("Username: %s" % self.db_user)
        print("Password: %s" % self.db_pass)
        print("Table Prefix: %s" % self.db_pref)
        print("Host: %s" % self.db_host)

    # End show_creds

    def purge_dirlist(self, dirlist):
        '''
        Remove cms directories from the dirlist
        '''

        if len(self.cms_directories) == 0:
            LOGGER.debug("No directories to purge for CMS type %s", self.type)
            return dirlist

        # Because python can't handle just removing it from the list
        # while iterating over it, and it won't just pass the value
        new_dirlist = []
        new_dirlist += dirlist

        for the_directory in dirlist:
            for expression in self.cms_directories:
                if None is not re.search(expression, the_directory):
                    new_dirlist.remove(the_directory)
                    continue

        # By default, we purge none
        return new_dirlist

    # CMS virtual functions. They should typically be overridden

    def post_tables_hook(self):
        '''Perform operatons which need a working database'''
        LOGGER.debug(
            "CMS type %s has nothing to do after tables are available",
            self.type,
        )
        return True

    def get_siteurl(self):
        '''
        Get the siteurl for the CMS.
        If the CMS doesn't have this ability, use tempurl
        '''

        self.siteurl = self.tempurl

        return self.siteurl


# End CMS class


class UnknownCMS(CMS):
    '''
    Class for unkown CMS installations
    '''

    def setup(self):

        self.type = 'Unknown'

        self.config = ''

        # Define db settings as empty
        self.db_name_data = VariableData('const', '')
        self.db_user_data = VariableData('const', '')
        self.db_pass_data = VariableData('const', '')
        self.db_pref_data = VariableData('const', '')

        return True


# End of UnknownCMS Class


def load_modules(cms_find: 'CMSFind', include_list=None):
    '''
    Load additional CMS modules,
    accepting a list to include only certain modules
    '''
    mod_names = [x for x in dir(cms_tools.mods) if not x.startswith('_')]
    available_mods = []

    for mod_name in mod_names:
        if include_list is not None and mod_name not in include_list:
            LOGGER.debug("Skipping module %s", mod_name)
            continue

        getattr(cms_tools.mods, mod_name).register_cms(cms_find)
        LOGGER.debug("Registered module: %s", mod_name)
        available_mods.append(mod_name)

    return available_mods


class CMSFind:
    '''
    Class to search for CMS in a directory
    '''

    interactivity = False
    quick_search = {}

    def __init__(self, start_path=None, search_depth=0, interactivity=0):
        '''
        Initiate CMSFind object,
        setting search parameters and interactivity
        '''
        self.interactivity = interactivity
        self.start_path = start_path
        self.search_depth = search_depth

    def add_quick(self, search_token, class_reference):
        '''
        Add informaiton for a quick search.
        search_token is searched in index.php
        class_reference is used to create the new CMS
        '''

        if search_token in self.quick_search:
            return False

        self.quick_search[search_token] = class_reference
        return None

    def quick(self, the_directory):
        '''
        Perform a quick search for the CMS.
        Simply search index.php for specified string.
        Return CMS object if one is found, or None.
        '''

        index_filename = os.path.join(the_directory, "index.php")

        # If no index.php, asume it isn't a CMS
        if not os.path.isfile(index_filename):
            return None

        index_file = readfile(index_filename)

        for search_token, class_ref in self.quick_search.items():
            if None is not re.search(search_token, index_file):
                new_cms = class_ref(the_directory, self.interactivity)
                if new_cms.status > CMSStatus.uninitiated:
                    return new_cms
                if new_cms.status < 0:
                    return new_cms

        return UnknownCMS(the_directory)

    def find_in_path(self, current_path, depth):
        '''
        Generator to return CMS installations.
        Starting at path, find CMS instances down to depth subdirectories.
        '''

        if 0 == depth:
            return

        if not os.path.isdir(current_path):
            return

        # Strip out directories that should never be searched
        # These are technically, regex, so I use them as such
        for exclude_rx in [
            r"virtfs",
            r"\.[^/]*",
            r"cgi-bin",
            r"www",
            r"mail",
            r"etc",
            r"access-logs",
            r"tmp",
            r"bin",
            r"perl[0-9]*",
            r"php",
            r"ssl",
            r"cache",
            r"cpanel[^/]*",
            r"\bbac?k\b",
            r"backup",
            r"\bold\b",
            r"\bnew\b",
            r"\bte?mp\b",
            r"xmlrpc",
        ]:
            if re.search("/%s$" % exclude_rx, current_path):
                return

        LOGGER.debug(
            "checking path: %s, Depth from bottom: %d", current_path, depth
        )

        subdirs = os.listdir(current_path)

        new_cms = self.quick(current_path)
        if None is not new_cms:
            subdirs = new_cms.purge_dirlist(subdirs)
            yield new_cms

        for the_directory in subdirs:
            next_path = os.path.join(current_path, the_directory)
            if os.path.isdir(current_path):
                yield from self.find_in_path(next_path, depth - 1)

    def find_cms(self):
        '''
        Main find function, starting from start path and depth,
        returning only known CMS
        '''
        assert self.start_path is not None, "Start path not set"

        if self.search_depth < 1:
            LOGGER.error("Attempting to search with a depth less than 1")
            return

        for the_cms in self.find_in_path(self.start_path, self.search_depth):
            if "Unknown" != the_cms.type:
                yield the_cms


# End CMSFind

# import late to avoid circular import
import cms_tools.mods  # pylint: disable=wrong-import-position