Da

✏️ Editing: migrator_utils.py

import os
import pipes
import re
from collections import namedtuple
from contextlib import contextmanager
from functools import wraps

from parallels.core import MigrationError, MigrationNoContextError
from parallels.core import messages
from parallels.core.actions.migration_list.utils import get_migration_list_file_location
from parallels.core.logging import get_logger
from parallels.core.logging import log_context
from parallels.core.registry import Registry
from parallels.core.runners.base import BaseRunner
from parallels.core.runners.entities import ExecutionOptions
from parallels.core.utils import unix_utils, get_thirdparties_base_path
from parallels.core.utils.cagefs import is_cagefs_enabled_for_user, disable_cagefs, enable_cagefs
from parallels.core.utils.common import cached, is_ascii_string, safe_format, open_no_inherit, default, mkdir_p
from parallels.core.utils.common.ip import escape_ip
from parallels.core.utils.common_constants import SSH_DEFAULT_PORT
from parallels.core.utils.download_utils import download_zip
from parallels.core.utils.file_utils import files_have_equal_contents
from parallels.core.utils.locks.file_lock_exception import FileLockException
from parallels.core.utils.steps_profiler import get_default_steps_profiler
from parallels.core.utils.unix_utils import format_command

from parallels.core.utils.common.string_utils import safe_utf8_decode, safe_idn_encode

logger = get_logger(__name__)

def copy_directory_content_unix(
    source_ip, source_user, source_server, target_server, source_path, target_path,
    key_pathname, exclude=None, skip_if_source_dir_not_exists=False,
    rsync_additional_args=None,
    user_mapping=None,
    group_mapping=None,
    source_rsync_bin=None, source_port=SSH_DEFAULT_PORT,
    target_user=None,
    preserve_owner=False,
    preserve_group=False,
    preserve_perms=False
):
    """
    :type source_ip: str | unicode
    :type source_user: str | unicode
    :type source_server: parallels.core.connections.server.Server
    :type target_server: parallels.core.connections.server.Server
    :type source_path: str | unicode
    :type target_path: str| unicode
    :type key_pathname: str | unicode
    :type exclude: list[str | unicode] | None
    :type skip_if_source_dir_not_exists: bool
    :type rsync_additional_args: list[str | unicode] | None
    :type user_mapping: dict[str | unicode, str | unicode] | None
    :type group_mapping: dict[str | unicode, str | unicode] | None
    :type source_rsync_bin: str | unicode
    :type source_port: int | str | unicode
    :type target_user: str | unicode | None
    :type preserve_owner: bool
    :type preserve_group: bool
    :type preserve_perms: bool
    :rtype: None
    """
    exclude = default(exclude, [])
    user_mapping = default(user_mapping, {})
    group_mapping = default(group_mapping, {})

with source_server.runner() as runner_source:
        if runner_source.file_exists(source_path):
            _copy_directory_content_unix_plain(
                source_ip, source_user, source_server, target_server, source_path, target_path,
                key_pathname, exclude=exclude,
                rsync_additional_args=rsync_additional_args,
                user_mapping=user_mapping,
                group_mapping=group_mapping,
                source_rsync_bin=source_rsync_bin,
                source_port=source_port,
                target_user=target_user,
                preserve_owner=preserve_owner,
                preserve_group=preserve_group,
                preserve_perms=preserve_perms
            )
        else:
            if skip_if_source_dir_not_exists:
                # source directory does not exists, so there is nothing to copy
                return
            else:
                raise SourceDirectoryDoesNotExistError(
                    safe_format(messages.SOURCE_DIRECTORY_DOES_NOT_EXIST, path=source_path, server_ip=source_ip)
                )

class SourceDirectoryDoesNotExistError(MigrationError):
    pass

DEFAULT_RSYNC_PATH = '/usr/bin/rsync'

def detect_rsync_path(runner, node_info_string):
    if runner.file_exists(DEFAULT_RSYNC_PATH):
        return DEFAULT_RSYNC_PATH
    else:
        raise MigrationError(
            messages.UNABLE_TO_FIND_RSYNC_BINARY % (
                node_info_string, DEFAULT_RSYNC_PATH
            )
        )

