import asyncio

import functools

import getpass

import inspect

import io

import ipaddress

import os

import shlex

import socket

import sys

import tempfile

import time

from collections import OrderedDict

from functools import partial

from pathlib import Path

from types import TracebackType

from typing import TYPE_CHECKING, Any, AnyStr, Awaitable, Callable, Dict

from typing import Generic, List, Mapping, Optional, Sequence, Set, Tuple

from typing import Type, TypeVar, Union, cast

from typing_extensions import Protocol, Self

from .agent import SSHAgentClient, SSHAgentListener

from .auth import Auth, ClientAuth, KbdIntChallenge, KbdIntPrompts

from .auth import KbdIntResponse, PasswordChangeResponse

from .auth import get_supported_client_auth_methods, lookup_client_auth

from .auth import get_supported_server_auth_methods, lookup_server_auth

from .auth_keys import SSHAuthorizedKeys, read_authorized_keys

from .channel import SSHChannel, SSHClientChannel, SSHServerChannel

from .channel import SSHTCPChannel, SSHUNIXChannel, SSHTunTapChannel

from .channel import SSHX11Channel, SSHAgentChannel

from .client import SSHClient

from .compression import Compressor, Decompressor, get_compression_algs

from .compression import get_default_compression_algs, get_compression_params

from .compression import get_compressor, get_decompressor

from .config import ConfigPaths, SSHConfig, SSHClientConfig, SSHServerConfig

from .constants import DEFAULT_LANG, DEFAULT_PORT

from .constants import DISC_BY_APPLICATION

from .constants import EXTENDED_DATA_STDERR

from .constants import MSG_DISCONNECT, MSG_IGNORE, MSG_UNIMPLEMENTED, MSG_DEBUG

from .constants import MSG_SERVICE_REQUEST, MSG_SERVICE_ACCEPT, MSG_EXT_INFO

from .constants import MSG_CHANNEL_OPEN, MSG_CHANNEL_OPEN_CONFIRMATION

from .constants import MSG_CHANNEL_OPEN_FAILURE

from .constants import MSG_CHANNEL_FIRST, MSG_CHANNEL_LAST

from .constants import MSG_KEXINIT, MSG_NEWKEYS, MSG_KEX_FIRST, MSG_KEX_LAST

from .constants import MSG_USERAUTH_REQUEST, MSG_USERAUTH_FAILURE

from .constants import MSG_USERAUTH_SUCCESS, MSG_USERAUTH_BANNER

from .constants import MSG_USERAUTH_FIRST, MSG_USERAUTH_LAST

from .constants import MSG_GLOBAL_REQUEST, MSG_REQUEST_SUCCESS

from .constants import MSG_REQUEST_FAILURE

from .constants import OPEN_ADMINISTRATIVELY_PROHIBITED, OPEN_CONNECT_FAILED

from .constants import OPEN_UNKNOWN_CHANNEL_TYPE

from .encryption import Encryption, get_encryption_algs

from .encryption import get_default_encryption_algs

from .encryption import encryption_needs_mac

from .encryption import get_encryption_params, get_encryption

from .forward import SSHForwarder

from .gss import GSSBase, GSSClient, GSSServer, GSSError

from .kex import Kex, get_kex_algs, get_default_kex_algs

from .kex import expand_kex_algs, get_kex

from .keysign import KeySignPath, SSHKeySignKeyPair

from .keysign import find_keysign, get_keysign_keys

from .known_hosts import KnownHostsArg, match_known_hosts

from .listener import ListenKey, SSHListener

from .listener import SSHTCPClientListener, SSHUNIXClientListener

from .listener import TCPListenerFactory, UNIXListenerFactory

from .listener import create_tcp_forward_listener, create_unix_forward_listener

from .listener import create_socks_listener

from .logging import SSHLogger, logger

from .mac import get_mac_algs, get_default_mac_algs

from .misc import BytesOrStr, BytesOrStrDict, DefTuple, Env, EnvSeq, FilePath

from .misc import HostPort, IPNetwork, MaybeAwait, OptExcInfo, Options, SockAddr

