utils.py - dev - kotkov/eco

utils.py
90 строк3 KB
Arseny Kotkov
Replace "addr" and "host" code occurrences to "ip"
26 фев 2025, 07:40
26 фев 2025, 07:4068084d2
import time

import paramiko


class Socket:
    def __init__(self, ip=None, port=None):
        self._ip = ip
        self._port = port

    @property
    def ip(self):
        return self._ip

    @ip.setter
    def ip(self, value):
        self._ip = value

    @property
    def port(self):
        return self._port

    @port.setter
    def port(self, value):
        self._port = value


def execute_ssh_command(ip, username, password, command, port=22,
                        timeout=10):
    """
    Executes a bash command via SSH on a remote machine using password authentication.

    Args:
        hostname (str): The hostname or IP of the remote server.
        username (str): The SSH username.
        password (str): The SSH password.
        command (str): The bash command to execute.
        port (int, optional): The SSH port. Defaults to 22.
        timeout (int, optional): Timeout in seconds. Defaults to 10 seconds.

    Returns:
       tuple: (stdout, stderr, return_code)
       - stdout: standard output as string
       - stderr: standard error as string
       - return_code: exit code of the executed command as int or None on timeout

    Raises:
        paramiko.AuthenticationException: If authentication fails.
        paramiko.SSHException: If there is a general SSH error.
        TimeoutError: If the connection or command execution exceeds the timeout.
    """

    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(
        paramiko.AutoAddPolicy())  # Automatically add host key to known_hosts

    try:
        # Establish Connection
        client.connect(ip, port=port, username=username,
                       password=password, timeout=timeout)

        # Execute Command
        _, stdout, stderr = client.exec_command(command, timeout=timeout)

        # Read output and error
        stdout_str = stdout.read().decode("utf-8")
        stderr_str = stderr.read().decode("utf-8")

        # Wait for command to finish and get return code (handle timeout)
        channel = stdout.channel
        exit_status = None
        start = time.time()
        while not channel.exit_status_ready():
            if time.time() - start > timeout:
                channel.close()
                raise TimeoutError(
                    f"Timeout while waiting for command execution")
            time.sleep(0.1)
        exit_status = channel.recv_exit_status()

        return stdout_str, stderr_str, exit_status

    except paramiko.AuthenticationException as auth_err:
        raise auth_err
    except paramiko.SSHException as ssh_err:
        raise ssh_err
    except TimeoutError as timeout_err:
        raise timeout_err
    finally:
        client.close()