def create_rsync_command(
    key_pathname, source_runner, source_user, source_ip,
    source_filepath, target_filepath, exclude=None,
    rsync_additional_args=None,
    source_rsync_bin=None,
    source_port=SSH_DEFAULT_PORT, target_user=None,
    preserve_owner=False, preserve_group=False, preserve_perms=False
):
    """
    :type key_pathname: str | unicode
    :type source_runner: parallels.core.runners.base.BaseRunner
    :type source_user: str | unicode
    :type source_ip: str | unicode
    :type source_filepath: str | unicode
    :type target_filepath: str | unicode
    :type exclude: list[str | unicode] | None
    :type rsync_additional_args: list[str | unicode] | None
    :type source_rsync_bin: str | unicode | None
    :type source_port: int | str | unicode
    :type target_user: str | unicode | None
    :type preserve_owner: bool
    :type preserve_group: bool
    :type preserve_perms: bool
    :rtype: tuple[str, dict[str, str], dict[str, str]]
    """
    exclude = default(exclude, [])
    rsync_additional_args = default(rsync_additional_args, [])

umask_command = u'umask 022'
    # Compose base command
    rsync_command = (
        u'/usr/bin/rsync -e {ssh_command} --timeout=30 --recursive --links --times --devices --specials'
    )
    rsync_command = umask_command + u'; ' + rsync_command
    rsync_args = dict(
        ssh_command=format_command(
            u'ssh -i {key} -p {port} '
            u'-o PasswordAuthentication=no -o StrictHostKeyChecking=no -o GSSAPIAuthentication=no',
            key=key_pathname,
            port=source_port
        )
    )

# Add source rsync path argument
    if source_rsync_bin is None:
        # detect rsync binary location on the source node
        source_rsync_bin = detect_rsync_path(source_runner, source_ip)
        if source_rsync_bin != DEFAULT_RSYNC_PATH:  # non-default path
            rsync_command += u' --rsync-path={source_rsync_path}'
            rsync_args.update(dict(
                source_rsync_path=source_rsync_bin
            ))

# Add preserve owner, group, permissions options
    if preserve_owner:
        rsync_command += u' --owner'
    if preserve_group:
        rsync_command += u' --group'
    if preserve_perms:
        rsync_command += u' --perms'
    else:
        rsync_command += u' --chmod=Fu=rw,Fgo=r,Du=rwx,Dgo=rx'

# Add additional arguments
    for num, rsync_additional_arg in enumerate(rsync_additional_args):
        key = u'rsync_additional_arg_%s' % num
        rsync_command += u' {%s}' % key
        rsync_args[key] = rsync_additional_arg

# Add excludes
    for num, exclude_value in enumerate(exclude):
        key = u'exclude_%s' % num
        rsync_command += u' --exclude={%s}' % key
        rsync_args[key] = exclude_value

# Add source and destination paths
    rsync_command += u' {source_path} {target_path}'
    rsync_args.update(dict(
        source_path=u'{source_user}@{source_ip}:{source_filepath}'.format(
            source_user=source_user,
            source_ip=escape_ip(source_ip),
            source_filepath=pipes.quote(source_filepath)
        ),
        target_path=target_filepath
    ))

rsync_env = {
        # keep explicit arguments protection for compatibility with rsync < 3.2.4
        'RSYNC_OLD_ARGS': '1',
    }

if target_user is None:
        return rsync_command, rsync_args, rsync_env
    else:
        su_command = u'su {target_user} --shell /bin/sh -c {rsync_command}'
        su_args = dict(
            target_user=target_user,
            rsync_command=format_command(rsync_command, **rsync_args)
        )
        return su_command, su_args, rsync_env

def _copy_directory_content_unix_plain(
    source_ip, source_user, source_server, target_server,
    source_path, target_path, key_pathname,
    exclude=None, rsync_additional_args=None,
    user_mapping=None, group_mapping=None,
    source_rsync_bin=None, source_port=SSH_DEFAULT_PORT,
    target_user=None,
    preserve_owner=False, preserve_group=False, preserve_perms=False
):
    """
    :type source_ip: str | unicode
    :type source_user: str | unicode
    :type source_server: parallels.core.connections.server.Server
    :type target_server: parallels.core.connections.server.Server
    :type source_path: str | unicode
    :type target_path: str | unicode
    :type key_pathname: str | unicode
    :type exclude: list[str | unicode] | None
    :type rsync_additional_args: list[str | unicode] | None
    :type user_mapping: dict[str | unicode, str | unicode] | None
    :type group_mapping: dict[str | unicode, str | unicode] | None
    :type source_rsync_bin: str | unicode
    :type source_port: int | str | unicode
    :type target_user: str | unicode | None
    :type preserve_owner: bool
    :type preserve_group: bool
    :type preserve_perms: bool
    """
    exclude = default(exclude, [])
    user_mapping = default(user_mapping, {})
    group_mapping = default(group_mapping, {})