from .misc import ChannelListenError, ChannelOpenError, CompressionError

from .misc import DisconnectError, ConnectionLost, HostKeyNotVerifiable

from .misc import KeyExchangeFailed, IllegalUserName, MACError

from .misc import PasswordChangeRequired, PermissionDenied, ProtocolError

from .misc import ProtocolNotSupported, ServiceNotAvailable

from .misc import TermModesArg, TermSizeArg

from .misc import async_context_manager, construct_disc_error, encode_env

from .misc import get_symbol_names, ip_address, lookup_env, map_handler_name

from .misc import parse_byte_count, parse_time_interval, split_args

from .packet import Boolean, Byte, NameList, String, UInt32, PacketDecodeError

from .packet import SSHPacket, SSHPacketHandler, SSHPacketLogger

from .pattern import WildcardPattern, WildcardPatternList

from .pkcs11 import load_pkcs11_keys

from .process import PIPE, ProcessSource, ProcessTarget

from .process import SSHServerProcessFactory, SSHCompletedProcess

from .process import SSHClientProcess, SSHServerProcess

from .public_key import CERT_TYPE_HOST, CERT_TYPE_USER, KeyImportError

from .public_key import CertListArg, IdentityListArg, KeyListArg, SigningKey

from .public_key import KeyPairListArg, X509CertPurposes, SSHKey, SSHKeyPair

from .public_key import SSHCertificate, SSHOpenSSHCertificate

from .public_key import SSHX509Certificate, SSHX509CertificateChain

from .public_key import decode_ssh_public_key, decode_ssh_certificate

from .public_key import get_public_key_algs, get_default_public_key_algs

from .public_key import get_certificate_algs, get_default_certificate_algs

from .public_key import get_x509_certificate_algs

from .public_key import get_default_x509_certificate_algs

from .public_key import load_keypairs, load_default_keypairs

from .public_key import load_public_keys, load_default_host_public_keys

from .public_key import load_certificates

from .public_key import load_identities, load_default_identities

from .saslprep import saslprep, SASLPrepError

from .server import SSHServer

from .session import DataType, SSHClientSession, SSHServerSession

from .session import SSHTCPSession, SSHUNIXSession, SSHTunTapSession

from .session import SSHClientSessionFactory, SSHTCPSessionFactory

from .session import SSHUNIXSessionFactory, SSHTunTapSessionFactory

from .sftp import MIN_SFTP_VERSION, SFTPClient, SFTPServer

from .sftp import start_sftp_client

from .stream import SSHReader, SSHWriter, SFTPServerFactory

from .stream import SSHSocketSessionFactory, SSHServerSessionFactory

from .stream import SSHClientStreamSession, SSHServerStreamSession

from .stream import SSHTCPStreamSession, SSHUNIXStreamSession

from .stream import SSHTunTapStreamSession

from .subprocess import SSHSubprocessTransport, SSHSubprocessProtocol

from .subprocess import SubprocessFactory, SSHSubprocessWritePipe

from .tuntap import SSH_TUN_MODE_POINTTOPOINT, SSH_TUN_MODE_ETHERNET

from .tuntap import SSH_TUN_UNIT_ANY, create_tuntap

from .version import __version__

from .x11 import SSHX11ClientForwarder

from .x11 import SSHX11ClientListener, SSHX11ServerListener

from .x11 import create_x11_client_listener, create_x11_server_listener

if TYPE_CHECKING:

# pylint: disable=unused-import

from .crypto import X509NamePattern

_ClientFactory = Callable[[], SSHClient]

_ServerFactory = Callable[[], SSHServer]

_ProtocolFactory = Union[_ClientFactory, _ServerFactory]

_Conn = TypeVar('_Conn', bound='SSHConnection')

_Options = TypeVar('_Options', bound='SSHConnectionOptions')

_ServerHostKeysHandler = Optional[Callable[[List[SSHKey], List[SSHKey],

List[SSHKey], List[SSHKey]],

MaybeAwait[None]]]

class _TunnelProtocol(Protocol):

"""Wait for this tunnel to close"""

