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
docstr: Opens an SSH connection to the remote host.
docstr_tok: ["Opens","an","SSH","connection","to","the","remote","host","."]
code: def get_conn(self) -> paramiko.SSHClient:
      """Opens an SSH connection to the remote host."""
      self.log.debug(
          "Creating SSH client for conn_id: %s", self.ssh_conn_id
      )
      client = paramiko.SSHClient()
  
      if self.allow_host_key_change:
          self.log.warning(
              "Remote Identification Change is not verified. "
              "This won't protect against Man-In-The-Middle attacks"
          )
          # to avoid BadHostKeyException, skip loading host keys
          client.set_missing_host_key_policy(
              paramiko.MissingHostKeyPolicy
          )
      else:
          client.load_system_host_keys()
  
      if self.no_host_key_check:
          self.log.warning(
              "No Host Key Verification. This won't protect against Man-In-The-Middle attacks"
          )
          client.set_missing_host_key_policy(
              paramiko.AutoAddPolicy()
          )  # nosec B507
          # to avoid BadHostKeyException, skip loading and saving host keys
          known_hosts = os.path.expanduser("~/.ssh/known_hosts")
          if not self.allow_host_key_change and os.path.isfile(
              known_hosts
          ):
              client.load_host_keys(known_hosts)
  
      elif self.host_key is not None:
          # Get host key from connection extra if it not set or None then we fallback to system host keys
          client_host_keys = client.get_host_keys()
          if self.port == SSH_PORT:
              client_host_keys.add(
                  self.remote_host,
                  self.host_key.get_name(),
                  self.host_key,
              )
          else:
              client_host_keys.add(
                  f"[{self.remote_host}]:{self.port}",
                  self.host_key.get_name(),
                  self.host_key,
              )
  
      connect_kwargs: dict[str, Any] = {
          "hostname": self.remote_host,
          "username": self.username,
          "timeout": self.conn_timeout,
          "compress": self.compress,
          "port": self.port,
          "sock": self.host_proxy,
          "look_for_keys": self.look_for_keys,
          "banner_timeout": self.banner_timeout,
      }
  
      if self.password:
          password = self.password.strip()
          connect_kwargs.update(password=password)
  
      if self.pkey:
          connect_kwargs.update(pkey=self.pkey)
  
      if self.key_file:
          connect_kwargs.update(key_filename=self.key_file)
  
      if self.disabled_algorithms:
          connect_kwargs.update(
              disabled_algorithms=self.disabled_algorithms
          )
  
      def log_before_sleep(retry_state):
          return self.log.info(
              "Failed to connect. Sleeping before retry attempt %d",
              retry_state.attempt_number,
          )
  
      for attempt in Retrying(
          reraise=True,
          wait=wait_fixed(3) + wait_random(0, 2),
          stop=stop_after_attempt(3),
          before_sleep=log_before_sleep,
      ):
          with attempt:
              client.connect(**connect_kwargs)
  
      if self.keepalive_interval:
          # MyPy check ignored because "paramiko" isn't well-typed. The `client.get_transport()` returns
          # type "Transport | None" and item "None" has no attribute "set_keepalive".
          client.get_transport().set_keepalive(self.keepalive_interval)  # type: ignore[union-attr]
  
      if self.ciphers:
          # MyPy check ignored because "paramiko" isn't well-typed. The `client.get_transport()` returns
          # type "Transport | None" and item "None" has no method `get_security_options`".
          client.get_transport().get_security_options().ciphers = self.ciphers  # type: ignore[union-attr]
  
      self.client = client
      return client
  
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 4 for calling the SSHHook obj's underlying member method code with example usage: obj.get_conn() and returns: client, self