if not source_path.endswith('/'):
        source_path += '/'
    if not target_path.endswith('/'):
        target_path += '/'

# CageFS is a CloudLinux feature which allows to restrict access to files of other users.
    # When it is enabled, it is not possible to run "rsync" utility under user on the target server.
    # So, we temporary disable it for the user.
    disabled_cagefs = False

if target_user is not None:
        is_cagefs_enabled = is_cagefs_enabled_for_user(target_server, target_user)
        if is_cagefs_enabled:
            try:
                disabled_cagefs = True
                disable_cagefs(target_server, target_user)
            except Exception as e:
                logger.exception()
                logger.fwarn(
                    messages.FAILED_TO_DISABLE_CAGEFS_FOR_TARGET_USER,
                    username=target_user, server=target_server.description(),
                    reason=e
                )

try:
        with source_server.runner() as runner_source, target_server.runner() as runner_target:
            assert isinstance(runner_target, BaseRunner)
            cmd, args, env = create_rsync_command(
                key_pathname, runner_source, source_user,
                source_ip, source_path, target_path, exclude,
                rsync_additional_args,
                source_rsync_bin, source_port=source_port,
                target_user=target_user,
                preserve_owner=preserve_owner,
                preserve_group=preserve_group,
                preserve_perms=preserve_perms
            )
            # Define umask to overwrite the default mask, used by migrator.
            # For security reasons migrator can use special umask.
            # Calling rsync with the default umask can create files with unexpected permissions.
            runner_target.execute_command(cmd, args, ExecutionOptions(umask=000, environment_variables=env))

for mapping_source_user, mapping_target_user in user_mapping.items():
                logger.debug(messages.CHANGE_FILE_PERMISSIONS_ACCORDING_USER_MAP)
                unix_utils.map_copy_owner_user(
                    runner_source, runner_target, source_path, target_path, mapping_source_user, mapping_target_user
                )
            for mapping_source_group, mapping_target_group in group_mapping.items():
                logger.debug(messages.CHANGE_FILE_PERMISSIONS_ACCORDING_GROUP_MAP)
                unix_utils.map_copy_owner_group(
                    runner_source, runner_target, source_path, target_path, mapping_source_group, mapping_target_group
                )
    finally:
        if disabled_cagefs:
            try:
                enable_cagefs(target_server, target_user)
            except Exception as e:
                logger.exception()
                logger.fwarn(
                    messages.FAILED_TO_ENABLE_CAGEFS_FOR_TARGET_USER,
                    username=target_user, server=target_server.description(),
                    reason=e
                )

def trace(op_name, message):
    def decorator(method):
        @wraps(method)
        def wrapper(self, *args, **kw):
            with trace_step(op_name, message, profile=True):
                return method(self, *args, **kw)
        return wrapper
    return decorator

@contextmanager
def trace_step(op_name, message, profile=False, log_level='info', compound=True):
    if log_level == 'info':
        log_function = logger.info
    elif log_level == 'debug':
        log_function = logger.debug
    else:
        raise Exception("Invalid log level '%s'" % log_level)

if compound:
        # put an empty line before the block
        log_function(u"")
        log_function(messages.TRACE_START, message)
    else:
        log_function(message)
    try:
        with log_context(op_name):
            if profile:
                with get_default_steps_profiler().measure_time(
                    op_name, message
                ):
                    yield
            else:
                yield
        if compound:
            log_function(messages.TRACE_FINISH, message)
    except MigrationNoContextError:
        if compound:
            log_function(messages.TRACE_FINISH, message)
        raise
    except:
        logger.error(messages.TRACE_ABORT, message)
        raise

class DbServerInfo(namedtuple('DbServerInfo', ('dbtype', 'host', 'port', 'login', 'password'))):
    @staticmethod
    def from_plesk_api(db_server_info):
        """Convert from plesk_api.DbServerInfo to DbServerInfo"""
        return DbServerInfo(
            dbtype=db_server_info.dbtype,
            host=db_server_info.host,
            port=db_server_info.port,
            login=db_server_info.admin,
            password=db_server_info.password
        )

