class_docstr: Execute remote commands with Paramiko.
  
  .. seealso:: https://github.com/paramiko/paramiko
  
  This hook also lets you create ssh tunnel and serve as basis for SFTP file transfer.
  
  :param ssh_conn_id: :ref:`ssh connection id<howto/connection:ssh>` from airflow
      Connections from where all the required parameters can be fetched like
      username, password or key_file, though priority is given to the
      params passed during init.
  :param remote_host: remote host to connect
  :param username: username to connect to the remote_host
  :param password: password of the username to connect to the remote_host
  :param key_file: path to key file to use to connect to the remote_host
  :param port: port of remote host to connect (Default is paramiko SSH_PORT)
  :param conn_timeout: timeout (in seconds) for the attempt to connect to the remote_host.
      The default is 10 seconds. If provided, it will replace the `conn_timeout` which was
      predefined in the connection of `ssh_conn_id`.
  :param timeout: (Deprecated). timeout for the attempt to connect to the remote_host.
      Use conn_timeout instead.
  :param cmd_timeout: timeout (in seconds) for executing the command. The default is 10 seconds.
      Nullable, `None` means no timeout. If provided, it will replace the `cmd_timeout`
      which was predefined in the connection of `ssh_conn_id`.
  :param keepalive_interval: send a keepalive packet to remote host every
      keepalive_interval seconds
  :param banner_timeout: timeout to wait for banner from the server in seconds
  :param disabled_algorithms: dictionary mapping algorithm type to an
      iterable of algorithm identifiers, which will be disabled for the
      lifetime of the transport
  :param ciphers: list of ciphers to use in order of preference
code: def exec_ssh_client_command(
      self,
      ssh_client: paramiko.SSHClient,
      command: str,
      get_pty: bool,
      environment: dict | None,
      timeout: int | ArgNotSet | None = NOTSET,
  ) -> tuple[int, bytes, bytes]:
      self.log.info("Running command: %s", command)
  
      cmd_timeout: int | None
      if not isinstance(timeout, ArgNotSet):
          cmd_timeout = timeout
      elif not isinstance(self.cmd_timeout, ArgNotSet):
          cmd_timeout = self.cmd_timeout
      else:
          cmd_timeout = CMD_TIMEOUT
      del timeout  # Too easy to confuse with "timedout" below.
  
      # set timeout taken as params
      stdin, stdout, stderr = ssh_client.exec_command(
          command=command,
          get_pty=get_pty,
          timeout=cmd_timeout,
          environment=environment,
      )
      # get channels
      channel = stdout.channel
  
      # closing stdin
      stdin.close()
      channel.shutdown_write()
  
      agg_stdout = b""
      agg_stderr = b""
  
      # capture any initial output in case channel is closed already
      stdout_buffer_length = len(stdout.channel.in_buffer)
  
      if stdout_buffer_length > 0:
          agg_stdout += stdout.channel.recv(stdout_buffer_length)
  
      timedout = False
  
      # read from both stdout and stderr
      while (
          not channel.closed
          or channel.recv_ready()
          or channel.recv_stderr_ready()
      ):
          readq, _, _ = select([channel], [], [], cmd_timeout)
          if cmd_timeout is not None:
              timedout = not readq
          for recv in readq:
              if recv.recv_ready():
                  output = stdout.channel.recv(len(recv.in_buffer))
                  agg_stdout += output
                  for line in (
                      output.decode("utf-8", "replace")
                      .strip("\n")
                      .splitlines()
                  ):
                      self.log.info(line)
              if recv.recv_stderr_ready():
                  output = stderr.channel.recv_stderr(
                      len(recv.in_stderr_buffer)
                  )
                  agg_stderr += output
                  for line in (
                      output.decode("utf-8", "replace")
                      .strip("\n")
                      .splitlines()
                  ):
                      self.log.warning(line)
          if (
              stdout.channel.exit_status_ready()
              and not stderr.channel.recv_stderr_ready()
              and not stdout.channel.recv_ready()
          ) or timedout:
              stdout.channel.shutdown_read()
              try:
                  stdout.channel.close()
              except Exception:
                  # there is a race that when shutdown_read has been called and when
                  # you try to close the connection, the socket is already closed
                  # We should ignore such errors (but we should log them with warning)
                  self.log.warning(
                      "Ignoring exception on close", exc_info=True
                  )
              break
  
      stdout.close()
      stderr.close()
  
      if timedout:
          raise AirflowException("SSH command timed out")
  
      exit_status = stdout.channel.recv_exit_status()
  
      return exit_status, agg_stdout, agg_stderr
  
imports: from __future__ import annotations
  import os
  import warnings
  from base64 import decodebytes
  from functools import cached_property
  from io import StringIO
  from select import select
  from typing import Any, Sequence
  import paramiko
  from paramiko.config import SSH_PORT
  from sshtunnel import SSHTunnelForwarder
  from tenacity import Retrying, stop_after_attempt, wait_fixed, wait_random
  from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
  from airflow.hooks.base import BaseHook
  from airflow.utils.platform import getuser
  from airflow.utils.types import NOTSET, ArgNotSet
label_desc: Use image node_id 10 for calling the SSHHook obj's underlying member method code with example usage: obj.exec_ssh_client_command(ssh_client, command, get_pty, environment, timeout) and returns: exit_status,  agg_stdout,  agg_stderr