class _TunnelConnectorProtocol(_TunnelProtocol, Protocol):

"""Create an outbound tunnel connection"""

class _TunnelListenerProtocol(_TunnelProtocol, Protocol):

"""Create an inbound tunnel listener"""

_AcceptHandler = Optional[Callable[['SSHConnection'], MaybeAwait[None]]]

_ErrorHandler = Optional[Callable[['SSHConnection',

Optional[Exception]], None]]

_OpenHandler = Callable[[SSHPacket], Tuple[SSHClientChannel, SSHClientSession]]

_PacketHandler = Callable[[SSHPacket], None]

_AlgsArg = DefTuple[Union[str, Sequence[str]]]

_AuthArg = DefTuple[bool]

_AuthKeysArg = DefTuple[Union[None, str, List[str], SSHAuthorizedKeys]]

_ClientHostKey = Union[SSHKeyPair, SSHKeySignKeyPair]

_ClientKeysArg = Union[KeyListArg, KeyPairListArg]

_CNAMEArg = DefTuple[Union[Sequence[str], Sequence[Tuple[str, str]]]]

_GlobalRequest = Tuple[Optional[_PacketHandler], SSHPacket, bool]

_GlobalRequestResult = Tuple[int, SSHPacket]

_KeyOrCertOptions = Mapping[str, object]

_ListenerArg = Union[bool, SSHListener]

_ProxyCommand = Optional[Union[str, Sequence[str]]]

_RequestPTY = Union[bool, str]

_TCPServerHandlerFactory = Callable[[str, int], SSHSocketSessionFactory]

_UNIXServerHandlerFactory = Callable[[], SSHSocketSessionFactory]

_TunnelConnector = Union[None, str, _TunnelConnectorProtocol]

_TunnelListener = Union[None, str, _TunnelListenerProtocol]

_VersionArg = DefTuple[BytesOrStr]

SSHAcceptHandler = Callable[[str, int], MaybeAwait[bool]]

_USERAUTH_SERVICE = b'ssh-userauth'

_CONNECTION_SERVICE = b'ssh-connection'

_MAX_BANNER_LINES = 1024

_MAX_BANNER_LINE_LEN = 8192

_MAX_VERSION_LINE_LEN = 255

_MAX_USERNAME_LEN = 1024

_DEFAULT_REKEY_BYTES = 1 << 30 # 1 GiB

_DEFAULT_REKEY_SECONDS = 3600 # 1 hour

_DEFAULT_LOGIN_TIMEOUT = 120 # 2 minutes

_DEFAULT_KEEPALIVE_INTERVAL = 0 # disabled by default

_DEFAULT_KEEPALIVE_COUNT_MAX = 3

_DEFAULT_WINDOW = 2*1024*1024 # 2 MiB

_DEFAULT_MAX_PKTSIZE = 32768 # 32 kiB

_DEFAULT_LINE_HISTORY = 1000 # 1000 lines

_DEFAULT_MAX_LINE_LENGTH = 1024 # 1024 characters

async def _canonicalize_host(loop: asyncio.AbstractEventLoop,

return None

async def _open_proxy(

return conn

async def _open_tunnel(tunnels: object, options: _Options,

return None

async def _connect(options: _Options, config: DefTuple[ConfigPaths],

await conn.wait_closed()

async def _listen(options: _Options, config: DefTuple[ConfigPaths],

return SSHAcceptor(server, options)

def _validate_version(version: DefTuple[BytesOrStr]) -> bytes:

return version

def _expand_algs(alg_type: str, algs: str,

return matched

def _select_algs(alg_type: str, algs: _AlgsArg, config_algs: _AlgsArg,

raise ValueError(f'No {alg_type} algorithms selected')

def _select_host_key_algs(algs: _AlgsArg, config_algs: _AlgsArg,

possible_algs, default_algs)

def _validate_algs(config: SSHConfig, kex_algs_arg: _AlgsArg,

return kex_algs, enc_algs, mac_algs, cmp_algs, sig_algs

class SSHAcceptor:

self._options.update(**kwargs)

class SSHConnection(SSHPacketHandler, asyncio.Protocol):