@staticmethod
    def from_plesk_backup(db_server_info):
        """Convert plesks_migrator.data_model.DatabaseServer to DbServerInfo"""
        return DbServerInfo(
            dbtype=db_server_info.dbtype,
            host=db_server_info.host,
            port=db_server_info.port,
            login=db_server_info.login,
            password=db_server_info.password
        )

def get_package_root_path(package):
    """Get path to Python package root directory"""
    dirs = [p for p in package.__path__]
    assert all(d == dirs[0] for d in dirs)
    return dirs[0]

def get_package_extras_file_path(package, filename):
    """Get path to extras file of Python package

:type filename: str | unicode
    :rtype: str | unicode
    """
    return os.path.join(
        get_package_root_path(package), 'extras', filename
    )

def get_package_scripts_file_path(package, filename):
    """Get path to scripts file of Python package

:type filename: str | unicode
    :rtype: str | unicode
    """
    return os.path.join(
        get_package_root_path(package), 'scripts', filename
    )

def version_to_tuple(version):
    version = safe_utf8_decode(version)
    return tuple(map(int, (version.split('.'))))

def split_unix_path(path):
    """Split path by directory name and file name

Returns tuple (directory name, file name)

:rtype tuple
    """
    last_slash_pos = path.rfind('/')
    directory = path[:last_slash_pos]
    filename = path[last_slash_pos+1:]
    return directory, filename

def normalize_domain_name(domain_name):
    """Normalize domain name - convert to lowercase punycode, convert wildcard subdomains from "*" to "_" notation

Use this function for comparison of domain names.

:rtype str | unicode | None
    """
    if domain_name is None:
        return None

return safe_idn_encode(domain_name).lower().replace('*', '_')

@cached
def get_version():
    version_file_name = os.path.join(Registry.get_instance().get_base_dir(), 'version')
    if not os.path.exists(version_file_name):
        version = 'DEVELOPMENT'
    else:
        version = read_string_from_file(version_file_name)
    return version

def read_string_from_file(file_name):
    with open_no_inherit(file_name, 'r') as fp:
        return fp.read().strip()

def format_owner_login(owner_login):
    """Format login variable of admin/reseller/client into a string to be used in messages

Considers administrator (for which login is None, according to different models used in migrator)

:type owner_login: str | unicode
    :rtype: unicode
    """
    if owner_login is None:
        return u'administrator'
    else:
        return u"'%s'" % owner_login

def get_option_value_safe(options, option_name, default_value):
    """Read command line option in a safe way

If there is no command line option for a command, warning will be emitted and default value returned.
    If you will try to use option directly, migrator will completely fail.
    """
    if hasattr(options, option_name):
        return getattr(options, option_name)
    else:
        logger.warning(messages.NO_COMMAND_LINE_OPTION_WARNING.format(option_name=option_name))
        return default_value

def get_optional_option_value(options, option_name, default_value):
    """Read command line option, which could exist or not exist

If there is no command line option for a command, default value returned.
    """
    if hasattr(options, option_name):
        return getattr(options, option_name)
    else:
        return default_value

def is_locked():
    result = True
    migration_lock = Registry.get_instance().get_migration_lock()
    try:
        migration_lock.acquire()
        migration_lock.release()
        result = False
    except FileLockException:
        # assume that command is still in progress
        pass
    return result

def safe_idn_decode(domain_name):
    """Safely decode domain name from punycode to unicode

If domain name is already in unicode - return it as is. If any error occurred during decoding - return domain
    name as is.

:type domain_name: str | unicode
    :rtype: unicode
    """
    if not is_ascii_string(domain_name):
        return domain_name

try:
        return domain_name.encode().decode('idna')
    except:
        logger.exception()
        logger.warning(messages.UNABLE_TO_CONVERT_IDN_NAME.format(domain_name=domain_name))
        return domain_name

def is_unicode_domain(domain_name):
    """Whether specified domain is in unicode or in punycode (IDNA). For unicode form - return true.

:rtype: bool
    """
    return safe_idn_encode(domain_name) != domain_name

