Skip to content

ark_pty_ssh_connection

ArkPTYSSHConnection

Bases: ArkConnection

Source code in ark_sdk_python/common/connections/ssh/ark_pty_ssh_connection.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class ArkPTYSSHConnection(ArkConnection):
    __ANSI_STRIPPER: Any = re.compile(r'\x1b[^m]*m|\x1b\[\?2004[hl]')

    def __init__(self):
        super().__init__()
        self.__is_connected: bool = False
        self.__is_suspended: bool = False
        self.__ssh_client: Optional[pxssh.pxssh] = None

    @overrides
    def connect(self, connection_details: ArkConnectionDetails) -> None:
        """
        Performs SSH connection with given details or keys
        Saves the ssh session to be used for command executions

        Args:
            connection_details (ArkConnectionDetails): _description_

        Raises:
            ArkException: _description_
        """
        if self.__is_connected:
            return
        try:
            target_port = SSH_PORT
            user = None
            if connection_details.port:
                target_port = connection_details.port
            credentials_map = {}
            if connection_details.credentials:
                user = connection_details.credentials.user
                if connection_details.credentials.password:
                    credentials_map['password'] = connection_details.credentials.password.get_secret_value()
                elif connection_details.credentials.private_key_filepath:
                    path = Path(connection_details.credentials.private_key_filepath)
                    if not path.exists():
                        raise ArkException(f'Given private key path [{path}] does not exist')
                    credentials_map['ssh_key'] = connection_details.credentials.private_key_filepath
            self.__ssh_client = pxssh.pxssh()
            self.__ssh_client.login(
                server=connection_details.address,
                username=user,
                port=target_port,
                login_timeout=30,
                **credentials_map,
            )
            self.__is_connected = True
            self.__is_suspended = False
        except Exception as ex:
            raise ArkException(f'Failed to ssh connect [{str(ex)}]') from ex

    @overrides
    def disconnect(self) -> None:
        """
        Disconnects the ssh session
        """
        if not self.__is_connected:
            return
        self.__ssh_client.logout()
        self.__ssh_client = None
        self.__is_connected = False
        self.__is_suspended = False

    @overrides
    def suspend_connection(self) -> None:
        """
        Suspends execution of ssh commands
        """
        self.__is_suspended = True

    @overrides
    def restore_connection(self) -> None:
        """
        Restores execution of ssh commands
        """
        self.__is_suspended = False

    @overrides
    def is_suspended(self) -> bool:
        """
        Checks whether ssh commands can be executed or not

        Returns:
            bool: _description_
        """
        return self.__is_suspended

    @overrides
    def is_connected(self) -> bool:
        """
        Checks whether theres a ssh session connected

        Returns:
            bool: _description_
        """
        return self.__is_connected

    @overrides
    def run_command(self, command: ArkConnectionCommand) -> ArkConnectionResult:
        """
        Runs a command over ssh session, returning the result accordingly

        Args:
            command (ArkConnectionCommand): _description_

        Raises:
            ArkException: _description_

        Returns:
            ArkConnectionResult: _description_
        """
        if not self.__is_connected or self.__is_suspended:
            raise ArkException('Cannot run command while not being connected')
        self._logger.debug(f'Running command [{command.command}]')
        self.__ssh_client.sendline(command.command)
        self.__ssh_client.prompt()
        stdout = self.__ssh_client.before.decode()
        self.__ssh_client.sendline('echo $?')
        self.__ssh_client.prompt()
        exit_code_output = self.__ssh_client.before.decode()
        exit_code_output = ArkPTYSSHConnection.__ANSI_STRIPPER.sub('', exit_code_output)
        rc = int(exit_code_output.strip().splitlines()[-1])
        if rc != command.expected_rc and command.raise_on_error:
            raise ArkException(f'Failed to execute command [{command.command}] - [{rc}] - [{stdout}]')
        self._logger.debug(f'Command rc: [{rc}]')
        self._logger.debug(f'Command stdout: [{stdout}]')
        return ArkConnectionResult(stdout=stdout, rc=rc)

connect(connection_details)

Performs SSH connection with given details or keys Saves the ssh session to be used for command executions

Parameters:

Name Type Description Default
connection_details ArkConnectionDetails

description

 required

Raises:

Type Description
ArkException

