Da

✏️ Editing: windows_agent_remote_object.py

import ntpath
import os
import pickle
import socket
import ssl
import struct
import threading
import time

from parallels.core import messages, MigrationError, MigrationNoContextError
from parallels.core.logging import get_logger
from parallels.core.registry import Registry
from parallels.core.runners.windows.local import LocalWindowsRunner
from parallels.core.utils import get_thirdparties_base_path
from parallels.core.utils.common import mkdir_p, read_file_contents, safe_format, open_no_inherit, format_bytes, \
    format_percentage, safe_string_repr, optimize_values
from parallels.core.utils.download_utils import download_if_newer, download
from parallels.core.utils.message_utils import multi_line_message
from parallels.core.utils.migrator_utils import download_thirdparty_zip
from parallels.core.utils.paexec import RemoteNodeSettings, PAExecCommand, PAExecNetworkTimeoutException
from parallels.core.utils.plesk_utils import get_windows_util_path
from parallels.core.utils.process_handle import ProcessHandle
from parallels.core.utils.steps_profiler import sleep

logger = get_logger(__name__)

RPC_AGENT_VERSION = '1.22'

class WindowsAgentRemoteObject(object):
    """Object to work with remote RPC agent"""

def __init__(self, settings, proxy_to=None):
        """Class constructor

If proxy_to parameter is specified, then connect to another RPC agent server from current RPC agent server
        and proxy all requests to it. In that case communication will look like:
        [Plesk Migrator] <-----> [proxy RPC agent] <-----> [remote RPC agent]
        For more details check '_proxy_to_another_rpc_agent' function

:type settings: parallels.core.migrator_config.PhysicalWindowsServerConfig
        :type proxy_to: str | None
        """
        self._settings = settings
        self._proxy_to = proxy_to
        self._sockets = {}
        self._started_automatically = False
        remote_node_settings = RemoteNodeSettings(
            remote_server_ip=self._settings.ip,
            username=self._settings.windows_auth.username,
            password=self._settings.windows_auth.password
        )
        self._paexec_cmd = PAExecCommand(remote_node_settings)

def deploy_and_connect(self):
        logger.debug(
            messages.DEBUG_TRY_TO_CONNECT_TO_TRANSFER_AGENT,
            self._settings.ip, self._settings.agent_settings.port
        )
        if not self.try_connect():
            logger.debug(messages.CONNECTION_FAILED_MOST_LIKELY_THAT_PANEL)
            logger.info(messages.DEPLOY_AND_START_PANEL_MIGRATOR_TRANSFER, self._settings.ip)
            try:
                logger.debug(messages.TRY_START_PANEL_MIGRATOR_TRANSFER_AGENT)
                self.start()
            except PAExecNetworkTimeoutException:
                # Timeout when connecting to remote server with PAExec. That usually means permanent network issue:
                # IP not routed, not configured, firewall blocks connections, etc. We can't resolve that issue
                # automatically, so just print agent installation instructions and stop any other attempts.
                logger.debug(
                    messages.EXCEPTION_WHEN_TRYING_START_PANEL_MIGRATOR,
                    exc_info=True
                )
                self._print_agent_installation_instructions()
            except Exception:
                logger.debug(
                    messages.EXCEPTION_WHEN_TRYING_START_PANEL_MIGRATOR,
                    exc_info=True
                )

