Transport API¶

class HandledMessage:
    """Sentinel for messages handled internally by the transport."""

    msg_type = 0
    _data = b""

class PacketProfiler:
    """
    Optional per-stage packet timing. Activated by setting SPINDLEX_PROFILE=1.

    Records build / encrypt / socket-write latencies for every packet sent
    through _send_message, then reports median and P95 via summary().
    """

    __slots__ = ("build", "encrypt", "write")

    def __init__(self) -> None:
        self.build: list[float] = []
        self.encrypt: list[float] = []
        self.write: list[float] = []

    def record(self, build_s: float, encrypt_s: float, write_s: float) -> None:
        self.build.append(build_s)
        self.encrypt.append(encrypt_s)
        self.write.append(write_s)

    def reset(self) -> None:
        self.build.clear()
        self.encrypt.clear()
        self.write.clear()

    def summary(self, pct: int = 95) -> str:
        n = len(self.build)
        if n == 0:
            return "PacketProfiler: no samples"

        def _stats(samples: list[float]) -> tuple[float, float]:
            s = sorted(samples)
            med = statistics.median(s)
            idx = max(0, int(len(s) * pct / 100) - 1)
            return med * 1e6, s[idx] * 1e6

        b_med, b_p = _stats(self.build)
        e_med, e_p = _stats(self.encrypt)
        w_med, w_p = _stats(self.write)
        total_med = b_med + e_med + w_med
        total_p = b_p + e_p + w_p
        return (
            f"PacketProfiler ({n} packets) - median / P{pct} in µs:\n"
            f"  build:   {b_med:7.1f} / {b_p:7.1f}\n"
            f"  encrypt: {e_med:7.1f} / {e_p:7.1f}\n"
            f"  write:   {w_med:7.1f} / {w_p:7.1f}\n"
            f"  total:   {total_med:7.1f} / {total_p:7.1f}"
        )