description
Source code in ark_sdk_python/common/connections/ssh/ark_pty_ssh_connection.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@overrides
def connect(self, connection_details: ArkConnectionDetails) -> None:
    """
    Performs SSH connection with given details or keys
    Saves the ssh session to be used for command executions

    Args:
        connection_details (ArkConnectionDetails): _description_

    Raises:
        ArkException: _description_
    """
    if self.__is_connected:
        return
    try:
        target_port = SSH_PORT
        user = None
        if connection_details.port:
            target_port = connection_details.port
        credentials_map = {}
        if connection_details.credentials:
            user = connection_details.credentials.user
            if connection_details.credentials.password:
                credentials_map['password'] = connection_details.credentials.password.get_secret_value()
            elif connection_details.credentials.private_key_filepath:
                path = Path(connection_details.credentials.private_key_filepath)
                if not path.exists():
                    raise ArkException(f'Given private key path [{path}] does not exist')
                credentials_map['ssh_key'] = connection_details.credentials.private_key_filepath
        self.__ssh_client = pxssh.pxssh()
        self.__ssh_client.login(
            server=connection_details.address,
            username=user,
            port=target_port,
            login_timeout=30,
            **credentials_map,
        )
        self.__is_connected = True
        self.__is_suspended = False
    except Exception as ex:
        raise ArkException(f'Failed to ssh connect [{str(ex)}]') from ex

disconnect()

Disconnects the ssh session

Source code in ark_sdk_python/common/connections/ssh/ark_pty_ssh_connection.py 
67
68
69
70
71
72
73
74
75
76
77
@overrides
def disconnect(self) -> None:
    """
    Disconnects the ssh session
    """
    if not self.__is_connected:
        return
    self.__ssh_client.logout()
    self.__ssh_client = None
    self.__is_connected = False
    self.__is_suspended = False

is_connected()

Checks whether theres a ssh session connected

Returns:

Name Type Description
bool bool

description
Source code in ark_sdk_python/common/connections/ssh/ark_pty_ssh_connection.py 
103
104
105
106
107
108
109
110
111
@overrides
def is_connected(self) -> bool:
    """
    Checks whether theres a ssh session connected

    Returns:
        bool: _description_
    """
    return self.__is_connected

is_suspended()

Checks whether ssh commands can be executed or not

Returns:

Name Type Description
bool bool

description
Source code in ark_sdk_python/common/connections/ssh/ark_pty_ssh_connection.py 
 93
 94
 95
 96
 97
 98
 99
100
101
@overrides
def is_suspended(self) -> bool:
    """
    Checks whether ssh commands can be executed or not

    Returns:
        bool: _description_
    """
    return self.__is_suspended

restore_connection()

Restores execution of ssh commands

Source code in ark_sdk_python/common/connections/ssh/ark_pty_ssh_connection.py 
86
87
88
89
90
91
@overrides
def restore_connection(self) -> None:
    """
    Restores execution of ssh commands
    """
    self.__is_suspended = False

run_command(command)

Runs a command over ssh session, returning the result accordingly

Parameters:

Name Type Description Default
command ArkConnectionCommand

description

 required

Raises:

Type Description
ArkException

description

Returns:

Name Type Description
ArkConnectionResult ArkConnectionResult

description
Source code in ark_sdk_python/common/connections/ssh/ark_pty_ssh_connection.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@overrides
def run_command(self, command: ArkConnectionCommand) -> ArkConnectionResult:
    """
    Runs a command over ssh session, returning the result accordingly

    Args:
        command (ArkConnectionCommand): _description_

    Raises:
        ArkException: _description_

    Returns:
        ArkConnectionResult: _description_
    """
    if not self.__is_connected or self.__is_suspended:
        raise ArkException('Cannot run command while not being connected')
    self._logger.debug(f'Running command [{command.command}]')
    self.__ssh_client.sendline(command.command)
    self.__ssh_client.prompt()
    stdout = self.__ssh_client.before.decode()
    self.__ssh_client.sendline('echo $?')
    self.__ssh_client.prompt()
    exit_code_output = self.__ssh_client.before.decode()
    exit_code_output = ArkPTYSSHConnection.__ANSI_STRIPPER.sub('', exit_code_output)
    rc = int(exit_code_output.strip().splitlines()[-1])
    if rc != command.expected_rc and command.raise_on_error:
        raise ArkException(f'Failed to execute command [{command.command}] - [{rc}] - [{stdout}]')
    self._logger.debug(f'Command rc: [{rc}]')
    self._logger.debug(f'Command stdout: [{stdout}]')
    return ArkConnectionResult(stdout=stdout, rc=rc)

suspend_connection()

Suspends execution of ssh commands

Source code in ark_sdk_python/common/connections/ssh/ark_pty_ssh_connection.py 
79
80
81
82
83
84
@overrides
def suspend_connection(self) -> None:
    """
    Suspends execution of ssh commands
    """
    self.__is_suspended = True