# failed to start - then deploy and start
                try:
                    self._deploy_and_start()
                except Exception:
                    logger.debug(
                        messages.EXCEPTION_WHEN_TRYING_DEPLOY_AND_START,
                        exc_info=True
                    )
                    self._print_agent_installation_instructions()
                else:
                    if not self.try_connect_multiple():  # wait for agent to start
                        self._print_agent_connection_issue()
                    else:
                        self._started_automatically = True
            else:
                if not self.try_connect_multiple():
                    # started successfully, but failed to connect - then redeploy and start
                    try:
                        self._deploy_and_start()
                    except Exception:
                        logger.debug(
                            messages.EXCEPTION_WHEN_TRYING_DEPLOY_AND_START_1,
                            exc_info=True
                        )
                        self._print_agent_installation_instructions()
                    else:
                        if not self.try_connect_multiple():  # wait for agent to start
                            self._print_agent_connection_issue()
                        else:
                            self._started_automatically = True
                else:
                    self._started_automatically = True
                    # Here we successfully started agent that was already deployed.
                    # Check version of agent, and if it is lower than supported one,
                    # then redeploy
                    if not self._version_matches():
                        logger.fdebug(messages.DEBUG_RPC_AGENT_VERSION_MISMATCH, source_ip=self._settings.ip)
                        try:
                            self._deploy_and_start()
                        except Exception:
                            logger.debug(
                                messages.RPC_AGENT_DEPLOY_ERROR_ON_VERSION_MISMATCH,
                                exc_info=True
                            )
                            self._print_agent_installation_instructions_version_mismatch()
                        else:
                            if not self.try_connect_multiple():  # wait for agent to start
                                self._print_agent_connection_issue()
                            else:
                                self._started_automatically = True
        else:
            # Here we successfully connected to agent. Check version of agent,
            # and if it is lower than supported one, then redeploy
            if not self._version_matches():
                logger.fdebug(messages.DEBUG_RPC_AGENT_VERSION_MISMATCH, source_ip=self._settings.ip)
                try:
                    self._deploy_and_start()
                except Exception:
                    logger.debug(
                        messages.RPC_AGENT_DEPLOY_ERROR_ON_VERSION_MISMATCH,
                        exc_info=True
                    )
                    self._print_agent_installation_instructions_version_mismatch()
                else:
                    if not self.try_connect_multiple():  # wait for agent to start
                        self._print_agent_connection_issue()
                    else:
                        self._started_automatically = True

def try_connect_multiple_and_check_version(self):
        """Try connect to RPC agent that was deployed manually, check its version

If we failed to connect to RPC agent, raise exception.

If version of manually deployed RPC agent is not compatible with Plesk Migrator,
        raise exception (that should be handled by GUI to display comprehensive error message)
        """
        if not self.try_connect_multiple(attempts=3, interval=5):
            self._print_agent_connection_issue()

if not self._version_matches():
            raise RPCAgentConnectionError(
                safe_format(
                    messages.WINDOWS_AGENT_VERSION_MISMATCH,
                    hostname=self._settings.ip,
                    dist="%s\\run-panel-migrator-rpc-agent.exe" % get_thirdparties_base_path(),
                ),
                error_id='RPC_AGENT_VERSION_MISMATCH'
            )

def deploy(self):
        logger.debug(messages.DEBUG_UPLOAD_AND_START_AGENT)

RPCAgentBuilder().build()
        self._paexec_cmd.run(
            executable=r".\run-panel-migrator-rpc-agent.exe",
            # first, deploy agent executable to the server
            copy_program=True,
            # do not wait for archive to extract and agent to start
            do_not_wait=True,
        )

def start(self):
        self._paexec_cmd.run(
            executable=ntpath.join(self._settings.agent_settings.agent_path, "panel-migrator-rpc-agent.cmd"),
            # do not wait for start script to finish
            do_not_wait=True,
        )

def stop(self, force=False):
        # Shutdown agent only if it was started automatically.
        # In case agent was deployed/started by customer - don't stop it.
        if self._started_automatically or force:
            try:
                # stop rpc agent process
                self.__getattr__('stop')()
            except Exception:
                logger.exception()
                self._kill()

def shutdown(self):
        self.stop()
        for sock in optimize_values(self._sockets):
            sock.close()

def try_connect_multiple(self, attempts=6, interval=5):
        try:
            range_attempts = range(attempts)

for _ in range_attempts:
                if self._try_connect_raise_ssl_error():
                    return True
                sleep(interval, messages.TRY_CONNECT_AGENT_AT_S % self._settings.ip)
            return False
        except socket.error:
            return False

def try_connect(self):
        try:
            self.connect()
            return True
        except socket.error:
            logger.exception()
            return False

