diff --git a/.github/dependabot.yml b/.github/dependabot.yml index ad8b29e..e01bb2b 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -7,6 +7,13 @@ version: 2 updates: - package-ecosystem: "pub" directory: "/example/" + schedule: + interval: "weekly" + time: "09:00" + timezone: Europe/Madrid + + - package-ecosystem: "github-actions" + directory: "/" schedule: interval: "weekly" time: "09:00" diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d776e5..5f6bc6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,30 @@ +## [2.22.5] - 2026-07-30 +- Exported `src/ssh_userauth.dart` in `lib/dartssh2.dart` to expose `SSHUserInfoRequest`, `SSHUserInfoPrompt`, `SSHAuthMethod`, and `SSHChangePasswordResponse` [#188]. Thanks [@vicajilau]. + +## [2.22.4] - 2026-07-27 +- Advertised standard RFC 8731 key exchange name `curve25519-sha256` alongside legacy `curve25519-sha256@libssh.org` [#187]. Thanks [@nickn17]. + +## [2.22.3] - 2026-07-20 +- Fixed an SSH channel leak in `SftpClient.close()` by closing the underlying SSH channel and returning `Future` to allow awaiting channel teardown [#186]. Thanks [@keinstn]. + +## [2.22.2] - 2026-07-15 +- Added `flush()` to `SSHSocket`, `SSHClient`, and `SSHChannel` to allow force flushing of buffered outgoing data [#183]. Thanks [@vicajilau]. + +## [2.22.1] - 2026-07-13 +- Fixed a keepalive issue where overlapping pings could occur and caught errors during ping execution. Thanks [@vicajilau]. + +## [2.22.0] - 2026-07-03 +- Added optional `handshakeTimeout` and `authTimeout` to `SSHClient` to limit connection negotiation and user authentication times [#182]. Thanks [@GT-610]. + +## [2.21.1] - 2026-07-02 +- Fixed an `SSHTransport` busy-loop (100% CPU / ANR) that occurred when a partial packet remained in the read buffer [#179]. Thanks [@vicajilau]. + +## [2.21.0] - 2026-07-01 +- Added `SSHSession.waitForExit({Duration? timeout})` to await remote process exit status with an optional timeout [#176]. Thanks [@GT-610]. +- Hardened SOCKS5 dynamic forwarding (half-close streaming, dialing guards, timeout cancellation, malformed UTF-8 decoding, and buffer limits) [#175]. Thanks [@GT-610]. +- Hardened SSH agent channel frame validation (rejecting empty or oversized frames) and fallback RSA signature type checks [#175]. Thanks [@GT-610]. +- Improved EC private key parsing with proper ASN.1 OID curve detection, public point derivation validation, and robust comments decoding [#175]. Thanks [@GT-610]. + ## [2.20.0] - 2026-06-30 - **BREAKING**: Bumped the minimum Dart SDK constraint to `3.0.0` [#23]. Thanks [@vicajilau]. - **BREAKING**: Declared `OpenSSHKeyPair` as an `abstract mixin class` to comply with Dart 3.0 class modifier rules [#23]. Thanks [@vicajilau]. @@ -253,7 +280,15 @@ [#18]: https://github.com/TerminalStudio/dartssh2/issues/18 [#17]: https://github.com/TerminalStudio/dartssh2/issues/17 [#14]: https://github.com/TerminalStudio/dartssh2/pull/14 +[#175]: https://github.com/TerminalStudio/dartssh2/pull/175 +[#176]: https://github.com/TerminalStudio/dartssh2/pull/176 +[#179]: https://github.com/TerminalStudio/dartssh2/pull/179 +[#182]: https://github.com/TerminalStudio/dartssh2/pull/182 +[#183]: https://github.com/TerminalStudio/dartssh2/pull/183 +[#186]: https://github.com/TerminalStudio/dartssh2/pull/186 +[#187]: https://github.com/TerminalStudio/dartssh2/pull/187 [#1]: https://github.com/TerminalStudio/dartssh/pull/1/files +[#188]: https://github.com/TerminalStudio/dartssh2/issues/188 [@linhanyu]: https://github.com/linhanyu [@Migarl]: https://github.com/Migarl @@ -270,4 +305,5 @@ [@bradmartin333]: https://github.com/bradmartin333 [@Wackymax]: https://github.com/Wackymax [@gkc]: https://github.com/gkc -[@vicajilau]: https://github.com/vicajilau \ No newline at end of file +[@vicajilau]: https://github.com/vicajilau +[@GT-610]: https://github.com/GT-610 diff --git a/README.md b/README.md index 038e126..ff4a8e8 100644 --- a/README.md +++ b/README.md @@ -162,6 +162,24 @@ void main() async { `ident` defaults to `DartSSH_2.0`. +### Configure handshake and authentication timeouts + +You can specify optional timeouts for the transport handshake and user authentication: + +```dart +void main() async { + final client = SSHClient( + await SSHSocket.connect('localhost', 22), + username: '', + onPasswordRequest: () => '', + handshakeTimeout: const Duration(seconds: 15), + authTimeout: const Duration(seconds: 15), + ); +} +``` + +By default, these parameters are `null` (no timeout is enforced). Without these timeouts, the connection or authentication process could hang indefinitely if the remote server becomes unresponsive. + ### Spawn a shell on remote host ```dart @@ -277,6 +295,25 @@ void main() async { Processes killed by signals do not have an exit code, instead they have an exit signal property. +**Waiting for exit status with a timeout** + +Alternatively, you can wait for the remote process to report its exit status or exit signal with an optional timeout using `session.waitForExit()`: + +```dart +void main() async { + final session = await client.execute('sleep 5'); + + // Wait for the exit status to be reported (or up to 10 seconds). + final exitCode = await session.waitForExit(timeout: Duration(seconds: 10)); + + if (exitCode != null) { + print('Process exited with code: $exitCode'); + } else { + print('Process timed out or was terminated by a signal'); + } +} +``` + ### Forward connections on local port 8080 to the server ```dart diff --git a/lib/src/algorithm/ssh_kex_type.dart b/lib/src/algorithm/ssh_kex_type.dart index 88efb51..918b0f5 100644 --- a/lib/src/algorithm/ssh_kex_type.dart +++ b/lib/src/algorithm/ssh_kex_type.dart @@ -7,6 +7,15 @@ class SSHKexType extends SSHAlgorithm { digestFactory: SHA256Digest.new, ); + /// RFC 8731 name for the same algorithm as [x25519]. Servers hardened to a + /// single kex commonly offer only this spelling, and OpenSSH matches names + /// literally; without it the handshake dies with "no matching key exchange + /// method found". + static const x25519Rfc = SSHKexType._( + name: 'curve25519-sha256', + digestFactory: SHA256Digest.new, + ); + static const nistp256 = SSHKexType._( name: 'ecdh-sha2-nistp256', digestFactory: SHA256Digest.new, diff --git a/lib/src/dynamic_forward_io.dart b/lib/src/dynamic_forward_io.dart index 454e999..128067c 100644 --- a/lib/src/dynamic_forward_io.dart +++ b/lib/src/dynamic_forward_io.dart @@ -171,9 +171,8 @@ class _SocksConnection { return; } - _buffer.add(chunk); - try { + _buffer.add(chunk); await _consumeHandshake(); } catch (_) { await close(); diff --git a/lib/src/http/http_date.dart b/lib/src/http/http_date.dart index 12f8a33..26fd6fc 100644 --- a/lib/src/http/http_date.dart +++ b/lib/src/http/http_date.dart @@ -87,4 +87,4 @@ DateTime? parseHttpDate(String input) { } return null; -} \ No newline at end of file +} diff --git a/lib/src/kex/kex_nist.dart b/lib/src/kex/kex_nist.dart index 848d520..c9d8285 100644 --- a/lib/src/kex/kex_nist.dart +++ b/lib/src/kex/kex_nist.dart @@ -128,9 +128,10 @@ String _getNameByCurve(ECDomainParameters curve) { late BigInt x; do { x = decodeBigIntWithSign( - 1, - randomBytes((secretBits + 7) ~/ 8), - ) % curve.n; + 1, + randomBytes((secretBits + 7) ~/ 8), + ) % + curve.n; } while (x == BigInt.zero); final c = curve.G * x; diff --git a/lib/src/sftp/sftp_client.dart b/lib/src/sftp/sftp_client.dart index b354973..ecf05b4 100644 --- a/lib/src/sftp/sftp_client.dart +++ b/lib/src/sftp/sftp_client.dart @@ -258,12 +258,24 @@ class SftpClient { } /// Close the sftp session. - void close() { + /// + /// This also closes the underlying SSH channel that the sftp subsystem runs + /// on. Without this the channel is leaked: every [SSHClient.sftp] call opens + /// a fresh session channel, so an application that opens an sftp session per + /// operation would accumulate open channels on the connection until the + /// server refuses further `CHANNEL_OPEN`s. + Future close() async { + if (_done.isCompleted) return; + final error = SftpAbortError("Connection closed"); for (var waiter in _replyWaiters.values) { - waiter.completeError(SftpAbortError("Connection closed")); + waiter.completeError(error); } _replyWaiters.clear(); + if (!_handshake.isCompleted) { + _handshake.completeError(error, StackTrace.current); + } _done.complete(); + await _channel.close(); } void _closeError(Object error, [StackTrace? stackTrace]) { diff --git a/lib/src/socket/ssh_socket.dart b/lib/src/socket/ssh_socket.dart index 13a1d07..0c369ad 100644 --- a/lib/src/socket/ssh_socket.dart +++ b/lib/src/socket/ssh_socket.dart @@ -28,4 +28,7 @@ abstract class SSHSocket { Future close(); void destroy(); + + /// Force flush any buffered outgoing data. + Future flush() async {} } diff --git a/lib/src/socket/ssh_socket_io.dart b/lib/src/socket/ssh_socket_io.dart index 67a4c47..a32e598 100644 --- a/lib/src/socket/ssh_socket_io.dart +++ b/lib/src/socket/ssh_socket_io.dart @@ -38,6 +38,11 @@ class _SSHNativeSocket implements SSHSocket { _socket.destroy(); } + @override + Future flush() async { + await _socket.flush(); + } + @override String toString() { final address = '${_socket.remoteAddress.host}:${_socket.remotePort}'; diff --git a/lib/src/ssh_algorithm.dart b/lib/src/ssh_algorithm.dart index 652303f..e898bb8 100644 --- a/lib/src/ssh_algorithm.dart +++ b/lib/src/ssh_algorithm.dart @@ -92,6 +92,7 @@ class SSHAlgorithms { // Prefer modern KEX first; move legacy SHA-1/group1 variants to the end // as fallback-only to improve security defaults. this.kex = const [ + SSHKexType.x25519Rfc, SSHKexType.x25519, SSHKexType.nistp521, SSHKexType.nistp384, diff --git a/lib/src/ssh_channel.dart b/lib/src/ssh_channel.dart index 255e469..071a0bf 100644 --- a/lib/src/ssh_channel.dart +++ b/lib/src/ssh_channel.dart @@ -31,6 +31,8 @@ class SSHChannelController { final void Function(SSHMessage) sendMessage; + final Future Function()? onFlush; + SSHChannel get channel => SSHChannel(this); SSHChannelController({ @@ -41,6 +43,7 @@ class SSHChannelController { required this.remoteInitialWindowSize, required this.remoteMaximumPacketSize, required this.sendMessage, + this.onFlush, this.printDebug, }) { if (remoteInitialWindowSize > 0) { @@ -62,8 +65,13 @@ class SSHChannelController { /// A [StreamController] that accepts data from local end of the channel. final _localStream = StreamController(); + late final StreamSink _localSink = + _SSHChannelSink(_localStream); + late final _localStreamConsumer = SSHChannelDataConsumer(_localStream.stream); + SSHChannelData? _pendingUploadData; + /// Handler of channel requests from the remote side. late var _requestHandler = _defaultRequestHandler; @@ -83,6 +91,8 @@ class SSHChannelController { final _done = Completer(); + final _flushBarriers = >{}; + Future sendExec(String command) async { sendMessage( SSH_Message_Channel_Request.exec( @@ -233,6 +243,7 @@ class SSHChannelController { Future close() async { if (_done.isCompleted) return; + _failFlushBarriers(StateError('Channel closed before flush completed')); _localStreamConsumer.cancel(); _sendEOFIfNeeded(); @@ -250,6 +261,7 @@ class SSHChannelController { /// received. void destroy() { if (_done.isCompleted) return; + _failFlushBarriers(StateError('Channel destroyed before flush completed')); _remoteStream.close(); _localStreamConsumer.cancel(); _sendEOFIfNeeded(); @@ -409,13 +421,17 @@ class SSHChannelController { late final _uploadLoop = OnceSimultaneously(() async { while (true) { - if (_remoteWindow <= 0) { - return; + SSHChannelData? data; + if (_pendingUploadData != null) { + if (_remoteWindow <= 0) return; + data = _pendingUploadData; + _pendingUploadData = null; + } else { + final dataToRead = + _remoteWindow > 0 ? min(_remoteWindow, remoteMaximumPacketSize) : 1; + data = await _localStreamConsumer.read(dataToRead); } - final dataToRead = min(_remoteWindow, remoteMaximumPacketSize); - final data = await _localStreamConsumer.read(dataToRead); - if (data == null) { _sendEOFIfNeeded(); @@ -425,6 +441,19 @@ class SSHChannelController { return; } + if (data is _SSHChannelFlushBarrier) { + _flushBarriers.remove(data.completer); + if (!data.completer.isCompleted) { + data.completer.complete(); + } + continue; + } + + if (_remoteWindow <= 0) { + _pendingUploadData = data; + return; + } + if (_hasSentEOF) { return; } @@ -447,6 +476,26 @@ class SSHChannelController { _remoteWindow -= data.bytes.length; } }); + + Future flush() async { + if (!_done.isCompleted) { + final barrier = Completer(); + _flushBarriers.add(barrier); + _localStream.add(_SSHChannelFlushBarrier(barrier)); + _uploadLoop.activate(); + await barrier.future; + } + await onFlush?.call(); + } + + void _failFlushBarriers(Object error) { + for (final barrier in _flushBarriers) { + if (!barrier.isCompleted) { + barrier.completeError(error, StackTrace.current); + } + } + _flushBarriers.clear(); + } } class SSHChannel { @@ -464,7 +513,7 @@ class SSHChannel { /// A [StreamSink] that sends data to the remote side. Chucks must be /// equal to or less than [maximumPacketSize]. - StreamSink get sink => _controller._localStream.sink; + StreamSink get sink => _controller._localSink; Future get done => _controller._done.future; @@ -477,6 +526,9 @@ class SSHChannel { sink.add(SSHChannelData(data, type: type)); } + /// Force flush any buffered outgoing data on this channel to the socket. + Future flush() => _controller.flush(); + void setRequestHandler(SSHChannelRequestHandler handler) { _controller._requestHandler = handler; } @@ -547,6 +599,41 @@ class SSHChannelData { SSHChannelData(this.bytes, {this.type}); } +class _SSHChannelFlushBarrier extends SSHChannelData { + _SSHChannelFlushBarrier(this.completer) : super(Uint8List(0)); + + final Completer completer; +} + +class _SSHChannelSink implements StreamSink { + _SSHChannelSink(this._controller); + + final StreamController _controller; + + @override + void add(SSHChannelData data) { + _controller.add(data); + } + + @override + void addError(Object error, [StackTrace? stackTrace]) { + _controller.addError(error, stackTrace); + } + + @override + Future addStream(Stream stream) async { + await for (final data in stream) { + add(data); + } + } + + @override + Future close() => _controller.close(); + + @override + Future get done => _controller.done; +} + class SSHChannelExtendedDataType { static const stderr = 1; } diff --git a/lib/src/ssh_client.dart b/lib/src/ssh_client.dart index c4ccbb9..58c983a 100644 --- a/lib/src/ssh_client.dart +++ b/lib/src/ssh_client.dart @@ -125,11 +125,12 @@ class SSHRunResult { } class SSHClient { - /// RFC 4252 recommended authentication timeout period + /// Opt-in RFC 4252 authentication timeout preset. This is not applied + /// automatically; pass it as [authTimeout] to enable it. static const Duration defaultAuthTimeout = Duration(minutes: 10); - /// Default handshake timeout. Separates transport handshake timeout from - /// authentication timeout for better robustness. + /// Opt-in handshake timeout preset. This is not applied automatically; pass + /// it as [handshakeTimeout] to enable it. static const Duration defaultHandshakeTimeout = Duration(seconds: 30); /// RFC 4252 recommended maximum authentication attempts per session @@ -203,13 +204,15 @@ class SSHClient { /// Username of clinet host used for hostbased authentication. final String? userNameOnClientHost; - /// Auth timeout, 10m by default. - final Duration authTimeout; + /// Maximum time to wait for the SSH transport handshake to complete. This is + /// null unless explicitly provided; [defaultHandshakeTimeout] is an opt-in + /// preset. + final Duration? handshakeTimeout; - /// Handshake timeout, 30s by default. This only covers the SSH transport - /// handshake (version exchange, KEX, host key verification, NEWKEYS), and is - /// independent from [authTimeout]. - final Duration handshakeTimeout; + /// Maximum time to wait for authentication after the transport is ready. + /// This is null unless explicitly provided; [defaultAuthTimeout] is an + /// opt-in preset. + final Duration? authTimeout; /// Max auth attempts, 20 by default. final int maxAuthAttempts; @@ -250,12 +253,8 @@ class SSHClient { this.onUserauthBanner, this.onAuthenticated, this.keepAliveInterval = const Duration(seconds: 10), - - /// Authentication timeout period. RFC 4252 recommends 10 minutes. - this.authTimeout = defaultAuthTimeout, - - /// Handshake timeout period. Defaults to 30s. - this.handshakeTimeout = defaultHandshakeTimeout, + this.handshakeTimeout, + this.authTimeout, /// Maximum authentication attempts. RFC 4252 recommends 20 attempts. this.maxAuthAttempts = defaultMaxAuthAttempts, @@ -293,16 +292,14 @@ class SSHClient { _keyPairsLeft.addAll(identities!); } - // 初始化 hostbased 密钥队列 if (hostbasedIdentities != null) { _hostbasedKeyPairsLeft.addAll(hostbasedIdentities!); } - // 添加认证超时定时器 - _authTimeoutTimer = Timer(authTimeout, _onAuthTimeout); - - // 添加握手超时定时器（与认证超时分离） - _handshakeTimeoutTimer = Timer(handshakeTimeout, _onHandshakeTimeout); + final handshakeTimeout = this.handshakeTimeout; + if (handshakeTimeout != null) { + _handshakeTimeoutTimer = Timer(handshakeTimeout, _handleHandshakeTimeout); + } } static String _validateIdent(String ident) { @@ -366,6 +363,8 @@ class SSHClient { SSHAuthMethod? _currentAuthMethod; + var _transportReady = false; + /// A [Future] that completes when the client has authenticated, or /// completes with an error if the client could not authenticate. Future get authenticated => _authenticated.future; @@ -754,12 +753,17 @@ class SSHClient { /// Shutdown the entire SSH connection. Sessions and channels will also be /// closed immediately. void close() { - _authTimeoutTimer?.cancel(); _handshakeTimeoutTimer?.cancel(); + _authTimeoutTimer?.cancel(); _closeChannels(); _transport.close(); } + /// Force flush any buffered outgoing data to the socket. + Future flush() async { + await _transport.flush(); + } + /// Close all channels that are currently open. void _closeChannels() { for (final channel in _channels.values) { @@ -772,9 +776,14 @@ class SSHClient { void _handleTransportReady() { printDebug?.call('SSHClient._onTransportReady'); - // 握手完成，取消握手超时定时器 + _transportReady = true; _handshakeTimeoutTimer?.cancel(); _handshakeTimeoutTimer = null; + + final authTimeout = this.authTimeout; + if (authTimeout != null) { + _authTimeoutTimer = Timer(authTimeout, _handleAuthTimeout); + } _requestAuthentication(); } @@ -829,17 +838,6 @@ class SSHClient { } } - void _onHandshakeTimeout() { - // 若在握手阶段一直未就绪，则返回握手超时错误并关闭连接 - if (_authenticated.isCompleted) return; - final msg = - 'Handshake timed out after ${handshakeTimeout.inSeconds} seconds.'; - _authenticated.completeError(SSHHandshakeError(msg)); - // 认证阶段不会开始，取消其定时器以避免误触发 - _authTimeoutTimer?.cancel(); - _transport.closeWithError(SSHHandshakeError(msg)); - } - void _handlePacket(Uint8List payload) { try { _dispatchMessage(payload); @@ -852,6 +850,21 @@ class SSHClient { @visibleForTesting void handlePacket(Uint8List packet) => _handlePacket(packet); + @visibleForTesting + SSHChannelController acceptChannelForTesting({ + required SSHChannelId localChannelId, + required SSHChannelId remoteChannelId, + required int remoteInitialWindowSize, + required int remoteMaximumPacketSize, + }) { + return _acceptChannel( + localChannelId: localChannelId, + remoteChannelId: remoteChannelId, + remoteInitialWindowSize: remoteInitialWindowSize, + remoteMaximumPacketSize: remoteMaximumPacketSize, + ); + } + void _sendMessage(SSHMessage message) { printTrace?.call('-> $socket: $message'); _transport.sendPacket(message.encode()); @@ -928,11 +941,30 @@ class SSHClient { printTrace?.call('<- $socket: SSH_Message_Userauth_Success'); printDebug?.call('SSHClient._handleUserauthSuccess'); _authTimeoutTimer?.cancel(); + _authTimeoutTimer = null; _authenticated.complete(); onAuthenticated?.call(); _keepAlive?.start(); } + void _handleHandshakeTimeout() { + if (_authenticated.isCompleted || _transportReady) return; + + _handshakeTimeoutTimer = null; + final error = SSHHandshakeError('Handshake timed out'); + _authenticated.completeError(error, StackTrace.current); + } + + void _handleAuthTimeout() { + if (_authenticated.isCompleted) return; + + _authTimeoutTimer = null; + _authenticated.completeError( + SSHAuthAbortError('Authentication timed out'), + StackTrace.current, + ); + } + void _handleUserauthFailure(Uint8List payload) { final message = SSH_Message_Userauth_Failure.decode(payload); printTrace?.call('<- $socket: $message'); @@ -1677,6 +1709,7 @@ class SSHClient { remoteInitialWindowSize: remoteInitialWindowSize, remoteMaximumPacketSize: remoteMaximumPacketSize, sendMessage: _sendMessage, + onFlush: flush, printDebug: printDebug, ); @@ -1703,32 +1736,6 @@ class SSHClient { final replyCompleter = _channelOpenReplyWaiters.remove(id)!; replyCompleter.complete(message); } - - void _onAuthTimeout() { - if (!_authenticated.isCompleted) { - final attemptedMethods = []; - - if (_currentAuthMethod != null) { - attemptedMethods.add(_currentAuthMethod!.name); - } - - var timeoutMessage = - 'Authentication timed out after ${authTimeout.inSeconds} seconds.'; - - if (_authAttempts > 0) { - timeoutMessage += ' Made $_authAttempts authentication attempts.'; - - if (attemptedMethods.isNotEmpty) { - timeoutMessage += ' Methods tried: ${attemptedMethods.join(', ')}'; - } - } else { - timeoutMessage += ' No authentication attempts were made.'; - } - - _authenticated.completeError(SSHAuthAbortError(timeoutMessage)); - close(); - } - } } extension on SSHClient { diff --git a/lib/src/ssh_forward.dart b/lib/src/ssh_forward.dart index 2a03f2c..3e6b987 100644 --- a/lib/src/ssh_forward.dart +++ b/lib/src/ssh_forward.dart @@ -46,13 +46,27 @@ class SSHForwardChannel implements SSHSocket { final SSHChannel _channel; SSHForwardChannel(this._channel) { - _sinkController.stream - .map((data) => data is Uint8List ? data : Uint8List.fromList(data)) - .map((data) => SSHChannelData(data)) - .pipe(_channel.sink); + _sinkController.stream.listen( + (event) { + if (event is _SSHForwardData) { + final data = event.data is Uint8List + ? event.data as Uint8List + : Uint8List.fromList(event.data); + _channel.sink.add(SSHChannelData(data)); + return; + } + + final barrier = event as _SSHForwardFlushBarrier; + unawaited(_completeFlushBarrier(barrier.completer)); + }, + onError: _channel.sink.addError, + onDone: _channel.sink.close, + ); } - final _sinkController = StreamController>(); + final _sinkController = StreamController<_SSHForwardUploadEvent>(); + + late final StreamSink> _sink = _SSHForwardSink(_sinkController); /// Data received from the remote host. @override @@ -60,7 +74,7 @@ class SSHForwardChannel implements SSHSocket { /// Write to this sink to send data to the remote host. @override - StreamSink> get sink => _sinkController.sink; + StreamSink> get sink => _sink; /// Close our end of the channel. Returns a future that waits for the /// other side to close. @@ -76,6 +90,66 @@ class SSHForwardChannel implements SSHSocket { void destroy() { _channel.destroy(); } + + /// Force flush any buffered outgoing data. + @override + Future flush() async { + final barrier = Completer(); + _sinkController.add(_SSHForwardFlushBarrier(barrier)); + await barrier.future; + } + + Future _completeFlushBarrier(Completer barrier) async { + try { + await _channel.flush(); + barrier.complete(); + } catch (error, stackTrace) { + barrier.completeError(error, stackTrace); + } + } +} + +sealed class _SSHForwardUploadEvent {} + +class _SSHForwardData extends _SSHForwardUploadEvent { + _SSHForwardData(this.data); + + final List data; +} + +class _SSHForwardFlushBarrier extends _SSHForwardUploadEvent { + _SSHForwardFlushBarrier(this.completer); + + final Completer completer; +} + +class _SSHForwardSink implements StreamSink> { + _SSHForwardSink(this._controller); + + final StreamController<_SSHForwardUploadEvent> _controller; + + @override + void add(List data) { + _controller.add(_SSHForwardData(data)); + } + + @override + void addError(Object error, [StackTrace? stackTrace]) { + _controller.addError(error, stackTrace); + } + + @override + Future addStream(Stream> stream) async { + await for (final data in stream) { + add(data); + } + } + + @override + Future close() => _controller.close(); + + @override + Future get done => _controller.done; } class SSHX11Channel extends SSHForwardChannel { diff --git a/lib/src/ssh_keepalive.dart b/lib/src/ssh_keepalive.dart index a6bc75d..d60047f 100644 --- a/lib/src/ssh_keepalive.dart +++ b/lib/src/ssh_keepalive.dart @@ -9,6 +9,8 @@ class SSHKeepAlive { final Future Function() ping; + bool _isPinging = false; + SSHKeepAlive({ required this.ping, this.interval = const Duration(seconds: 10), @@ -16,7 +18,15 @@ class SSHKeepAlive { void start() { _timer ??= Timer.periodic(interval, (timer) async { - await ping(); + if (_isPinging) return; + _isPinging = true; + try { + await ping(); + } catch (_) { + // Ignore errors, the client transport will handle disconnection. + } finally { + _isPinging = false; + } }); } diff --git a/lib/src/ssh_session.dart b/lib/src/ssh_session.dart index af0d312..ccc0aba 100644 --- a/lib/src/ssh_session.dart +++ b/lib/src/ssh_session.dart @@ -32,9 +32,18 @@ class SSHSession { /// be available on the [stdout] and [stderr] streams at this time. Future get done => _channel.done; + /// The underlying SSH channel. + SSHChannel get channel => _channel; + SSHSession(this._channel) { _channel.setRequestHandler(_handleRequest); + done.then((_) { + if (!_exitCompleter.isCompleted) { + _exitCompleter.complete(_exitCode); + } + }); + _channelDataSubscription = _channel.stream.listen( _handleChannelData, onDone: _handleChannelDataDone, @@ -51,6 +60,8 @@ class SSHSession { SSHSessionExitSignal? _exitSignal; + final _exitCompleter = Completer(); + late final StreamSubscription _channelDataSubscription; late final _stdinController = StreamController(); @@ -71,6 +82,12 @@ class SSHSession { stdin.add(data); } + /// Force flush any buffered stdin data to the remote process. + Future flush() async { + await Future.microtask(() {}); + await _channel.flush(); + } + /// Inform remote process of the current window size. void resizeTerminal( int width, @@ -103,37 +120,16 @@ class SSHSession { _channel.close(); } - /// Wait for the remote process to exit. Returns the exit code of the remote - /// process, or null if the process has not yet exited. - Future waitForExit({Duration? timeout}) async { - final completer = Completer(); - - void checkExit() { - if (_exitCode != null && !completer.isCompleted) { - completer.complete(_exitCode); - } - } - - // Check the exit code immediately - checkExit(); - - final subscription = done.asStream().listen((_) => checkExit()); - Timer? timeoutTimer; + /// Waits for the remote process to report an exit status. + /// + /// Returns the exit status, or `null` if the process exited without reporting + /// one, was terminated by a signal, or [timeout] elapsed before it exited. + Future waitForExit({Duration? timeout}) { + Future wait = _exitCompleter.future; if (timeout != null) { - timeoutTimer = Timer(timeout, () { - subscription.cancel(); - if (!completer.isCompleted) { - completer.complete(null); - } - }); + wait = wait.timeout(timeout, onTimeout: () => null); } - - completer.future.then((_) { - timeoutTimer?.cancel(); - subscription.cancel(); - }); - - return completer.future; + return wait; } /// Deliver [signal] to the remote process. Some implementations may not @@ -146,6 +142,9 @@ class SSHSession { switch (request.requestType) { case SSHChannelRequestType.exitStatus: _exitCode = request.exitStatus!; + if (!_exitCompleter.isCompleted) { + _exitCompleter.complete(_exitCode); + } return true; case SSHChannelRequestType.exitSignal: _exitSignal = SSHSessionExitSignal( @@ -154,6 +153,9 @@ class SSHSession { errorMessage: request.errorMessage!, languageTag: request.languageTag!, ); + if (!_exitCompleter.isCompleted) { + _exitCompleter.complete(null); + } return true; } return false; diff --git a/lib/src/ssh_transport.dart b/lib/src/ssh_transport.dart index ba8948f..f4116df 100644 --- a/lib/src/ssh_transport.dart +++ b/lib/src/ssh_transport.dart @@ -126,6 +126,9 @@ class SSHTransport { /// Guards asynchronous packet processing to preserve message order. var _isProcessingData = false; + /// Tracks whether new socket data was received since packet processing started. + var _hasNewData = false; + /// Identification string sent by us without trailing \r\n. For example, /// "SSH-2.0-DartSSH_2.0". String get _localVersion => 'SSH-2.0-$version'; @@ -448,6 +451,12 @@ class SSHTransport { socket.destroy(); } + /// Force flush any buffered outgoing data to the socket. + Future flush() async { + await socket.flush(); + } + + /// Subscribes to the underlying socket stream to handle incoming data and status events. void _initSocket() { _socketSubscription = socket.stream.listen( _onSocketData, @@ -460,6 +469,7 @@ class SSHTransport { void _onSocketData(Uint8List data) { _buffer.add(data); + _hasNewData = true; _scheduleProcessData(); } @@ -479,6 +489,8 @@ class SSHTransport { } _isProcessingData = true; + final lengthBefore = _buffer.length; + _hasNewData = false; _processDataAsync().catchError((error, stackTrace) { if (error is SSHError) { @@ -489,7 +501,9 @@ class SSHTransport { }).whenComplete(() { _isProcessingData = false; if (_buffer.isNotEmpty && !isClosed) { - _scheduleProcessData(); + if (_hasNewData || _buffer.length < lengthBefore) { + _scheduleProcessData(); + } } }); } @@ -1517,6 +1531,7 @@ class SSHTransport { switch (_kexType) { case SSHKexType.x25519: + case SSHKexType.x25519Rfc: _kex = await SSHKexX25519.createAsync(); break; case SSHKexType.nistp256: diff --git a/pubspec.yaml b/pubspec.yaml index 2b88391..7a543a7 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -1,5 +1,5 @@ name: dartssh2 -version: 2.20.0 +version: 2.22.5 description: SSH and SFTP client written in pure Dart, aiming to be feature-rich as well as easy to use. homepage: https://github.com/TerminalStudio/dartssh2 diff --git a/test/src/algorithm/ssh_cipher_type_test.dart b/test/src/algorithm/ssh_cipher_type_test.dart index 900e2f1..239b9e2 100644 --- a/test/src/algorithm/ssh_cipher_type_test.dart +++ b/test/src/algorithm/ssh_cipher_type_test.dart @@ -105,9 +105,13 @@ void main() { test('Default values are set correctly', () { final algorithms = SSHAlgorithms(); + expect(SSHKexType.x25519Rfc.name, 'curve25519-sha256'); + expect(algorithms.kex.toNameList().first, 'curve25519-sha256'); + expect( algorithms.kex, equals([ + SSHKexType.x25519Rfc, SSHKexType.x25519, SSHKexType.nistp521, SSHKexType.nistp384, diff --git a/test/src/http/http_client_chunked_test.dart b/test/src/http/http_client_chunked_test.dart index c7f2ddb..c25677f 100644 --- a/test/src/http/http_client_chunked_test.dart +++ b/test/src/http/http_client_chunked_test.dart @@ -45,6 +45,9 @@ class _FakeSocket implements SSHSocket { @override Future get done => _controller.done; + @override + Future flush() async {} + @override StreamSink> get sink => _NullSink(); diff --git a/test/src/http/http_client_test.dart b/test/src/http/http_client_test.dart index 42e1195..327ce79 100644 --- a/test/src/http/http_client_test.dart +++ b/test/src/http/http_client_test.dart @@ -331,4 +331,7 @@ class _FakeSocket implements SSHSocket { } unawaited(_sinkController.close()); } + + @override + Future flush() async {} } diff --git a/test/src/sftp/sftp_client_protocol_test.dart b/test/src/sftp/sftp_client_protocol_test.dart index e40eeb8..e0c73ff 100644 --- a/test/src/sftp/sftp_client_protocol_test.dart +++ b/test/src/sftp/sftp_client_protocol_test.dart @@ -98,6 +98,37 @@ void main() { harness.dispose(); }); + test('close aborts a pending handshake', () async { + final harness = _SftpHarness(); + await harness.nextOutgoingPacket(); + + final handshakeExpectation = expectLater( + harness.client.handshake, + throwsA(isA()), + ); + final closeFuture = harness.client.close(); + harness.closeRemote(); + + await closeFuture; + await handshakeExpectation; + harness.dispose(); + }); + + test('close closes the underlying channel', () async { + final harness = _SftpHarness(); + await harness.nextOutgoingPacket(); + harness.sendResponsePacket(SftpVersionPacket(3)); + await harness.client.handshake; + + final closeFuture = harness.client.close(); + // The server acks the channel close, letting the teardown complete. + harness.closeRemote(); + await closeFuture; + + await expectLater(harness.channelDone, completes); + harness.dispose(); + }); + test('request waiter is registered before packet is sent', () async { final harness = _SftpHarness(); await harness.nextOutgoingPacket(); @@ -905,6 +936,14 @@ class _SftpHarness { Future nextOutgoingPacket() => _outgoing.stream.first; + Future get channelDone => _controller.channel.done; + + void closeRemote() { + _controller.handleMessage( + SSH_Message_Channel_Close(recipientChannel: _controller.localId), + ); + } + void respondDuringOutbound(SftpPacket Function(Uint8List payload) responder) { _outboundResponder = responder; } diff --git a/test/src/socket/dynamic_forward_io_test.dart b/test/src/socket/dynamic_forward_io_test.dart index 6698a28..3dccf09 100644 --- a/test/src/socket/dynamic_forward_io_test.dart +++ b/test/src/socket/dynamic_forward_io_test.dart @@ -369,6 +369,69 @@ void main() { expect(dialedHosts[0], '192.168.1.2'); expect(dialedHosts[1], contains(':')); }); + + test('closes remote sink when client EOF arrives during streaming', + () async { + late _DialedTunnel dialed; + + final forward = await startDynamicForward( + bindHost: '127.0.0.1', + bindPort: 0, + options: const SSHDynamicForwardOptions(), + dial: (_, __) async { + dialed = _DialedTunnel.create(); + return dialed.channel; + }, + ); + + final client = await Socket.connect(forward.host, forward.port); + final incoming = client.asBroadcastStream(); + addTearDown(() async { + await client.close(); + await forward.close(); + dialed.dispose(); + }); + + await _sendGreeting(client, incoming); + final reply = + await _sendConnectDomain(client, incoming, 'example.com', 443); + expect(reply[1], 0x00); + + // Send some data then close client side (half-close / EOF). + client.add(utf8.encode('data')); + await client.close(); + await dialed.remoteEof.timeout(const Duration(seconds: 1)); + }); + + test('handles handshake buffer overflow gracefully', () async { + final forward = await startDynamicForward( + bindHost: '127.0.0.1', + bindPort: 0, + options: const SSHDynamicForwardOptions(), + dial: (_, __) async => _DialedTunnel.create().channel, + ); + addTearDown(() => forward.close()); + + final client = await Socket.connect(forward.host, forward.port); + addTearDown(() => client.close()); + final incoming = client.asBroadcastStream(); + final clientDone = incoming.drain(); + + // Send a valid greeting but then flood the handshake buffer beyond + // kMaxHandshakeSize (32768). The server should close the connection + // rather than keep buffering indefinitely. + await _sendGreeting(client, incoming); + final huge = Uint8List(33000); + client.add(huge); + + // The overflow victim must be closed before the forward is reused. + await clientDone.timeout(const Duration(seconds: 1)); + + // Verify the forward still accepts new connections. + final client2 = await Socket.connect(forward.host, forward.port); + addTearDown(() => client2.close()); + await _sendGreeting(client2, client2.asBroadcastStream()); + }); }); } @@ -436,14 +499,21 @@ Future _readAtLeast( } class _DialedTunnel { - _DialedTunnel._(this.channel, this._controller, this.sentToRemote); + _DialedTunnel._( + this.channel, + this._controller, + this.sentToRemote, + this.remoteEof, + ); final SSHForwardChannel channel; final SSHChannelController _controller; final List sentToRemote; + final Future remoteEof; factory _DialedTunnel.create() { final sentToRemote = []; + final remoteEof = Completer(); final controller = SSHChannelController( localId: 1, @@ -455,6 +525,9 @@ class _DialedTunnel { sendMessage: (message) { if (message is SSH_Message_Channel_Data) { sentToRemote.addAll(message.data); + } else if (message is SSH_Message_Channel_EOF && + !remoteEof.isCompleted) { + remoteEof.complete(); } }, ); @@ -463,6 +536,7 @@ class _DialedTunnel { SSHForwardChannel(controller.channel), controller, sentToRemote, + remoteEof.future, ); } diff --git a/test/src/ssh_agent_test.dart b/test/src/ssh_agent_test.dart index 2a653f2..5a7acd0 100644 --- a/test/src/ssh_agent_test.dart +++ b/test/src/ssh_agent_test.dart @@ -289,4 +289,76 @@ void main() { controller.destroy(); }); + + test('SSHAgentChannel closes on invalid frame length (zero)', () async { + final handler = _RecordingAgentHandler(Uint8List.fromList([1])); + + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (_) {}, + ); + + SSHAgentChannel( + controller.channel, + handler, + printDebug: (_) {}, + ); + + // Frame with length = 0 (invalid). + final zeroFrame = Uint8List.fromList([0, 0, 0, 0]); + controller.handleMessage( + SSH_Message_Channel_Data( + recipientChannel: controller.localId, + data: zeroFrame, + ), + ); + + await controller.channel.done; + + // The channel should have been destroyed, no requests processed. + expect(handler.requests, isEmpty); + + controller.destroy(); + }); + + test('SSHAgentChannel closes on frame length exceeding maxFrameSize', + () async { + final handler = _RecordingAgentHandler(Uint8List.fromList([1])); + + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (_) {}, + ); + + SSHAgentChannel( + controller.channel, + handler, + printDebug: (_) {}, + ); + + // Frame claiming a length > maxFrameSize (256 * 1024 = 262144). + final oversizedFrame = Uint8List.fromList([0, 4, 0, 1]); + controller.handleMessage( + SSH_Message_Channel_Data( + recipientChannel: controller.localId, + data: oversizedFrame, + ), + ); + + await controller.channel.done; + + expect(handler.requests, isEmpty); + + controller.destroy(); + }); } diff --git a/test/src/ssh_auth_abort_error_test.dart b/test/src/ssh_auth_abort_error_test.dart index e400574..fd0b774 100644 --- a/test/src/ssh_auth_abort_error_test.dart +++ b/test/src/ssh_auth_abort_error_test.dart @@ -75,6 +75,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _RecordingSink implements StreamSink> { diff --git a/test/src/ssh_client_forward_dynamic_test.dart b/test/src/ssh_client_forward_dynamic_test.dart index 1df5076..8ed211a 100644 --- a/test/src/ssh_client_forward_dynamic_test.dart +++ b/test/src/ssh_client_forward_dynamic_test.dart @@ -68,6 +68,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _NoopSink implements StreamSink> { diff --git a/test/src/ssh_client_ident_test.dart b/test/src/ssh_client_ident_test.dart index ea6c57b..2b792de 100644 --- a/test/src/ssh_client_ident_test.dart +++ b/test/src/ssh_client_ident_test.dart @@ -100,6 +100,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _RecordingSink implements StreamSink> { diff --git a/test/src/ssh_client_run_with_result_test.dart b/test/src/ssh_client_run_with_result_test.dart index 417f939..d7601f4 100644 --- a/test/src/ssh_client_run_with_result_test.dart +++ b/test/src/ssh_client_run_with_result_test.dart @@ -8,6 +8,65 @@ import 'package:dartssh2/src/ssh_channel.dart'; import 'package:test/test.dart'; void main() { + group('SSHSession.waitForExit', () { + test('returns exit status after channel closes', () async { + final harness = _SessionHarness(); + + final exit = harness.session.waitForExit(); + harness.sendExitStatus(7); + harness.close(); + + expect(await exit, 7); + + harness.dispose(); + }); + + test('returns exit status before channel closes', () async { + final harness = _SessionHarness(); + + final exit = harness.session.waitForExit(); + harness.sendExitStatus(7); + + expect(await exit, 7); + + harness.dispose(); + }); + + test('returns null when timeout elapses before exit', () async { + final harness = _SessionHarness(); + + final exit = await harness.session.waitForExit( + timeout: const Duration(milliseconds: 10), + ); + + expect(exit, isNull); + expect(harness.session.exitCode, isNull); + + harness.dispose(); + }); + + test('returns null when channel closes without exit status', () async { + final harness = _SessionHarness(); + + final exit = harness.session.waitForExit(); + harness.close(); + + expect(await exit, isNull); + + harness.dispose(); + }); + + test('returns existing exit status immediately', () async { + final harness = _SessionHarness(); + + harness.sendExitStatus(3); + + expect(await harness.session.waitForExit(), 3); + + harness.dispose(); + }); + }); + group('SSHClient.runWithResult', () { test('captures stdout/stderr and exit status', () async { final harness = _SessionHarness(); @@ -225,6 +284,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _NoopSink implements StreamSink> { diff --git a/test/src/ssh_client_test.dart b/test/src/ssh_client_test.dart index f952fc0..3f4b052 100644 --- a/test/src/ssh_client_test.dart +++ b/test/src/ssh_client_test.dart @@ -4,6 +4,7 @@ library; import 'dart:convert'; import 'package:dartssh2/dartssh2.dart'; +import 'package:dartssh2/src/ssh_channel.dart'; import 'package:test/test.dart'; import '../test_utils.dart'; @@ -271,4 +272,32 @@ void main() { client.close(); }); }); + + group('SSHClient.flush', () { + test('can flush client, session, channel, and forward channel', () async { + final client = await getTestClient(); + await client.authenticated; + + await client.flush(); + + final session = await client.execute('echo flush_test'); + await session.flush(); + + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (msg) {}, + onFlush: () async {}, + ); + final forwardChannel = SSHForwardChannel(controller.channel); + await forwardChannel.flush(); + + await session.done; + client.close(); + }); + }); } diff --git a/test/src/ssh_client_timeout_test.dart b/test/src/ssh_client_timeout_test.dart new file mode 100644 index 0000000..41ba404 --- /dev/null +++ b/test/src/ssh_client_timeout_test.dart @@ -0,0 +1,106 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:mirrors'; +import 'dart:typed_data'; + +import 'package:dartssh2/dartssh2.dart'; +import 'package:test/test.dart'; + +void main() { + final clientLibrary = reflectClass(SSHClient).owner as LibraryMirror; + Symbol privateSymbol(String name) => + MirrorSystem.getSymbol(name, clientLibrary); + + group('SSHClient timeouts', () { + test('fails authentication future when handshake times out', () async { + final socket = _FakeSSHSocket(); + final client = SSHClient( + socket, + username: 'demo', + handshakeTimeout: const Duration(milliseconds: 10), + ); + + await expectLater( + client.authenticated, + throwsA(isA()), + ); + + client.close(); + }); + + test('fails authentication future when auth times out', () async { + final socket = _FakeSSHSocket(); + final client = SSHClient( + socket, + username: 'demo', + authTimeout: const Duration(milliseconds: 10), + ); + + reflect(client).invoke(privateSymbol('_handleTransportReady'), const []); + + await expectLater( + client.authenticated, + throwsA(isA()), + ); + + client.close(); + }); + }); +} + +class _FakeSSHSocket implements SSHSocket { + final _inputController = StreamController(); + final _doneCompleter = Completer(); + final _sink = _RecordingSink(); + + @override + Stream get stream => _inputController.stream; + + @override + StreamSink> get sink => _sink; + + @override + Future get done => _doneCompleter.future; + + @override + Future close() async { + if (!_doneCompleter.isCompleted) { + _doneCompleter.complete(); + } + await _inputController.close(); + } + + @override + void destroy() { + if (!_doneCompleter.isCompleted) { + _doneCompleter.complete(); + } + unawaited(_inputController.close()); + } + + @override + Future flush() async {} +} + +class _RecordingSink implements StreamSink> { + @override + void add(List data) { + latin1.decode(data); + } + + @override + void addError(Object error, [StackTrace? stackTrace]) {} + + @override + Future addStream(Stream> stream) async { + await for (final data in stream) { + add(data); + } + } + + @override + Future close() async {} + + @override + Future get done async {} +} diff --git a/test/src/ssh_flush_test.dart b/test/src/ssh_flush_test.dart new file mode 100644 index 0000000..2ad377e --- /dev/null +++ b/test/src/ssh_flush_test.dart @@ -0,0 +1,179 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:dartssh2/dartssh2.dart'; +import 'package:dartssh2/src/message/base.dart'; +import 'package:dartssh2/src/ssh_channel.dart'; +import 'package:test/test.dart'; + +void main() { + group('SSHSocket base class', () { + test('default flush implementation completes normally', () async { + final socket = _TestSSHSocket(); + await expectLater(socket.flush(), completes); + }); + }); + + group('SSHNativeSocket', () { + test('native socket flush works', () async { + final server = await ServerSocket.bind('127.0.0.1', 0); + final socket = await SSHSocket.connect('127.0.0.1', server.port); + await socket.flush(); + await socket.close(); + await server.close(); + }); + }); + + group('SSHTransport.flush', () { + test('delegates to socket.flush', () async { + final socket = _FakeSSHSocket(); + final transport = SSHTransport(socket); + await transport.flush(); + expect(socket.flushCount, 1); + transport.close(); + }); + }); + + group('SSHClient.flush and channel callback delegation', () { + test('delegates to transport and sets up onFlush', () async { + final socket = _FakeSSHSocket(); + final client = SSHClient(socket, username: 'demo'); + await client.flush(); + expect(socket.flushCount, 1); + + final channelController = client.acceptChannelForTesting( + localChannelId: 1, + remoteChannelId: 2, + remoteInitialWindowSize: 1024, + remoteMaximumPacketSize: 1024, + ); + + expect(channelController.onFlush, isNotNull); + await channelController.flush(); + expect(socket.flushCount, 2); + + client.close(); + }); + }); + + group('SSHChannel.flush', () { + test('delegates to controller.flush', () async { + var flushed = false; + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (msg) {}, + onFlush: () async { + flushed = true; + }, + ); + final channel = controller.channel; + await channel.flush(); + expect(flushed, isTrue); + }); + }); + + group('SSHSession.flush', () { + test('exposes channel and delegates flush', () async { + var flushed = false; + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (msg) {}, + onFlush: () async { + flushed = true; + }, + ); + final channel = controller.channel; + final session = SSHSession(channel); + expect(session.channel, channel); + + await session.flush(); + expect(flushed, isTrue); + }); + }); + + group('SSHForwardChannel.flush', () { + test('flushes after data exactly exhausts the remote window', () async { + var flushed = false; + final sentData = []; + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 3, + remoteInitialWindowSize: 3, + sendMessage: (message) { + if (message is SSH_Message_Channel_Data) { + sentData.addAll(message.data); + } + }, + onFlush: () async { + flushed = true; + }, + ); + final forwardChannel = SSHForwardChannel(controller.channel); + forwardChannel.sink.add([1, 2, 3]); + await forwardChannel.flush().timeout(const Duration(seconds: 1)); + + expect(sentData, [1, 2, 3]); + expect(flushed, isTrue); + }); + }); +} + +class _TestSSHSocket extends SSHSocket { + @override + Stream get stream => throw UnimplementedError(); + + @override + StreamSink> get sink => throw UnimplementedError(); + + @override + Future get done => throw UnimplementedError(); + + @override + Future close() => throw UnimplementedError(); + + @override + void destroy() => throw UnimplementedError(); +} + +class _FakeSSHSocket implements SSHSocket { + int flushCount = 0; + final _streamController = StreamController(); + final _sinkController = StreamController>(); + + @override + Stream get stream => _streamController.stream; + + @override + StreamSink> get sink => _sinkController.sink; + + @override + Future get done => _streamController.done; + + @override + Future close() async { + await _streamController.close(); + await _sinkController.close(); + } + + @override + void destroy() {} + + @override + Future flush() async { + flushCount++; + } +} diff --git a/test/src/ssh_keepalive_test.dart b/test/src/ssh_keepalive_test.dart new file mode 100644 index 0000000..16815ac --- /dev/null +++ b/test/src/ssh_keepalive_test.dart @@ -0,0 +1,84 @@ +import 'dart:async'; + +import 'package:dartssh2/src/ssh_keepalive.dart'; +import 'package:test/test.dart'; + +void main() { + group('SSHKeepAlive', () { + test('calls ping at specified interval', () async { + var pingCount = 0; + final completer = Completer(); + + final keepAlive = SSHKeepAlive( + interval: const Duration(milliseconds: 10), + ping: () async { + pingCount++; + if (pingCount >= 3) { + completer.complete(); + } + }, + ); + + keepAlive.start(); + await completer.future; + keepAlive.stop(); + + expect(pingCount, greaterThanOrEqualTo(3)); + }); + + test('prevents overlapping pings', () async { + var pingCount = 0; + var activePings = 0; + var maxActivePings = 0; + final completer = Completer(); + + final keepAlive = SSHKeepAlive( + interval: const Duration(milliseconds: 10), + ping: () async { + pingCount++; + activePings++; + if (activePings > maxActivePings) { + maxActivePings = activePings; + } + // Sleep longer than the interval to cause an overlapping tick + await Future.delayed(const Duration(milliseconds: 50)); + activePings--; + if (pingCount >= 2 && !completer.isCompleted) { + completer.complete(); + } + }, + ); + + keepAlive.start(); + // Wait for a few intervals. If overlapping wasn't prevented, activePings would exceed 1. + await Future.delayed(const Duration(milliseconds: 100)); + keepAlive.stop(); + + expect(maxActivePings, equals(1)); + }); + + test('handles ping errors and resets isPinging status', () async { + var pingCount = 0; + final completer = Completer(); + + final keepAlive = SSHKeepAlive( + interval: const Duration(milliseconds: 10), + ping: () async { + pingCount++; + if (pingCount == 1) { + throw Exception('ping failed'); + } + if (pingCount == 2) { + completer.complete(); + } + }, + ); + + keepAlive.start(); + await completer.future; + keepAlive.stop(); + + expect(pingCount, equals(2)); + }); + }); +} diff --git a/test/src/ssh_transport_aead_test.dart b/test/src/ssh_transport_aead_test.dart index 89ae7bc..8afeaa1 100644 --- a/test/src/ssh_transport_aead_test.dart +++ b/test/src/ssh_transport_aead_test.dart @@ -789,6 +789,9 @@ class _CaptureSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _CaptureSink implements StreamSink> { diff --git a/test/src/ssh_transport_version_test.dart b/test/src/ssh_transport_version_test.dart index ce37450..a92c274 100644 --- a/test/src/ssh_transport_version_test.dart +++ b/test/src/ssh_transport_version_test.dart @@ -43,6 +43,62 @@ void main() { client.close(); }); + + test('does not busy loop on partial packet after handshake', () async { + final socket = _FakeSSHSocket(); + final client = SSHClient( + socket, + username: 'demo', + ); + + // Complete the version exchange. + socket.addIncoming('SSH-2.0-OpenSSH_3.6.1p2\r\n'); + await _pumpUntil(() => client.remoteVersion != null); + + // Send the first 4 bytes of a packet indicating a length of 100. + socket.addRawIncoming(Uint8List.fromList([0, 0, 0, 100])); + + // Wait a moment. If there is a microtask busy loop, the delayed future + // will never run and the test will timeout. + await Future.delayed(const Duration(milliseconds: 50)); + + client.close(); + }); + + test('reschedules processing when more data remains in the buffer', + () async { + final socket = _FakeSSHSocket(); + final client = SSHClient( + socket, + username: 'demo', + ); + + // Send the version banner followed by an invalid packet length in the + // same chunk. Processing must continue into the queued remainder and + // surface the packet error instead of stopping after the banner. + socket.addRawIncoming( + Uint8List.fromList([ + ...latin1.encode('SSH-2.0-OpenSSH_3.6.1p2\r\n'), + 0, + 0, + 0, + 0, + ]), + ); + + await expectLater( + client.authenticated, + throwsA( + predicate((error) { + return error is SSHAuthAbortError && error.reason is SSHPacketError; + }), + ), + ); + + expect(client.remoteVersion, 'SSH-2.0-OpenSSH_3.6.1p2'); + + client.close(); + }); }); } @@ -74,6 +130,10 @@ class _FakeSSHSocket implements SSHSocket { _inputController.add(Uint8List.fromList(latin1.encode(data))); } + void addRawIncoming(Uint8List data) { + _inputController.add(data); + } + @override Future close() async { if (!_doneCompleter.isCompleted) { @@ -89,6 +149,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _RecordingSink implements StreamSink> { diff --git a/test/src/ssh_userauth_export_test.dart b/test/src/ssh_userauth_export_test.dart new file mode 100644 index 0000000..683f788 --- /dev/null +++ b/test/src/ssh_userauth_export_test.dart @@ -0,0 +1,25 @@ +import 'package:dartssh2/dartssh2.dart'; +import 'package:test/test.dart'; + +void main() { + test( + 'SSHUserInfoRequest and related userauth classes are exported by dartssh2.dart', + () { + final prompt = SSHUserInfoPrompt('Password:', false); + expect(prompt.promptText, equals('Password:')); + expect(prompt.echo, isFalse); + + final request = SSHUserInfoRequest('Title', 'Instruction', [prompt]); + expect(request.name, equals('Title')); + expect(request.instruction, equals('Instruction')); + expect(request.prompts.length, equals(1)); + expect(request.prompts.first.promptText, equals('Password:')); + + final changePasswordResponse = SSHChangePasswordResponse('old', 'new'); + expect(changePasswordResponse.oldPassword, equals('old')); + expect(changePasswordResponse.newPassword, equals('new')); + + expect( + SSHAuthMethod.keyboardInteractive.name, equals('keyboard-interactive')); + }); +}