Da

✏️ Editing: checker.py

import paramiko

from parallels.core.logging import get_logger
from parallels.core import messages, MigrationNoContextError, safe_format
from parallels.core.utils import ssh_utils
from parallels.plesk.utils.xml_rpc.plesk.core import PleskError

logger = get_logger(__name__)

class ConnectionChecker(object):
    def check(self, check_list):
        """check_list contains items that should be checked here. Possible options are: ssh_auth, windows_auth."""
        for server, checks in check_list:
            server_options = server.node_settings
            if hasattr(server_options, "ssh_auth") and "ssh_auth" in checks: 
                self.check_ssh(server.description(), server_options)
            if hasattr(server_options, "windows_auth") and "windows_auth" in checks:
                with server.runner() as runner:
                    runner.check(server.description())

@staticmethod
    def check_ssh(server_name, server_options):
        logger.debug(messages.CHECK_SSH_CONNECTION_FOR_S_SERVER % server_name)
        try:
            with ssh_utils.connect(server_options) as ssh_client:
                ssh_client.get_transport().open_session()
        except paramiko.AuthenticationException:
            logger.exception()
            raise MigrationNoContextError(
                messages.UNABLE_LOGIN_S_SSH_SERVICE_DUE % (
                    server_name
                )
            )
        except (paramiko.SSHException, EnvironmentError) as err:
            logger.exception()
            raise MigrationNoContextError(messages.CHECKER_FAILED_TO_CONNECT_BY_SSH % (server_name, err))
        except Exception as err:
            logger.exception()
            if "q must be exactly 160, 224, or 256 bits long" in str(err):
                message = messages.CHECKER_FAILED_TO_CHECK_OUTDATED_SSH_CONNECTION % (server_name, err)
            else:
                message = messages.CHECKER_FAILED_TO_CHECK_SSH_CONNECTION % (server_name, err)
            raise MigrationNoContextError(message)

@staticmethod
    def check_plesk_api(plesk_server):
        """
        :type plesk_server: parallels.core.connections.plesk_server.PleskServer
        """
        logger.debug(messages.DEBUG_CHECK_API_CONNECTION % plesk_server.description())
        try:
            plesk_server.run_test_api_request()
        except PleskError as e:
            logger.exception()
            if e.code == 1001:
                raise PleskConnectionCheckError(
                    messages.FAILED_TO_CHECK_PLESK_API_INVALID_CREDENTIALS.format(
                        server=plesk_server.description()
                    )
                )
            elif e.code == 1030:
                raise PleskConnectionCheckError(
                    messages.FAILED_TO_CHECK_PLESK_API_EXPIRED_LICENSE.format(
                        server=plesk_server.description()
                    )
                )
            else:
                raise PleskConnectionCheckError(
                    safe_format(
                        messages.FAILED_TO_CONNECT_BY_PLESK_API,
                        server=plesk_server.description(), error=e
                    )
                )
        except EnvironmentError as e:
            logger.exception()
            raise PleskConnectionCheckError(
                safe_format(
                    messages.FAILED_TO_CONNECT_BY_PLESK_API,
                    server=plesk_server.description(), error=e
                )
            )
        except Exception as e:
            logger.exception()
            raise PleskConnectionCheckError(
                safe_format(
                    messages.FAILED_TO_CHECK_PLESK_API_CONNECTION,
                    server=plesk_server.description(), error=e
                )
            )

class PleskConnectionCheckError(MigrationNoContextError):
    pass

📁 Sid Gifari File Manager