def connect(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(30)
        sock = self._wrap_ssl(sock)
        # to avoid dropping of idle connections turn on keep alive feature and set time 5 min and interval 1 sec
        # see https://msdn.microsoft.com/en-us/library/dd877220%28v=vs.85%29.aspx for details
        sock.ioctl(socket.SIO_KEEPALIVE_VALS, (1, self._settings.agent_settings.tcp_keepalive_time, 1000))

logger.debug(messages.DEBUG_CONNECT_TO_AGENT, self._settings.ip)
        sock.connect((
            self._settings.ip, int(self._settings.agent_settings.port)
        ))

# set timeout only at connection time to avoid issues with remote commands that are executed for a long time
        sock.settimeout(None)

self.set_socket(sock)
        if self._proxy_to is not None:
            self._proxy_to_another_rpc_agent(self._proxy_to)

def reconnect(self):
        logger.debug(messages.DEBUG_RECONNECT_TO_AGENT, self._settings.ip)
        # close socket
        sock = self.get_socket(autoconnect=False)
        if sock is not None:
            try:
                sock.close()
            except Exception:
                logger.debug(messages.EXCEPTION_WHEN_CLOSING_SOCKET, exc_info=True)
        # try to connect
        self.connect()

def is_configured_manually(self):
        return not self._started_automatically

def __getattr__(self, attr):
        """One of the most important functions there, where actual execution of functions on remote server happens

For example, if you call remote_object.sh_unchecked('echo 123'), it will be translated to
        remote_object.__getattr__('sh_unchecked')('echo 123')
        """
        return self._get_remote_function(attr)

def test_connection(self):
        """Test connection to the server"""
        self._get_remote_function('test_connection', multiple_attempts=False)()

def _get_remote_function(self, remote_function_name, multiple_attempts=True):
        def run_remote_function(*args, **kwargs):
            def run():
                command = pickle.dumps(
                    (remote_function_name, args, kwargs)
                )
                self.get_socket().sendall(struct.pack("I", len(command)))
                self.get_socket().sendall(command)
                length = self._receive(4)
                length, = struct.unpack("I", length)
                if length == 0:
                    raise MigrationError(
                        safe_format(
                            messages.WINDOWS_FAILED_TO_EXECUTE_REMOTE_COMMAND, server_ip=self._settings.ip
                        )
                    )
                else:
                    result = pickle.loads(self._receive(length))
                    return result

if multiple_attempts:
                return self._run_multiple_attempts(run)
            else:
                return run()

return run_remote_function

GET_FILE_LOG_INTERVAL = 10

def get_file(self, remote_filename, local_filename):
        """Get file from the remote server to the local server

:type remote_filename: str
        :type local_filename: str
        """
        command = pickle.dumps(
            ('get_file_stream', [], {'filename': remote_filename})
        )
        self.get_socket().sendall(struct.pack("I", len(command)))
        self.get_socket().sendall(command)

file_meta_length = self._receive(4)
        file_meta_length, = struct.unpack("I", file_meta_length)
        file_meta = pickle.loads(self._receive(file_meta_length))

file_size = file_meta['size']
        copied_bytes = 0

start_time = time.time()

last_log_time = start_time
        last_log_size = 0

with open_no_inherit(local_filename, 'wb') as fp:
            read_buffer_size = 1024
            while copied_bytes < file_size:
                data = self.get_socket().recv(read_buffer_size)
                if not data: break
                fp.write(data)
                copied_bytes += len(data)
                current_time = time.time()
                if current_time > last_log_time + self.GET_FILE_LOG_INTERVAL:
                    if current_time - last_log_time == 0:
                        speed = 0
                    else:
                        speed = int((copied_bytes - last_log_size) / (current_time - last_log_time))

logger.fdebug(
                        messages.DEBUG_FILE_TRANSFER_PROGRESS,
                        percentage=format_percentage(copied_bytes, file_size),
                        interval=int(current_time - last_log_time),
                        bytes_human_readable=format_bytes(copied_bytes),
                        bytes=copied_bytes,
                        remote_filename=remote_filename,
                        local_filename=local_filename,
                        speed_human_readable=format_bytes(speed),
                        speed=speed
                    )

last_log_time = current_time
                    last_log_size = copied_bytes

def get_file_chunk(self, filename, offset, size):
        """Get chunk of file with specified size (in bytes) starting at specified offset

:type filename: str
        :type offset: int
        :type size: int
        :rtype: str
        """
        command = pickle.dumps(
            ('get_file_chunk', [], {'filename': filename, 'offset': offset, 'size': size})
        )
        self.get_socket().sendall(struct.pack("I", len(command)))
        self.get_socket().sendall(command)

length = self._receive(4)
        length, = struct.unpack("I", length)
        data = self._receive(length)
        return data

PIPE_STDOUT_PACKET = 10
    PIPE_STDERR_PACKET = 20
    PIPE_FINAL_PACKET = 30
    PIPE_STDIN_PACKET = 40

def piped_sh_unchecked(
        self, cmd_str, args,
        consume_stdout_function, consume_stderr_function,
        finalize_function, error_function,
        read_stdin=False
    ):
        """Execute command, and read stdout/stderr dynamically, as it goes

Consume stdout/stderr function is called once we get the next portion of data from command's stdout/stderr.
        The only argument it accepts is that data (as byte string).

Once the command is finished, finalize function is called. The only argument is exit code of the command.

:type cmd_str: str
        :type args: dict | None
        :type consume_stdout_function: (str) -> None
        :type consume_stderr_function: (str) -> None
        :type error_function: (str) -> None
        :type finalize_function: (int) -> None
        :type read_stdin: bool
        :rtype: parallels.core.utils.process_handle.ProcessHandle | None
        """
        full_logging = self._settings.agent_settings.piped_command_full_logging
        sock = self.get_socket()

if not read_stdin:
            self._execute_remote_piped_command(
                sock, cmd_str, args,
                consume_stdout_function, consume_stderr_function,
                finalize_function, error_function,
                read_stdin, full_logging
            )
        else:
            thread_output = threading.Thread(
                target=lambda: self._execute_remote_piped_command(
                    sock, cmd_str, args,
                    consume_stdout_function, consume_stderr_function,
                    finalize_function, error_function,
                    read_stdin, full_logging
                )
            )
            thread_output.start()
            return PipedCommandProcessHandle(
                sock=sock, read_output_thread=thread_output, full_logging=full_logging
            )

def _execute_remote_piped_command(
        self, sock, cmd_str, args,
        consume_stdout_function, consume_stderr_function, finalize_function, error_function,
        read_stdin, full_logging
    ):
        """Run command on remote server, receive stdout/stderr and call provided callbacks

:type cmd_str: str
        :type args: dict | None
        :type consume_stdout_function: (bytes) -> None
        :type consume_stderr_function: (bytes) -> None
        :type finalize_function: (int) -> None
        :type error_function: (str) -> None
        :type read_stdin: bool
        :type full_logging: bool
        :rtype: None
        """
        try:
            sender = SocketSender(sock)
            receiver = SocketReceiver(sock)

command = pickle.dumps(
                ('piped_sh_unchecked', [], {'cmd_str': cmd_str, 'args': args, 'read_stdin': read_stdin})
            )
            sender.send_bytes(command)

while True:
                packet_type = receiver.receive_int()

if full_logging:
                    logger.fdebug(messages.DEBUG_PIPED_PACKET_TYPE, packet_type=packet_type)

if packet_type in (self.PIPE_STDOUT_PACKET, self.PIPE_STDERR_PACKET):
                    data = receiver.receive_bytes()

if packet_type == self.PIPE_STDOUT_PACKET:
                        if full_logging:
                            logger.fdebug(messages.DEBUG_PIPED_STDOUT, stdout=repr(data))
                        consume_stdout_function(data)
                    else:
                        if full_logging:
                            logger.fdebug(messages.DEBUG_PIPED_STDERR, stderr=repr(data))
                        consume_stderr_function(data)
                elif packet_type == self.PIPE_FINAL_PACKET:
                    exit_code = receiver.receive_int()
                    logger.fdebug(messages.COMMAND_IN_PIPED_MODE_FINISHED, exit_code=exit_code)
                    finalize_function(exit_code)
                    return
                elif packet_type is None:
                    logger.fdebug(messages.RPC_AGENT_PROTOCOL_NO_PACKET)
                    error_function(messages.RPC_AGENT_PROTOCOL_NO_PACKET)
                    return
                else:
                    logger.fdebug(messages.RPC_AGENT_PROTOCOL_INVALID_PACKET, packet_type=packet_type)
                    error_function(safe_format(messages.RPC_AGENT_PROTOCOL_INVALID_PACKET, packet_type=packet_type))
                    return
        except Exception as e:
            logger.exception()
            error_function(safe_format("{exception}", exception=e))
            return

def _version_matches(self):
        try:
            logger.fdebug(messages.DEBUG_GET_RPC_AGENT_VERSION, source_ip=self._settings.ip)
            version = self.get_version()
            logger.fdebug(
                messages.DEBUG_RPC_AGENT_VERSION, version=version, source_ip=self._settings.ip
            )
        except Exception:
            logger.exception()
            logger.fdebug(messages.DEBUG_RPC_AGENT_VERSION_UNKNOWN, source_ip=self._settings.ip)
            # Version before 1.9: there were no such method, so get_version call will fail.
            # Versions before 1.9 are not compatible with migrator anyway, so return False.
            return False

return version == RPC_AGENT_VERSION

def _proxy_to_another_rpc_agent(self, hostname, port=10155):
        """Connect to another RPC agent server from current RPC agent server, proxy all requests

Communication will look like:
        [Plesk Migrator] <-----> [proxy RPC agent] <-----> [remote RPC agent]

Once this function is called, all operations will be executed on specified remote RPC agent server.
        There will be no way to execute commands on a server which is used as a proxy within this object.
        Create another instance if you need to execute commands both on a proxy server and on remote RPC agent server.

:type hostname: str
        :type port: int
        """
        # There is a special command "rpc_agent_proxy" in RPC agent.
        # Check RPC agent sources (panel-migrator-rpc-agent.py) for more details.
        command = pickle.dumps(('rpc_agent_proxy', [hostname, port], {}))
        self.get_socket().sendall(struct.pack("I", len(command)))
        self.get_socket().sendall(command)

success = self._receive(4)
        success, = struct.unpack("I", success)
        if success == 0:
            raise RPCAgentConnectionError(safe_format(
                messages.RPC_PROXY_FAILED_CONNECTION, remote_server=hostname, proxy_server=self._settings.ip
            ))

# Once the "rpc_agent_proxy" command is sent, we are working with remote RPC agent with the socket.
        # So the first operation we should perform is to set up SSL with remote RPC agent.
        if self._settings.agent_settings.use_ssl:
            try:
                sock = self.get_socket(autoconnect=False)
                # Shutdown old SSL with proxy RPC agent
                sock = sock.unwrap()
                # Set up new SSL with remote RPC agent
                sock = self._wrap_ssl(sock)
                self.set_socket(sock)
            except Exception as e:
                # Issue when setting up SSL connection: most probably an RPC agent running on the remote server
                # has wrong SSL certificates: recommend to redeploy the agent
                raise RPCAgentConnectionError(safe_format(
                    messages.RPC_PROXY_FAILED_CONNECTION_SSL_ISSUE,
                    remote_server=hostname, proxy_server=self._settings.ip,
                    reason=e
                ))

if not self._version_matches():
            raise RPCAgentConnectionError(
                safe_format(
                    messages.WINDOWS_AGENT_VERSION_MISMATCH,
                    hostname=hostname,
                    dist="%s\\run-panel-migrator-rpc-agent.exe" % get_thirdparties_base_path(),
                ),
                error_id='RPC_AGENT_VERSION_MISMATCH'
            )

def get_socket(self, autoconnect=True):
        if threading.current_thread() not in self._sockets:
            if autoconnect:
                self.connect()
            else:
                return None
        return self._sockets[threading.current_thread()]

def set_socket(self, sock):
        self._sockets[threading.current_thread()] = sock

def _deploy_and_start(self):
        logger.debug(messages.TRY_STOP_PANEL_MIGRATOR_TRANSFER_AGENT)
        try:
            self._kill()
        except Exception:
            logger.debug(messages.EXCEPTION_WHEN_TRYING_STOP_PANEL_MIGRATOR, exc_info=True)
        logger.debug(messages.DEPLOY_PANEL_MIGRATOR_TRANSFER_AGENT_AT, self._settings.ip)
        self.deploy()

def _kill(self):
        # terminate rpc agent process
        self._paexec_cmd.run(
            executable="cmd.exe",
            arguments="/c taskkill /F /IM panel-migrator-rpc-agent.exe",
            is_unchecked=True
        )

def _print_agent_installation_instructions(self):
        """Print agent installation instructions - in case we failed to deploy or start agent"""
        raise RPCAgentConnectionError(
            safe_format(
                messages.WINDOWS_AGENT_INSTALL_INSTRUCTIONS,
                source_ip=self._settings.ip,
                dist="%s\\run-panel-migrator-rpc-agent.exe" % get_thirdparties_base_path(),
            ),
            error_id='RPC_AGENT_FAILED_TO_INSTALL'
        )

def _print_agent_installation_instructions_version_mismatch(self):
        """Print agent install instructions - RPC agent of incompatible version, failed to deploy/start new agent"""
        raise RPCAgentConnectionError(
            safe_format(
                messages.WINDOWS_AGENT_INSTALL_INSTRUCTIONS_VERSION_MISMATCH,
                source_ip=self._settings.ip,
                dist="%s\\run-panel-migrator-rpc-agent.cmd" % get_thirdparties_base_path(),
            ),
            error_id='RPC_AGENT_FAILED_TO_INSTALL_VERSION_MISMATCH'
        )

def _print_agent_connection_issue(self):
        """Print agent connection issue, that is more likely caused by firewall"""
        raise RPCAgentConnectionError(
            safe_format(
                messages.WINDOWS_AGENT_CONNECTION_ISSUE,
                source_ip=self._settings.ip,
                port=self._settings.agent_settings.port,
            ),
            error_id='RPC_AGENT_FAILED_TO_CONNECT'
        )

def _try_connect_raise_ssl_error(self):
        try:
            self.connect()
            return True
        except socket.error as e:
            if 'SSL' in safe_string_repr(e):
                # If issue was caused by any SSL problem, we need to redeploy agent,
                # so raise exception immediately - it should be handled in caller function
                raise
            logger.exception()
            return False

def _wrap_ssl(self, sock):
        """Wrap socket with SSL wrapper according to current settings"""
        if self._settings.agent_settings.use_ssl:
            # import there to avoid dependency on OpenSSL on Linux
            from parallels.core.utils.ssl_utils import SSLKeys

client_key = self._settings.agent_settings.client_key
            client_cert = self._settings.agent_settings.client_cert
            server_cert = self._settings.agent_settings.server_cert

ssl_files = [client_key, client_cert, server_cert]

if any([f is None for f in ssl_files]):
                ssl_keys = SSLKeys.get_instance()
                if client_key is None or client_cert is None or server_cert is None:
                    ssl_keys.generate_keys_lazy()

if client_key is None:
                    client_key = ssl_keys.migrator_server_key_filename
                if client_cert is None:
                    client_cert = ssl_keys.migrator_server_crt_filename
                if server_cert is None:
                    server_cert = ssl_keys.source_server_crt_filename

logger.fdebug(
                messages.DEBUG_SSL_CLIENT_KEY,
                ip=self._settings.ip,
                ssl_client_key=client_key, ssl_client_cert=client_cert, ssl_server_cert=server_cert
            )

sock = ssl.wrap_socket(
                sock,
                keyfile=client_key,
                certfile=client_cert,
                cert_reqs=ssl.CERT_REQUIRED,
                ca_certs=server_cert
            )
        else:
            logger.debug(messages.SSL_IS_DISABLED_FOR_AGENT_AT, self._settings.ip)

return sock

def _run_multiple_attempts(self, run_function, max_attempts=5, interval_between_attempts=10):
        for attempt in range(0, max_attempts):
            try:
                if attempt > 0:
                    # reconnect on 2nd and later attempts
                    self.reconnect()

result = run_function()
                if attempt > 0:
                    logger.info(
                        safe_format(
                            messages.REMOTE_OPERATION_OK_AFTER_ATTEMPTS,
                            host=self._settings.ip, attempts=attempt + 1
                        )
                    )
                return result
            except socket.error as e:
                logger.exception()
                if attempt >= max_attempts - 1:
                    raise e
                else:
                    logger.error(
                        safe_format(
                            messages.REMOTE_OPERATION_FAILED_RETRY,
                            interval_between_attempts=interval_between_attempts
                        )
                    )
                    sleep(interval_between_attempts, messages.SLEEP_RETRY_RUNNING_REMOTE_COMMAND)

def _receive(self, size):
        b = b''
        while len(b) < size:
            r = self.get_socket().recv(size - len(b))
            if not r:
                return b
            b += r
        return b

class RPCAgentBuilder(object):
    """Class to create RPC agent executables.

RPC agent executable is an application that could be uploaded on source server
    and launched by simple double click on it.

Directory structure:
    $var_dir/rpc-agent/
        build/ - directory for temporary files used when building agent
        certificates/ - directory for SSL keys and certificates used by agent
        agent-built - file indicating that agent was successfully built
    $var_dir/thirdparties/
        7zsd.sfx - used to build agent
        panel-migrator-rpc-agent/ - agent Python sources and binaries
        run-panel-migrator-rpc-agent.exe - agent executable itself (results of the current class)

Build schema:
    1)
    RPC agent sources (panel-migrator-rpc-agent_*.zip)
        +
    Certificates (*.crt, *.key)
        =
    7 ZIP archive (panel-migrator-rpc-agent-installer.7z)

2)
    7 ZIP archive (panel-migrator-rpc-agent-installer.7z)
        +
    Self-extractable text config  (config.txt)
        +
    Self-extractable binary (7zsd.sfx)
        =
    Self-extractable executable file

Self-extractable binary (7zsd.sfx) was taken from LZMA SDK (http://www.7-zip.org/download.html) -
    it is modified version of standard 7zip self-extractable binary, but allows more options necessary
    for unattended unpack and run.

The function returns path to built RPC agent on filesystem.

:rtype: str
    """
    def build(self):
        """Build RPC agent executable (if it does not exist)

Returns path to RPC agent executable.

:rtype: str
        """
        runner = LocalWindowsRunner()

# Download latest version of 7zip
        sevenzip_sfx_path = os.path.join(get_thirdparties_base_path(), '7zsd.sfx')
        sevenzip_sfx_storage_url = 'http://autoinstall.plesk.com/panel-migrator/thirdparties/7zsd.sfx'
        sevenzip_updated = download_if_newer(sevenzip_sfx_storage_url, sevenzip_sfx_path) is not None

if not sevenzip_updated and \
                runner.file_exists(self._agent_built_file_path()) and \
                runner.file_exists(self._agent_executable_path()):
            # Agent is already built
            return self._agent_executable_path()

# Import there to avoid dependency on OpenSSL on Linux
        from parallels.core.utils.ssl_utils import SSLKeys

ssl_keys = SSLKeys.get_instance()
        ssl_keys.generate_keys_lazy()

# Download RPC agent Python sources
        agent_python_sources = download_thirdparty_zip(
            'panel-migrator-rpc-agent_{version}.zip'.format(version=RPC_AGENT_VERSION),
            'panel-migrator-rpc-agent', messages.DOWNLOAD_RPC_AGENT
        )

archive_path = self._build_path('panel-migrator-rpc-agent-installer.7z')
        config_path = self._build_path('config.txt')

# Clean up temporary files that may exist from the previous run. Remove RPC agent executable.
        runner.remove_file(archive_path)
        runner.remove_file(config_path)
        runner.remove_file(self._agent_executable_path())

# Create archive with agent out of Python sources and certificates
        command = (
            r'{seven_zip_bin} a -m0=LZMA {archive_path} '
            r'{source_server_key} {source_server_crt} {migrator_server_crt}'
        )

args = dict(
            seven_zip_bin=get_windows_util_path(runner, '7z.exe'),
            archive_path=archive_path,
            source_server_key=ssl_keys.source_server_key_filename,
            source_server_crt=ssl_keys.source_server_crt_filename,
            migrator_server_crt=ssl_keys.migrator_server_crt_filename
        )
        for i, filename in enumerate(runner.get_files_list(agent_python_sources)):
            command += ' {file%s}' % i
            args['file%s' % i] = ntpath.join(agent_python_sources, filename)

runner.execute_command(command, args)

# Create configuration file for self-extractable archive
        config_text = multi_line_message(r"""
            ;!@Install@!UTF-8!
            InstallPath="%SYSTEMDRIVE%\\plesk_migrator\\rpc-agent"
            RunProgram="panel-migrator-rpc-agent.cmd"
            ;!@InstallEnd@!
        """).replace('\n', '\r\n')
        runner.upload_file_content(config_path, config_text)

# Create self-extractable archive, which is single-click RPC agent executable
        rpc_agent_executable_contents = b''.join([
            read_file_contents(sevenzip_sfx_path),
            read_file_contents(config_path),
            read_file_contents(archive_path)
        ])
        runner.upload_file_content(self._agent_executable_path(), rpc_agent_executable_contents, encode=False)

# Clean up temporary files
        runner.remove_file(archive_path)
        runner.remove_file(config_path)

runner.upload_file_content(self._agent_built_file_path(), messages.RPC_AGENT_BUILT_INDICATOR)

return self._agent_executable_path()

def build_and_print_path(self):
        """Build RPC agent executable (if it does not exist) and print path to console

:rtype: None
        """
        self.build()
        print(self._agent_executable_path())

@staticmethod
    def _agent_built_file_path():
        """Get path to a file indicating whether agent was successfully built

:rtype: str
        """
        return os.path.join(Registry.get_instance().get_var_dir(), 'rpc-agent', 'agent-built')

@staticmethod
    def _agent_executable_path():
        """Get path of RPC agent executable (which is the main result of this class)

:rtype: str
        """
        return os.path.join(get_thirdparties_base_path(), 'run-panel-migrator-rpc-agent.exe')

@staticmethod
    def _build_dir():
        """Create (if not exists) and get path of build directory

:rtype: str
        """
        directory = os.path.join(Registry.get_instance().get_var_dir(), 'rpc-agent', 'build')
        if not os.path.exists(directory):
            mkdir_p(directory)
        return directory

def _build_path(self, base_name):
        """Get path to temporary build file

:type base_name: str
        :rtype : str
        """
        return os.path.join(self._build_dir(), base_name)

class PipedCommandProcessHandle(ProcessHandle):
    def __init__(self, sock, read_output_thread, full_logging):
        """
        :type sock: socket.SocketType
        :type read_output_thread: threading.Thread
        :type full_logging: bool
        """
        self._sock = sock
        self._read_output_thread = read_output_thread
        self._full_logging = full_logging

def write_stdin(self, data: bytes) -> None:
        """Write data to stdin of the process
        """
        sender = SocketSender(self._sock)
        if self._full_logging:
            logger.fdebug(messages.DEBUG_SEND_STDIN_DATA_PACKET, length=len(data))
        sender.send_int(WindowsAgentRemoteObject.PIPE_STDIN_PACKET)
        sender.send_bytes(data)

def close_stdin(self):
        """Close stdin of the process

:rtype: None
        """
        sender = SocketSender(self._sock)
        if self._full_logging:
            logger.fdebug(messages.DEBUG_SEND_STDIN_FINAL_PACKET)
        sender.send_int(WindowsAgentRemoteObject.PIPE_FINAL_PACKET)
        if self._full_logging:
            logger.fdebug(messages.DEBUG_WAIT_FOR_OUTPUT_THREAD)
        self._read_output_thread.join()
        if self._full_logging:
            logger.fdebug(messages.DEBUG_OUTPUT_THREAD_FINISHED)

class SocketSender(object):
    """Helper class to write data to provided socket"""

def __init__(self, sock):
        """
        :type sock: socket.SocketType
        """
        self._sock = sock
        self._lock = threading.Lock()

def send_int(self, value):
        """Send integer to the socket

:type value: int
        :rtype: None
        """
        self._sock.sendall(struct.pack("I", value))

@property
    def lock(self):
        """Lock to avoid multiple threads to write data to the socket at the same time

:rtype: threading.Lock
        """
        return self._lock

def send_bytes(self, value: bytes) -> None:
        """Send string (sequence of bytes) to the socket, along with its length.
        """
        self.send_int(len(value))
        self._sock.sendall(value)

class SocketReceiver(object):
    """Helper class to read data from provided socket"""

def __init__(self, sock):
        """
        :type sock: socket.SocketType
        """
        self._sock = sock

def receive_int(self) -> int | None:
        """Receive integer from the socket
        """
        data = self.receive(4)
        if len(data) == 0:
            return None

value, = struct.unpack("I", data)
        return value

def receive_bytes(self) -> bytes:
        """Receive string (sequence of bytes) of specified size from the socket
        """
        length = self.receive_int()
        return self.receive(length)

def receive(self, size: int) -> bytes:
        """Receive string (sequence of bytes) of specified size from the socket
        """
        b = b''
        while len(b) < size:
            r = self._sock.recv(size - len(b))
            if not r:
                return b
            b += r
        return b

class RPCAgentConnectionError(MigrationNoContextError):
    pass

📁 Sid Gifari File Manager