from _typeshed import FileDescriptorOrPath
from collections.abc import Callable, Iterable, Mapping, Sequence
from logging import Logger
from socket import socket
from threading import Condition, Event, Lock, Thread
from typing import Any, Protocol, TypeAlias, type_check_only

from paramiko.auth_handler import AuthHandler, AuthOnlyHandler, _InteractiveCallback
from paramiko.channel import Channel
from paramiko.message import Message
from paramiko.packet import Packetizer
from paramiko.pkey import PKey
from paramiko.proxy import ProxyCommand
from paramiko.server import ServerInterface, SubsystemHandler
from paramiko.sftp_client import SFTPClient
from paramiko.util import ClosingContextManager

_Addr: TypeAlias = tuple[str, int]
_SocketLike: TypeAlias = str | _Addr | socket | Channel | ProxyCommand

@type_check_only
class _KexEngine(Protocol):
    def start_kex(self) -> None: ...
    def parse_next(self, ptype: int, m: Message) -> None: ...

class Transport(Thread, ClosingContextManager):
    daemon: bool
    sock: socket | Channel

    packetizer: Packetizer
    local_version: str
    remote_version: str
    local_cipher: str
    local_kex_init: bytes | None
    local_mac: str | None
    local_compression: str | None
    session_id: bytes | None
    host_key_type: str | None
    host_key: PKey | None

    kex_engine: _KexEngine | None
    H: bytes | None
    K: int | None

    initial_kex_done: bool
    in_kex: bool
    authenticated: bool
    lock: Lock

    channel_events: dict[int, Event]
    channels_seen: dict[int, bool]
    default_max_packet_size: int
    default_window_size: int

    saved_exception: Exception | None
    clear_to_send: Event
    clear_to_send_lock: Lock
    clear_to_send_timeout: float
    log_name: str
    logger: Logger
    auth_handler: AuthHandler | None
    global_response: Message | None
    completion_event: Event | None
    banner_timeout: float
    handshake_timeout: float
    auth_timeout: float
    channel_timeout: float
    disabled_algorithms: Mapping[str, Iterable[str]] | None
    server_sig_algs: bool

    server_mode: bool
    server_object: ServerInterface | None
    server_key_dict: dict[str, PKey]
    server_accepts: list[Channel]
    server_accept_cv: Condition
    subsystem_table: dict[str, tuple[type[SubsystemHandler], tuple[Any, ...], dict[str, Any]]]

    def __init__(
        self,
        sock: _SocketLike,
        default_window_size: int = 2097152,
        default_max_packet_size: int = 32768,
        disabled_algorithms: Mapping[str, Iterable[str]] | None = None,
        server_sig_algs: bool = True,
        strict_kex: bool = True,
        packetizer_class: type[Packetizer] | None = None,
    ) -> None: ...
    @property
    def preferred_ciphers(self) -> Sequence[str]: ...
    @property
    def preferred_macs(self) -> Sequence[str]: ...
    @property
    def preferred_keys(self) -> Sequence[str]: ...
    @property
    def preferred_pubkeys(self) -> Sequence[str]: ...
    @property
    def preferred_kex(self) -> Sequence[str]: ...
    @property
    def preferred_compression(self) -> Sequence[str]: ...
    def atfork(self) -> None: ...
    def get_security_options(self) -> SecurityOptions: ...
    def start_client(self, event: Event | None = None, timeout: float | None = None) -> None: ...
    def start_server(self, event: Event | None = None, server: ServerInterface | None = None) -> None: ...
    def add_server_key(self, key: PKey) -> None: ...
    def get_server_key(self) -> PKey | None: ...
    @staticmethod
    def load_server_moduli(filename: FileDescriptorOrPath | None = None) -> bool: ...
    def close(self) -> None: ...
    def get_remote_server_key(self) -> PKey: ...
    def is_active(self) -> bool: ...
    def open_session(
        self, window_size: int | None = None, max_packet_size: int | None = None, timeout: float | None = None
    ) -> Channel: ...
    def open_x11_channel(self, src_addr: _Addr | None = None) -> Channel: ...
    def open_forward_agent_channel(self) -> Channel: ...
    def open_forwarded_tcpip_channel(self, src_addr: _Addr, dest_addr: _Addr) -> Channel: ...
    def open_channel(
        self,
        kind: str,
        dest_addr: _Addr | None = None,
        src_addr: _Addr | None = None,
        window_size: int | None = None,
        max_packet_size: int | None = None,
        timeout: float | None = None,
    ) -> Channel: ...
    def request_port_forward(
        self, address: str, port: int, handler: Callable[[Channel, _Addr, _Addr], object] | None = None
    ) -> int: ...
    def cancel_port_forward(self, address: str, port: int) -> None: ...
    def open_sftp_client(self) -> SFTPClient | None: ...
    def send_ignore(self, byte_count: int | None = None) -> None: ...
    def renegotiate_keys(self) -> None: ...
    def set_keepalive(self, interval: float) -> None: ...
    def global_request(self, kind: str, data: Iterable[Any] | None = None, wait: bool = True) -> Message | None: ...
    def accept(self, timeout: float | None = None) -> Channel | None: ...
    def connect(
        self, hostkey: PKey | None = None, username: str = "", password: str | None = None, pkey: PKey | None = None
    ) -> None: ...
    def get_exception(self) -> Exception | None: ...
    def set_subsystem_handler(self, name: str, handler: type[SubsystemHandler], *larg: Any, **kwarg: Any) -> None: ...
    def is_authenticated(self) -> bool: ...
    def get_username(self) -> str | None: ...
    def get_banner(self) -> bytes | None: ...
    def auth_none(self, username: str) -> list[str]: ...
    def auth_password(self, username: str, password: str, event: Event | None = None, fallback: bool = True) -> list[str]: ...
    def auth_publickey(self, username: str, key: PKey, event: Event | None = None) -> list[str]: ...
    def auth_interactive(self, username: str, handler: _InteractiveCallback, submethods: str = "") -> list[str]: ...
    def auth_interactive_dumb(
        self, username: str, handler: _InteractiveCallback | None = None, submethods: str = ""
    ) -> list[str]: ...
    def set_log_channel(self, name: str) -> None: ...
    def get_log_channel(self) -> str: ...
    def set_hexdump(self, hexdump: bool) -> None: ...
    def get_hexdump(self) -> bool: ...
    def use_compression(self, compress: bool = True) -> None: ...
    def getpeername(self) -> tuple[str, int]: ...
    def stop_thread(self) -> None: ...
    def run(self) -> None: ...