def safe_mail_idn_decode(email_address):
    """
    :type email_address: str | unicode
    :rtype: str | unicode
    """
    email_address_parts = email_address.split('@')
    if len(email_address_parts) != 2:
        return email_address
    mail_name, domain_name = email_address_parts
    return '@'.join([mail_name, safe_idn_decode(domain_name)])

def path_startswith(path1, path2):
    path1_items = [i for i in re.split('[/\\\\]', path1) if i != '']
    path2_items = [i for i in re.split('[/\\\\]', path2) if i != '']
    return path1_items == path2_items[:len(path1_items)]

def is_running_from_cli():
    """Check if migrator is running from command-line interface (comparing to GUI)

:rtype: bool
    """
    messages_ui = os.getenv('MIGRATOR_MESSAGES_UI')
    return messages_ui != 'gui'

def ip_file_path(ip_address):
    """Get file path path which corresponds to specified IP address,

Function replaces symbols which are valid in IP address, but not valid in filesystem path

:type ip_address: str | unicode
    :rtype: str | unicode
    """
    return ip_address.replace(':', '_').replace('.', '_')

def download_thirdparty_zip(remote_name, local_name, log_message):
    """Download thirdparty ZIP archive and unpack to a directory

If it was already downloaded before - skip.
    Returns path to a local directory with thirdparty.

:type remote_name: str | unicode
    :type local_name: str | unicode
    :type log_message: str | unicode
    :rtype: str | unicode
    """
    dir_path = os.path.join(get_thirdparties_base_path(), local_name)

if not os.path.exists(get_thirdparties_base_path()):
        mkdir_p(get_thirdparties_base_path())

if not os.path.exists(dir_path):
        url = 'http://autoinstall.plesk.com/panel-migrator/thirdparties/%s' % remote_name
        logger.info(log_message.format(url=url))
        download_zip(url, dir_path)

return dir_path

def is_cache_enabled_and_valid(global_context, cache_migration_list_file):
    """Whether caching is enabled and cache is valid for different files and models like converted dump, etc

By default caching is disabled for 'transfer-accounts' and 'check' commands, but enabled for post-migration
    commands like 'copy-content', 'copy-web-content', 'test-sites', etc. It allows to run post-migration
    operations fast once migration was performed, but keep data up to date when running migration itself.
    Caching is performed by migration list: if migration list file changed (even by single whitespace),
    then cache is considered invalid.

Cache migration list file is a file which contains contents of migration list file by the moment when cache
    was updated last time.

:type global_context: parallels.core.global_context.GlobalMigrationContext
    :type cache_migration_list_file: str | unicode
    :rtype: bool
    """
    if not _is_caching_enabled(global_context):
        return False

migration_list_file = get_migration_list_file_location(global_context)
    if migration_list_file is None:
        return False

if not os.path.exists(cache_migration_list_file):
        return False

if not os.path.exists(migration_list_file):
        return False

if not files_have_equal_contents(migration_list_file, cache_migration_list_file):
        return False

return True

def _is_caching_enabled(global_context):
    """Whether caching is enabled for different files and models like converted dump, existing objects model, etc

By default caching is disabled for 'transfer-accounts' and 'check' commands, but enabled for post-migration
    commands like 'copy-content', 'copy-web-content', 'test-sites', etc.

:type global_context: parallels.core.global_context.GlobalMigrationContext
    :rtype: bool
    """
    if hasattr(global_context.options, 'use_cache'):
        return global_context.options.use_cache
    else:
        return False

def remote_lookup_cmd(runner, *args, **kwargs):
    """
    Return the first command found under the specified `basedir` from remote server.

This function searches for commands (file names) provided in `args` within the directory
    specified by the `basedir` keyword argument. It returns the first command that exists
    and is executable. If no such command is found, the function returns `None`.

Commands are prioritized in the order they are passed in `args`. The default `basedir`
    is `/usr/bin`, but this can be overridden by providing a `basedir` keyword argument.

:param runner: The runner object to execute the command on the remote server.
    :param args: One or more command names to search for.
    :param kwargs: Optional keyword arguments. Supported keys:
        - basedir (str): The base directory to search for commands. Defaults to `/usr/bin`.
    :return: The full path to the first executable command found, or `None` if no command is found.
    :rtype: str | None
    """
    basedir = kwargs.get("basedir", "/usr/bin")

for path in [os.path.join(basedir, arg) for arg in args]:
        if runner.file_exists(path):
            return path

return None

📁 Sid Gifari File Manager