diff --git a/lib/core/security/ssh_tunnel_manager.dart b/lib/core/security/ssh_tunnel_manager.dart index 084cd408..eadd657c 100644 --- a/lib/core/security/ssh_tunnel_manager.dart +++ b/lib/core/security/ssh_tunnel_manager.dart @@ -143,6 +143,7 @@ class SshTunnelManager { final SshClientBuilder _buildClient; final Map _sessions = {}; + final Map> _pendingTunnels = {}; /// Number of pooled tunnel sessions currently open. @visibleForTesting @@ -249,193 +250,266 @@ class SshTunnelManager { ); } - // 2. Connect to SSH Bastion (with optional Jump Host) - SSHClient? jumpClient; - final SSHSocket bastionSocket; - - if (config.jumpHost != null && config.jumpHost!.trim().isNotEmpty) { - final jumpHost = config.jumpHost!.trim(); - final jumpPort = config.jumpPort ?? 22; - final jumpUser = config.jumpUsername ?? config.username; - - final rawJumpSocket = await _dial( - jumpHost, - jumpPort, - Duration(seconds: config.connectTimeoutSeconds), - what: 'SSH jump host', - ); - - jumpClient = _buildClient( - rawJumpSocket, - username: jumpUser, - onPasswordRequest: () => - secrets.jumpPassword ?? secrets.password ?? '', + // Single-flight: if another caller is already dialing this poolKey, wait for it + final pending = _pendingTunnels[poolKey]; + if (pending != null) { + await pending; + // After pending finishes, re-call openTunnel to increment refCount and return handle + return openTunnel( + config: config, + secrets: secrets, + remoteHost: remoteHost, + remotePort: remotePort, ); - try { - await jumpClient.authenticated; - } catch (e) { - unawaited(jumpClient.close()); - throw _sshFailure(e, jumpUser, jumpHost, jumpPort); - } + } - // Forward through jump host to target bastion (returns SSHForwardChannel which implements SSHSocket) - try { - bastionSocket = - await jumpClient.forwardLocal(config.host.trim(), config.port); - } catch (e) { - unawaited(jumpClient.close()); - throw SshConnectionException( - 'The SSH jump host $jumpHost:$jumpPort could not reach the bastion ' - '${config.host.trim()}:${config.port}: $e', - ); + final future = _dialAndOpenTunnel( + poolKey: poolKey, + config: config, + secrets: secrets, + remoteHost: remoteHost, + remotePort: remotePort, + ); + _pendingTunnels[poolKey] = future; + try { + return await future; + } finally { + if (identical(_pendingTunnels[poolKey], future)) { + unawaited(_pendingTunnels.remove(poolKey)); } - } else { - bastionSocket = await _dial( - config.host.trim(), - config.port, - Duration(seconds: config.connectTimeoutSeconds), - what: 'SSH server', - ); } + } - // Prepare identities for private key auth - List identities = []; - if (config.authType == SshAuthType.privateKey) { - String? keyContent = secrets.privateKey; - if ((keyContent == null || keyContent.isEmpty) && - config.privateKeyPath != null && - config.privateKeyPath!.trim().isNotEmpty) { - final file = File(config.privateKeyPath!.trim()); - if (await file.exists()) { + Future _dialAndOpenTunnel({ + required String poolKey, + required SshTunnelConfig config, + required SshTunnelSecrets secrets, + required String remoteHost, + required int remotePort, + }) async { + SSHClient? jumpClient; + SSHSocket? jumpSocket; + SSHSocket? bastionSocket; + SSHClient? client; + ServerSocket? serverSocket; + Timer? keepAliveTimer; + + try { + // Resolve the private key first: a missing or unreadable key must fail + // before any socket is opened. + // Prepare identities for private key auth + List identities = []; + if (config.authType == SshAuthType.privateKey) { + String? keyContent = secrets.privateKey; + if ((keyContent == null || keyContent.isEmpty) && + config.privateKeyPath != null && + config.privateKeyPath!.trim().isNotEmpty) { + final keyPath = config.privateKeyPath!.trim(); + final file = File(keyPath); + if (!await file.exists()) { + throw SshAuthenticationException( + 'SSH private key file not found at path: $keyPath', + ); + } keyContent = await file.readAsString(); + if (keyContent.trim().isEmpty) { + throw SshAuthenticationException( + 'SSH private key file is empty at path: $keyPath', + ); + } } - } - if (keyContent != null && keyContent.isNotEmpty) { - try { - identities = SSHKeyPair.fromPem( - keyContent, - secrets.passphrase, - ); - } catch (e) { - throw SshAuthenticationException( - 'Failed to parse private key: $e', + if (keyContent != null && keyContent.isNotEmpty) { + try { + identities = SSHKeyPair.fromPem( + keyContent, + secrets.passphrase, + ); + } catch (e) { + throw SshAuthenticationException( + 'Failed to parse private key: $e', + ); + } + } else if (secrets.privateKey == null && + (config.privateKeyPath == null || + config.privateKeyPath!.trim().isEmpty)) { + throw const SshAuthenticationException( + 'No SSH private key or key path was provided', ); } } - } - // Host key verification (MitM protection) - String? observedFingerprint; - Future handleVerifyHostKey(String type, Uint8List fingerprint) async { - observedFingerprint = formatFingerprint(fingerprint); - final pinned = config.knownHostFingerprint; - if (pinned == null || pinned.trim().isEmpty) { - return true; // TOFU / accept-new - } - if (!fingerprintMatches(pinned, fingerprint)) { - debugPrint( - 'SSH Host Key Mismatch! Expected: ${pinned.trim()}, ' - 'Actual: $observedFingerprint', + // 2. Connect to SSH Bastion (with optional Jump Host) + if (config.jumpHost != null && config.jumpHost!.trim().isNotEmpty) { + final jumpHost = config.jumpHost!.trim(); + final jumpPort = config.jumpPort ?? 22; + final jumpUser = config.jumpUsername ?? config.username; + + jumpSocket = await _dial( + jumpHost, + jumpPort, + Duration(seconds: config.connectTimeoutSeconds), + what: 'SSH jump host', ); - return false; - } - return true; - } - final client = _buildClient( - bastionSocket, - username: config.username.trim(), - onPasswordRequest: () => secrets.password ?? '', - identities: identities.isNotEmpty ? identities : null, - onVerifyHostKey: handleVerifyHostKey, - ); + jumpClient = _buildClient( + jumpSocket, + username: jumpUser, + onPasswordRequest: () => + secrets.jumpPassword ?? secrets.password ?? '', + ); + try { + await jumpClient.authenticated; + } catch (e) { + throw _sshFailure(e, jumpUser, jumpHost, jumpPort); + } - try { - await client.authenticated; - } catch (e) { - unawaited(client.close()); - unawaited(jumpClient?.close()); - if (observedFingerprint != null && - config.knownHostFingerprint != null && - config.knownHostFingerprint!.isNotEmpty) { - throw SshHostKeyMismatchException( - 'SSH host key verification failed for ${config.host}. ' - 'Observed fingerprint: $observedFingerprint', + // Forward through jump host to target bastion (returns SSHForwardChannel which implements SSHSocket) + try { + bastionSocket = + await jumpClient.forwardLocal(config.host.trim(), config.port); + } catch (e) { + throw SshConnectionException( + 'The SSH jump host $jumpHost:$jumpPort could not reach the bastion ' + '${config.host.trim()}:${config.port}: $e', + ); + } + } else { + bastionSocket = await _dial( + config.host.trim(), + config.port, + Duration(seconds: config.connectTimeoutSeconds), + what: 'SSH server', ); } - throw _sshFailure(e, config.username, config.host, config.port); - } - // Zero sensitive in-memory credentials immediately after successful authentication - secrets.zero(); + // Host key verification (MitM protection) + String? observedFingerprint; + Future handleVerifyHostKey(String type, Uint8List fingerprint) async { + observedFingerprint = formatFingerprint(fingerprint); + final pinned = config.knownHostFingerprint; + if (pinned == null || pinned.trim().isEmpty) { + return true; // TOFU / accept-new + } + if (!fingerprintMatches(pinned, fingerprint)) { + debugPrint( + 'SSH Host Key Mismatch! Expected: ${pinned.trim()}, ' + 'Actual: $observedFingerprint', + ); + return false; + } + return true; + } - // 3. Start local ephemeral port forwarding on 127.0.0.1:0 - final serverSocket = await ServerSocket.bind( - InternetAddress.loopbackIPv4, - 0, - ); - final localPort = serverSocket.port; + client = _buildClient( + bastionSocket, + username: config.username.trim(), + onPasswordRequest: () => secrets.password ?? '', + identities: identities.isNotEmpty ? identities : null, + onVerifyHostKey: handleVerifyHostKey, + ); - serverSocket.listen( - (clientSocket) async { - try { - final forward = await client.forwardLocal(remoteHost, remotePort); - unawaited(forward.stream.cast>().pipe(clientSocket).catchError((_) {})); - unawaited(clientSocket.cast>().pipe(forward.sink).catchError((_) {})); - } catch (e) { - clientSocket.destroy(); + try { + await client.authenticated; + } catch (e) { + if (observedFingerprint != null && + config.knownHostFingerprint != null && + config.knownHostFingerprint!.isNotEmpty) { + throw SshHostKeyMismatchException( + 'SSH host key verification failed for ${config.host}. ' + 'Observed fingerprint: $observedFingerprint', + ); } - }, - onError: (_) {}, - ); + throw _sshFailure(e, config.username, config.host, config.port); + } - // 4. Setup keep-alive ping timer - Timer? keepAliveTimer; - _PooledTunnelSession? session; - if (config.keepAliveIntervalSeconds > 0) { - var failedPings = 0; - keepAliveTimer = Timer.periodic( - Duration(seconds: config.keepAliveIntervalSeconds), - (_) async { - if (client.isClosed) { - if (session != null) { - await _dropZombieSession(session); - } - return; - } + // Zero sensitive in-memory credentials immediately after successful authentication + secrets.zero(); + + // 3. Start local ephemeral port forwarding on 127.0.0.1:0 + serverSocket = await ServerSocket.bind( + InternetAddress.loopbackIPv4, + 0, + ); + final localPort = serverSocket.port; + + serverSocket.listen( + (clientSocket) async { try { - await client.ping(); - failedPings = 0; - } catch (_) { - failedPings++; - if (failedPings >= _maxKeepAlivePingFailures) { + final forward = await client!.forwardLocal(remoteHost, remotePort); + unawaited(forward.stream.cast>().pipe(clientSocket).catchError((_) {})); + unawaited(clientSocket.cast>().pipe(forward.sink).catchError((_) {})); + } catch (e) { + clientSocket.destroy(); + } + }, + onError: (_) {}, + ); + + // 4. Setup keep-alive ping timer + _PooledTunnelSession? session; + if (config.keepAliveIntervalSeconds > 0) { + var failedPings = 0; + keepAliveTimer = Timer.periodic( + Duration(seconds: config.keepAliveIntervalSeconds), + (_) async { + if (client!.isClosed) { if (session != null) { await _dropZombieSession(session); } + return; } - } - }, - ); - } + try { + await client.ping(); + failedPings = 0; + } catch (_) { + failedPings++; + if (failedPings >= _maxKeepAlivePingFailures) { + if (session != null) { + await _dropZombieSession(session); + } + } + } + }, + ); + } - session = _PooledTunnelSession( - poolKey: poolKey, - client: client, - serverSocket: serverSocket, - localPort: localPort, - jumpClient: jumpClient, - keepAliveTimer: keepAliveTimer, - ); - _sessions[poolKey] = session; + session = _PooledTunnelSession( + poolKey: poolKey, + client: client, + serverSocket: serverSocket, + localPort: localPort, + jumpClient: jumpClient, + keepAliveTimer: keepAliveTimer, + ); + _sessions[poolKey] = session; - return SshTunnelHandle( - localHost: '127.0.0.1', - localPort: localPort, - remoteHost: remoteHost, - remotePort: remotePort, - onRelease: () => _releaseSession(session!), - ); + return SshTunnelHandle( + localHost: '127.0.0.1', + localPort: localPort, + remoteHost: remoteHost, + remotePort: remotePort, + onRelease: () => _releaseSession(session!), + ); + } catch (_) { + keepAliveTimer?.cancel(); + try { + await serverSocket?.close(); + } catch (_) {} + try { + unawaited(client?.close()); + } catch (_) {} + try { + bastionSocket?.destroy(); + } catch (_) {} + try { + unawaited(jumpClient?.close()); + } catch (_) {} + try { + jumpSocket?.destroy(); + } catch (_) {} + rethrow; + } } /// Number of consecutive failed keep-alive pings before considering the diff --git a/test/core/security/ssh_tunnel_manager_test.dart b/test/core/security/ssh_tunnel_manager_test.dart index 46bf40d7..9c34eb43 100644 --- a/test/core/security/ssh_tunnel_manager_test.dart +++ b/test/core/security/ssh_tunnel_manager_test.dart @@ -790,4 +790,127 @@ void main() { expect(await _accepts(handle.localPort), isTrue); }); }); + + group('robustness and resource hygiene (#1312)', () { + test('missing private key file fails clearly and reports the path', () async { + final missingPath = '/nonexistent/path/to/id_ed25519_${DateTime.now().microsecondsSinceEpoch}'; + await expectLater( + open( + config: _config( + authType: SshAuthType.privateKey, + privateKeyPath: missingPath, + ), + secrets: SshTunnelSecrets(), + ), + throwsA( + isA().having( + (e) => e.message, + 'message', + allOf( + contains('not found'), + contains(missingPath), + ), + ), + ), + ); + expect(manager.activeSessionCount, 0); + expect(server.connects, isEmpty); + }); + + test('empty private key file fails clearly with descriptive message', () async { + final tempDir = await Directory.systemTemp.createTemp('empty_key_test_'); + final emptyFile = File(p.join(tempDir.path, 'id_empty'))..writeAsStringSync(' \n'); + try { + await expectLater( + open( + config: _config( + authType: SshAuthType.privateKey, + privateKeyPath: emptyFile.path, + ), + secrets: SshTunnelSecrets(), + ), + throwsA( + isA().having( + (e) => e.message, + 'message', + allOf( + contains('empty'), + contains(emptyFile.path), + ), + ), + ), + ); + } finally { + tempDir.deleteSync(recursive: true); + } + expect(manager.activeSessionCount, 0); + expect(server.connects, isEmpty); + }); + + test('failure during bastion authentication cleans up socket and jump host', () async { + server.userPasswords['jumper'] = 'jump-pass'; + server.userPasswords['deploy'] = 'real-secret'; + + await expectLater( + open( + config: _config(jumpHost: 'jump.example', jumpUsername: 'jumper'), + secrets: SshTunnelSecrets( + password: 'wrong-bastion-password', + jumpPassword: 'jump-pass', + ), + ), + throwsA(isA()), + ); + + expect(manager.activeSessionCount, 0); + expect(server.clients, hasLength(2)); + expect(server.clients.every((c) => c.isClosed), isTrue); + expect(server.sockets.every((s) => s.isClosed), isTrue); + }); + + test('failure during jump host forwardLocal closes jump client and socket', () async { + server.userPasswords['jumper'] = 'jump-pass'; + server.failForwardLocal = StateError('forwarding disabled on jump host'); + + await expectLater( + open( + config: _config(jumpHost: 'jump.example', jumpUsername: 'jumper'), + secrets: SshTunnelSecrets( + password: 'bastion-password', + jumpPassword: 'jump-pass', + ), + ), + throwsA(isA()), + ); + + expect(manager.activeSessionCount, 0); + expect(server.clients, hasLength(1)); + expect(server.clients.single.isClosed, isTrue); + expect(server.sockets.single.isClosed, isTrue); + }); + + test('concurrent openTunnel calls for the same target dial once and share session', () async { + server.connectDelay = const Duration(milliseconds: 100); + + final future1 = open(); + final future2 = open(); + + final results = await Future.wait([future1, future2]); + final h1 = results[0]; + final h2 = results[1]; + + expect(h1.localPort, h2.localPort); + expect(manager.activeSessionCount, 1); + expect(server.clients, hasLength(1)); + expect(server.connects, hasLength(1)); + + await h1.release(); + expect(manager.activeSessionCount, 1); + expect(await _accepts(h2.localPort), isTrue); + + await h2.release(); + expect(manager.activeSessionCount, 0); + expect(await _accepts(h2.localPort), isFalse); + }); + }); } diff --git a/test/support/fake_ssh.dart b/test/support/fake_ssh.dart index 7f0244bc..2e92b396 100644 --- a/test/support/fake_ssh.dart +++ b/test/support/fake_ssh.dart @@ -53,6 +53,12 @@ class FakeSshServer { /// Every client the manager built, in order. final clients = []; + /// Every socket created by connect, in order. + final sockets = []; + + /// If non-null, forwardLocal throws this error. + Object? failForwardLocal; + Future connect( String host, int port, { @@ -62,7 +68,9 @@ class FakeSshServer { final delay = connectDelay; if (delay != null) await Future.delayed(delay); if (failConnect) throw StateError('connect to $host:$port failed'); - return FakeSshSocket(); + final s = FakeSshSocket(); + sockets.add(s); + return s; } SSHClient build( @@ -94,6 +102,9 @@ class FakeSshSocket implements SSHSocket { // ignore: close_sinks final _out = StreamController>(); + var isClosed = false; + var isDestroyed = false; + @override Stream get stream => _in.stream; @@ -104,10 +115,15 @@ class FakeSshSocket implements SSHSocket { Future get done => Future.value(); @override - Future close() async {} + Future close() async { + isClosed = true; + } @override - void destroy() {} + void destroy() { + isDestroyed = true; + isClosed = true; + } @override Future flush() async {} @@ -218,6 +234,8 @@ class FakeSshClient implements SSHClient { int localPort = 0, }) async { if (_closed) throw StateError('client closed'); + final fail = server.failForwardLocal; + if (fail != null) throw fail; forwards.add((host: remoteHost, port: remotePort)); return _EchoForwardChannel(received); }