class SecurityOptions:
    __slots__ = "_transport"
    def __init__(self, transport: Transport) -> None: ...

    @property
    def ciphers(self) -> Sequence[str]: ...
    @ciphers.setter
    def ciphers(self, x: Sequence[str]) -> None: ...

    @property
    def digests(self) -> Sequence[str]: ...
    @digests.setter
    def digests(self, x: Sequence[str]) -> None: ...

    @property
    def key_types(self) -> Sequence[str]: ...
    @key_types.setter
    def key_types(self, x: Sequence[str]) -> None: ...

    @property
    def kex(self) -> Sequence[str]: ...
    @kex.setter
    def kex(self, x: Sequence[str]) -> None: ...

    @property
    def compression(self) -> Sequence[str]: ...
    @compression.setter
    def compression(self, x: Sequence[str]) -> None: ...

class ChannelMap:
    def __init__(self) -> None: ...
    def put(self, chanid: int, chan: Channel) -> None: ...
    def get(self, chanid: int) -> Channel: ...
    def delete(self, chanid: int) -> None: ...
    def values(self) -> list[Channel]: ...
    def __len__(self) -> int: ...

class ServiceRequestingTransport(Transport):
    def ensure_session(self) -> None: ...
    def get_auth_handler(self) -> AuthOnlyHandler: ...
    def auth_password(self, username: str, password: str, fallback: bool = True) -> list[str]: ...  # type: ignore[override]
    def auth_publickey(self, username: str, key: PKey) -> list[str]: ...  # type: ignore[override]