class Transport:
    """
    SSH transport layer implementation.

    Manages SSH protocol handshake, key exchange, authentication,
    and secure packet transmission according to RFC 4251-4254.
    """

    def __init__(
        self,
        sock: socket.socket,
        rekey_bytes_limit: Optional[int] = None,
        rekey_time_limit: Optional[float] = None,
    ) -> None:
        """
        Initialize transport with socket connection.

        Args:
            sock: Connected socket for SSH communication
            rekey_bytes_limit: Number of bytes before rekeying (default: 1GB)
            rekey_time_limit: Seconds before rekeying (default: 1 hour)
        """
        self._socket = sock
        self._active = False
        self._server_mode = False
        self._channels: dict[int, Channel] = {}
        self._next_channel_id = 0

        # Connection state
        self._authenticated = False
        self._session_id: Optional[bytes] = None
        self._server_version: Optional[str] = None
        self._client_version: Optional[str] = None

        # Crypto state
        self._crypto_backend = default_crypto_backend
        self._encryption_key_c2s: Optional[bytes] = None
        self._encryption_key_s2c: Optional[bytes] = None
        self._mac_key_c2s: Optional[bytes] = None
        self._mac_key_s2c: Optional[bytes] = None
        self._iv_c2s: Optional[bytes] = None
        self._iv_s2c: Optional[bytes] = None
        self._cipher_c2s: Optional[str] = None
        self._cipher_s2c: Optional[str] = None
        self._mac_c2s: Optional[str] = None
        self._mac_s2c: Optional[str] = None

        # Active crypto state (updated only on NEWKEYS)
        self._mac_key_in_active: Optional[bytes] = None
        self._mac_key_out_active: Optional[bytes] = None
        self._mac_in_active: Optional[str] = None
        self._mac_out_active: Optional[str] = None

        # Rekeying policy (configurable)
        self._rekey_bytes_limit = rekey_bytes_limit or (
            1024 * 1024 * 1024
        )  # 1GB default
        self._rekey_time_limit = rekey_time_limit or 3600  # 1 hour default

        # Cipher instances
        self._encryptor_instance: Optional[Any] = None
        self._decryptor_instance: Optional[Any] = None
        self._chacha20_key_out: Optional[bytes] = None
        self._chacha20_key_in: Optional[bytes] = None

        self._sequence_number_in = 0
        self._sequence_number_out = 0
        self._packet_buffer = bytearray()
        # Lock ordering contract - to prevent ABBA deadlock, always acquire in this order:
        #   1. _read_lock  (socket-level serialisation; held while reading a packet)
        #   2. _lock       (state-level serialisation; held while mutating transport state)
        # Never acquire _read_lock while already holding _lock.
        self._lock = threading.RLock()
        self._read_lock = threading.RLock()
        self._kex_condition = threading.Condition(self._lock)
        self._server_host_key_blob: Optional[bytes] = None

        self._kex_in_progress = False
        self._kex = KeyExchange(self)
        self._bytes_since_rekey = 0
        self._last_rekey_time = time.time()

        # Timeouts
        self._connect_timeout: float = float(DEFAULT_CONNECT_TIMEOUT)
        self._auth_timeout: float = float(DEFAULT_AUTH_TIMEOUT)

        # Authentication state
        self._userauth_service_requested = False

        # Server interface for authentication callbacks
        self._server_interface: Optional[Any] = None

        # Port forwarding
        self._port_forwarding_manager: Optional[PortForwardingManager] = None

        # Message dispatching
        self._message_queue: deque[Message] = deque()
        self._timeout = 10.0

        self._logger = logging.getLogger(__name__)
        self._strict_kex = False
        self._stop_event = threading.Event()
        self._kex_thread: Optional[threading.Thread] = None
        self._server_key: Optional[Any] = None

        self._buffer_size = 32768
        env_buffer_size = os.environ.get("SPINDLEX_BUFFER_SIZE")
        if env_buffer_size:
            try:
                self._buffer_size = max(4096, int(env_buffer_size))
            except (ValueError, TypeError):
                pass

        self._profiler: Optional[PacketProfiler] = (
            PacketProfiler() if os.environ.get("SPINDLEX_PROFILE") == "1" else None
        )

    def get_timeout(self) -> Optional[float]:
        """
        Get transport timeout.

        Returns:
            Current timeout in seconds, or None if no timeout
        """
        return self._socket.gettimeout()

    def set_timeout(self, timeout: float) -> None:
        """Set default timeout for transport operations."""
        self._timeout = timeout
        if self._socket:
            self._socket.settimeout(timeout)

    def set_rekey_policy(
        self, bytes_limit: Optional[int] = None, time_limit: Optional[float] = None
    ) -> None:
        """
        Configure rekeying thresholds.

        Args:
            bytes_limit: Number of bytes before rekeying (default: 1GB)
            time_limit: Seconds before rekeying (default: 1 hour)
        """
        if bytes_limit is not None:
            self._rekey_bytes_limit = bytes_limit
        if time_limit is not None:
            self._rekey_time_limit = time_limit

    def start_client(self, timeout: Optional[float] = None) -> None:
        """
        Start SSH client transport.

        Args:
            timeout: Handshake timeout in seconds

        Raises:
            TransportException: If client start fails
        """
        if timeout is not None:
            self._connect_timeout = timeout

        try:
            with self._read_lock:
                with self._lock:
                    if self._active:
                        raise TransportException("Transport already active")

                    self._server_mode = False
                    self._client_version = create_version_string()

                    # Set socket timeout for handshake
                    old_timeout = self._socket.gettimeout()
                    self._socket.settimeout(self._connect_timeout)

                    try:
                        # Perform SSH handshake
                        self._logger.debug("Starting handshake...")
                        self._do_handshake()
                        self._logger.debug("Handshake complete.")

                    finally:
                        # Restore original socket timeout
                        try:
                            if self._socket and self._socket.fileno() != -1:
                                self._socket.settimeout(old_timeout)
                        except (OSError, AttributeError):
                            pass

                # Start key exchange (WITHOUT holding _lock)
                self._logger.debug("Starting KEX...")
                self._start_kex()
                self._logger.debug("KEX complete.")

                with self._lock:
                    self._active = True

        except (OSError, struct.error, SSHException) as e:
            self.close()
            if isinstance(e, SSHException):
                raise
            raise TransportException(f"Client start failed: {e}") from e

    def start_server(self, server_key: Any, timeout: Optional[float] = None) -> None:
        """
        Start SSH server transport.

        Args:
            server_key: Server's private key
            timeout: Handshake timeout in seconds

        Raises:
            TransportException: If server start fails
        """
        if timeout is not None:
            self._connect_timeout = timeout

        try:
            with self._read_lock:
                with self._lock:
                    if self._active:
                        raise TransportException("Transport already active")

                    self._server_mode = True
                    self._server_key = server_key
                    self._server_version = create_version_string()

                    # Set socket timeout for handshake
                    old_timeout = self._socket.gettimeout()
                    self._socket.settimeout(self._connect_timeout)

                    try:
                        # Perform SSH handshake
                        self._logger.debug("Starting handshake...")
                        self._do_handshake()
                        self._logger.debug("Handshake complete.")

                    finally:
                        # Restore original socket timeout
                        try:
                            if self._socket and self._socket.fileno() != -1:
                                self._socket.settimeout(old_timeout)
                        except (OSError, AttributeError):
                            pass

                # Start key exchange (WITHOUT holding _lock)
                self._logger.debug("Starting KEX...")
                self._start_kex()
                self._logger.debug("KEX complete.")

                with self._lock:
                    self._active = True

        except (OSError, struct.error, SSHException) as e:
            self.close()
            if isinstance(e, SSHException):
                raise
            raise TransportException(f"Server start failed: {e}") from e

    def auth_password(self, username: str, password: str) -> bool:
        """
        Authenticate using password.

        Args:
            username: Username for authentication
            password: Password for authentication

        Returns:
            True if authentication successful

        Raises:
            AuthenticationException: If authentication fails
        """
        if not self._active:
            raise AuthenticationException("Transport not active")

        if self._authenticated:
            return True

        # Request ssh-userauth service if not already done
        if not self._userauth_service_requested:
            self._request_userauth_service()

        from ..auth.password import PasswordAuth

        auth = PasswordAuth(self)
        msg = auth.authenticate(username, password)
        return self._handle_auth_response_message(msg)

    def auth_publickey(self, username: str, key: Any) -> bool:
        """
        Authenticate using public key.

        Args:
            username: Username for authentication
            key: Private key for authentication

        Returns:
            True if authentication successful

        Raises:
            AuthenticationException: If authentication fails
        """
        if not self._active:
            raise AuthenticationException("Transport not active")

        if self._authenticated:
            return True

        # Request ssh-userauth service if not already done
        if not self._userauth_service_requested:
            self._request_userauth_service()

        from ..auth.publickey import PublicKeyAuth

        auth = PublicKeyAuth(self)
        msg = auth.authenticate(username, key)
        return self._handle_auth_response_message(msg)

    def auth_keyboard_interactive(self, username: str, handler: Any) -> bool:
        """
        Authenticate using keyboard-interactive method.

        Args:
            username: Username for authentication
            handler: Callback function to handle prompts

        Returns:
            True if authentication successful

        Raises:
            AuthenticationException: If authentication fails
        """
        if not self._active:
            raise AuthenticationException("Transport not active")

        if self._authenticated:
            return True

        try:
            # Request ssh-userauth service if not already done
            if not self._userauth_service_requested:
                self._request_userauth_service()

            from ..auth.keyboard_interactive import KeyboardInteractiveAuth

            # Send initial keyboard-interactive request
            auth_request = UserAuthRequestMessage(
                username=username,
                service=SERVICE_CONNECTION,
                method=AUTH_KEYBOARD_INTERACTIVE,
                method_data=self._build_keyboard_interactive_data(),
            )

            self._send_message(auth_request)

            # Perform interactive authentication
            ki_auth = KeyboardInteractiveAuth(self)
            result = ki_auth.authenticate(username, handler)

            if result:
                self._authenticated = True

            return result

        except (OSError, struct.error, SSHException) as e:
            if isinstance(e, AuthenticationException):
                raise
            raise AuthenticationException(
                f"Keyboard-interactive authentication failed: {e}"
            )

    def _build_keyboard_interactive_data(self) -> bytes:
        """Build keyboard-interactive authentication method data."""
        data = bytearray()
        data.extend(write_string(""))  # language tag
        data.extend(write_string(""))  # submethods
        return bytes(data)

    def auth_gssapi(
        self,
        username: str,
        gss_host: Optional[str] = None,
        gss_deleg_creds: bool = False,
    ) -> bool:
        """
        Authenticate using GSSAPI method.

        Args:
            username: Username for authentication
            gss_host: GSSAPI hostname (optional)
            gss_deleg_creds: Whether to delegate credentials

        Returns:
            True if authentication successful

        Raises:
            AuthenticationException: If authentication fails
        """
        if not self._active:
            raise AuthenticationException("Transport not active")

        if self._authenticated:
            return True

        try:
            from ..auth.gssapi import GSSAPIAuth

            # Create GSSAPI authenticator
            gssapi_auth = GSSAPIAuth(self)

            # Perform GSSAPI authentication
            result = gssapi_auth.authenticate(username, gss_host, gss_deleg_creds)

            # Clean up GSSAPI resources
            gssapi_auth.cleanup()

            return result

        except ImportError:
            raise AuthenticationException("GSSAPI authentication not available")
        except (OSError, struct.error, SSHException) as e:
            if isinstance(e, AuthenticationException):
                raise
            raise AuthenticationException(f"GSSAPI authentication failed: {e}") from e

    def _request_userauth_service(self) -> None:
        """Request ssh-userauth service."""
        if self._userauth_service_requested:
            return

        service_request = ServiceRequestMessage(SERVICE_USERAUTH)
        self._send_message(service_request)

        # Wait for service accept
        msg = self._expect_message(MSG_SERVICE_ACCEPT)
        if not isinstance(msg, ServiceAcceptMessage):
            raise AuthenticationException(
                f"Expected SERVICE_ACCEPT, got {type(msg).__name__}"
            )

        if msg.service_name != SERVICE_USERAUTH:
            raise AuthenticationException(f"Service not accepted: {msg.service_name}")

        self._userauth_service_requested = True

    def _handle_auth_response_message(self, msg: Message) -> bool:
        """
        Handle a specific authentication response message.

        Args:
            msg: MSG_USERAUTH_SUCCESS or MSG_USERAUTH_FAILURE

        Returns:
            True if success

        Raises:
            AuthenticationException: If failure or unexpected message
        """
        if isinstance(msg, UserAuthSuccessMessage):
            self._authenticated = True
            return True
        elif isinstance(msg, UserAuthFailureMessage):
            # Check if partial success
            if msg.partial_success:
                raise AuthenticationException(
                    f"Partial success - additional methods required: {', '.join(msg.authentications)}"
                )
            else:
                return False
        elif msg.msg_type == MSG_USERAUTH_SUCCESS:
            self._authenticated = True
            return True
        elif msg.msg_type == MSG_USERAUTH_FAILURE:
            # Re-unpack as UserAuthFailureMessage to check partial_success
            auth_msg: UserAuthFailureMessage
            if not isinstance(msg, UserAuthFailureMessage):
                auth_msg = UserAuthFailureMessage.unpack(msg.pack())  # type: ignore[assignment]
            else:
                auth_msg = msg

            if auth_msg.partial_success:
                raise AuthenticationException(
                    f"Partial success - additional methods required: {', '.join(auth_msg.authentications)}"
                )
            return False
        else:
            raise AuthenticationException(
                f"Unexpected authentication response: {type(msg).__name__} (type {msg.msg_type})"
            )

    def _handle_auth_response(self) -> bool:
        """Receive and handle authentication response message."""
        msg = self._expect_message(MSG_USERAUTH_SUCCESS, MSG_USERAUTH_FAILURE)
        return self._handle_auth_response_message(msg)

    def open_channel(
        self, kind: str, dest_addr: Optional[tuple[str, int]] = None
    ) -> Channel:
        """
        Open new SSH channel.

        Args:
            kind: Channel type (session, direct-tcpip, etc.)
            dest_addr: Destination address for forwarding channels

        Returns:
            New Channel instance

        Raises:
            TransportException: If channel creation fails
        """
        if not self._active:
            raise TransportException("Transport not active")

        if not self._authenticated:
            raise TransportException("Transport not authenticated")

        with self._lock:
            # Check channel limit
            if len(self._channels) >= MAX_CHANNELS:
                raise TransportException("Maximum number of channels reached")

            # Find next available channel ID (recycling IDs)
            channel_id = self._next_channel_id
            while channel_id in self._channels:
                channel_id = (channel_id + 1) % MAX_CHANNELS

            self._next_channel_id = (channel_id + 1) % MAX_CHANNELS

            # Create channel instance
            channel = Channel(self, channel_id)

            # Register channel BEFORE sending open request to avoid race
            self._channels[channel_id] = channel

        try:
            # Build channel open message
            type_specific_data = b""
            if kind == CHANNEL_DIRECT_TCPIP and dest_addr:
                type_specific_data = self._build_direct_tcpip_data(dest_addr)

            open_msg = ChannelOpenMessage(
                channel_type=kind,
                sender_channel=channel_id,
                initial_window_size=DEFAULT_WINDOW_SIZE,
                maximum_packet_size=DEFAULT_MAX_PACKET_SIZE,
                type_specific_data=type_specific_data,
            )

            # Send channel open request
            self._send_message(open_msg)

            # Wait for response (CRITICAL: release lock before calling _expect_message)
            response = self._expect_message(
                MSG_CHANNEL_OPEN_CONFIRMATION,
                MSG_CHANNEL_OPEN_FAILURE,
                channel_id=channel_id,
            )

            if isinstance(response, ChannelOpenConfirmationMessage):
                # Channel opened successfully
                with self._lock:
                    channel._remote_channel_id = response.sender_channel
                    channel._remote_window_size = response.initial_window_size
                    channel._remote_max_packet_size = response.maximum_packet_size
                    channel._local_window_size = DEFAULT_WINDOW_SIZE
                    channel._local_max_packet_size = DEFAULT_MAX_PACKET_SIZE

                return channel

            elif isinstance(response, ChannelOpenFailureMessage):
                # Channel open failed - remove from channels
                with self._lock:
                    if channel_id in self._channels:
                        del self._channels[channel_id]
                raise TransportException(
                    f"Channel open failed: {response.description} (code: {response.reason_code})"
                )

            else:
                # Unexpected response - remove from channels
                with self._lock:
                    if channel_id in self._channels:
                        del self._channels[channel_id]
                raise TransportException(
                    f"Unexpected response to channel open: {type(response).__name__}"
                )

        except (OSError, struct.error, SSHException) as e:
            # Cleanup on error
            with self._lock:
                if channel_id in self._channels:
                    del self._channels[channel_id]
            if isinstance(e, SSHException):
                raise
            raise TransportException(f"Failed to open channel: {e}") from e

    def _build_direct_tcpip_data(self, dest_addr: tuple[str, int]) -> bytes:
        """Build type-specific data for direct-tcpip channel."""
        try:
            sockname = self._socket.getsockname()
            originator_ip = sockname[0]
            originator_port = sockname[1]
        except OSError:
            originator_ip = "127.0.0.1"
            originator_port = 0
        data = bytearray()
        data.extend(write_string(dest_addr[0]))  # destination host
        data.extend(write_uint32(dest_addr[1]))  # destination port
        data.extend(write_string(originator_ip))  # originator IP
        data.extend(write_uint32(originator_port))  # originator port (RFC 4254 §7.2)
        return bytes(data)

    def _close_channel(self, channel_id: int) -> None:
        """
        Close channel and remove from channels dict.

        Args:
            channel_id: Channel ID to close
        """
        with self._lock:
            if channel_id in self._channels:
                channel = self._channels[channel_id]

                # Send channel close message if not already closed
                if not channel.closed and channel._remote_channel_id is not None:
                    try:
                        close_msg = ChannelCloseMessage(channel._remote_channel_id)
                        self._send_message(close_msg)
                    except (OSError, TransportException):
                        pass  # Ignore errors during close

                # Remove from channels dict
                del self._channels[channel_id]

    def _handle_channel_message(self, msg: Message) -> None:
        """
        Handle channel-related messages.

        Args:
            msg: Channel message to handle
        """
        # Handle messages that don't have a recipient_channel field first
        if msg.msg_type == MSG_CHANNEL_OPEN:
            self._handle_channel_open(msg)
            return

        if msg.msg_type == MSG_GLOBAL_REQUEST:
            self._handle_global_request(msg)
            return

        # Other channel messages (91-100) have recipient_channel as first uint32
        recipient_channel = None
        if len(msg._data) >= 4:
            recipient_channel, _ = read_uint32(msg._data, 0)

        if recipient_channel is not None:
            channel = None
            with self._lock:
                channel = self._channels.get(recipient_channel)

            if channel:
                try:
                    if msg.msg_type == MSG_CHANNEL_DATA:
                        self._handle_channel_data(msg)
                    elif msg.msg_type == MSG_CHANNEL_EXTENDED_DATA:
                        self._handle_channel_extended_data(msg)
                    elif msg.msg_type == MSG_CHANNEL_EOF:
                        self._handle_channel_eof(msg)
                    elif msg.msg_type == MSG_CHANNEL_CLOSE:
                        self._handle_channel_close(msg)
                    elif msg.msg_type == MSG_CHANNEL_WINDOW_ADJUST:
                        self._handle_channel_window_adjust(msg)
                    elif msg.msg_type == MSG_CHANNEL_SUCCESS:
                        self._handle_channel_success(msg)
                    elif msg.msg_type == MSG_CHANNEL_FAILURE:
                        self._handle_channel_failure(msg)
                    elif msg.msg_type == MSG_CHANNEL_REQUEST:
                        self._handle_channel_request(msg)
                    else:
                        self._logger.debug(
                            f"Unhandled channel message type {msg.msg_type} for channel {recipient_channel}"
                        )
                except Exception as e:
                    self._logger.error(
                        f"Error handling channel message {msg.msg_type}: {e}"
                    )
            else:
                self._logger.debug(
                    f"Message type {msg.msg_type} for unknown channel {recipient_channel}"
                )

    def _handle_channel_open(self, msg: Message) -> None:
        """
        Handle incoming channel open request.

        Args:
            msg: Channel open message
        """
        # Parse the remote sender_channel up-front so we always have a valid
        # id to reference in a failure response, even if later fields are
        # malformed. If even this fails we cannot reply safely - the peer
        # will time out, which is the expected behaviour for a garbage frame.
        sender_channel: Optional[int] = None
        try:
            if not isinstance(msg, ChannelOpenMessage):
                msg = ChannelOpenMessage.unpack(msg.pack())

            channel_type = msg.channel_type  # type: ignore[attr-defined]
            sender_channel = msg.sender_channel  # type: ignore[attr-defined]
            initial_window_size = msg.initial_window_size  # type: ignore[attr-defined]
            maximum_packet_size = msg.maximum_packet_size  # type: ignore[attr-defined]
            type_specific_data = msg.type_specific_data  # type: ignore[attr-defined]

            if channel_type == CHANNEL_SESSION:
                self._handle_session_open(
                    sender_channel, initial_window_size, maximum_packet_size
                )
            elif channel_type == CHANNEL_FORWARDED_TCPIP:
                self._handle_forwarded_tcpip_open(
                    sender_channel,
                    initial_window_size,
                    maximum_packet_size,
                    type_specific_data,
                )
            else:
                # Unknown channel type - send failure
                failure_msg = ChannelOpenFailureMessage(
                    recipient_channel=sender_channel,
                    reason_code=SSH_OPEN_UNKNOWN_CHANNEL_TYPE,
                    description=f"Unknown channel type: {channel_type}",
                    language="",
                )
                self._send_message(failure_msg)

        except (
            OSError,
            struct.error,
            ValueError,
            UnicodeDecodeError,
            SSHException,
        ) as e:
            # Do not leak internal exception details to the remote peer.
            self._logger.error("Channel open failed: %s", e)
            if sender_channel is None:
                # No valid sender id parsed - can't send a targeted failure.
                return
            try:
                failure_msg = ChannelOpenFailureMessage(
                    recipient_channel=sender_channel,
                    reason_code=SSH_OPEN_CONNECT_FAILED,
                    description="Channel open failed",
                    language="",
                )
                self._send_message(failure_msg)
            except (TransportException, OSError) as send_err:
                self._logger.debug(
                    "Could not send channel open failure response: %s", send_err
                )

    def _handle_session_open(
        self, sender_channel: int, initial_window_size: int, maximum_packet_size: int
    ) -> None:
        """Handle session channel open request."""
        # 1. Check with server interface
        if self._server_interface:
            result = self._server_interface.check_channel_request(
                CHANNEL_SESSION, sender_channel
            )
            if result != 0:
                failure_msg = ChannelOpenFailureMessage(
                    sender_channel, result, "Administratively prohibited", ""
                )
                self._send_message(failure_msg)
                return

        # 2. Create channel
        with self._lock:
            if len(self._channels) >= MAX_CHANNELS:
                failure_msg = ChannelOpenFailureMessage(
                    sender_channel,
                    SSH_OPEN_RESOURCE_SHORTAGE,
                    "Too many open channels",
                    "",
                )
                self._send_message(failure_msg)
                return

            channel_id = self._next_channel_id
            self._next_channel_id = (self._next_channel_id + 1) % MAX_CHANNELS

            channel = Channel(self, channel_id)
            channel._remote_channel_id = sender_channel
            channel._remote_window_size = initial_window_size
            channel._remote_max_packet_size = maximum_packet_size
            channel._local_window_size = DEFAULT_WINDOW_SIZE
            channel._local_max_packet_size = MAX_PACKET_SIZE
            self._channels[channel_id] = channel

        # 3. Send confirmation
        confirm_msg = ChannelOpenConfirmationMessage(
            recipient_channel=sender_channel,
            sender_channel=channel_id,
            initial_window_size=channel._local_window_size,
            maximum_packet_size=channel._local_max_packet_size,
        )
        self._send_message(confirm_msg)

        # 4. Notify server interface
        if self._server_interface:
            self._server_interface.on_channel_opened(channel)

    def _handle_forwarded_tcpip_open(
        self,
        sender_channel: int,
        initial_window_size: int,
        maximum_packet_size: int,
        type_specific_data: bytes,
    ) -> None:
        """
        Handle forwarded-tcpip channel open.

        Args:
            sender_channel: Remote channel ID
            initial_window_size: Initial window size
            maximum_packet_size: Maximum packet size
            type_specific_data: Channel type specific data
        """
        try:
            # Parse forwarded-tcpip data
            offset = 0
            connected_address_bytes, offset = read_string(type_specific_data, offset)
            connected_port, offset = read_uint32(type_specific_data, offset)
            originator_address_bytes, offset = read_string(type_specific_data, offset)
            originator_port, offset = read_uint32(type_specific_data, offset)

            connected_address = connected_address_bytes.decode(SSH_STRING_ENCODING)
            originator_address = originator_address_bytes.decode(SSH_STRING_ENCODING)

            # Create local channel
            with self._lock:
                if len(self._channels) >= MAX_CHANNELS:
                    failure_msg = ChannelOpenFailureMessage(
                        sender_channel,
                        SSH_OPEN_RESOURCE_SHORTAGE,
                        "Too many open channels",
                        "",
                    )
                    self._send_message(failure_msg)
                    return

                channel_id = self._next_channel_id
                self._next_channel_id = (self._next_channel_id + 1) % MAX_CHANNELS

                channel = Channel(self, channel_id)
                channel._remote_channel_id = sender_channel
                channel._remote_window_size = initial_window_size
                channel._remote_max_packet_size = maximum_packet_size
                channel._local_window_size = DEFAULT_WINDOW_SIZE
                channel._local_max_packet_size = DEFAULT_MAX_PACKET_SIZE

                self._channels[channel_id] = channel

            # Send confirmation
            confirm_msg = ChannelOpenConfirmationMessage(
                recipient_channel=sender_channel,
                sender_channel=channel_id,
                initial_window_size=DEFAULT_WINDOW_SIZE,
                maximum_packet_size=DEFAULT_MAX_PACKET_SIZE,
                type_specific_data=b"",
            )
            self._send_message(confirm_msg)

            # Handle the forwarded connection
            if self._port_forwarding_manager:
                origin_addr = (originator_address, originator_port)
                dest_addr = (connected_address, connected_port)
                self._port_forwarding_manager.handle_forwarded_connection(
                    channel, origin_addr, dest_addr
                )

        except (
            OSError,
            struct.error,
            ValueError,
            UnicodeDecodeError,
            SSHException,
        ) as e:
            # Send failure response
            failure_msg = ChannelOpenFailureMessage(
                recipient_channel=sender_channel,
                reason_code=SSH_OPEN_CONNECT_FAILED,
                description=f"Forwarded connection failed: {e}",
                language="",
            )
            self._send_message(failure_msg)

    def _handle_channel_data(self, msg: Message) -> None:
        """Handle channel data message."""
        if isinstance(msg, ChannelDataMessage):
            channel = None
            with self._lock:
                channel = self._channels.get(msg.recipient_channel)

            if channel:
                channel._handle_data(msg.data)
            else:
                self._logger.debug(f"Data for UNKNOWN channel {msg.recipient_channel}")

    def _handle_channel_extended_data(self, msg: Message) -> None:
        """Handle channel extended data message."""
        # Parse extended data message
        data = msg._data
        offset = 0
        recipient_channel, offset = read_uint32(data, offset)
        data_type, offset = read_uint32(data, offset)
        message_data, offset = read_string(data, offset)

        channel = None
        with self._lock:
            channel = self._channels.get(recipient_channel)

        if channel:
            channel._handle_extended_data(data_type, message_data)

    def _handle_channel_eof(self, msg: Message) -> None:
        """Handle channel EOF message."""
        recipient_channel, _ = read_uint32(msg._data, 0)

        channel = None
        with self._lock:
            channel = self._channels.get(recipient_channel)

        if channel:
            channel._handle_eof()

    def _handle_channel_close(self, msg: Message) -> None:
        """Handle channel close message."""
        if isinstance(msg, ChannelCloseMessage):
            channel = None
            with self._lock:
                channel = self._channels.get(msg.recipient_channel)

            if channel:
                channel._handle_close()
                # Remove from channels dict
                with self._lock:
                    if msg.recipient_channel in self._channels:
                        del self._channels[msg.recipient_channel]

    def _handle_channel_window_adjust(self, msg: Message) -> None:
        """Handle channel window adjust message."""
        data = msg._data
        offset = 0
        recipient_channel, offset = read_uint32(data, offset)
        bytes_to_add, offset = read_uint32(data, offset)

        channel = None
        with self._lock:
            channel = self._channels.get(recipient_channel)

        if channel:
            channel._handle_window_adjust(bytes_to_add)

    def _handle_channel_success(self, msg: Message) -> None:
        """Handle channel success message."""
        recipient_channel, _ = read_uint32(msg._data, 0)

        channel = None
        with self._lock:
            channel = self._channels.get(recipient_channel)

        if channel:
            channel._handle_request_success()

    def _handle_channel_failure(self, msg: Message) -> None:
        """Handle channel failure message."""
        recipient_channel, _ = read_uint32(msg._data, 0)

        channel = None
        with self._lock:
            channel = self._channels.get(recipient_channel)

        if channel:
            channel._handle_request_failure()

    def _handle_channel_request(self, msg: Message) -> None:
        """Handle channel request message."""
        # Parse channel request message
        data = msg._data
        offset = 0
        recipient_channel, offset = read_uint32(data, offset)
        request_type_bytes, offset = read_string(data, offset)
        want_reply, offset = read_boolean(data, offset)

        request_type = request_type_bytes.decode(SSH_STRING_ENCODING)
        request_data = bytes(data[offset:]) if offset < len(data) else b""

        channel = None
        with self._lock:
            channel = self._channels.get(recipient_channel)

        if channel:
            # Handle channel request
            success = channel._handle_request(request_type, request_data)

            # Send reply if requested
            if want_reply:
                if success:
                    reply_msg = Message(MSG_CHANNEL_SUCCESS)
                else:
                    reply_msg = Message(MSG_CHANNEL_FAILURE)

                if channel._remote_channel_id is not None:
                    reply_msg.add_uint32(channel._remote_channel_id)
                    self._send_message(reply_msg)

    def _handle_exit_signal_request(self, channel: Channel, data: bytes) -> None:
        """Handle exit signal request data."""
        try:
            offset = 0
            signal_name_bytes, offset = read_string(data, offset)
            core_dumped, offset = read_boolean(data, offset)
            error_message_bytes, offset = read_string(data, offset)
            language_tag_bytes, offset = read_string(data, offset)

            signal_name = signal_name_bytes.decode(SSH_STRING_ENCODING)
            error_message = error_message_bytes.decode(
                SSH_STRING_ENCODING, errors="replace"
            )
            language_tag = language_tag_bytes.decode(
                SSH_STRING_ENCODING, errors="replace"
            )

            channel._handle_exit_signal(
                signal_name, core_dumped, error_message, language_tag
            )
        except (struct.error, ValueError, UnicodeDecodeError, IndexError):
            # Ignore malformed exit signal data
            pass

    def _send_channel_data(self, channel_id: int, data: bytes) -> None:
        """
        Send data through channel.

        Args:
            channel_id: Local channel ID
            data: Data to send
        """
        with self._lock:
            if channel_id not in self._channels:
                raise TransportException(f"Channel {channel_id} not found")

            channel = self._channels[channel_id]

            # Defensive check only - the channel layer owns _remote_window_size
            # and is the single decrement point (mirrors the async path in
            # AsyncChannel.send). Transport must not decrement here, otherwise
            # the window would be debited twice per send.
            if len(data) > channel._remote_window_size:
                raise TransportException("Remote window size exceeded")

            if len(data) > channel._remote_max_packet_size:
                raise TransportException("Remote max packet size exceeded")

            if channel._remote_channel_id is not None:
                data_msg = ChannelDataMessage(channel._remote_channel_id, data)
                self._send_message(data_msg)

    def _send_channel_window_adjust(self, channel_id: int, bytes_to_add: int) -> None:
        """
        Send channel window adjust message.

        Args:
            channel_id: Local channel ID
            bytes_to_add: Number of bytes to add to window
        """
        with self._lock:
            if channel_id not in self._channels:
                return

            channel = self._channels[channel_id]

            # Build window adjust message
            if channel._remote_channel_id is not None:
                msg = Message(MSG_CHANNEL_WINDOW_ADJUST)
                msg.add_uint32(channel._remote_channel_id)
                msg.add_uint32(bytes_to_add)

                self._send_message(msg)

            # Update local window size
            channel._local_window_size += bytes_to_add

    def _send_channel_request(
        self, channel_id: int, request_type: str, want_reply: bool, data: bytes
    ) -> None:
        """
        Send channel request message.

        Args:
            channel_id: Local channel ID
            request_type: Type of request
            want_reply: Whether reply is wanted
            data: Request-specific data
        """
        with self._lock:
            if channel_id not in self._channels:
                raise TransportException(f"Channel {channel_id} not found")

            channel = self._channels[channel_id]

            # Build channel request message
            if channel._remote_channel_id is not None:
                msg = Message(MSG_CHANNEL_REQUEST)
                msg.add_uint32(channel._remote_channel_id)
                msg.add_string(request_type)
                msg.add_boolean(want_reply)
                if data:
                    msg._data.extend(data)

                self._send_message(msg)

    def _send_channel_eof(self, channel_id: int) -> None:
        """
        Send channel EOF message.

        Args:
            channel_id: Local channel ID
        """
        with self._lock:
            if channel_id not in self._channels:
                return

            channel = self._channels[channel_id]

            # Build EOF message
            if channel._remote_channel_id is not None:
                msg = Message(MSG_CHANNEL_EOF)
                msg.add_uint32(channel._remote_channel_id)

                self._send_message(msg)

    def _send_global_request(
        self, request_name: str, want_reply: bool, data: bytes = b""
    ) -> bool:
        """
        Send global request message.

        Args:
            request_name: Name of the global request
            want_reply: Whether to wait for reply
            data: Request-specific data

        Returns:
            True if request succeeded (when want_reply=True)

        Raises:
            TransportException: If request fails
        """
        if not self._active:
            raise TransportException("Transport not active")

        try:
            # Build global request message
            msg = Message(MSG_GLOBAL_REQUEST)
            msg.add_string(request_name)
            msg.add_boolean(want_reply)
            if data:
                msg._data.extend(data)

            # Send request
            self._send_message(msg)

            if want_reply:
                # Wait for response
                response = self._expect_message(
                    MSG_REQUEST_SUCCESS, MSG_REQUEST_FAILURE
                )

                if response.msg_type == MSG_REQUEST_SUCCESS:
                    return True
                elif response.msg_type == MSG_REQUEST_FAILURE:
                    return False
                else:
                    raise TransportException(
                        f"Unexpected response to global request: {type(response).__name__}"
                    )

            return True  # No reply requested

        except (OSError, struct.error, SSHException) as e:
            if isinstance(e, SSHException):
                raise
            raise TransportException(f"Failed to send global request: {e}") from e

    def _handle_global_request(self, msg: Message) -> None:
        """
        Handle incoming global request.

        Args:
            msg: Global request message
        """
        try:
            # Parse global request message
            data = msg._data
            offset = 0

            request_name_bytes, offset = read_string(data, offset)
            want_reply, offset = read_boolean(data, offset)
            request_data = bytes(data[offset:]) if offset < len(data) else b""

            request_name = request_name_bytes.decode(SSH_STRING_ENCODING)

            # Handle specific request types
            success = False

            if request_name == "tcpip-forward":
                success = self._handle_tcpip_forward_request(request_data)
            elif request_name == "cancel-tcpip-forward":
                success = self._handle_cancel_tcpip_forward_request(request_data)
            else:
                # Unknown request type
                success = False

            # Send reply if requested
            if want_reply:
                if success:
                    reply_msg = Message(MSG_REQUEST_SUCCESS)
                else:
                    reply_msg = Message(MSG_REQUEST_FAILURE)

                self._send_message(reply_msg)

        except (OSError, struct.error, ValueError, UnicodeDecodeError, SSHException):
            # Send failure reply if requested
            if want_reply:
                try:
                    reply_msg = Message(MSG_REQUEST_FAILURE)
                    self._send_message(reply_msg)
                except (OSError, TransportException):
                    pass

    def _handle_tcpip_forward_request(self, data: bytes) -> bool:
        """
        Handle tcpip-forward global request.

        Args:
            data: Request data

        Returns:
            True if request should be accepted
        """
        try:
            offset = 0
            bind_address_bytes, offset = read_string(data, offset)
            bind_port, offset = read_uint32(data, offset)

            bind_address = bind_address_bytes.decode(SSH_STRING_ENCODING)

            # Delegate to server interface for validation
            if self._server_mode and self._server_interface:
                return bool(
                    self._server_interface.check_port_forward_request(
                        bind_address, bind_port
                    )
                )

            # Default to reject if no server interface
            return False

        except (struct.error, ValueError, UnicodeDecodeError, IndexError):
            return False

    def _handle_cancel_tcpip_forward_request(self, data: bytes) -> bool:
        """
        Handle cancel-tcpip-forward global request.

        Args:
            data: Request data

        Returns:
            True if request should be accepted
        """
        try:
            offset = 0
            bind_address_bytes, offset = read_string(data, offset)
            bind_port, offset = read_uint32(data, offset)

            bind_address = bind_address_bytes.decode(SSH_STRING_ENCODING)

            # Delegate to server interface for validation
            if self._server_mode and self._server_interface:
                return bool(
                    self._server_interface.check_port_forward_cancel_request(
                        bind_address, bind_port
                    )
                )

            # Default to reject if no server interface
            return False

        except (struct.error, ValueError, UnicodeDecodeError, IndexError):
            return False

    def __enter__(self) -> "Transport":
        return self

    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        self.close()

    def close(self) -> None:
        """Close transport and cleanup resources."""
        kex_thread: Optional[threading.Thread] = None
        # Snapshot the channel list under the transport lock, then release it
        # before calling Channel.close(). Channel.close() takes the channel
        # lock and then re-enters _close_channel which takes the transport
        # lock - holding the transport lock here while calling channel.close()
        # would invert the order taken by any concurrent caller of
        # Channel.close() and deadlock the two threads.
        with self._lock:
            self._active = False
            self._stop_event.set()
            kex_thread = self._kex_thread
            channels_snapshot = list(self._channels.values())
            sock = self._socket

        for channel in channels_snapshot:
            try:
                channel.close()
            except (OSError, SSHException):
                pass

        with self._lock:
            self._channels.clear()

        if sock is not None:
            try:
                sock.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
            try:
                sock.close()
            except OSError:
                pass

        # Join kex thread outside lock to avoid deadlock.
        # Socket closure above ensures the thread unblocks promptly.
        if (
            kex_thread is not None
            and kex_thread.is_alive()
            and kex_thread != threading.current_thread()
        ):
            kex_thread.join(timeout=5.0)

    def _do_handshake(self) -> None:
        """
        Perform SSH protocol handshake and version negotiation.

        Raises:
            TransportException: If handshake fails
            ProtocolException: If protocol error occurs
        """
        try:
            # RFC 4253, Section 4.2: Both sides MUST send an identification string.
            # It is recommended to send it before receiving the peer's string to avoid deadlock.
            self._send_version()
            self._recv_version()

        except (OSError, struct.error, SSHException) as e:
            if isinstance(e, SSHException):
                raise
            raise TransportException(f"Error during version exchange: {e}") from e

    def _send_version(self) -> None:
        """Send SSH version string."""
        version_string = create_version_string()
        if self._server_mode:
            self._server_version = version_string
        else:
            self._client_version = version_string

        version_line = version_string + "\r\n"
        self._socket.sendall(version_line.encode(SSH_STRING_ENCODING))

    def _recv_version(self) -> None:
        """Receive and validate SSH version string."""
        version_line = b""
        _MAX_BANNER_LINES = 20

        # RFC 4253: The server MAY send other lines of data before
        # sending the version string. ... The identification string
        # MUST start with 'SSH-'.
        banner_lines = 0
        while True:
            if banner_lines > _MAX_BANNER_LINES:
                raise ProtocolException(
                    "Too many banner lines before SSH version string"
                )
            banner_lines += 1

            # _recv_bytes is internally buffered (32 KB reads), so this per-byte
            # loop reads from memory on all but the first iteration.
            current_line = b""
            while True:
                char = self._recv_bytes(1)
                if not char:
                    raise TransportException("Connection closed during banner read")

                current_line += char
                if char == b"\n":
                    # Remove trailing CRLF or LF
                    if current_line.endswith(b"\r\n"):
                        current_line = current_line[:-2]
                    else:
                        current_line = current_line[:-1]
                    break

                if len(current_line) > MAX_VERSION_LINE_LENGTH:
                    raise ProtocolException("Version line too long")

            if current_line.startswith(b"SSH-"):
                version_line = current_line
                break

            self._logger.debug(f"Ignoring non-SSH line: {repr(current_line)}")

        try:
            version_string = version_line.decode(SSH_STRING_ENCODING)
        except UnicodeDecodeError:
            raise ProtocolException("Invalid version string encoding")

        self._remote_version = version_string.strip()
        self._logger.debug(f"Remote version: {self._remote_version}")

        # Parse and validate version
        try:
            protocol_version, software_version = parse_version_string(version_string)
        except ValueError as e:
            raise ProtocolException(f"Invalid version string: {e}") from e

        if not is_supported_version(protocol_version):
            raise ProtocolException(f"Unsupported protocol version: {protocol_version}")

        if self._server_mode:
            self._client_version = version_string
        else:
            self._server_version = version_string

    def _start_kex(self) -> None:
        """
        Start key exchange process in the background.
        Includes a 30-second watchdog to prevent hanging on stalled connections.
        """
        with self._lock:
            self._kex_in_progress = True
            self._kex_thread = threading.current_thread()
            self._kex_condition.notify_all()

        timeout = 30.0

        # Set a temporary timeout for key exchange
        old_timeout = self._socket.gettimeout()
        self._socket.settimeout(timeout)

        try:
            # Send KEXINIT message with our supported algorithms
            self._send_kexinit()

            # Receive server's KEXINIT message
            self._recv_kexinit()

            # Now perform key exchange using KeyExchange class
            self._logger.debug("Starting DH exchange phase...")
            self._kex.start_kex()
            self._logger.debug("Rekeying handshake complete.")

        except (OSError, struct.error, SSHException) as e:
            # _kex_in_progress / _active are reset under the lock in finally
            with self._lock:
                self._active = False
            try:
                self.close()
            except (OSError, SSHException):
                pass
            if isinstance(e, SSHException):
                raise
            raise TransportException(f"Rekeying failed or timed out: {e}") from e
        finally:
            # Clear flags and reset byte counter atomically under the lock so
            # _check_rekey / _send_packet observers see a consistent snapshot.
            with self._lock:
                self._kex_in_progress = False
                self._kex_thread = None
                self._bytes_since_rekey = 0
                self._last_rekey_time = time.time()
                self._kex_condition.notify_all()

            try:
                if self._socket and self._socket.fileno() != -1:
                    self._socket.settimeout(old_timeout)
            except (OSError, AttributeError):
                pass

    def _send_kexinit(self) -> None:
        """Send KEXINIT message with supported algorithms."""
        cookie = self._crypto_backend.generate_random(KEX_COOKIE_SIZE)

        # Use algorithms from CipherSuite to ensure consistency
        cipher_suite = self._kex._cipher_suite

        # Strict KEX (Terrapin defense)
        kex_algorithms = list(cipher_suite.KEX_ALGORITHMS)
        if self._server_mode:
            # Server sends kex-strict-s
            kex_algorithms = [
                a for a in kex_algorithms if a != "kex-strict-c-v00@openssh.com"
            ]
            if "kex-strict-s-v00@openssh.com" not in kex_algorithms:
                kex_algorithms.append("kex-strict-s-v00@openssh.com")
            # Servers don't send ext-info-c
            kex_algorithms = [a for a in kex_algorithms if a != "ext-info-c"]
        else:
            # Client: append client-side signaling tokens, strip server-only token
            for token in cipher_suite.KEX_SIGNAL_TOKENS:
                if token not in kex_algorithms:
                    kex_algorithms.append(token)
            kex_algorithms = [
                a for a in kex_algorithms if a != "kex-strict-s-v00@openssh.com"
            ]

        kexinit_msg = KexInitMessage(
            cookie=cookie,
            kex_algorithms=kex_algorithms,
            server_host_key_algorithms=cipher_suite.HOST_KEY_ALGORITHMS,
            encryption_algorithms_client_to_server=cipher_suite.ENCRYPTION_ALGORITHMS,
            encryption_algorithms_server_to_client=cipher_suite.ENCRYPTION_ALGORITHMS,
            mac_algorithms_client_to_server=cipher_suite.MAC_ALGORITHMS,
            mac_algorithms_server_to_client=cipher_suite.MAC_ALGORITHMS,
            compression_algorithms_client_to_server=[COMPRESS_NONE],
            compression_algorithms_server_to_client=[COMPRESS_NONE],
        )

        self._client_kexinit_blob = kexinit_msg.pack()
        self._send_message(kexinit_msg)

    def _recv_kexinit(self) -> None:
        """Receive and process KEXINIT message."""
        msg = self._expect_message(MSG_KEXINIT)

        if not isinstance(msg, KexInitMessage):
            raise ProtocolException(f"Expected KEXINIT, got {type(msg).__name__}")

        # Store peer's KEXINIT for algorithm negotiation
        self._peer_kexinit = msg
        self._logger.debug(f"Peer KEX algorithms: {msg.kex_algorithms}")

        # Check for strict KEX marker from peer
        self._peer_strict_kex_version = None
        for algo in msg.kex_algorithms:
            if algo in (
                "kex-strict-s-v00@openssh.com",
                "kex-strict-c-v00@openssh.com",
            ):
                self._peer_strict_kex_version = "v00"
                break

        if self._peer_strict_kex_version:
            self._strict_kex = True
            self._logger.debug("Strict KEX mode enabled (Terrapin defense)")

    def _check_rekey(self) -> None:
        """Check if rekeying is needed and start it if so."""
        if not self._active:
            return

        with self._lock:
            if self._kex_in_progress:
                return

            # Check byte limit, time limit, or sequence number (rekey every 2^31 packets)
            if (
                self._bytes_since_rekey >= self._rekey_bytes_limit
                or (time.time() - self._last_rekey_time) >= self._rekey_time_limit
                or self._sequence_number_out >= REKEY_SEQUENCE_THRESHOLD
                or self._sequence_number_in >= REKEY_SEQUENCE_THRESHOLD
            ):
                self._logger.debug(
                    f"Triggering rekeying: bytes={self._bytes_since_rekey}, limit={self._rekey_bytes_limit}"
                )
                # Trigger rekeying in a separate thread to avoid blocking current I/O
                # _kex_thread is set inside _start_kex() under self._lock (single owner)
                self._kex_in_progress = True
                kex_thread = threading.Thread(
                    target=self._start_kex, name="RekeyingThread", daemon=True
                )
                kex_thread.start()

    def _send_message(self, message: Message) -> None:
        """
        Send SSH message.

        Args:
            message: Message to send

        Raises:
            TransportException: If send fails
        """
        try:
            payload = message.pack()

            with self._lock:
                if self._profiler is not None:
                    t0 = perf_counter()
                    packet = self._build_packet(payload)
                    t1 = perf_counter()
                    if self._encryptor_instance or getattr(
                        self, "_cipher_out_active", None
                    ):
                        packet = self._encrypt_packet(packet)
                    t2 = perf_counter()
                    self._socket.sendall(packet)
                    t3 = perf_counter()
                    self._profiler.record(t1 - t0, t2 - t1, t3 - t2)
                else:
                    packet = self._build_packet(payload)

                    # Encrypt if we have an active cipher
                    if self._encryptor_instance or getattr(
                        self, "_cipher_out_active", None
                    ):
                        packet = self._encrypt_packet(packet)

                    self._socket.sendall(packet)

                # Track bytes sent for rekeying
                if message.msg_type not in [
                    MSG_KEXINIT,
                    MSG_NEWKEYS,
                ] and not (MSG_KEXDH_INIT <= message.msg_type <= MSG_KEXDH_REPLY):
                    self._bytes_since_rekey += len(packet)
                    self._check_rekey()

                # If this was a NEWKEYS message, activate encryption AFTER sending it
                if message.msg_type == MSG_NEWKEYS:
                    self._activate_outbound_encryption()

                self._sequence_number_out = (self._sequence_number_out + 1) & 0xFFFFFFFF

                # Strict-KEX (Terrapin defense, RFC): after sending NEWKEYS,
                # reset outbound sequence to 0 so the next packet uses nonce 0.
                if message.msg_type == MSG_NEWKEYS and self._strict_kex:
                    self._sequence_number_out = 0
                    self._logger.debug("Sequence number (out) reset for strict KEX")

        except (OSError, struct.error, TransportException) as e:
            if isinstance(e, TransportException):
                raise
            raise TransportException(f"Failed to send message: {e}") from e

    def _dispatch_packet(
        self, packet: bytes, single_pump: bool = False
    ) -> Optional[Message]:
        """
        Parse a raw SSH packet and dispatch it.

        Returns the message for the caller to consume, HandledMessage() if the
        packet was handled internally and single_pump=True, or None if it was
        handled internally and the caller should loop for the next packet.
        """
        payload = extract_message_from_packet(packet)

        with self._lock:
            msg = Message.unpack(payload)

            # Track bytes received for rekeying (unless it's KEX)
            if msg.msg_type not in [
                MSG_KEXINIT,
                MSG_NEWKEYS,
            ] and not (MSG_KEXDH_INIT <= msg.msg_type <= MSG_KEXDH_REPLY):
                self._bytes_since_rekey += len(packet)
                self._check_rekey()

            # ALWAYS increment sequence number for EVERY packet received
            self._sequence_number_in = (self._sequence_number_in + 1) & 0xFFFFFFFF

            if msg.msg_type in [MSG_IGNORE, MSG_DEBUG, MSG_EXT_INFO]:
                return HandledMessage() if single_pump else None  # type: ignore[return-value]

            if msg.msg_type == MSG_DISCONNECT:
                try:
                    d_msg = DisconnectMessage.unpack(payload)
                    reason = getattr(d_msg, "description", "Unknown")
                    code = getattr(d_msg, "reason_code", 0)
                except (struct.error, ValueError, UnicodeDecodeError, IndexError):
                    raise TransportException("Disconnected by peer")
                raise TransportException(f"Disconnected: {reason} (code: {code})")

            if msg.msg_type == MSG_NEWKEYS:
                self._activate_inbound_encryption()

            if msg.msg_type == MSG_KEXINIT and not self._kex_in_progress:
                # Peer initiated rekeying - set flag, queue message, start KEX thread.
                self._kex_in_progress = True
                self._message_queue.append(msg)
                self._kex_thread = threading.Thread(target=self._start_kex, daemon=True)
                self._kex_thread.start()
                return HandledMessage() if single_pump else None  # type: ignore[return-value]

            if (
                msg.msg_type == MSG_GLOBAL_REQUEST  # 80
                or msg.msg_type == MSG_CHANNEL_OPEN  # 90
                or (msg.msg_type >= 93 and msg.msg_type <= 100)
            ):
                self._handle_channel_message(msg)
                return HandledMessage() if single_pump else None  # type: ignore[return-value]

            # Server-side specific messages
            if self._server_mode:
                if msg.msg_type == MSG_SERVICE_REQUEST:
                    self._handle_service_request(msg)
                    return HandledMessage() if single_pump else None  # type: ignore[return-value]
                if msg.msg_type == MSG_USERAUTH_REQUEST:
                    self._handle_userauth_request(msg)
                    return HandledMessage() if single_pump else None  # type: ignore[return-value]

            return msg

    def _read_message(self, single_pump: bool = False) -> Optional[Message]:
        """
        Read next message from socket and dispatch if needed.
        Does NOT check the message queue.
        """
        try:
            while True:
                # If rekeying is in progress, only the rekeying thread is allowed to read from the socket.
                # Other threads must wait and check the queue (handled in _recv_message and _expect_message).
                if (
                    self._kex_in_progress
                    and not getattr(self, "_is_async", False)
                    and threading.current_thread() != self._kex_thread
                ):
                    # We should not be here if called from _recv_message or _expect_message
                    # as they have their own yielding loops, but for safety:
                    return None

                # Hold _read_lock while reading a complete packet to prevent
                # multiple threads from interleaving partial reads.
                with self._read_lock:
                    packet = self._recv_packet()
                    if not packet:
                        if not self._active:
                            return None
                        raise TransportException("Empty packet received")

                    msg = self._dispatch_packet(packet, single_pump)
                    if msg is not None:
                        return msg
                    # None means handled internally - loop to read the next packet.

        except (OSError, struct.error, SSHException, ProtocolException) as e:
            if isinstance(e, (SSHException, ProtocolException)):
                raise
            raise TransportException(f"Failed to receive message: {e}") from e

    def _pump(self) -> Optional[Union[Message, type[HandledMessage]]]:
        """
        Read next message and either handle it or queue it.
        This is used for background message processing to ensure no
        messages are lost when multiple threads are waiting for messages.

        Returns:
            The message read, or HandledMessage if it was handled internally.
        """
        # First check if we already have messages in the queue
        with self._lock:
            if self._message_queue:
                return self._message_queue.popleft()

        msg = self._read_message(single_pump=True)
        if msg:
            return msg
        return None

    def _recv_message(self) -> Message:
        """
        Receive SSH message, checking the queue first.
        """
        while True:
            while True:
                with self._lock:
                    if self._message_queue:
                        return self._message_queue.popleft()

                    # If no rekeying or we are the rekeying thread, proceed to read
                    if (
                        not self._kex_in_progress
                        or threading.current_thread() == self._kex_thread
                    ):
                        break

                # Release _lock before waiting so the kex thread can acquire it
                # (Event.wait does NOT release locks, causing starvation otherwise)
                if self._stop_event.wait(0.1):
                    raise TransportException("Transport is stopping")

            # Inner loop exited via break - safe to read from socket now
            msg = self._read_message()
            if msg is not None:
                return msg

    def _expect_message(
        self, *allowed_types: int, channel_id: Optional[int] = None
    ) -> Message:
        """
        Receive next message and ensure it's one of the allowed types.
        Messages of other types are queued for later processing.
        """
        while True:
            # 1. Check queue for allowed message
            while True:
                with self._lock:
                    for i, queued in enumerate(self._message_queue):
                        if queued.msg_type in allowed_types:
                            # If channel_id is specified, check if it matches
                            if channel_id is not None:
                                msg_channel_id = getattr(
                                    queued, "recipient_channel", None
                                )
                                if msg_channel_id is None and len(queued._data) >= 4:
                                    # Fallback for messages not fully parsed or without attribute
                                    msg_channel_id, _ = read_uint32(queued._data, 0)

                                if msg_channel_id != channel_id:
                                    continue

                            del self._message_queue[i]
                            return queued

                    # If no rekeying or we are the rekeying thread, proceed to read
                    if (
                        not self._kex_in_progress
                        or threading.current_thread() == self._kex_thread
                    ):
                        break

                # Release _lock before waiting so the kex thread can acquire it
                # (Event.wait does NOT release locks, causing starvation otherwise)
                if self._stop_event.wait(0.1):
                    raise TransportException("Transport is stopping")

            # 2. Not in queue, read from socket
            try:
                read_msg = self._read_message()
            except TransportException:
                if not self._active:
                    raise TransportException("Transport closed")
                raise

            if read_msg is None:
                continue

            if read_msg.msg_type in allowed_types:
                # Check channel_id if specified
                if channel_id is not None:
                    msg_channel_id = getattr(read_msg, "recipient_channel", None)
                    if msg_channel_id is None and len(read_msg._data) >= 4:
                        msg_channel_id, _ = read_uint32(read_msg._data, 0)

                    if msg_channel_id != channel_id:
                        # Not for this channel, queue it for others and continue looking
                        with self._lock:
                            if len(self._message_queue) >= MAX_QUEUE_SIZE:
                                raise TransportException(
                                    "Message queue size limit exceeded"
                                )
                            self._message_queue.append(read_msg)
                        continue
                return read_msg

            # 3. Not what we wanted, queue it for others
            with self._lock:
                if len(self._message_queue) >= MAX_QUEUE_SIZE:
                    raise TransportException("Message queue size limit exceeded")
                self._message_queue.append(read_msg)

    def get_server_host_key(self) -> Optional[Any]:
        """
        Get server's public host key.

        Returns:
            PKey object or None if not available
        """
        if not self._server_host_key_blob:
            return None

        from ..crypto.pkey import PKey

        try:
            return PKey.from_string(self._server_host_key_blob)
        except (ValueError, struct.error, SSHException):
            return None

    def _activate_outbound_encryption(self) -> None:
        """Activate outbound encryption using negotiated parameters."""
        if self._server_mode:
            cipher_name = self._cipher_s2c
            key = self._encryption_key_s2c
            iv = self._iv_s2c
            mac_name = self._mac_s2c
            mac_key = self._mac_key_s2c
        else:
            cipher_name = self._cipher_c2s
            key = self._encryption_key_c2s
            iv = self._iv_c2s
            mac_name = self._mac_c2s
            mac_key = self._mac_key_c2s

        if not cipher_name or not key:
            return

        if iv is None:
            raise TransportException("Encryption parameters not fully negotiated")

        self._cipher_out_active = cipher_name
        self._encryption_key_out_active = key
        self._iv_out_active = iv

        self._logger.info(f"Activating outbound encryption: {cipher_name}")

        if cipher_name == "chacha20-poly1305@openssh.com":
            self._chacha20_key_out = key
            self._encryptor_instance = None
            self._mac_out_active = None
            self._mac_key_out_active = None
            return

        encryptor = self._crypto_backend.create_cipher(cipher_name, key, iv)
        self._logger.debug(
            f"Activating {cipher_name}: key_len={len(key)}, iv_len={len(iv)}"
        )
        # AES-CTR needs separate encryptor instance for state
        self._encryptor_instance = encryptor.encryptor()

        self._mac_out_active = mac_name
        self._mac_key_out_active = mac_key

        # NOTE: strict-KEX outbound sequence reset happens in _send_message
        # AFTER the unconditional ``seq_out += 1`` that follows this call,
        # so the next packet after NEWKEYS goes out with sequence 0 as
        # required by the kex-strict-*@openssh.com extension. Resetting
        # here would be overwritten by the +1 and produce seq=1.

    def _activate_inbound_encryption(self) -> None:
        """Activate inbound encryption using negotiated parameters."""
        if self._server_mode:
            cipher_name = self._cipher_c2s
            key = self._encryption_key_c2s
            iv = self._iv_c2s
            mac_name = self._mac_c2s
            mac_key = self._mac_key_c2s
        else:
            cipher_name = self._cipher_s2c
            key = self._encryption_key_s2c
            iv = self._iv_s2c
            mac_name = self._mac_s2c
            mac_key = self._mac_key_s2c

        if not cipher_name or not key:
            return

        if iv is None:
            raise TransportException("Encryption parameters not fully negotiated")

        self._cipher_in_active = cipher_name
        self._encryption_key_in_active = key
        self._iv_in_active = iv

        self._logger.info(f"Activating inbound encryption: {cipher_name}")

        if cipher_name == "chacha20-poly1305@openssh.com":
            self._chacha20_key_in = key
            self._decryptor_instance = None
            self._mac_in_active = None
            self._mac_key_in_active = None
        else:
            decryptor = self._crypto_backend.create_cipher(cipher_name, key, iv)
            # AES-CTR needs separate decryptor instance for state
            self._decryptor_instance = decryptor.decryptor()

            self._mac_in_active = mac_name
            self._mac_key_in_active = mac_key

        # Strict-KEX (Terrapin defense): after receiving NEWKEYS, reset inbound
        # sequence to 0 so the next received packet is verified with nonce 0.
        if self._strict_kex:
            self._sequence_number_in = 0
            self._logger.debug("Sequence number (in) reset for strict KEX")

    def _encrypt_packet(self, packet: bytes) -> bytes:
        """Encrypt SSH packet and add MAC if needed."""
        if getattr(self, "_cipher_out_active", None) == "chacha20-poly1305@openssh.com":
            if self._chacha20_key_out is None:
                raise TransportException("ChaCha20 outbound key not initialised")
            return self._crypto_backend.chacha20_poly1305_encrypt(
                self._chacha20_key_out,
                self._sequence_number_out,
                packet[:PACKET_LENGTH_SIZE],
                packet[PACKET_LENGTH_SIZE:],
            )

        if self._encryptor_instance:
            # AES-CTR or similar
            encrypted = self._encryptor_instance.update(packet)

            # Add MAC
            if self._mac_out_active and self._mac_key_out_active:
                mac_data = (
                    struct.pack(">I", self._sequence_number_out & 0xFFFFFFFF) + packet
                )
                mac = self._crypto_backend.compute_mac(
                    self._mac_out_active, self._mac_key_out_active, mac_data
                )
                return bytes(encrypted + mac)
            return bytes(encrypted)

        return packet

    def _build_packet(self, payload: bytes) -> bytes:
        """
        Build SSH packet from payload.

        Args:
            payload: Message payload

        Returns:
            Complete SSH packet
        """
        cipher_name = getattr(self, "_cipher_out_active", None) or getattr(
            self, "_cipher_c2s", None
        )

        if cipher_name:
            if cipher_name not in _CIPHER_BLOCK_SIZES:
                raise TransportException(f"Unknown cipher: {cipher_name}")
            block_size = _CIPHER_BLOCK_SIZES[cipher_name]
        else:
            block_size = 8

        # For AEAD ciphers (e.g. chacha20-poly1305) the 4-byte length field is
        # encrypted separately, so only the body (packet_length bytes) must be a
        # multiple of block_size.  For unencrypted packets and CTR/stream ciphers
        # the entire packet (4-byte prefix + body) is one unit, so the prefix is
        # included in the alignment.  Use _cipher_out_active (the currently
        # active encryption cipher) not the negotiated _cipher_c2s, because
        # NEWKEYS is sent before encryption is activated and must use standard
        # alignment even though _cipher_c2s may already be set to chacha20.
        active_cipher = getattr(self, "_cipher_out_active", None)
        length_overhead = 0 if active_cipher in _AEAD_CIPHERS else PACKET_LENGTH_SIZE
        padding_length = block_size - (
            (len(payload) + PADDING_LENGTH_SIZE + length_overhead) % block_size
        )

        if padding_length < MIN_PADDING_SIZE:
            padding_length += block_size

        # Generate random padding
        padding = self._crypto_backend.generate_random(padding_length)

        # Build packet into a pre-allocated bytearray to avoid repeated copies
        packet_length = PADDING_LENGTH_SIZE + len(payload) + padding_length
        packet = bytearray(
            PACKET_LENGTH_SIZE + PADDING_LENGTH_SIZE + len(payload) + padding_length
        )
        struct.pack_into(">IB", packet, 0, packet_length, padding_length)
        payload_start = PACKET_LENGTH_SIZE + PADDING_LENGTH_SIZE
        packet[payload_start : payload_start + len(payload)] = payload
        packet[payload_start + len(payload) :] = padding

        return bytes(packet)

    def _recv_packet(self) -> bytes:
        """
        Receive complete SSH packet.

        Returns:
            Complete SSH packet

        Raises:
            TransportException: If receive fails
        """
        if getattr(self, "_cipher_in_active", None) == "chacha20-poly1305@openssh.com":
            enc_length = self._recv_bytes(PACKET_LENGTH_SIZE)
            if len(enc_length) < PACKET_LENGTH_SIZE:
                if not self._active:
                    return b""
                raise TransportException("Short read while receiving packet length")
            if self._chacha20_key_in is None:
                raise TransportException("ChaCha20 inbound key not initialised")
            plain_length = self._crypto_backend.chacha20_poly1305_decrypt_length(
                self._chacha20_key_in, self._sequence_number_in, enc_length
            )
            packet_length = struct.unpack(">I", plain_length)[0]
            # For AEAD ciphers the minimum valid body is one block (8 bytes):
            # padding_len(1) + payload(>=1) + padding(>=4), aligned to 8.
            if packet_length < 8 or packet_length > MAX_PACKET_SIZE:
                raise ProtocolException(f"Invalid packet length: {packet_length}")
            enc_body = self._recv_bytes(packet_length)
            tag = self._recv_bytes(16)
            plain_body = self._crypto_backend.chacha20_poly1305_decrypt_body(
                self._chacha20_key_in,
                self._sequence_number_in,
                enc_length,
                enc_body,
                tag,
            )
            return bytes(plain_length + plain_body)

        if self._decryptor_instance:
            # AES-CTR: length is encrypted
            encrypted_length = self._recv_bytes(PACKET_LENGTH_SIZE)
            if len(encrypted_length) < PACKET_LENGTH_SIZE:
                if not self._active:
                    return b""
                raise TransportException("Short read while receiving packet length")

            length_data = self._decryptor_instance.update(encrypted_length)
            packet_length = struct.unpack(">I", length_data)[0]

            # Validate length
            if (
                packet_length < MIN_PACKET_SIZE - PACKET_LENGTH_SIZE
                or packet_length > MAX_PACKET_SIZE
            ):
                raise ProtocolException(f"Invalid packet length: {packet_length}")

            # Read rest of packet (encrypted)
            encrypted_payload = self._recv_bytes(packet_length)
            packet_payload = self._decryptor_instance.update(encrypted_payload)

            # Verify MAC
            if self._mac_in_active and self._mac_key_in_active:
                # Get mac length from CipherSuite
                mac_info = self._kex._cipher_suite.get_mac_info(self._mac_in_active)
                mac_len = mac_info["digest_len"]

                received_mac = self._recv_bytes(mac_len)
                mac_data = (
                    struct.pack(">I", self._sequence_number_in & 0xFFFFFFFF)
                    + length_data
                    + packet_payload
                )
                expected_mac = self._crypto_backend.compute_mac(
                    self._mac_in_active, self._mac_key_in_active, mac_data
                )

                if not hmac.compare_digest(received_mac, expected_mac):
                    raise TransportException("MAC verification failed")

            return bytes(length_data + packet_payload)

        # Unencrypted
        # Read packet length
        length_data = self._recv_bytes(PACKET_LENGTH_SIZE)
        packet_length = struct.unpack(">I", length_data)[0]

        # Validate packet length
        if packet_length < MIN_PACKET_SIZE - PACKET_LENGTH_SIZE:
            raise ProtocolException(f"Invalid packet length: {packet_length}")

        if packet_length > MAX_PACKET_SIZE - PACKET_LENGTH_SIZE:
            raise ProtocolException(f"Packet too large: {packet_length}")

        # Read rest of packet
        packet_data = self._recv_bytes(packet_length)

        # Verify MAC if present (even if unencrypted)
        if self._mac_in_active and self._mac_key_in_active:
            mac_info = self._kex._cipher_suite.get_mac_info(self._mac_in_active)
            mac_len = mac_info["digest_len"]
            received_mac = self._recv_bytes(mac_len)

            mac_data = (
                struct.pack(">I", self._sequence_number_in & 0xFFFFFFFF)
                + length_data
                + packet_data
            )
            expected_mac = self._crypto_backend.compute_mac(
                self._mac_in_active, self._mac_key_in_active, mac_data
            )

            if not hmac.compare_digest(received_mac, expected_mac):
                raise TransportException("MAC verification failed")

        # Return complete packet
        return length_data + packet_data

    def _recv_bytes(self, length: int) -> bytes:
        """
        Receive exact number of bytes from socket using internal buffering.

        Args:
            length: Number of bytes to receive

        Returns:
            Received bytes

        Raises:
            TransportException: If receive fails
        """
        # Locking model:
        # * ``self._lock`` guards ``self._packet_buffer`` and is held only for
        #   short, non-blocking buffer slices.
        # * ``self._read_lock`` serializes the actual blocking ``socket.recv``
        #   call so two threads cannot race the kernel for the same socket.
        # Fast path: if the buffer already has enough bytes, consume under
        # ``self._lock`` only — no need to acquire ``_read_lock`` at all.
        # Slow path: acquire ``_read_lock``, re-check the buffer (another
        # thread may have filled it while we waited), then block in recv().
        # Buffer consumption in the slow path therefore happens under both
        # ``_read_lock`` and ``self._lock``.
        while True:
            with self._lock:
                if len(self._packet_buffer) >= length:
                    data = bytes(self._packet_buffer[:length])
                    del self._packet_buffer[:length]
                    return data

            with self._read_lock:
                # Re-check buffer after acquiring _read_lock: another reader
                # may have refilled it while we waited.
                with self._lock:
                    if len(self._packet_buffer) >= length:
                        data = bytes(self._packet_buffer[:length])
                        del self._packet_buffer[:length]
                        return data
                    short_by = length - len(self._packet_buffer)

                # Read at least _buffer_size to leverage buffering. The blocking
                # ``recv`` happens with ``_read_lock`` held but ``self._lock``
                # released, so other threads can both send messages and serve
                # themselves from the existing buffer.
                try:
                    to_read = max(self._buffer_size, short_by)
                    chunk = self._socket.recv(to_read)
                except socket.timeout:
                    raise TransportException("Timeout receiving data")
                except OSError as e:
                    if not self._active:
                        return b""
                    raise TransportException(f"Socket error: {e}") from e

                if not chunk:
                    if not self._active or self._stop_event.is_set():
                        return b""
                    self._logger.debug("Socket closed while receiving")
                    raise TransportException("Connection closed unexpectedly")

                with self._lock:
                    self._packet_buffer += chunk

    @property
    def active(self) -> bool:
        """Check if transport is active."""
        return self._active

    @property
    def server_mode(self) -> bool:
        """Check if transport is in server mode."""
        return self._server_mode

    @property
    def authenticated(self) -> bool:
        """Check if transport is authenticated."""
        return self._authenticated

    @property
    def session_id(self) -> Optional[bytes]:
        """Get session ID."""
        return self._session_id

    def set_server_interface(self, server_interface: Any) -> None:
        """
        Set server interface for authentication callbacks.

        Args:
            server_interface: Server interface implementing authentication methods
        """
        self._server_interface = server_interface

    def get_server_interface(self) -> Optional[Any]:
        """
        Get server interface.

        Returns:
            Server interface or None if not set
        """
        return self._server_interface

    def _handle_service_request(self, msg: Message) -> None:
        """Handle service request message (server mode)."""
        try:
            service_name_bytes, _ = read_string(msg._data, 0)
            service_name = service_name_bytes.decode(SSH_STRING_ENCODING)

            if service_name == SERVICE_USERAUTH:
                accept_msg = ServiceAcceptMessage(SERVICE_USERAUTH)
                self._send_message(accept_msg)
            else:
                self._logger.warning(f"Rejecting unsupported service: {service_name}")
        except (
            OSError,
            struct.error,
            ValueError,
            UnicodeDecodeError,
            SSHException,
        ) as e:
            self._logger.error(f"Error handling service request: {e}")

    def _handle_userauth_request(self, msg: Message) -> None:
        """Handle user authentication request message (server mode)."""
        if not self._server_interface:
            self._send_message(UserAuthFailureMessage(["password", "publickey"], False))
            return

        username = ""
        try:
            # Unpack request
            auth_req = UserAuthRequestMessage._unpack_data(bytes(msg._data))
            username = auth_req.username
            method = auth_req.method

            result = AUTH_FAILED
            if method == AUTH_PASSWORD:
                offset = 0
                # Read boolean (False) before password
                change_requested, offset = read_boolean(auth_req.method_data, offset)
                password_bytes, offset = read_string(auth_req.method_data, offset)
                password = password_bytes.decode(SSH_STRING_ENCODING)
                result = self._server_interface.check_auth_password(username, password)
            elif method == AUTH_PUBLICKEY:
                offset = 0
                has_signature, offset = read_boolean(auth_req.method_data, offset)
                algo_name_bytes, offset = read_string(auth_req.method_data, offset)
                algo_name = algo_name_bytes.decode(SSH_STRING_ENCODING)
                key_blob, offset = read_string(auth_req.method_data, offset)

                from ..crypto.pkey import PKey

                try:
                    key = PKey.from_string(key_blob)
                except (ValueError, struct.error, SSHException):
                    # Invalid key blob
                    self._send_message(UserAuthFailureMessage(["publickey"], False))
                    return

                if not has_signature:
                    # Client is just querying if the key is acceptable
                    if self._server_interface.check_auth_publickey(username, key):
                        # Send PK_OK to indicate key is acceptable
                        pk_ok = Message(MSG_USERAUTH_PK_OK)
                        pk_ok._data.extend(write_string(algo_name))
                        pk_ok._data.extend(write_string(key_blob))
                        self._send_message(pk_ok)
                        return
                    else:
                        result = AUTH_FAILED
                else:
                    # Full authentication request with signature
                    signature, offset = read_string(auth_req.method_data, offset)

                    # Build data that was signed:
                    # string session_id, byte MSG_USERAUTH_REQUEST, string username,
                    # string service, string "publickey", boolean TRUE,
                    # string algo_name, string key_blob
                    signed_data = write_string(self._session_id or b"")
                    signed_data += write_byte(MSG_USERAUTH_REQUEST)
                    signed_data += write_string(username)
                    signed_data += write_string(auth_req.service)
                    signed_data += write_string(AUTH_PUBLICKEY)
                    signed_data += write_boolean(True)
                    signed_data += write_string(algo_name)
                    signed_data += write_string(key_blob)

                    if key.verify(signature, signed_data):
                        result = self._server_interface.check_auth_publickey(
                            username, key
                        )
                    else:
                        self._logger.warning(
                            f"Public key signature verification failed for user {username}"
                        )
                        result = AUTH_FAILED

            # Send response
            if result == AUTH_SUCCESSFUL:
                self._authenticated = True
                self._server_interface.on_authentication_successful(username, method)
                self._send_message(UserAuthSuccessMessage())
            else:
                self._server_interface.on_authentication_failed(username, method)
                allowed_methods = self._server_interface.get_allowed_auths(username)
                self._send_message(UserAuthFailureMessage(allowed_methods, False))

        except (
            OSError,
            struct.error,
            ValueError,
            UnicodeDecodeError,
            SSHException,
        ) as e:
            self._logger.error(f"Error handling userauth request: {e}")
            allowed = self._server_interface.get_allowed_auths(username)
            self._send_message(UserAuthFailureMessage(allowed, False))

    def get_port_forwarding_manager(self) -> "PortForwardingManager":
        """
        Get port forwarding manager.

        Returns:
            Port forwarding manager instance
        """
        if self._port_forwarding_manager is None:
            from .forwarding import PortForwardingManager

            self._port_forwarding_manager = PortForwardingManager(self)

        return self._port_forwarding_manager