feat!: keep the SOCKSSocket transport private

Release as 2.0.0 and document migration to managed I/O, teardown, and
TLS. Remove close-time flush retries for operations that callers can no
longer start outside the wrapper. Tests use the wrapper APIs instead of
the removed getter.

BREAKING CHANGE: remove SOCKSSocket.socket; use the wrapper APIs and
built-in TLS instead.

Author: sneurlax

CHANGELOG.md

@@ -1,5 +1,7 @@
-## Unreleased
+## 2.0.0
 
+- **Breaking:** Remove `SOCKSSocket.socket` so transport operations cannot bypass the wrapper's write queue and lifecycle tracking. Use `destroy()`, `close()`, `write()`, `outputStream`, and `inputStream` instead.
+- **Breaking:** Caller-managed TLS through `SecureSocket.secure(socks.socket)` is no longer supported. Use `SOCKSSocket.create(sslEnabled: true, securityContext: ...)` to negotiate TLS during `connectTo()`.
 - Add `destroy()` to abort without draining output; writes it cuts short fail instead of reporting success (#2). It also ends a connect, TLS handshake or `reconnect()` in flight, and a later `close()` completes normally even when it failed a close that was already draining.
 - Add `closeOnPeerEof` to `create()`; pass `false` to keep writing after the peer half-closes (#3).
 - `close()` now ends a `reconnect()` or TLS handshake in flight instead of being overtaken by the reconnect or waiting for the handshake deadline.

README.md

@@ -97,6 +97,25 @@ By default the connection closes once the peer closes its side. Pass `closeOnPee
 
 `destroy()` aborts a connection without draining output; pending writes fail, and a connect, TLS handshake or `reconnect()` in flight ends with `SocksCancelledException`. Use `close()` to drain accepted output before closing; it ends a connect or `reconnect()` in flight the same way.
 
+## Migrating to 2.0.0
+
+`SOCKSSocket.socket` has been removed. All transport operations now go through the wrapper so writes and teardown share the same lifecycle tracking.
+
+| Previous operation | Replacement |
+| --- | --- |
+| `socks.socket.destroy()` | `socks.destroy()` |
+| `socks.socket.close()` | `await socks.close()` |
+| `socks.socket.add(bytes)` | `socks.outputStream.add(bytes)` |
+| `socks.socket.addStream(source)` | `await socks.outputStream.addStream(source)` |
+| Reading the underlying socket | `socks.inputStream` or `socks.listen(...)` |
+| `SecureSocket.secure(socks.socket, ...)` | Set `sslEnabled: true` and, if needed, `securityContext` on `SOCKSSocket.create(...)` |
+
+For an awaited binary write, use `await socks.outputStream.addStream(Stream.value(bytes))`. The output sink accepts one stream at a time. For text, use `await socks.write(text)`. To drain and finish the connection, use `await socks.close()`.
+
+Built-in TLS starts during `connectTo()`, after the SOCKS handshake. Upgrading an established plaintext application session to TLS is not exposed by `SOCKSSocket`.
+
+`SocksConnection.start` still returns a `ConnectionTask<Socket>` for `HttpClient.connectionFactory`; that separate API is unchanged.
+
 ## HttpClient Connections
 
 `SocksConnection.start` returns a cancellable `ConnectionTask<Socket>` for `HttpClient.connectionFactory`; see `example/http/http_connection.dart`. Pass `tlsHost` for HTTPS. Requires Dart 3.5.

lib/socks_socket.dart

@@ -131,12 +131,7 @@ class SOCKSSocket {
   /// The underlying Socket that connects to the SOCKS5 proxy server.
   late Socket _socksSocket;
 
-  /// Getter for the underlying Socket that connects to the SOCKS5 proxy server.
-  ///
-  /// To abort the connection, call [destroy] rather than `socket.destroy()`:
-  /// dart:io completes a pending flush on a destroyed [Socket] normally, so
-  /// writes cut short that way would be reported as sent.
-  Socket get socket => sslEnabled ? _secureSocksSocket : _socksSocket;
+  Socket get _socket => sslEnabled ? _secureSocksSocket : _socksSocket;
 
   /// A wrapper around the _socksSocket that enables SSL connections.
   late Socket _secureSocksSocket;
@@ -402,7 +397,6 @@ class SOCKSSocket {
       _channel = RawChannel(raw);
       _socksSocket = _channel!.socket();
     } else {
-      // Keep a native Socket so callers can use SecureSocket.secure(socket).
       _socksSocket = await _connectProxy(Socket.startConnect);
       _nativeSocketOpen = true;
       _nativeSocketNeedsDestroy = true;
@@ -807,8 +801,8 @@ class SOCKSSocket {
         Error.throwWithStackTrace(_writeFailure!, _writeFailureStack!);
       }
       try {
-        socket.add(bytes);
-        await socket.flush().timeout(_operationTimeout);
+        _socket.add(bytes);
+        await _socket.flush().timeout(_operationTimeout);
         // A transport destroyed mid-flush can still complete the flush.
         if (_writeFailure != null) {
           Error.throwWithStackTrace(_writeFailure!, _writeFailureStack!);
@@ -869,32 +863,6 @@ class SOCKSSocket {
     _closeFuture = (_closeFuture ?? _close()).catchError((Object _) {});
   }
 
-  Future<void> _flushBeforeClose(Socket transport) async {
-    final elapsed = Stopwatch()..start();
-    while (true) {
-      final remaining = _operationTimeout - elapsed.elapsed;
-      if (remaining <= Duration.zero) {
-        throw TimeoutException(
-            'SOCKS5 output did not drain before close', _operationTimeout);
-      }
-      final Future flushing;
-      try {
-        flushing = transport.flush();
-      } on StateError {
-        // The public native socket may have an active flush/addStream outside
-        // _writeTail. IOSink rejects another flush synchronously while bound,
-        // and exposes no future for that operation. Keep its transport alive
-        // and retry within one deadline; wrapping it would break manual TLS.
-        const interval = Duration(milliseconds: 10);
-        await Future<void>.delayed(remaining < interval ? remaining : interval);
-        continue;
-      }
-      // Async failures belong to the flush itself and must not be retried.
-      await flushing.timeout(remaining);
-      return;
-    }
-  }
-
   /// Ends a connect in flight. The TLS handshake cannot notice a destroyed
   /// transport on its own, so it waits for this or for its deadline.
   void _signalCancel(SocksCancelledException error) {
@@ -915,7 +883,9 @@ class SOCKSSocket {
       await _writeTail;
       final open = _nativeSocketOpen || (channel != null && !channel.isClosed);
       if (open && _writeFailure == null) {
-        await _flushBeforeClose(upgraded ? _secureSocksSocket : _socksSocket);
+        await (upgraded ? _secureSocksSocket : _socksSocket)
+            .flush()
+            .timeout(_operationTimeout);
         flushed = true;
       }
     } catch (error, stack) {

pubspec.yaml

@@ -1,6 +1,6 @@
 name: socks_socket
 description: SOCKS version 5 sockets for Dart and Flutter, *eg.* ElectrumX and/or Fulcrum over Tor via socket(s).
-version: 1.4.0
+version: 2.0.0
 repository: https://github.com/cypherstack/socks_socket
 
 environment:

test/connection_lifecycle_test.dart

@@ -82,7 +82,7 @@ void main() {
   });
 
   test('peer EOF still delivers accepted writes', () async {
-    const size = 1024 * 1024;
+    const size = 8 * 1024 * 1024;
     final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
     addTearDown(server.close);
     final received = Completer<int>();
@@ -119,8 +119,6 @@ void main() {
     addTearDown(() => uploader.close().catchError((_) {}));
     await uploader.connect();
     await uploader.connectTo('localhost', 80);
-    // Linux SO_SNDBUF.
-    uploader.socket.setRawOption(RawSocketOption.fromInt(1, 7, 65536));
     final input = uploader.inputStream.drain<void>();
     await uploader.write('A' * size);
     await input;

test/peer_eof_test.dart

@@ -9,52 +9,30 @@ import 'helpers/tunnel_peer.dart';
 
 const _deadline = tunnelDeadline;
 
-// SOL_SOCKET, SO_SNDBUF and SO_LINGER differ between Linux and macOS.
+// SOL_SOCKET and SO_LINGER differ between Linux and macOS.
 final _solSocket = Platform.isMacOS ? 0xffff : 1;
-final _soSndbuf = Platform.isMacOS ? 0x1001 : 7;
 final _soLinger = Platform.isMacOS ? 0x80 : 13;
 
 void main() {
-  test('peer EOF drains a pending underlying socket flush', () async {
-    final (client, peer) = await connectTunnel();
-    final eof = client.inputStream.drain<void>();
-    // Linux SO_SNDBUF: force the upload to remain pending while reads pause.
-    client.socket.setRawOption(RawSocketOption.fromInt(1, 7, 4096));
-    peer.input.pause();
-    final payload = List<int>.generate(512 * 1024, (i) => i % 251);
-    client.socket.add(payload);
-    var flushed = false;
-    final flushing = client.socket.flush().then((_) => flushed = true);
-    flushing.ignore();
-    await Future<void>.delayed(const Duration(milliseconds: 20));
-    expect(flushed, isFalse, reason: 'The regression needs a pending flush');
-    await peer.socket.close();
-    await eof.timeout(_deadline);
-    peer.input.resume();
-    await flushing.timeout(_deadline);
-    await client.close().timeout(_deadline);
-    await peer.done.future.timeout(_deadline);
-    expect(peer.bytes.takeBytes(), payload);
-  }, testOn: 'linux');
-
   for (final tls in [false, true]) {
-    test('peer EOF drains underlying socket.addStream (TLS=$tls)', () async {
+    test('peer EOF drains a pending write (TLS=$tls)', () async {
       final (client, peer) = await connectTunnel(tls: tls);
       final eof = client.inputStream.drain<void>();
-      final source = StreamController<List<int>>();
-      addTearDown(source.close);
-      final uploading = client.socket.addStream(source.stream);
-      uploading.ignore();
-      source.add([65]);
-      await peer.firstChunk.future.timeout(_deadline);
+      peer.input.pause();
+      // Exceed the transport buffers without accessing the client's socket.
+      final payload = 'A' * (8 * 1024 * 1024);
+      var flushed = false;
+      final writing = client.write(payload).then((_) => flushed = true);
+      writing.ignore();
+      await Future<void>.delayed(const Duration(milliseconds: 200));
+      expect(flushed, isFalse, reason: 'The regression needs a pending write');
       await peer.socket.close();
       await eof.timeout(_deadline);
-      source.add([66]);
-      await source.close();
-      await uploading.timeout(_deadline);
+      peer.input.resume();
+      await writing.timeout(_deadline);
       await client.close().timeout(_deadline);
       await peer.done.future.timeout(_deadline);
-      expect(peer.bytes.takeBytes(), [65, 66]);
+      expect(peer.bytes.takeBytes(), payload.codeUnits);
     });
 
     test('peer EOF drains an accepted outputStream (TLS=$tls)', () async {
@@ -117,24 +95,21 @@ void main() {
     expect(peer.bytes.length, 0);
   });
 
-  test('reset after peer EOF cancels a paused underlying upload source',
-      () async {
+  test('reset after peer EOF cancels a paused outputStream source', () async {
     final (client, peer) = await connectTunnel();
     final eof = client.inputStream.drain<void>();
     final source = StreamController<List<int>>();
     addTearDown(() async {
-      client.socket.destroy();
+      client.destroy();
       await source.close();
     });
-    // Shrink SO_SNDBUF to keep the source paused in an in-flight write.
-    client.socket
-        .setRawOption(RawSocketOption.fromInt(_solSocket, _soSndbuf, 4096));
-    final uploading = client.socket.addStream(source.stream);
-    final failed = expectLater(uploading, throwsA(isA<SocketException>()));
+    final uploading = client.outputStream.addStream(source.stream);
+    final failed =
+        expectLater(uploading, throwsA(isA<SocksConnectionException>()));
     source.add([65]);
     await peer.firstChunk.future.timeout(_deadline);
     peer.input.pause();
-    source.add(List<int>.filled(512 * 1024, 66));
+    source.add(List<int>.filled(8 * 1024 * 1024, 66));
     await Future<void>.delayed(const Duration(milliseconds: 20));
     expect(source.isPaused, isTrue);
     await peer.socket.close();
@@ -144,31 +119,29 @@ void main() {
         Uint8List.view(Int32List.fromList([1, 0]).buffer)));
     peer.socket.destroy();
     await failed.timeout(_deadline);
-    await client.close().timeout(_deadline);
+    await expectLater(client.close().timeout(_deadline),
+        throwsA(isA<SocksConnectionException>()));
     expect(source.hasListener, isFalse);
     await source.close().timeout(_deadline);
   }, testOn: 'linux || mac-os');
 
-  for (final direct in [false, true]) {
-    test('close times out when an upload never ends (direct=$direct)',
-        () async {
-      final (client, peer) = await connectTunnel(
-        operationTimeout: const Duration(milliseconds: 100),
-      );
-      final eof = client.inputStream.drain<void>();
-      final source = StreamController<List<int>>();
-      addTearDown(source.close);
-      final output = direct ? client.socket : client.outputStream;
-      output.addStream(source.stream).ignore();
-      source.add([65]);
-      await peer.firstChunk.future.timeout(_deadline);
-      await peer.socket.close();
-      await eof.timeout(_deadline);
-      await expectLater(
-          client.close().timeout(_deadline), throwsA(isA<TimeoutException>()));
-      await peer.done.future.timeout(_deadline);
-      expect(client.state, ConnectionState.disconnected);
-      expect(source.hasListener, isFalse);
-    });
-  }
+  test('close times out when an outputStream upload never ends', () async {
+    final (client, peer) = await connectTunnel(
+      operationTimeout: const Duration(milliseconds: 100),
+    );
+    final eof = client.inputStream.drain<void>();
+    final source = StreamController<List<int>>();
+    addTearDown(source.close);
+    final output = client.outputStream;
+    output.addStream(source.stream).ignore();
+    source.add([65]);
+    await peer.firstChunk.future.timeout(_deadline);
+    await peer.socket.close();
+    await eof.timeout(_deadline);
+    await expectLater(
+        client.close().timeout(_deadline), throwsA(isA<TimeoutException>()));
+    await peer.done.future.timeout(_deadline);
+    expect(client.state, ConnectionState.disconnected);
+    expect(source.hasListener, isFalse);
+  });
 }

test/manual_tls_test.dart test/tls_configuration_test.darttest/manual_tls_test.dart renamed to test/tls_configuration_test.dart

@@ -8,7 +8,7 @@ import 'helpers/mock_socks_server.dart';
 import 'helpers/test_certificates.dart';
 
 void main() {
-  test('legacy plain socket supports a caller-managed TLS upgrade', () async {
+  test('SOCKSSocket manages TLS with a custom security context', () async {
     final certificates = TestCertificates.generate();
     final proxy = MockSocksServer()
       ..sslEnabled = true
@@ -18,16 +18,14 @@ void main() {
     final socket = await SOCKSSocket.create(
       proxyHost: InternetAddress.loopbackIPv4.address,
       proxyPort: proxy.port,
+      sslEnabled: true,
+      securityContext: certificates.clientContext(),
     );
     addTearDown(socket.close);
     await socket.connect();
     await socket.connectTo('localhost', 443);
-    final secured = await SecureSocket.secure(socket.socket,
-        host: 'localhost', context: certificates.clientContext());
-    addTearDown(secured.destroy);
-    final reply = secured.expand((b) => b).take(10).toList();
-    secured.add(utf8.encode('manual TLS'));
-    await secured.flush();
-    expect(utf8.decode(await reply), 'manual TLS');
+    final reply = socket.inputStream.expand((b) => b).take(10).toList();
+    await socket.write('custom TLS');
+    expect(utf8.decode(await reply), 'custom TLS');
   });
 }