diff --git a/.gitignore b/.gitignore index 3cceda5..86410c5 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,6 @@ # Avoid committing pubspec.lock for library packages; see # https://dart.dev/guides/libraries/private-files#pubspeclock. pubspec.lock + +# IDE files +.idea/ diff --git a/.idea/.gitignore b/.idea/.gitignore deleted file mode 100644 index 26d3352..0000000 --- a/.idea/.gitignore +++ /dev/null @@ -1,3 +0,0 @@ -# Default ignored files -/shelf/ -/workspace.xml diff --git a/.idea/libraries/Dart_Packages.xml b/.idea/libraries/Dart_Packages.xml deleted file mode 100644 index ee95adb..0000000 --- a/.idea/libraries/Dart_Packages.xml +++ /dev/null @@ -1,582 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/.idea/libraries/Dart_SDK.xml b/.idea/libraries/Dart_SDK.xml deleted file mode 100644 index fcbce96..0000000 --- a/.idea/libraries/Dart_SDK.xml +++ /dev/null @@ -1,29 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml deleted file mode 100644 index 639900d..0000000 --- a/.idea/misc.xml +++ /dev/null @@ -1,6 +0,0 @@ - - - - - - \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml deleted file mode 100644 index f49cd95..0000000 --- a/.idea/modules.xml +++ /dev/null @@ -1,8 +0,0 @@ - - - - - - - - \ No newline at end of file diff --git a/.idea/socks_socket.iml b/.idea/socks_socket.iml deleted file mode 100644 index dff7377..0000000 --- a/.idea/socks_socket.iml +++ /dev/null @@ -1,13 +0,0 @@ - - - - - - - - - - - - - \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml deleted file mode 100644 index 35eb1dd..0000000 --- a/.idea/vcs.xml +++ /dev/null @@ -1,6 +0,0 @@ - - - - - - \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 8eb0f8b..91e5e66 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,28 @@ +## 2.0.0 + +- **Breaking:** Remove `SOCKSSocket.socket` so transport operations cannot bypass the wrapper's write queue and lifecycle tracking. Use `destroy()`, `closeOutput()`, `close()`, `write()`, `outputStream`, and `inputStream` instead. `closeOutput()` preserves the native socket's write-side half-close behavior. +- **Breaking:** Caller-managed TLS through `SecureSocket.secure(socks.socket)` is no longer supported. Use `SOCKSSocket.create(sslEnabled: true, securityContext: ...)` to negotiate TLS during `connectTo()`. +- **Breaking:** Remove the `SOCKSSocket()` constructor, deprecated since 1.2.0; it never awaited the proxy connection. Use `SOCKSSocket.create()`. +- **Breaking:** Remove `responseController` and `subscription`, which exposed the stream controller and subscription behind `inputStream`; closing or cancelling them bypassed the lifecycle tracking. Use `inputStream`. +- 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 `closeOutput()` to drain and half-close the sending direction while preserving input for protocols that respond after request EOF. +- Add `closeOnPeerEof` to `create()`; pass `false` to keep writing after the peer half-closes (#3). +- `closeOutput()` rejects new writes as soon as draining starts, preserves accepted uploads, and aborts the connection and pending output if draining fails or times out. +- `close()` now ends a `reconnect()` or TLS handshake in flight instead of being overtaken by the reconnect or waiting for the handshake deadline. +- Give the greeting and username/password authentication exchanges separate handshake timeout budgets, preserving the previous slow-proxy behavior. +- Roll back `outputStream.addStream()` bookkeeping if the supplied stream rejects its subscription, so later output shutdown cannot wait forever. +- Make `close()` consistently rethrow a recorded write failure whether it came from `write()` or `outputStream`. +- Make `closeOutput()` rethrow the same recorded write failure as `close()` instead of replacing it with a state error. +- Release the underlying socket after callers close both input and output, while still allowing writes between input cancellation and output shutdown. +- Validate `SOCKSSocket.create()` proxy and timeout arguments before connecting. +- The handshake timeout message reports sub-second deadlines in milliseconds instead of "0 seconds". +- Fail a handshake read or write at once when the socket reports its error synchronously, as macOS does for a reset peer, instead of waiting for the handshake deadline. `SocksConnection` shares the fix. +- Document that `reconnect()` works after `cancel()`. +- Drive `SOCKSSocket` from one private lifecycle phase instead of a dozen flags. No API change. `state` now reads `disconnected` as soon as `close()` starts draining, not only once it completes. +- Carry the plain connection over the same raw channel as TLS and `SocksConnection`, so transport errors read the same on both transports. +- Read handshake replies straight from the channel, as `SocksConnection` does. A greeting or authentication reply with trailing bytes still fails the handshake, and a stray byte before the CONNECT reply now fails it too instead of being dropped. Handshake failure messages come from the shared protocol code. +- Migrate the Android example to Flutter's declarative Gradle plugins and grant release builds network access. + ## 1.4.0 - Add `SocksConnection.start` for `HttpClient.connectionFactory` (Dart 3.5). diff --git a/README.md b/README.md index c84eb47..7b1542f 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # SOCKS sockets - [![Pub](https://img.shields.io/pub/v/socks_socket.svg)](https://pub.dev/packages/socks_socket) - [![GitHub](https://img.shields.io/github/license/stackdump/socks_socket)]( +[![Pub](https://img.shields.io/pub/v/socks_socket.svg)](https://pub.dev/packages/socks_socket) +[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE) SOCKS version 5 sockets for Dart and Flutter, *eg.* ElectrumX and/or Fulcrum over Tor via socket(s). @@ -25,12 +25,18 @@ SOCKS version 5 sockets for Dart and Flutter, *eg.* ElectrumX and/or Fulcrum ove See `socks_socket.dart` itself for properties and methods and the example for reference. ```dart +import 'dart:io'; + import 'package:socks_socket/socks.dart'; +// The SOCKS5 port of your Tor instance, for example Tor.instance.port when +// Tor runs in-app through package:tor_ffi_plugin. +const proxyPort = 9050; + // Instantiate a socks socket at localhost and on the port selected by the tor service. var socksSocket = await SOCKSSocket.create( proxyHost: InternetAddress.loopbackIPv4.address, - proxyPort: Tor.instance.port, + proxyPort: proxyPort, sslEnabled: true, // For SSL connections. ); @@ -54,7 +60,7 @@ await socksSocket.sendServerFeaturesCommand(); // Configure custom timeouts for slow networks like Tor. var socksSocket = await SOCKSSocket.create( proxyHost: InternetAddress.loopbackIPv4.address, - proxyPort: Tor.instance.port, + proxyPort: proxyPort, sslEnabled: true, handshakeTimeout: Duration(seconds: 60), operationTimeout: Duration(seconds: 45), @@ -67,7 +73,7 @@ var socksSocket = await SOCKSSocket.create( // Use isolationToken to request a separate Tor circuit. var socksSocket = await SOCKSSocket.create( proxyHost: InternetAddress.loopbackIPv4.address, - proxyPort: Tor.instance.port, + proxyPort: proxyPort, sslEnabled: true, isolationToken: 'wallet-btc-001', ); @@ -91,7 +97,31 @@ await socksSocket.write('{"jsonrpc":"2.0","method":"server.ping","id":1}', newline: true); ``` -`cancel()` or `close()` during connect spends the instance; create a new one. Peer close is noticed, and `state` updated, only while `inputStream` has a listener. +If `connect()` fails, the instance is spent: `state` is `error`, another `connect()` throws, and `reconnect()` has no target. Create a new instance. After a failed `connectTo()`, `reconnect()` retries that target. After `cancel()` or `close()` during connect, `reconnect()` opens a new connection once a target is known. `reconnect()` closes the current connection first, which ends `inputStream`; a `close()` or `destroy()` while it is in progress cancels it, including one made from that stream's `onDone`. Peer close is noticed, and `state` updated, only while `inputStream` has a listener. + +By default the connection closes once the peer closes its side. Pass `closeOnPeerEof: false` to `create()` to keep writing after a peer half-close until you call `close()`. + +`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. When a protocol requires request EOF before the response, use `closeOutput()`, consume the response from `inputStream`, then call `close()`. + +## Migrating to 2.0.0 + +`SOCKSSocket.socket`, `responseController` and `subscription` have 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.closeOutput()` to preserve input (the native half-close behavior), or `await socks.close()` to close both directions | +| `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(...)` | +| `socks.responseController`, `socks.subscription` | `socks.inputStream`; pause or cancel your own subscription to it | +| `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 signal request EOF while continuing to receive, use `await socks.closeOutput()`. It immediately rejects new writes while draining accepted output; a drain timeout or transport failure aborts the connection and fails pending output. To drain and finish the entire 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` for `HttpClient.connectionFactory`; that separate API is unchanged. ## HttpClient Connections diff --git a/example/README.md b/example/README.md index 5a75e39..c7718a0 100644 --- a/example/README.md +++ b/example/README.md @@ -1,22 +1,18 @@ -# socks_socket_example +# socks_socket example -Demonstrates how to use the socks_socket package. Uses cypherstack/tor for the tor service. +A Flutter app that starts Tor in-process with `tor_ffi_plugin`, opens a +`SOCKSSocket` through it, and talks to an ElectrumX server over the proxy. It +also shows the same request through a general-purpose SOCKS5 client package +for comparison. -## TODO +Run it with `flutter run` on a desktop or mobile target. The first start +downloads Tor's consensus, which can take a minute. -- Show how/why the other packages available are not suitable for this use case. +Two smaller, non-Flutter examples live beside it: -## Getting Started - -Run as in `flutter run`. - -This project is a starting point for a Flutter application. - -A few resources to get you started if this is your first Flutter project: - -- [Lab: Write your first Flutter app](https://docs.flutter.dev/get-started/codelab) -- [Cookbook: Useful Flutter samples](https://docs.flutter.dev/cookbook) - -For help getting started with Flutter development, view the -[online documentation](https://docs.flutter.dev/), which offers tutorials, -samples, guidance on mobile development, and a full API reference. +- `socks_socket_example.dart`: the socket API without the UI. It still needs + Flutter for the Tor plugin; run it from a Flutter project, or point + `proxyPort` at a Tor you run yourself. +- `http/http_connection.dart`: `SocksConnection.start` as an + `HttpClient.connectionFactory`, runnable with plain Dart against any local + SOCKS5 proxy. See the main README for the command. diff --git a/example/android/app/build.gradle b/example/android/app/build.gradle index 4c908eb..587878c 100644 --- a/example/android/app/build.gradle +++ b/example/android/app/build.gradle @@ -1,3 +1,10 @@ +plugins { + id "com.android.application" + id "kotlin-android" + // The Flutter Gradle Plugin must follow the Android and Kotlin plugins. + id "dev.flutter.flutter-gradle-plugin" +} + def localProperties = new Properties() def localPropertiesFile = rootProject.file('local.properties') if (localPropertiesFile.exists()) { @@ -6,11 +13,6 @@ if (localPropertiesFile.exists()) { } } -def flutterRoot = localProperties.getProperty('flutter.sdk') -if (flutterRoot == null) { - throw new GradleException("Flutter SDK not found. Define location with flutter.sdk in the local.properties file.") -} - def flutterVersionCode = localProperties.getProperty('flutter.versionCode') if (flutterVersionCode == null) { flutterVersionCode = '1' @@ -21,10 +23,6 @@ if (flutterVersionName == null) { flutterVersionName = '1.0' } -apply plugin: 'com.android.application' -apply plugin: 'kotlin-android' -apply from: "$flutterRoot/packages/flutter_tools/gradle/flutter.gradle" - android { namespace "com.example.example" compileSdkVersion flutter.compileSdkVersion @@ -66,7 +64,3 @@ android { flutter { source '../..' } - -dependencies { - implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlin_version" -} diff --git a/example/android/app/src/main/AndroidManifest.xml b/example/android/app/src/main/AndroidManifest.xml index 19b862e..2c04d0a 100644 --- a/example/android/app/src/main/AndroidManifest.xml +++ b/example/android/app/src/main/AndroidManifest.xml @@ -1,4 +1,6 @@ + + properties.load(reader) } +plugins { + id "dev.flutter.flutter-plugin-loader" version "1.0.0" + id "com.android.application" version "7.3.0" apply false + id "org.jetbrains.kotlin.android" version "1.7.10" apply false +} -def flutterSdkPath = properties.getProperty("flutter.sdk") -assert flutterSdkPath != null, "flutter.sdk not set in local.properties" -apply from: "$flutterSdkPath/packages/flutter_tools/gradle/app_plugin_loader.gradle" +include ':app' diff --git a/example/lib/main.dart b/example/lib/main.dart index a14b185..7802cbe 100644 --- a/example/lib/main.dart +++ b/example/lib/main.dart @@ -5,7 +5,7 @@ import 'dart:io'; import 'package:flutter/material.dart'; import 'package:path_provider/path_provider.dart'; -import 'package:socks_socket/socks_socket.dart'; +import 'package:socks_socket/socks.dart'; // Imports needed for tor usage: import 'package:socks5_proxy/socks_client.dart'; // Just for example; can use any socks5 proxy package, pick your favorite. import 'package:tor_ffi_plugin/tor_ffi_plugin.dart'; @@ -180,7 +180,12 @@ class _MyAppState extends State { 'bitcoin.stackwallet.com', 50002); // Send a server features command to the connected socket, see method for more specific usage example.. + final response = socksSocket.inputStream + .transform(utf8.decoder) + .transform(const LineSplitter()) + .first; await socksSocket.sendServerFeaturesCommand(); + print(await response); // You should see a server response printed to the console. // diff --git a/example/macos/Runner/DebugProfile.entitlements b/example/macos/Runner/DebugProfile.entitlements index dddb8a3..3ba6c12 100644 --- a/example/macos/Runner/DebugProfile.entitlements +++ b/example/macos/Runner/DebugProfile.entitlements @@ -6,6 +6,8 @@ com.apple.security.cs.allow-jit + com.apple.security.network.client + com.apple.security.network.server diff --git a/example/macos/Runner/Release.entitlements b/example/macos/Runner/Release.entitlements index 852fa1a..7a2230d 100644 --- a/example/macos/Runner/Release.entitlements +++ b/example/macos/Runner/Release.entitlements @@ -4,5 +4,9 @@ com.apple.security.app-sandbox + com.apple.security.network.client + + com.apple.security.network.server + diff --git a/example/pubspec.yaml b/example/pubspec.yaml index c5c90b7..f880af6 100644 --- a/example/pubspec.yaml +++ b/example/pubspec.yaml @@ -34,12 +34,9 @@ dependencies: socks_socket: path: .. tor_ffi_plugin: - # When depending on this package from a real application you should use: - # tor_ffi_plugin: ^x.y.z - # See https://dart.dev/tools/pub/dependencies#version-constraints - # The example app is bundled with the plugin so we use a path dependency on - # the parent directory to use the current plugin's version. - git: https://github.com/cypherstack/tor # TODO: rework example app to use user-provided Tor proxy. + git: + url: https://github.com/cypherstack/tor + ref: 5586bfa3c48e495bc00f9b56fa18869daf768573 # native-0.1.0 # The following adds the Cupertino Icons font to your application. # Use with the CupertinoIcons class for iOS style icons. diff --git a/example/socks_socket_example.dart b/example/socks_socket_example.dart index e9b7c21..4c9c9a7 100644 --- a/example/socks_socket_example.dart +++ b/example/socks_socket_example.dart @@ -2,7 +2,7 @@ // // See the [Arti package](https://pub.dev/packages/arti) for a pure Dart implementation. -import 'dart:async'; +import 'dart:convert'; import 'dart:io'; import 'package:socks_socket/socks_socket.dart'; @@ -25,6 +25,11 @@ Future main() async { // Send a server features command to the connected socket, see method for // more specific usage example.. + final response = socksSocket.inputStream + .transform(utf8.decoder) + .transform(const LineSplitter()) + .first; await socksSocket.sendServerFeaturesCommand(); + print(await response); await socksSocket.close(); } diff --git a/lib/socks_socket.dart b/lib/socks_socket.dart index 043f02e..41ca9a1 100644 --- a/lib/socks_socket.dart +++ b/lib/socks_socket.dart @@ -74,49 +74,53 @@ enum SocksSocketState { typedef ConnectionState = SocksSocketState; +/// Where a [SOCKSSocket] is in its life. [SOCKSSocket.state] is derived from it. +enum _Phase { + /// A TCP connection to the proxy is open; [SOCKSSocket.connect] may run. + idle, + + /// [SOCKSSocket.connect] is negotiating the greeting. + greeting, + + /// The greeting succeeded; [SOCKSSocket.connectTo] may run. + greeted, + + /// [SOCKSSocket.connectTo] is requesting the target, then negotiating TLS. + requesting, + connected, + + /// [SOCKSSocket.closeOutput] is draining accepted output; input stays open. + closingOutput, + + /// Output is shut down, but [SOCKSSocket.inputStream] may still receive. + receiving, + + /// [SOCKSSocket.close] is draining accepted output. + closing, + + /// [SOCKSSocket.cancel] ended a connect; [SOCKSSocket.reconnect] may run. + cancelled, + + /// Closed or destroyed; [SOCKSSocket.reconnect] may run. + closed, + + /// The handshake or the transport failed; [SOCKSSocket.reconnect] may run. + failed, +} + /// A SOCKS5 socket. /// -/// A Dart 3 Socket wrapper that implements the SOCKS5 protocol. Now with SSL! -/// -/// Properties: -/// - [proxyHost]: The host of the SOCKS5 proxy server. -/// - [proxyPort]: The port of the SOCKS5 proxy server. -/// - [_socksSocket]: The underlying Socket that connects to the SOCKS5 proxy -/// server. -/// - [_responseController]: A StreamController that listens to the -/// [_socksSocket] and broadcasts the response. -/// -/// Methods: -/// - connect: Connects to the SOCKS5 proxy server. -/// - connectTo: Connects to the specified [domain] and [port] through the -/// SOCKS5 proxy server. -/// - write: Converts [object] to a String by invoking [Object.toString] and -/// sends the encoding of the result to the socket. -/// - sendServerFeaturesCommand: Sends the server.features command to the -/// proxy server. -/// - close: Closes the connection to the Tor proxy. -/// /// Usage: /// ```dart -/// // Instantiate a socks socket at localhost and on the port selected by the -/// // tor service. -/// var socksSocket = await SOCKSSocket.create( -/// proxyHost: InternetAddress.loopbackIPv4.address, -/// proxyPort: tor.port, -/// // sslEnabled: true, // For SSL connections. -/// ); -/// -/// // Connect to the socks instantiated above. -/// await socksSocket.connect(); -/// -/// // Connect to bitcoincash.stackwallet.com on port 50001 via socks socket. -/// await socksSocket.connectTo( -/// 'bitcoincash.stackwallet.com', 50001); -/// -/// // Send a server features command to the connected socket, see method for -/// // more specific usage example.. -/// await socksSocket.sendServerFeaturesCommand(); -/// await socksSocket.close(); +/// final socks = await SOCKSSocket.create( +/// proxyHost: InternetAddress.loopbackIPv4.address, +/// proxyPort: tor.port, +/// // sslEnabled: true, // Negotiate TLS with the target after CONNECT. +/// ); +/// await socks.connect(); +/// await socks.connectTo('bitcoincash.stackwallet.com', 50001); +/// await socks.sendServerFeaturesCommand(); +/// await socks.close(); /// ``` /// /// See also: @@ -128,144 +132,64 @@ class SOCKSSocket { /// The port of the SOCKS5 proxy server. final int proxyPort; - /// 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. - Socket get socket => sslEnabled ? _secureSocksSocket : _socksSocket; - - /// A wrapper around the _socksSocket that enables SSL connections. - late Socket _secureSocksSocket; - - bool _nativeSocketOpen = false; - bool _nativeSocketNeedsDestroy = false; - - RawChannel? _channel; - - RawChannel? _secureChannel; + /// Whether [connectTo] negotiates TLS with the target. + final bool sslEnabled; - /// A StreamController that listens to the _socksSocket and broadcasts. - late StreamController> _responseController = - _newResponseController(); + /// Timeout for the TCP connect, each handshake step, and the TLS upgrade. + final Duration _handshakeTimeout; - /// Bumped on teardown so a replaced connection's callbacks are ignored. - int _generation = 0; + /// Timeout for flushing a write, and for draining output in [close]. + final Duration _operationTimeout; - StreamController> _newResponseController() { - final generation = _generation; - return StreamController>.broadcast( - onListen: () { - if (generation == _generation) _resumeApplicationInput(); - }, - onCancel: () { - if (generation == _generation) _pauseApplicationInput(); - }, - ); - } + /// Accept bad certificates (testing only). + final bool _allowBadCertificates; - List? _pendingApplicationData; - bool _applicationInputPaused = false; + final SecurityContext? _securityContext; - /// Broadcast: each handshake step listens and cancels in turn. - StreamController>? _handshakeResponses; + final bool? _requireIsolation; - bool get _handshaking => _state != SocksSocketState.connected; + final bool _closeOnPeerEof; - void _resumeApplicationInput() { - if (_state != SocksSocketState.connected || - !responseController.hasListener) { - return; - } - final pending = _pendingApplicationData; - _pendingApplicationData = null; - if (pending != null && pending.isNotEmpty) { - _responseController.add(pending); - } - if (_applicationInputPaused) { - _applicationInputPaused = false; - _subscription?.resume(); - } - } + /// Tor circuit isolation token (RFC 1929 username/password auth). + String? _isolationToken; - void _pauseApplicationInput() { - if (_state == SocksSocketState.connected) _pauseInput(); - } + /// Target of the last [connectTo], for [reconnect]. + String? _targetDomain; + int? _targetPort; - void _pauseInput() { - if (!_applicationInputPaused) { - _applicationInputPaused = true; - _subscription?.pause(); - } - } + _Phase _phase = _Phase.idle; - /// A StreamController that listens to the _secureSocksSocket and broadcasts. - late StreamController> _secureResponseController = - _newResponseController(); + /// The connection to the proxy: a raw TCP channel until [connectTo] has + /// negotiated TLS, then the TLS channel. + RawChannel? _channel; - /// Getter for the StreamController that listens to the _socksSocket and - /// broadcasts, or the _secureSocksSocket and broadcasts if SSL is enabled. - StreamController> get responseController => - sslEnabled ? _secureResponseController : _responseController; + /// [_channel] as a [Socket], once connected. + Socket? _transport; - /// A StreamSubscription that listens to the _socksSocket or the - /// _secureSocksSocket if SSL is enabled. + /// Reads [_transport] into [_input]; paused while [_input] has no listener. StreamSubscription>? _subscription; - /// Getter for the StreamSubscription that listens to the _socksSocket or the - /// _secureSocksSocket if SSL is enabled. - StreamSubscription>? get subscription => _subscription; - - /// Is SSL enabled? - final bool sslEnabled; - - /// Current connection state. - SocksSocketState _state = SocksSocketState.disconnected; - - /// Peer close or failure is noticed only while [inputStream] has a listener; - /// until then a [write] to a dead peer may appear to succeed. - SocksSocketState get state => _state; + /// The proxy TCP connect in flight, so [close] and [destroy] can end it. + CancellableRawSocketConnect? _proxyConnect; - /// Target domain for [reconnect]. - String? _targetDomain; + /// Ends the handshake in flight; see [_signalCancel]. + Completer? _cancel; - /// Target port for [reconnect]. - int? _targetPort; - - /// Timeout for SOCKS5 handshake and TCP connect. - final Duration _handshakeTimeout; - - /// Timeout for socket flush after [write]. - final Duration _operationTimeout; - - static const int _maxHandshakeBuffer = 1024; + late StreamController> _input = _newInput(); _SocketOutputSink? _outputSink; Future _writeTail = Future.value(); Object? _writeFailure; StackTrace? _writeFailureStack; - Future? _closeFuture; - bool _closing = false; - bool _peerReadClosed = false; - /// Accept bad certificates (testing only). - final bool _allowBadCertificates; - - final SecurityContext? _securityContext; + Future? _closeFuture; + Future? _closeOutputFuture; - /// Cancel signal for in-flight operations. - Completer? _cancelCompleter; - bool _cancelled = false; - bool _greetingStarted = false; - bool _greetingComplete = false; - bool _requestStarted = false; + /// Set by [close] and [destroy]; a [reconnect] in flight stops at its next + /// step instead of opening a connection the caller has already given up. + bool _closeRequested = false; bool _reconnecting = false; - /// Tor circuit isolation token (RFC 1929 username/password auth). - String? _isolationToken; - - final bool? _requireIsolation; - - /// Private constructor. SOCKSSocket._( this.proxyHost, this.proxyPort, @@ -275,39 +199,14 @@ class SOCKSSocket { this._operationTimeout, this._allowBadCertificates, this._securityContext, - this._requireIsolation); - - /// Provides a stream of data as List. - /// - /// Reads pause while there is no listener; earlier data is kept. - Stream> get inputStream => sslEnabled - ? _secureResponseController.stream - : _responseController.stream; - - StreamSink> get outputStream => _outputSink ??= _newOutputSink(); - - _SocketOutputSink _newOutputSink() { - final generation = _generation; - final sink = _SocketOutputSink(_queueSinkWrite, (error, stack) { - if (generation == _generation) _failWrites(error, stack); - }); - if (_writeFailure != null) { - sink._stop(_writeFailure!, _writeFailureStack!); - } else if (_closing) { - sink.close().ignore(); - } - return sink; - } - - Future _queueSinkWrite(List data) { - if (!_canQueueWrite(allowClosing: true)) { - throw StateError( - 'Cannot write: socket is not connected (state: $_state)'); - } - return _queueWrite(data, allowClosing: true); - } + this._requireIsolation, + this._closeOnPeerEof); /// Creates a SOCKS5 socket to the specified [proxyHost] and [proxyPort]. + /// + /// By default the connection closes once the peer closes its side. With + /// [closeOnPeerEof] false, peer EOF only ends [inputStream]: the socket stays + /// [SocksSocketState.connected] and keeps writing until [close]. static Future create({ required String proxyHost, required int proxyPort, @@ -318,7 +217,18 @@ class SOCKSSocket { bool allowBadCertificates = false, SecurityContext? securityContext, bool? requireIsolation, + bool closeOnPeerEof = true, }) async { + if (proxyHost.isEmpty) throw ArgumentError.value(proxyHost, 'proxyHost'); + if (proxyPort < 1 || proxyPort > 65535) { + throw ArgumentError.value(proxyPort, 'proxyPort'); + } + if (handshakeTimeout <= Duration.zero) { + throw ArgumentError.value(handshakeTimeout, 'handshakeTimeout'); + } + if (operationTimeout <= Duration.zero) { + throw ArgumentError.value(operationTimeout, 'operationTimeout'); + } _checkIsolationToken(isolationToken); if (requireIsolation == true && isolationToken == null) { throw ArgumentError( @@ -326,9 +236,7 @@ class SOCKSSocket { 'requireIsolation', ); } - - // Create a SOCKS socket instance. - var instance = SOCKSSocket._( + final instance = SOCKSSocket._( proxyHost, proxyPort, sslEnabled, @@ -337,282 +245,136 @@ class SOCKSSocket { operationTimeout, allowBadCertificates, securityContext, - requireIsolation); - - // Initialize the SOCKS socket. + requireIsolation, + closeOnPeerEof); await instance._init(); - - // Return the SOCKS socket instance. return instance; } - /// Deprecated. Does not await _init(); use [SOCKSSocket.create]. - @Deprecated('Use SOCKSSocket.create() instead') - SOCKSSocket({ - required this.proxyHost, - required this.proxyPort, - required this.sslEnabled, - }) : _isolationToken = null, - _handshakeTimeout = const Duration(seconds: 30), - _operationTimeout = const Duration(seconds: 30), - _allowBadCertificates = false, - _securityContext = null, - _requireIsolation = null { - _init(); + static void _checkIsolationToken(String? token) { + if (token != null) encodeSocksCredentials(token, token); } - /// Initializes the SOCKS socket. + /// Current connection state. /// - /// This method is a private method that is called by the constructor. + /// Peer close or failure is noticed only while [inputStream] has a listener; + /// until then a [write] to a dead peer may appear to succeed. With + /// `closeOnPeerEof: false`, peer close leaves the state unchanged. + SocksSocketState get state => switch (_phase) { + _Phase.idle || + _Phase.closing || + _Phase.cancelled || + _Phase.closed => + SocksSocketState.disconnected, + _Phase.greeting || + _Phase.greeted || + _Phase.requesting => + SocksSocketState.connecting, + _Phase.connected || + _Phase.closingOutput || + _Phase.receiving => + SocksSocketState.connected, + _Phase.failed => SocksSocketState.error, + }; + + bool get _connecting => + _phase == _Phase.greeting || + _phase == _Phase.greeted || + _phase == _Phase.requesting; + + /// Whether [close], [cancel] or [destroy] has given the connection up. + bool get _givenUp => + _phase == _Phase.closing || + _phase == _Phase.cancelled || + _phase == _Phase.closed; + + /// Data from the target, as List. /// - /// Returns: - /// A Future that resolves to void. - Future _init() async { - // Connect to the SOCKS proxy server. - _applicationInputPaused = false; - _cancelled = false; - _greetingStarted = false; - _greetingComplete = false; - _requestStarted = false; - _peerReadClosed = false; - if (sslEnabled) { - final raw = await RawSocket.connect( - proxyHost, - proxyPort, - timeout: _handshakeTimeout, - ); - _channel = RawChannel(raw); - _socksSocket = _channel!.socket(); - } else { - // Keep a native Socket so callers can use SecureSocket.secure(socket). - _socksSocket = await Socket.connect( - proxyHost, - proxyPort, - timeout: _handshakeTimeout, - ); - _nativeSocketOpen = true; - _nativeSocketNeedsDestroy = true; - _socksSocket.done.then((_) { - _nativeSocketOpen = false; - }, onError: (Object _) { - _nativeSocketOpen = false; - }); - } - _secureChannel = null; - final responses = _responseController; - final handshake = _handshakeResponses = StreamController.broadcast(); - - // Listen to the socket. - _subscription = _socksSocket.listen( - (data) { - (_handshaking ? handshake : _responseController).add(data); - }, - onError: (Object e, StackTrace stack) { - final error = e is SocksException - ? e - : SocksConnectionException( - message: 'SOCKS5 connection error: $e', - ); - if (_handshaking && !handshake.isClosed) handshake.addError(error); - if (!responses.isClosed) responses.addError(error); - if (_state == SocksSocketState.connected) _failWrites(error, stack); + /// Reads pause while there is no listener; earlier data is kept. + Stream> get inputStream => _input.stream; + + StreamSink> get outputStream => _outputSink ??= _newOutputSink(); + + StreamController> _newInput() { + late final StreamController> input; + input = StreamController>.broadcast( + onListen: () { + if (identical(input, _input)) _subscription?.resume(); }, - onDone: () { - // Close the response controller when the socket is closed. - if (!handshake.isClosed) handshake.close(); - if (!_sslUpgraded) { - if (_state == SocksSocketState.connected) _peerClosed(); - if (!responses.isClosed) responses.close(); - } + onCancel: () { + if (identical(input, _input)) _subscription?.pause(); }, ); + return input; } - /// The peer stopped sending; writes already accepted are still delivered. - void _peerClosed() { - _peerReadClosed = true; - _state = SocksSocketState.disconnected; - close().ignore(); - } - - void _destroyTransport() { - // After the TLS upgrade the secure channel owns and drains the raw socket. - (_secureChannel ?? _channel)?.destroy(); - // Sink completion does not cancel a paused addStream source after a reset. - // Destroy every native transport we own, even when its done has completed. - if (_nativeSocketNeedsDestroy) { - _nativeSocketNeedsDestroy = false; - _nativeSocketOpen = false; - _socksSocket.destroy(); - } - } - - /// Closes controllers directly; a paused input subscription never sees done. - void _abandonConnection() { - _destroyTransport(); - if (!_responseController.isClosed) _responseController.close(); - if (!_secureResponseController.isClosed) { - _secureResponseController.close(); + _SocketOutputSink _newOutputSink() { + late final _SocketOutputSink sink; + sink = _SocketOutputSink(_queueSinkWrite, (error, stack) { + if (identical(sink, _outputSink)) _fail(error, stack, _Phase.failed); + }); + if (_writeFailure != null) { + sink._stop(_writeFailure!, _writeFailureStack!); + } else if (_phase == _Phase.closingOutput || + _phase == _Phase.receiving || + _phase == _Phase.closing || + _phase == _Phase.closed) { + sink.close().ignore(); } + return sink; } - /// Accumulates [expectedLength] bytes from the response stream. - Future> _waitForResponse(int expectedLength) => - _waitForReply((_) => expectedLength, allowTrailing: false); - - /// Accumulates a variable-length SOCKS5 connect response. - Future> _waitForConnectResponse() => - _waitForReply(socksConnectReplyLength, allowTrailing: true); - - Future> _waitForReply( - int? Function(List buffer) lengthOf, { - required bool allowTrailing, - }) { - final completer = Completer>(); - final buffer = []; - late final StreamSubscription> sub; - - void fail(String reason) { - sub.cancel(); - if (!completer.isCompleted) { - completer.completeError( - SocksHandshakeException( - proxyHost: proxyHost, - proxyPort: proxyPort, - message: 'SOCKS5 $reason (proxy: $proxyHost:$proxyPort).', - ), - ); - } - } + // --------------------------------------------------------------------------- + // Connecting. - sub = _handshakeResponses!.stream.listen( - (data) { - buffer.addAll(data); - final expectedLength = lengthOf(buffer); - if (buffer.length >= _maxHandshakeBuffer && - (expectedLength == null || buffer.length < expectedLength)) { - fail('handshake buffer overflow'); - return; - } - if (expectedLength == null || buffer.length < expectedLength) return; - if (expectedLength < 0) { - fail('reply is malformed'); - } else if (!allowTrailing && buffer.length > expectedLength) { - fail('reply has unexpected trailing data'); - } else { - sub.cancel(); - if (!completer.isCompleted) { - if (allowTrailing && expectedLength > 0 && !sslEnabled) { - _pendingApplicationData = buffer.sublist(expectedLength); - _pauseInput(); - } - completer.complete(buffer.sublist(0, expectedLength)); - } - } - }, - onError: (e) { - sub.cancel(); - if (!completer.isCompleted) { - completer.completeError(e); - } - }, - onDone: () { - if (!completer.isCompleted) { - completer.completeError( - SocksConnectionException( - message: 'Connection closed before SOCKS5 response received ' - '(proxy: $proxyHost:$proxyPort).', - ), - ); - } - }, - ); - - final dataFuture = completer.future.timeout( - _handshakeTimeout, - onTimeout: () { - sub.cancel(); - throw TimeoutException('SOCKS5 handshake timed out after ' - '${_handshakeTimeout.inSeconds} seconds.'); - }, - ); - - final cancel = _cancelCompleter; - if (cancel != null) { - return Future.any>([dataFuture, cancel.future]); + /// Opens the TCP connection to the proxy. + Future _init() async { + final operation = _proxyConnect = CancellableRawSocketConnect(); + if (_closeRequested) operation.cancel(); + final RawSocket raw; + try { + raw = await operation.connect(proxyHost, proxyPort, _handshakeTimeout); + } finally { + if (identical(_proxyConnect, operation)) _proxyConnect = null; } - return dataFuture; - } - - static void _checkIsolationToken(String? token) { - if (token != null) encodeSocksCredentials(token, token); + _channel = RawChannel(raw); + _cancel = Completer()..future.ignore(); } - Future _writeHandshake(List bytes) async => - _socksSocket.add(bytes); - - /// Connects to the SOCKS socket. - /// - /// Returns: - /// A Future that resolves to void. + /// Negotiates the SOCKS5 greeting with the proxy. Future connect() async { _checkNotReconnecting(); return _connect(); } Future _connect() async { - if (_greetingStarted || - _closing || - _state != SocksSocketState.disconnected) { + if (_phase != _Phase.idle) { throw StateError( 'Cannot connect: use reconnect() for another connection'); } - _greetingStarted = true; - _cancelCompleter = Completer()..future.ignore(); - _state = SocksSocketState.connecting; + _phase = _Phase.greeting; + final channel = _channel!; try { await negotiateSocks( - write: _writeHandshake, - read: _waitForResponse, + write: channel.write, + read: (count) => _readReply(channel, count), + exchange: (request, count) => _handshake(() async { + await channel.write(request); + return _readReply(channel, count); + }), credentials: _isolationToken == null ? null : encodeSocksCredentials(_isolationToken!, _isolationToken!), allowNoAuthFallback: !(_requireIsolation ?? _isolationToken != null), ); - if (_cancelled) { - throw SocksCancelledException(message: 'SOCKS5 connection cancelled.'); - } - _greetingComplete = true; - } on SocksProtocolFailure catch (error) { - _cancelCompleter = null; - if (_closing) throw _closedBeforeConnected(); - _state = SocksSocketState.error; - _abandonConnection(); - throw SocksHandshakeException( - proxyHost: proxyHost, - proxyPort: proxyPort, - message: 'SOCKS5 handshake failed: ${error.message}'); - } catch (e) { - _cancelCompleter = null; - if (e is SocksCancelledException) rethrow; - if (_closing) throw _closedBeforeConnected(); - _state = SocksSocketState.error; - _abandonConnection(); - rethrow; + _phase = _Phase.greeted; + } catch (error) { + throw _handshakeFailed(error, 'SOCKS5 handshake failed: '); } } - SocksCancelledException _closedBeforeConnected() => SocksCancelledException( - message: 'SOCKS5 connection closed before it was established.'); - - /// Connects to the specified [domain] and [port] through the SOCKS socket. - /// - /// Parameters: - /// - [domain]: The domain to connect to. - /// - [port]: The port to connect to. - /// - /// Returns: - /// A Future that resolves to void. + /// Asks the proxy to connect to [domain]:[port], then negotiates TLS with + /// the target if [sslEnabled]. Future connectTo(String domain, int port) async { _checkNotReconnecting(); return _connectTo(domain, port); @@ -623,57 +385,25 @@ class SOCKSSocket { } Future _connectTo(String domain, int port) async { - if (_cancelled) { + if (_phase == _Phase.cancelled) { throw SocksCancelledException(message: 'SOCKS5 connection cancelled.'); } - if (_state != SocksSocketState.connecting || - !_greetingComplete || - _requestStarted || - _closing) { + if (_phase != _Phase.greeted) { throw StateError( 'Cannot connectTo: must call connect() and await it before one connectTo()'); } - final request = encodeSocksDestination(domain, port); - - _requestStarted = true; + _phase = _Phase.requesting; _targetDomain = domain; _targetPort = port; - + final channel = _channel!; try { - _socksSocket.add(request); - final response = await _waitForConnectResponse(); - try { - checkSocksConnectReply(response); - } on SocksProtocolFailure catch (error) { - if (error.replyCode != null) { - final replyCode = SocksReplyCode.fromByte(error.replyCode!); - throw SocksRequestException( - replyCode: replyCode, - proxyHost: proxyHost, - proxyPort: proxyPort, - targetDomain: domain, - targetPort: port, - message: 'SOCKS5 request failed (proxy: $proxyHost:$proxyPort, ' - 'target: $domain:$port, reply: ' - '${replyCode?.description ?? "unknown (0x${error.replyCode!.toRadixString(16).padLeft(2, '0')})"})', - ); - } - throw SocksHandshakeException( - proxyHost: proxyHost, proxyPort: proxyPort, message: error.message); - } - - // Upgrade to SSL if needed. + await _handshake(() async { + await channel.write(request); + checkSocksConnectReply(await _readConnectReply(channel)); + }); if (sslEnabled) { - final channel = _channel!; - final deadline = Completer()..future.ignore(); - final timer = Timer(_handshakeTimeout, () { - channel.destroy(); - deadline.completeError( - TimeoutException('SOCKS5 SSL handshake timed out after ' - '${_handshakeTimeout.inSeconds} seconds.')); - }); - final handshake = RawSecureSocket.secure( + final upgrade = RawSecureSocket.secure( channel.raw, subscription: channel.detach(), host: domain, @@ -681,73 +411,133 @@ class SOCKSSocket { onBadCertificate: _allowBadCertificates ? (_) => true : null, ); try { - // Upgrade to SSL. - final secured = await Future.any([ - handshake, - deadline.future, - if (_cancelCompleter != null) _cancelCompleter!.future, - ]); - final secureChannel = _secureChannel = RawChannel(secured); - _secureSocksSocket = secureChannel.socket(); - _sslUpgraded = true; - final responses = _secureResponseController; - - // Listen to the secure socket. - _subscription = _secureSocksSocket.listen( - (data) { - // Add the data to the response controller. - responses.add(data); - }, - onError: (Object e, StackTrace stack) { - final error = e is SocksException - ? e - : SocksConnectionException( - message: 'SOCKS5 connection error: $e', - ); - if (!responses.isClosed) responses.addError(error); - if (_state == SocksSocketState.connected) { - _failWrites(error, stack); - } - }, - onDone: () { - if (_state == SocksSocketState.connected) _peerClosed(); - if (!responses.isClosed) responses.close(); - }, - ); - _pauseInput(); - } catch (e) { - handshake.then((s) => s.close(), onError: (_) {}); - if (_cancelCompleter?.isCompleted == true) { - throw SocksCancelledException( - message: 'SOCKS5 connection cancelled during SSL upgrade.', - ); - } + _channel = RawChannel(await _handshake(() => upgrade)); + } catch (_) { + // The upgrade may still succeed after a timeout or cancellation. + upgrade.then((s) => s.close(), onError: (_) {}); rethrow; - } finally { - timer.cancel(); } } + } catch (error) { + throw _handshakeFailed(error, '', domain: domain, port: port); + } + final transport = _transport = _channel!.socket(); + _subscription = transport.listen( + _input.add, + onError: (Object e, StackTrace stack) { + final error = e is SocksException + ? e + : SocksConnectionException(message: 'SOCKS5 connection error: $e'); + if (!_input.isClosed) _input.addError(error); + if (_phase == _Phase.connected || + _phase == _Phase.closingOutput || + _phase == _Phase.receiving) { + _fail(error, stack, _Phase.failed); + } + }, + onDone: () { + if ((_phase == _Phase.connected || + _phase == _Phase.closingOutput || + _phase == _Phase.receiving) && + _closeOnPeerEof) { + _ensureClosed().ignore(); + } + if (!_input.isClosed) _input.close(); + }, + ); + _phase = _Phase.connected; + _cancel = null; + if (!_input.hasListener) _subscription!.pause(); + } + + static String _describe(Duration duration) => + duration.inMilliseconds % 1000 == 0 + ? '${duration.inSeconds} s' + : '${duration.inMilliseconds} ms'; - // Check if cancelled between handshake completion and state transition. - if (_cancelCompleter?.isCompleted == true) { - throw SocksCancelledException( - message: 'SOCKS5 connection cancelled.', + /// Runs one handshake [step] against the handshake deadline and the cancel + /// signal. A cancellation that lands as the step completes still wins. + Future _handshake(Future Function() step) async { + final cancel = _cancel!; + final result = await Future.any([step(), cancel.future]).timeout( + _handshakeTimeout, + onTimeout: () => throw TimeoutException( + 'SOCKS5 handshake timed out after ${_describe(_handshakeTimeout)}.'), + ); + if (cancel.isCompleted) await cancel.future; + return result; + } + + /// Reads a fixed-size greeting or authentication reply. Bytes beyond it + /// have no place in the protocol here, so they fail the handshake. + Future> _readReply(RawChannel channel, int count) async { + final reply = await channel.read(count); + if (channel.raw.available() > 0) { + throw const SocksProtocolFailure('reply has unexpected trailing data'); + } + return reply; + } + + /// Reads the variable-length CONNECT reply. Bytes after it belong to the + /// target and stay in the socket for [inputStream]. + Future> _readConnectReply(RawChannel channel) async { + final reply = [...await channel.read(4)]; + var length = socksConnectReplyLength(reply); + if (length == null) { + reply.addAll(await channel.read(1)); + length = socksConnectReplyLength(reply); + } + if (length != null && length > reply.length) { + reply.addAll(await channel.read(length - reply.length)); + } + return reply; + } + + /// Tears down after a failed handshake step and maps [error] to what the + /// caller sees. Cancellation, [close] and [destroy] report as such. + Object _handshakeFailed(Object error, String prefix, + {String? domain, int? port}) { + _cancel = null; + if (error is SocksCancelledException) return error; + if (_phase == _Phase.cancelled) { + return SocksCancelledException(message: 'SOCKS5 connection cancelled.'); + } + if (_givenUp) return _closedBeforeConnected(); + _abandon(_Phase.failed); + if (error is SocksProtocolFailure) { + final replyCode = error.replyCode; + if (replyCode != null) { + final reply = SocksReplyCode.fromByte(replyCode); + return SocksRequestException( + replyCode: reply, + proxyHost: proxyHost, + proxyPort: proxyPort, + targetDomain: domain!, + targetPort: port!, + message: 'SOCKS5 request failed (proxy: $proxyHost:$proxyPort, ' + 'target: $domain:$port, reply: ' + '${reply?.description ?? "unknown (0x${replyCode.toRadixString(16).padLeft(2, '0')})"})', ); } - - _state = SocksSocketState.connected; - _cancelCompleter = null; - _resumeApplicationInput(); - } catch (e) { - _cancelCompleter = null; - if (e is SocksCancelledException) rethrow; - if (_closing) throw _closedBeforeConnected(); - _state = SocksSocketState.error; - _abandonConnection(); - rethrow; + return SocksHandshakeException( + proxyHost: proxyHost, + proxyPort: proxyPort, + message: '$prefix${error.message}'); + } + if (error is SocketException) { + return SocksConnectionException( + message: 'Connection closed before SOCKS5 response received ' + '(proxy: $proxyHost:$proxyPort): ${error.message}'); } + return error; } + SocksCancelledException _closedBeforeConnected() => SocksCancelledException( + message: 'SOCKS5 connection closed before it was established.'); + + // --------------------------------------------------------------------------- + // Writing. + /// Writes [object] to the socket. If [newline] is true, appends '\n'. Future write(Object? object, {bool newline = false}) async { if (object == null) return; @@ -755,36 +545,42 @@ class SOCKSSocket { await _queueWrite(newline ? [...data, 0x0A] : data); } - bool _canQueueWrite({bool allowClosing = false}) { - if (_closing && !allowClosing) return false; - // close() rejects new sink operations, but an already accepted addStream - // still feeds chunks after read-side EOF until it finishes or times out. - return _state == SocksSocketState.connected || - (allowClosing && - _closing && - _peerReadClosed && - _state == SocksSocketState.disconnected); - } + /// close() and closeOutput() reject new sink operations, but an addStream + /// they accepted still feeds chunks while output drains. + Future _queueSinkWrite(List data) => _queueWrite(data, + draining: _phase == _Phase.closing || _phase == _Phase.closingOutput); - Future _queueWrite(List data, {bool allowClosing = false}) { - if (!_canQueueWrite(allowClosing: allowClosing)) { - return Future.error( - StateError('Cannot write: socket is not connected (state: $_state)')); + /// Throws synchronously when not writable, so a bound stream ends with the + /// [StateError] instead of failing the connection. + Future _queueWrite(List data, {bool draining = false}) { + if (!(_phase == _Phase.connected || draining)) { + throw StateError('Cannot write: socket is not connected (state: $state)'); } + final transport = _transport!; final bytes = List.of(data); - final generation = _generation; final writing = _writeTail.then((_) async { if (_writeFailure != null) { Error.throwWithStackTrace(_writeFailure!, _writeFailureStack!); } try { - socket.add(bytes); - await socket.flush().timeout(_operationTimeout); + transport.add(bytes); + await transport.flush().timeout(_operationTimeout); + // A transport destroyed mid-flush can still complete the flush. + if (_writeFailure != null) { + Error.throwWithStackTrace(_writeFailure!, _writeFailureStack!); + } } catch (e, stack) { + // A failure that already ended the connection, such as destroy(), + // is the cause; report it rather than the transport's symptom. + if (_writeFailure != null) { + Error.throwWithStackTrace(_writeFailure!, _writeFailureStack!); + } final error = e is SocksException || e is TimeoutException ? e : SocksConnectionException(message: 'SOCKS5 write failed: $e'); - if (generation == _generation) _failWrites(error, stack); + if (identical(transport, _transport)) { + _fail(error, stack, _Phase.failed); + } Error.throwWithStackTrace(error, stack); } }); @@ -793,124 +589,170 @@ class SOCKSSocket { return writing; } - void _failWrites(Object error, StackTrace stack) { - _writeFailure ??= error; - _writeFailureStack ??= stack; - _state = SocksSocketState.error; - _outputSink?._stop(error, stack); - _abandonConnection(); + // --------------------------------------------------------------------------- + // Ending. + + /// Drains accepted output, then closes the connection. + /// + /// Rethrows a write failure that already ended the connection. A [close] + /// or [destroy] while [reconnect] is in progress cancels the reconnect. + Future close() { + _closeRequested = true; + return _ensureClosed(); } - /// Whether SSL upgrade consumed the raw socket. - bool _sslUpgraded = false; + Future _ensureClosed() => _closeFuture ??= _close(); - /// Closes the connection to the Tor proxy. + /// Drains accepted output and shuts down only the sending direction. /// - /// Returns: - /// A Future that resolves to void. - Future close() => _closeFuture ??= _close(); - - Future _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.delayed(remaining < interval ? remaining : interval); - continue; - } - // Async failures belong to the flush itself and must not be retried. - await flushing.timeout(remaining); - return; + /// Rejects new writes immediately while an already accepted upload drains. + /// [inputStream] remains open so a peer that waits for request EOF before + /// replying can still deliver its response. Call [close] after consuming + /// that response to release the remaining connection resources. + /// A drain timeout or transport failure aborts the connection and fails + /// pending output; [close] also reports that failure. + Future closeOutput() { + final closing = _closeOutputFuture; + if (closing != null) return closing; + _throwWriteFailure(); + if (_phase == _Phase.closingOutput || _phase == _Phase.receiving) { + return Future.value(); + } + if (_phase != _Phase.connected) { + throw StateError( + 'Cannot close output: socket is not connected (state: $state)'); } + return _closeOutputFuture = _closeOutput(); } - Future _close() async { - _closing = true; - final upgraded = sslEnabled && _sslUpgraded; - final channel = upgraded ? _secureChannel : _channel; - var flushed = false; - // Ensure all data is sent before closing. + Future _closeOutput() async { + _phase = _Phase.closingOutput; + final transport = _transport!; try { await _outputSink?.close().timeout(_operationTimeout); await _writeTail; - final open = _nativeSocketOpen || (channel != null && !channel.isClosed); - if (open && _writeFailure == null) { - await _flushBeforeClose(upgraded ? _secureSocksSocket : _socksSocket); - flushed = true; - } + _throwWriteFailure(); + await transport.flush().timeout(_operationTimeout); + await transport.close(); + if (_phase == _Phase.closingOutput) _phase = _Phase.receiving; } catch (error, stack) { - _outputSink?._stop(error, stack); + // Preserve a failure already recorded by a write or destroy(). + _throwWriteFailure(); + _fail(error, stack, _Phase.failed); rethrow; - } finally { - _generation++; - _state = SocksSocketState.disconnected; - if (flushed) { - await (upgraded ? _secureSocksSocket : _socksSocket) - .close() - .then((_) {}, onError: (_) {}); - } - await _subscription?.cancel(); - _destroyTransport(); - _closeHandshakeResponses(); - if (!_secureResponseController.isClosed) { - _secureResponseController.close(); - } - if (!_responseController.isClosed) { - _responseController.close(); - } - _sslUpgraded = false; } } + /// Tears the connection down at once, without draining output. + /// + /// Writes not yet delivered, through [write] or [outputStream], fail with a + /// [SocksConnectionException], and [inputStream] ends. A connect or + /// [reconnect] in flight fails with [SocksCancelledException]. A later + /// [close] completes normally. + void destroy() { + _closeRequested = true; + _signalCancel( + SocksCancelledException(message: 'SOCKS5 connection destroyed.'), + ); + _fail( + SocksConnectionException(message: 'SOCKS5 connection destroyed'), + StackTrace.current, + _Phase.closed, + ); + // A close already draining fails once its writes do; keep its outcome for + // the caller that started it, but let a later close() complete normally. + _closeFuture = (_closeFuture ?? _close()).catchError((Object _) {}); + } + /// Cancels an in-flight connect operation. No-op if not connecting. - /// The instance cannot be reused afterwards; create a new socket. + /// Once a target is known, [reconnect] opens a new connection; to end a + /// reconnect in progress, use [close] or [destroy]. Future cancel() async { - if (_state != SocksSocketState.connecting) return; + if (!_connecting) return; + _signalCancel( + SocksCancelledException(message: 'SOCKS5 connection cancelled.'), + ); + _abandon(_Phase.cancelled); + _outputSink?.close().ignore(); + } - final c = _cancelCompleter; - if (c == null || c.isCompleted) 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) { + _proxyConnect?.cancel(); + final cancel = _cancel; + if (cancel != null && !cancel.isCompleted) cancel.completeError(error); + } - _cancelled = true; - _state = SocksSocketState.disconnected; - c.completeError( - SocksCancelledException( - message: 'SOCKS5 connection cancelled.', - ), - ); + /// Records a failure, fails pending output, and tears down into [phase]. + void _fail(Object error, StackTrace stack, _Phase phase) { + _writeFailure ??= error; + _writeFailureStack ??= stack; + _outputSink?._stop(error, stack); + _abandon(phase); + } - // Destroy socket to force pending operations to fail. - _destroyTransport(); + /// Destroys the transport and ends [inputStream] without draining; a paused + /// input subscription would never see done otherwise. A closed socket stays + /// closed: failures its teardown provokes are not news. + void _abandon(_Phase phase) { + if (_phase != _Phase.closed) _phase = phase; + _channel?.destroy(); + if (!_input.isClosed) _input.close(); + } - await _subscription?.cancel(); - _outputSink?.close().ignore(); - _closeHandshakeResponses(); - if (!_responseController.isClosed) _responseController.close(); - if (!_secureResponseController.isClosed) { - _secureResponseController.close(); + Future _close() async { + _signalCancel(_closedBeforeConnected()); + _phase = switch (_phase) { + _Phase.connected || + _Phase.closingOutput || + _Phase.receiving => + _Phase.closing, + _Phase.failed => _Phase.failed, + _ => _Phase.closed, + }; + final channel = _channel; + final transport = _transport; + var flushed = false; + try { + final closingOutput = _closeOutputFuture; + if (closingOutput != null) { + await closingOutput; + } else { + await _outputSink?.close().timeout(_operationTimeout); + await _writeTail; + _throwWriteFailure(); + if (transport != null && channel != null && !channel.isClosed) { + await transport.flush().timeout(_operationTimeout); + flushed = true; + } + } + } catch (error, stack) { + _outputSink?._stop(error, stack); + rethrow; + } finally { + _phase = _Phase.closed; + if (flushed) { + await transport!.close().then((_) {}, onError: (_) {}); + } + await _subscription?.cancel(); + channel?.destroy(); + if (!_input.isClosed) _input.close(); } - - _state = SocksSocketState.disconnected; } - void _closeHandshakeResponses() { - final handshake = _handshakeResponses; - if (handshake != null && !handshake.isClosed) handshake.close(); + void _throwWriteFailure() { + if (_writeFailure != null) { + Error.throwWithStackTrace(_writeFailure!, _writeFailureStack!); + } } /// Reconnects to the previously connected target. /// + /// Closes the current connection first, which ends [inputStream]. A [close] + /// or [destroy] while the reconnect is in progress cancels it with + /// [SocksCancelledException], including one made from that stream's onDone. + /// /// Throws [StateError] if [connectTo] was never called. Future reconnect({String? isolationToken}) async { if (_reconnecting) throw StateError('Reconnect is already in progress'); @@ -927,75 +769,74 @@ class SOCKSSocket { throw StateError('Cannot reconnect: no target known. ' 'Call connectTo() before reconnect().'); } - _checkIsolationToken(isolationToken); - if (isolationToken != null) { - _isolationToken = isolationToken; - } + if (isolationToken != null) _isolationToken = isolationToken; + _closeRequested = false; try { - await close(); + await _ensureClosed(); } catch (_) { // Connection may already be broken. } + _checkReconnectAborted(); // Broadcast controllers can't be reused after close. - _pendingApplicationData = null; - _responseController = _newResponseController(); - _secureResponseController = _newResponseController(); + _input = _newInput(); + _transport = null; + _subscription = null; _outputSink = null; _writeTail = Future.value(); _writeFailure = null; _writeFailureStack = null; _closeFuture = null; - _closing = false; + _closeOutputFuture = null; + _phase = _Phase.idle; try { await _init(); + _checkReconnectAborted(); await _connect(); + _checkReconnectAborted(); await _connectTo(_targetDomain!, _targetPort!); + _checkReconnectAborted(); } catch (error) { - _state = error is SocksCancelledException - ? SocksSocketState.disconnected - : SocksSocketState.error; - _abandonConnection(); + final cancelled = error is SocksCancelledException || _closeRequested; + _abandon(cancelled ? _Phase.closed : _Phase.failed); + // A cancelled TCP connect fails with a SocketException. + if (cancelled && error is! SocksCancelledException) { + throw _reconnectCancelled(); + } rethrow; } } + void _checkReconnectAborted() { + if (_closeRequested) throw _reconnectCancelled(); + } + + SocksCancelledException _reconnectCancelled() => SocksCancelledException( + message: 'SOCKS5 reconnect cancelled: the connection was closed.'); + StreamSubscription> listen( void Function(List data)? onData, { Function? onError, void Function()? onDone, bool? cancelOnError, - }) { - return sslEnabled - ? _secureResponseController.stream.listen( - onData, - onError: onError, - onDone: onDone, - cancelOnError: cancelOnError, - ) - : _responseController.stream.listen( - onData, - onError: onError, - onDone: onDone, - cancelOnError: cancelOnError, - ); - } + }) => + _input.stream.listen( + onData, + onError: onError, + onDone: onDone, + cancelOnError: cancelOnError, + ); /// Sends the server.features command to the proxy server. /// /// This demos how to send the server.features command. Use as an example /// for sending other commands. - /// - /// Returns: - /// A Future that resolves to void. Future sendServerFeaturesCommand() async { - // The server.features command. const String command = '{"jsonrpc":"2.0","id":"0","method":"server.features","params":[]}'; - await write(command, newline: true); } } @@ -1029,9 +870,11 @@ class _SocketOutputSink implements StreamSink> { void _endStream([Object? error, StackTrace? stack]) { final completed = _streamDone; - _source?.cancel(); + final source = _source; _source = null; _streamDone = null; + // The source's onCancel may call destroy(), which ends up back here. + source?.cancel(); if (completed == null) return; if (error == null) { completed.complete(); @@ -1061,8 +904,13 @@ class _SocketOutputSink implements StreamSink> { @override Future addStream(Stream> stream) { _checkOpen(); - final completed = _streamDone = Completer(); - final source = _source = stream.listen(null, cancelOnError: true); + final completed = Completer(); + // listen() is allowed to throw synchronously (for example when a + // single-subscription stream was already consumed). Do not publish the + // pending stream state until the subscription was accepted. + final source = stream.listen(null, cancelOnError: true); + _streamDone = completed; + _source = source; source ..onData((List data) { source.pause(); diff --git a/lib/src/connection_socket.dart b/lib/src/connection_socket.dart index 6e82463..53f1986 100644 --- a/lib/src/connection_socket.dart +++ b/lib/src/connection_socket.dart @@ -3,6 +3,61 @@ import 'dart:convert'; import 'dart:io'; import 'dart:typed_data'; +abstract interface class RawSocketConnectTask { + Future get socket; + void cancel(); +} + +final class _IoRawSocketConnectTask implements RawSocketConnectTask { + final ConnectionTask _task; + _IoRawSocketConnectTask(this._task); + + @override + Future get socket => _task.socket; + + @override + void cancel() => _task.cancel(); +} + +typedef RawSocketStarter = Future Function( + String host, int port); + +Future _startRawSocket(String host, int port) async => + _IoRawSocketConnectTask(await RawSocket.startConnect(host, port)); + +/// One cancellable attempt to open a raw socket. +/// +/// Cancellation is sticky, including while [RawSocket.startConnect] is still +/// creating its [ConnectionTask]. This class is public only within `src/` so +/// the timing-sensitive behavior can be tested without relying on a host's TCP +/// listen-backlog behavior. +class CancellableRawSocketConnect { + final RawSocketStarter _start; + RawSocketConnectTask? _task; + bool _cancelled = false; + + CancellableRawSocketConnect([RawSocketStarter start = _startRawSocket]) + : _start = start; + + Future connect(String host, int port, Duration timeout) async { + final task = _task = await _start(host, port); + if (_cancelled) task.cancel(); + try { + return await task.socket.timeout(timeout, onTimeout: () { + task.cancel(); + throw SocketException('Connection timed out, host: $host, port: $port'); + }); + } finally { + _task = null; + } + } + + void cancel() { + _cancelled = true; + _task?.cancel(); + } +} + class RawChannel { final RawSocket raw; late final StreamSubscription _subscription; @@ -11,6 +66,7 @@ class RawChannel { Completer? _writable; bool _closed = false; bool _readClosed = false; + bool _writeClosed = false; bool _detached = false; bool _streaming = false; Object? _error; @@ -37,11 +93,7 @@ class RawChannel { } }, onCancel: () { - if (!_readClosed && !_closed && !_detached) { - _readClosed = true; - raw.readEventsEnabled = false; - raw.shutdown(SocketDirection.receive); - } + _closeRead(); }, ); _subscription = raw.listen(_event, onError: (Object error) { @@ -70,6 +122,7 @@ class RawChannel { _readable?.complete(); _readable = null; if (_streaming && !_incoming.isClosed) _incoming.close(); + if (_writeClosed) destroy(); } else if (event == RawSocketEvent.closed) { destroy(); } @@ -79,14 +132,15 @@ class RawChannel { final bytes = Uint8List(count); var offset = 0; while (offset < count) { - if (_closed) { - throw _error ?? const SocketException('SOCKS transport closed'); - } + _checkOpen(); final chunk = raw.read(count - offset); if (chunk != null) { bytes.setRange(offset, offset + chunk.length, chunk); offset += chunk.length; } else { + // A failed read reports its error, and closes this channel, before + // returning; a waiter created now would never be completed. + _checkOpen(); if (_readClosed) { throw const SocketException('Incomplete SOCKS response'); } @@ -101,22 +155,27 @@ class RawChannel { Future write(List bytes) async { var offset = 0; while (offset < bytes.length) { - if (_closed) { - throw _error ?? const SocketException('SOCKS transport closed'); - } + _checkOpen(); offset += raw.write(bytes, offset); + // A secure socket reports a failed write synchronously, closing this + // channel before write() returns; a plain socket defers the report. + _checkOpen(); // Plain sockets may buffer the send; wait until writable before flushing. if (offset < bytes.length || raw is! RawSecureSocket) { final ready = _writable = Completer(); raw.writeEventsEnabled = true; await ready.future; - if (_closed) { - throw _error ?? const SocketException('SOCKS transport closed'); - } + _checkOpen(); } } } + void _checkOpen() { + if (_closed) { + throw _error ?? const SocketException('SOCKS transport closed'); + } + } + StreamSubscription detach() { _detached = true; raw.readEventsEnabled = false; @@ -130,6 +189,27 @@ class RawChannel { return raw is RawSecureSocket ? _TlsSocket(this) : _ConnectionSocket(this); } + void _closeRead() { + if (_readClosed || _closed || _detached) return; + _readClosed = true; + raw.readEventsEnabled = false; + try { + raw.shutdown(SocketDirection.receive); + } finally { + if (_writeClosed) destroy(); + } + } + + void _closeWrite() { + if (_writeClosed || _closed) return; + _writeClosed = true; + try { + raw.shutdown(SocketDirection.send); + } finally { + if (_readClosed) destroy(); + } + } + void destroy() { if (_closed) return; _closed = true; @@ -187,9 +267,7 @@ class _RawConsumer implements StreamConsumer> { } @override - Future close() async { - if (!channel._closed) channel.raw.shutdown(SocketDirection.send); - } + Future close() async => channel._closeWrite(); } class _ConnectionSocket extends StreamView implements Socket { diff --git a/lib/src/socks_protocol.dart b/lib/src/socks_protocol.dart index 305bf51..be6d5b3 100644 --- a/lib/src/socks_protocol.dart +++ b/lib/src/socks_protocol.dart @@ -63,12 +63,18 @@ int? socksConnectReplyLength(List reply) { Future negotiateSocks({ required Future Function(List) write, required Future> Function(int) read, + Future> Function(List, int)? exchange, List? credentials, bool allowNoAuthFallback = false, }) async { + Future> requestReply(List request, int replyLength) async { + if (exchange != null) return exchange(request, replyLength); + await write(request); + return read(replyLength); + } + final methods = credentials == null ? [0] : [if (allowNoAuthFallback) 0, 2]; - await write([5, methods.length, ...methods]); - final greeting = await read(2); + final greeting = await requestReply([5, methods.length, ...methods], 2); if (greeting[0] != 5) { throw const SocksProtocolFailure('invalid reply version'); } @@ -84,8 +90,7 @@ Future negotiateSocks({ throw const SocksProtocolFailure('proxy rejected authentication method'); } if (greeting[1] == 2) { - await write(credentials!); - final auth = await read(2); + final auth = await requestReply(credentials!, 2); if (auth[0] != 1 || auth[1] != 0) { throw const SocksProtocolFailure('username/password auth rejected', rejected: true); diff --git a/pubspec.yaml b/pubspec.yaml index 28044e2..35b4b9f 100644 --- a/pubspec.yaml +++ b/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: diff --git a/test/close_output_test.dart b/test/close_output_test.dart new file mode 100644 index 0000000..4455cd1 --- /dev/null +++ b/test/close_output_test.dart @@ -0,0 +1,160 @@ +import 'dart:async'; + +import 'package:socks_socket/socks_socket.dart'; +import 'package:test/test.dart'; + +import 'helpers/tunnel_peer.dart'; + +void main() { + for (final tls in [false, true]) { + test('closeOutput rejects a racing write and a new output sink (TLS=$tls)', + () async { + final (client, peer) = await connectTunnel(tls: tls); + final response = client.inputStream.first; + response.ignore(); + await client.write('request'); + + final closing = client.closeOutput(); + closing.ignore(); + expect(() => client.outputStream.add([0]), throwsStateError); + final late = Future.microtask(() => client.write('late')); + await expectLater(late, throwsStateError); + await closing.timeout(tunnelDeadline); + await peer.done.future.timeout(tunnelDeadline); + expect(peer.bytes.takeBytes(), 'request'.codeUnits); + + peer.socket.add([42]); + await peer.socket.flush(); + expect(await response.timeout(tunnelDeadline), [42]); + await client.close().timeout(tunnelDeadline); + }); + + test('closeOutput rejects new writes while draining an upload (TLS=$tls)', + () async { + final (client, peer) = await connectTunnel(tls: tls); + final source = StreamController>(); + addTearDown(() async { + client.destroy(); + await source.close(); + }); + final response = client.inputStream.first; + response.ignore(); + final output = client.outputStream; + final uploading = output.addStream(source.stream); + uploading.ignore(); + source.add([65]); + await peer.firstChunk.future.timeout(tunnelDeadline); + + final closing = client.closeOutput(); + closing.ignore(); + await expectLater(client.write('late'), throwsStateError); + expect(() => output.add([0]), throwsStateError); + expect(() => output.addStream(Stream.value([0])), throwsStateError); + source.add([66]); + await source.close(); + await Future.wait([uploading, closing, client.closeOutput()]) + .timeout(tunnelDeadline); + await peer.done.future.timeout(tunnelDeadline); + expect(peer.bytes.takeBytes(), [65, 66]); + + peer.socket.add([42]); + await peer.socket.flush(); + expect(await response.timeout(tunnelDeadline), [42]); + expect(client.state, SocksSocketState.connected); + await client.close().timeout(tunnelDeadline); + }); + + test('closeOutput timeout cancels an unfinished upload (TLS=$tls)', + () async { + final (client, peer) = await connectTunnel( + tls: tls, + operationTimeout: const Duration(milliseconds: 100), + ); + final source = StreamController>(); + addTearDown(() async { + client.destroy(); + await source.close(); + }); + final inputDone = client.inputStream.drain(); + final output = client.outputStream; + final uploading = output.addStream(source.stream); + uploading.ignore(); + source.add([65]); + await peer.firstChunk.future.timeout(tunnelDeadline); + + await expectLater(client.closeOutput().timeout(tunnelDeadline), + throwsA(isA())); + expect(source.hasListener, isFalse); + await expectLater(uploading, throwsA(isA())); + await expectLater(output.done, throwsA(isA())); + await inputDone.timeout(tunnelDeadline); + await peer.done.future.timeout(tunnelDeadline); + expect(client.state, SocksSocketState.error); + await expectLater(client.write('late'), throwsStateError); + await expectLater(client.close().timeout(tunnelDeadline), + throwsA(isA())); + }); + + for (final peerEof in [false, true]) { + test( + '${peerEof ? 'peer EOF' : 'close()'} during closeOutput drains the upload (TLS=$tls)', + () async { + final (client, peer) = await connectTunnel(tls: tls); + final inputDone = client.inputStream.drain(); + final source = StreamController>(); + addTearDown(() async { + client.destroy(); + await source.close(); + }); + final uploading = client.outputStream.addStream(source.stream); + uploading.ignore(); + source.add([65]); + await peer.firstChunk.future.timeout(tunnelDeadline); + final closingOutput = client.closeOutput(); + closingOutput.ignore(); + if (peerEof) { + await peer.socket.close(); + await inputDone.timeout(tunnelDeadline); + } + final closing = client.close(); + closing.ignore(); + expect(client.state, SocksSocketState.disconnected); + source.add([66]); + await source.close(); + await Future.wait([uploading, closingOutput, closing]) + .timeout(tunnelDeadline); + await inputDone.timeout(tunnelDeadline); + await peer.done.future.timeout(tunnelDeadline); + expect(peer.bytes.takeBytes(), [65, 66]); + }); + } + + test( + 'destroy() interrupts closeOutput and a later close succeeds (TLS=$tls)', + () async { + final (client, peer) = await connectTunnel(tls: tls); + final source = StreamController>(); + addTearDown(() async { + client.destroy(); + await source.close(); + }); + final inputDone = client.inputStream.drain(); + final uploading = client.outputStream.addStream(source.stream); + uploading.ignore(); + source.add([65]); + await peer.firstChunk.future.timeout(tunnelDeadline); + final closingOutput = client.closeOutput(); + closingOutput.ignore(); + client.destroy(); + + final destroyed = throwsA(isA() + .having((error) => error.message, 'message', contains('destroyed'))); + await expectLater(uploading, destroyed); + await expectLater(closingOutput, destroyed); + await client.close().timeout(tunnelDeadline); + await inputDone.timeout(tunnelDeadline); + expect(source.hasListener, isFalse); + expect(client.state, SocksSocketState.disconnected); + }); + } +} diff --git a/test/connection_edges_test.dart b/test/connection_edges_test.dart index 1bf9ecf..951b711 100644 --- a/test/connection_edges_test.dart +++ b/test/connection_edges_test.dart @@ -23,6 +23,8 @@ void main() { expect( await proxy.applicationData.future.timeout(const Duration(seconds: 2)), [99]); + await socket.close(); + await proxy.disconnected.future.timeout(const Duration(seconds: 2)); }); test('upload stream errors close transport and retain the original error', diff --git a/test/connection_lifecycle_test.dart b/test/connection_lifecycle_test.dart index af5da1c..4a351ff 100644 --- a/test/connection_lifecycle_test.dart +++ b/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(); @@ -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(); await uploader.write('A' * size); await input; diff --git a/test/connection_socket_test.dart b/test/connection_socket_test.dart index 53fe48f..1da9f79 100644 --- a/test/connection_socket_test.dart +++ b/test/connection_socket_test.dart @@ -1,5 +1,6 @@ import 'dart:async'; import 'dart:io'; +import 'dart:typed_data'; import 'package:socks_socket/src/connection_socket.dart'; import 'package:test/test.dart'; @@ -9,6 +10,8 @@ class _PendingWriteSocket extends Stream implements RawSocket { final writeStarted = Completer(); bool pendingWrite = false; bool closedWithPendingWrite = false; + bool closed = false; + final shutdowns = []; @override bool readEventsEnabled = false; @@ -39,11 +42,13 @@ class _PendingWriteSocket extends Stream implements RawSocket { @override void shutdown(SocketDirection direction) { + shutdowns.add(direction); closedWithPendingWrite |= pendingWrite; } @override Future close() async { + closed = true; closedWithPendingWrite |= pendingWrite; events.close(); return this; @@ -53,7 +58,141 @@ class _PendingWriteSocket extends Stream implements RawSocket { dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); } +class _FakeRawSocketConnectTask implements RawSocketConnectTask { + final Completer result; + bool cancelled = false; + + _FakeRawSocketConnectTask(this.result); + + @override + Future get socket => result.future; + + @override + void cancel() { + cancelled = true; + if (!result.isCompleted) { + result.completeError(const SocketException('cancelled')); + } + } +} + +/// A socket that fails its next read or write synchronously: dart:io delivers +/// the error through the event stream and closes the socket before the call +/// returns, so no event follows. A plain socket reports a read failure this +/// way, which is how macOS surfaces a reset peer, and a secure socket reports +/// a write after the connection closed this way. +class _ResetSocket extends Stream implements RawSocket { + final events = StreamController(sync: true); + var reads = 0; + + @override + bool writeEventsEnabled = false; + bool _readEventsEnabled = false; + + @override + bool get readEventsEnabled => _readEventsEnabled; + + @override + set readEventsEnabled(bool enabled) { + _readEventsEnabled = enabled; + if (enabled) { + scheduleMicrotask(() { + if (_readEventsEnabled && events.hasListener) { + events.add(RawSocketEvent.read); + } + }); + } + } + + @override + StreamSubscription listen( + void Function(RawSocketEvent)? onData, { + Function? onError, + void Function()? onDone, + bool? cancelOnError, + }) => + events.stream.listen(onData, + onError: onError, onDone: onDone, cancelOnError: cancelOnError); + + @override + Uint8List? read([int? len]) { + // Nothing to read the first time; the reset arrives with the read event. + if (reads++ == 0) return null; + _reset('Read failed'); + return null; + } + + @override + int write(List buffer, [int offset = 0, int? count]) { + _reset('Write failed'); + return 0; + } + + void _reset(String what) { + events.addError(SocketException('$what: Connection reset by peer')); + scheduleMicrotask(() { + if (events.hasListener) events.add(RawSocketEvent.closed); + events.close(); + }); + } + + @override + Future close() async => this; + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + void main() { + for (final inputFirst in [true, false]) { + test( + 'channel closes after both directions close (${inputFirst ? 'input' : 'output'} first)', + () async { + final raw = _PendingWriteSocket(); + final socket = RawChannel(raw).socket(); + final input = socket.listen((_) {}); + + if (inputFirst) { + await input.cancel(); + expect(raw.shutdowns, [SocketDirection.receive]); + expect(raw.closed, isFalse); + await socket.close(); + } else { + await socket.close(); + expect(raw.shutdowns, [SocketDirection.send]); + expect(raw.closed, isFalse); + await input.cancel(); + } + + expect(raw.shutdowns, + containsAll([SocketDirection.receive, SocketDirection.send])); + expect(raw.closed, isTrue); + }); + } + + for (final delayedStart in [false, true]) { + test( + 'cancellable connect cancels ${delayedStart ? 'before' : 'after'} the task is created', + () async { + final start = Completer(); + final socket = Completer(); + socket.future.ignore(); + final task = _FakeRawSocketConnectTask(socket); + final connector = CancellableRawSocketConnect((_, __) async { + if (delayedStart) await start.future; + return task; + }); + final connecting = connector.connect(InternetAddress.loopbackIPv4.address, + 1080, const Duration(seconds: 2)); + final failed = expectLater(connecting, throwsA(isA())); + if (!delayedStart) await Future.value(); + connector.cancel(); + if (delayedStart) start.complete(); + await failed.timeout(const Duration(seconds: 1)); + expect(task.cancelled, isTrue); + }); + } + test('flush followed by close delivers the complete TCP payload', () async { final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); addTearDown(server.close); @@ -102,4 +241,27 @@ void main() { socket.destroy(); await flushed.timeout(const Duration(seconds: 2)); }); + + test('a read fails at once when the socket reports a reset synchronously', + () async { + final raw = _ResetSocket(); + final channel = RawChannel(raw); + await expectLater( + channel.read(2).timeout(const Duration(seconds: 2)), + throwsA(isA() + .having((e) => e.message, 'message', contains('reset')))); + expect(raw.reads, 2); + expect(channel.isClosed, isTrue); + }); + + test('a write fails at once when the socket reports a reset synchronously', + () async { + final raw = _ResetSocket(); + final channel = RawChannel(raw); + await expectLater( + channel.write([1, 2, 3]).timeout(const Duration(seconds: 2)), + throwsA(isA() + .having((e) => e.message, 'message', contains('reset')))); + expect(channel.isClosed, isTrue); + }); } diff --git a/test/create_arguments_test.dart b/test/create_arguments_test.dart new file mode 100644 index 0000000..a64d38e --- /dev/null +++ b/test/create_arguments_test.dart @@ -0,0 +1,50 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:socks_socket/socks_socket.dart'; +import 'package:test/test.dart'; + +import 'helpers/tunnel_peer.dart'; + +void main() { + final host = InternetAddress.loopbackIPv4.address; + + test('create() rejects an empty proxy host', () async { + await expectLater(SOCKSSocket.create(proxyHost: '', proxyPort: 1080), + throwsArgumentError); + }); + + for (final port in [0, 65536]) { + test('create() rejects proxy port $port', () async { + await expectLater(SOCKSSocket.create(proxyHost: host, proxyPort: port), + throwsArgumentError); + }); + } + + test('create() rejects a non-positive handshake timeout', () async { + await expectLater( + SOCKSSocket.create( + proxyHost: host, proxyPort: 1080, handshakeTimeout: Duration.zero), + throwsArgumentError); + }); + + test('create() rejects a non-positive operation timeout', () async { + await expectLater( + SOCKSSocket.create( + proxyHost: host, + proxyPort: 1080, + operationTimeout: const Duration(seconds: -1)), + throwsArgumentError); + }); + + test('a sub-second handshake timeout is reported in milliseconds', () async { + final server = await TunnelServer.start(); + server.holdGreeting = TunnelHold(); + final client = await server.createClient( + handshakeTimeout: const Duration(milliseconds: 500)); + await expectLater( + client.connect(), + throwsA(isA() + .having((e) => e.message, 'message', contains('500 ms')))); + }); +} diff --git a/test/destroy_test.dart b/test/destroy_test.dart new file mode 100644 index 0000000..c4694cb --- /dev/null +++ b/test/destroy_test.dart @@ -0,0 +1,68 @@ +import 'dart:async'; + +import 'package:socks_socket/socks_socket.dart'; +import 'package:test/test.dart'; + +import 'helpers/tunnel_peer.dart'; + +final _destroyed = isA() + .having((e) => e.message, 'message', contains('destroyed')); + +void main() { + for (final tls in [false, true]) { + group('destroy() (TLS=$tls)', () { + test('fails a pending outputStream write', () async { + final (client, peer) = await connectTunnel(tls: tls); + // The peer stops reading, so the write stays in flight. + peer.input.pause(); + final uploading = client.outputStream + .addStream(Stream.value(List.filled(8 * 1024 * 1024, 1))); + final failed = expectLater(uploading, throwsA(_destroyed)); + await Future.delayed(const Duration(milliseconds: 200)); + client.destroy(); + await failed.timeout(tunnelDeadline); + expect(client.state, SocksSocketState.disconnected); + await client.close().timeout(tunnelDeadline); + }); + + test('fails a pending write()', () async { + final (client, peer) = await connectTunnel(tls: tls); + peer.input.pause(); + final writing = client.write('A' * (8 * 1024 * 1024)); + final failed = expectLater(writing, throwsA(_destroyed)); + await Future.delayed(const Duration(milliseconds: 200)); + client.destroy(); + await failed.timeout(tunnelDeadline); + }); + + test('rejects later writes', () async { + final (client, _) = await connectTunnel(tls: tls); + client.destroy(); + await expectLater(client.write('A'), throwsStateError); + expect(() => client.outputStream.add([65]), throwsStateError); + await client.close().timeout(tunnelDeadline); + }); + + test('ends inputStream', () async { + final (client, _) = await connectTunnel(tls: tls); + final input = client.inputStream.drain(); + client.destroy(); + await input.timeout(tunnelDeadline); + }); + + test('may be called from the upload source\'s onCancel', () async { + final (client, peer) = await connectTunnel(tls: tls); + peer.input.pause(); + final source = StreamController>(onCancel: client.destroy); + source.add(List.filled(8 * 1024 * 1024, 1)); + final uploading = client.outputStream.addStream(source.stream); + final failed = + expectLater(uploading, throwsA(isA())); + await Future.delayed(const Duration(milliseconds: 200)); + client.destroy(); + await failed.timeout(tunnelDeadline); + await client.close().timeout(tunnelDeadline); + }); + }); + } +} diff --git a/test/helpers/tunnel_peer.dart b/test/helpers/tunnel_peer.dart new file mode 100644 index 0000000..3ec3d9e --- /dev/null +++ b/test/helpers/tunnel_peer.dart @@ -0,0 +1,212 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:socks_socket/socks_socket.dart'; +import 'package:socks_socket/src/connection_socket.dart'; +import 'package:test/test.dart'; + +import 'test_certificates.dart'; + +const tunnelDeadline = Duration(seconds: 5); + +class TunnelPeer { + final Socket socket; + final bytes = BytesBuilder(copy: false); + final firstChunk = Completer(); + final done = Completer(); + late final StreamSubscription> input; + + TunnelPeer(this.socket) { + input = socket.listen((chunk) { + bytes.add(chunk); + if (!firstChunk.isCompleted) firstChunk.complete(); + }, onError: (Object error, StackTrace stack) { + done.completeError(error, stack); + }, onDone: () { + if (!done.isCompleted) done.complete(); + }); + done.future.ignore(); + } +} + +/// A point where [TunnelServer] waits until the test lets it go on. +class TunnelHold { + final reached = Completer(); + final release = Completer(); + + /// For [TunnelServer.holdTls]: how many bytes of the server's handshake + /// reply reach the client before the hold. + final int after; + + TunnelHold({this.after = 0}); + + Future _wait() { + if (!reached.isCompleted) reached.complete(); + return release.future; + } +} + +/// A SOCKS5 server that accepts every CONNECT and hands back the tunnel end. +class TunnelServer { + final RawServerSocket _server; + final TestCertificates? certificates; + final _peers = StreamController(); + late final _accepted = StreamIterator(_peers.stream); + late final int port = _server.port; + int connections = 0; + + /// Delays the greeting reply, so a client stays in connect(). + TunnelHold? holdGreeting; + + /// Delays the CONNECT reply, so a client stays in connectTo(). + TunnelHold? holdConnect; + + /// Stalls the TLS handshake after [TunnelHold.after] bytes of the server's + /// reply, so a client stays in connectTo() after the SOCKS reply. + TunnelHold? holdTls; + + TunnelServer._(this._server, this.certificates) { + _server.listen(_serve); + } + + bool get tls => certificates != null; + + static Future start({bool tls = false}) async { + final raw = await RawServerSocket.bind(InternetAddress.loopbackIPv4, 0); + final server = + TunnelServer._(raw, tls ? TestCertificates.generate() : null); + addTearDown(server.close); + return server; + } + + Future _serve(RawSocket raw) async { + connections++; + addTearDown(raw.close); + try { + final channel = RawChannel(raw); + final greeting = await channel.read(2); + await channel.read(greeting[1]); + await holdGreeting?._wait(); + await channel.write([5, 0]); + final request = await channel.read(5); + await channel.read(request[4] + 2); + await holdConnect?._wait(); + await channel.write([5, 0, 0, 1, 127, 0, 0, 1, 0, 0]); + final Socket transport; + final hold = holdTls; + if (hold != null) { + transport = await _stalledTls(channel, hold); + } else if (tls) { + final secured = await RawSecureSocket.secureServer( + raw, + certificates!.serverContext(), + subscription: channel.detach(), + ); + transport = RawChannel(secured).socket(); + } else { + transport = channel.socket(); + } + addTearDown(transport.destroy); + _peers.add(TunnelPeer(transport)); + } catch (_) { + // The client gave up on this connection; nextPeer() times out if a + // test expected it. + raw.close(); + } + } + + /// Relays the TLS handshake through a local server and withholds its reply + /// after [TunnelHold.after] bytes until the hold is released. + Future _stalledTls(RawChannel channel, TunnelHold hold) async { + final backend = await SecureServerSocket.bind( + InternetAddress.loopbackIPv4, 0, certificates!.serverContext()); + addTearDown(backend.close); + final accepted = backend.first; + accepted.ignore(); + final front = channel.socket(); + final upstream = + await Socket.connect(InternetAddress.loopbackIPv4, backend.port); + addTearDown(upstream.destroy); + front.listen(upstream.add, + onError: (Object _) => upstream.destroy(), onDone: upstream.destroy); + var forwarded = 0; + late final StreamSubscription> reply; + reply = upstream.listen((chunk) async { + final held = hold.reached.isCompleted; + final keep = held ? chunk.length : hold.after - forwarded; + if (keep > 0) front.add(chunk.sublist(0, keep)); + forwarded += keep; + if (held) return; + reply.pause(); + await hold._wait(); + if (keep < chunk.length) front.add(chunk.sublist(keep)); + reply.resume(); + }, onError: (Object _) => front.destroy(), onDone: front.destroy); + return accepted; + } + + /// The next tunnel the server completed. + Future nextPeer() async { + if (!await _accepted.moveNext().timeout(tunnelDeadline)) { + throw StateError('Tunnel server closed'); + } + return _accepted.current; + } + + Future createClient({ + Duration handshakeTimeout = const Duration(seconds: 30), + Duration operationTimeout = tunnelDeadline, + bool closeOnPeerEof = true, + }) async { + final client = await SOCKSSocket.create( + proxyHost: InternetAddress.loopbackIPv4.address, + proxyPort: port, + sslEnabled: tls, + securityContext: certificates?.clientContext(), + handshakeTimeout: handshakeTimeout, + operationTimeout: operationTimeout, + closeOnPeerEof: closeOnPeerEof, + ); + addTearDown(() => client.close().catchError((_) {})); + return client; + } + + /// A connected client and its tunnel end. + Future<(SOCKSSocket, TunnelPeer)> connect({ + Duration handshakeTimeout = const Duration(seconds: 30), + Duration operationTimeout = tunnelDeadline, + bool closeOnPeerEof = true, + }) async { + final client = await createClient( + handshakeTimeout: handshakeTimeout, + operationTimeout: operationTimeout, + closeOnPeerEof: closeOnPeerEof, + ); + await client.connect(); + await client.connectTo('localhost', 443); + return (client, await nextPeer()); + } + + Future close() async { + for (final hold in [holdGreeting, holdConnect, holdTls]) { + if (hold != null && !hold.release.isCompleted) hold.release.complete(); + } + await _server.close(); + await _accepted.cancel(); + // Without a listener the controller's close() would never complete. + _peers.close().ignore(); + } +} + +Future<(SOCKSSocket, TunnelPeer)> connectTunnel({ + bool tls = false, + Duration operationTimeout = tunnelDeadline, + bool closeOnPeerEof = true, +}) async { + final server = await TunnelServer.start(tls: tls); + return server.connect( + operationTimeout: operationTimeout, + closeOnPeerEof: closeOnPeerEof, + ); +} diff --git a/test/input_lifecycle_test.dart b/test/input_lifecycle_test.dart index c7588ef..94bfa72 100644 --- a/test/input_lifecycle_test.dart +++ b/test/input_lifecycle_test.dart @@ -44,21 +44,16 @@ void main() { await socket.connect(); await socket.connectTo('localhost', 443); final peer = await accepted.future; - expect(socket.subscription!.isPaused, isTrue); peer.add([1]); await peer.flush(); expect(await socket.inputStream.first, [1]); - expect(socket.subscription!.isPaused, isTrue); peer.add([2]); await peer.flush(); expect(await socket.inputStream.first, [2]); - expect(socket.subscription!.isPaused, isTrue); final first = socket.inputStream.listen((_) {}); final second = socket.inputStream.listen((_) {}); await first.cancel(); - expect(socket.subscription!.isPaused, isFalse); await second.cancel(); - expect(socket.subscription!.isPaused, isTrue); }); test( @@ -83,7 +78,6 @@ void main() { await socket.reconnect(); final reply = socket.inputStream.first; await stale.cancel(); - expect(socket.subscription!.isPaused, isFalse); await socket.write('A'); expect(await reply.timeout(const Duration(seconds: 2)), [65]); }); diff --git a/test/output_lifecycle_test.dart b/test/output_lifecycle_test.dart index 54a710b..cf8d4cb 100644 --- a/test/output_lifecycle_test.dart +++ b/test/output_lifecycle_test.dart @@ -94,6 +94,7 @@ void main() { socket.outputStream.addError(error); await done; expect(socket.state, ConnectionState.error); + expect(() => socket.closeOutput(), throwsA(same(error))); }); }); } @@ -121,6 +122,31 @@ void main() { await socket.outputStream.close(); }); + test('a rejected addStream does not leave output shutdown pending', () async { + final proxy = MockSocksServer(); + await proxy.start(); + addTearDown(proxy.stop); + final socket = await SOCKSSocket.create( + proxyHost: InternetAddress.loopbackIPv4.address, + proxyPort: proxy.port, + ); + addTearDown(() => socket.close().catchError((_) {})); + await socket.connect(); + await socket.connectTo('localhost', 443); + + final source = StreamController>(); + addTearDown(source.close); + final first = source.stream.listen((_) {}); + addTearDown(first.cancel); + expect( + () => socket.outputStream.addStream(source.stream), throwsStateError); + + final reply = socket.inputStream.first; + socket.outputStream.add([65]); + expect(await reply.timeout(const Duration(seconds: 2)), [65]); + await socket.close().timeout(const Duration(seconds: 2)); + }); + test('output during the handshake does not abort it', () async { final proxy = MockSocksServer() ..responseDelay = const Duration(milliseconds: 100); @@ -245,6 +271,7 @@ void main() { } expect(socket.state, ConnectionState.error); await expectLater(socket.write('B'), throwsStateError); + await expectLater(socket.close(), throwsA(isA())); }); test('a write failing after reconnect does not affect the new connection', @@ -322,7 +349,7 @@ void main() { proxyPort: server.port, operationTimeout: const Duration(milliseconds: 100), ); - addTearDown(socket.close); + addTearDown(() => socket.close().catchError((_) {})); await socket.connect(); await socket.connectTo('localhost', 443); final inputDone = socket.inputStream.drain(); @@ -334,5 +361,6 @@ void main() { await inputDone; expect(socket.state, ConnectionState.error); await expectLater(socket.write('Z'), throwsStateError); + await expectLater(socket.close(), throwsA(isA())); }); } diff --git a/test/peer_eof_test.dart b/test/peer_eof_test.dart index 029e629..1bbf831 100644 --- a/test/peer_eof_test.dart +++ b/test/peer_eof_test.dart @@ -3,131 +3,62 @@ import 'dart:io'; import 'dart:typed_data'; import 'package:socks_socket/socks_socket.dart'; -import 'package:socks_socket/src/connection_socket.dart'; import 'package:test/test.dart'; -import 'helpers/test_certificates.dart'; +import 'helpers/tunnel_peer.dart'; -const _deadline = Duration(seconds: 5); +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; -class _Peer { - final Socket socket; - final bytes = BytesBuilder(copy: false); - final firstChunk = Completer(); - final done = Completer(); - late final StreamSubscription> input; +void main() { + for (final tls in [false, true]) { + test('closeOutput preserves a response sent after request EOF (TLS=$tls)', + () async { + final (client, peer) = await connectTunnel(tls: tls); + final response = client.inputStream.first; + await client.write('request'); + await client.closeOutput().timeout(_deadline); + await client.closeOutput().timeout(_deadline); + await expectLater(client.write('late'), throwsStateError); + expect(() => client.outputStream.add([0]), throwsStateError); + await peer.done.future.timeout(_deadline); + expect(peer.bytes.takeBytes(), 'request'.codeUnits); - _Peer(this.socket) { - input = socket.listen((chunk) { - bytes.add(chunk); - if (!firstChunk.isCompleted) firstChunk.complete(); - }, onError: (Object error, StackTrace stack) { - done.completeError(error, stack); - }, onDone: () { - if (!done.isCompleted) done.complete(); + peer.socket.add([42]); + await peer.socket.flush(); + expect(await response.timeout(_deadline), [42]); + expect(client.state, SocksSocketState.connected); + await client.close().timeout(_deadline); + expect(client.state, SocksSocketState.disconnected); }); - done.future.ignore(); } -} - -Future<(SOCKSSocket, _Peer)> _connect({ - bool tls = false, - Duration operationTimeout = _deadline, -}) async { - final certificates = tls ? TestCertificates.generate() : null; - final server = await RawServerSocket.bind(InternetAddress.loopbackIPv4, 0); - addTearDown(server.close); - final accepted = Completer<_Peer>(); - server.listen((raw) async { - addTearDown(raw.close); - try { - final channel = RawChannel(raw); - final greeting = await channel.read(2); - await channel.read(greeting[1]); - await channel.write([5, 0]); - final request = await channel.read(5); - await channel.read(request[4] + 2); - await channel.write([5, 0, 0, 1, 127, 0, 0, 1, 0, 0]); - final Socket transport; - if (tls) { - final secured = await RawSecureSocket.secureServer( - raw, - certificates!.serverContext(), - subscription: channel.detach(), - ); - transport = RawChannel(secured).socket(); - } else { - transport = channel.socket(); - } - addTearDown(transport.destroy); - accepted.complete(_Peer(transport)); - } catch (error, stack) { - accepted.completeError(error, stack); - } - }); - final client = await SOCKSSocket.create( - proxyHost: InternetAddress.loopbackIPv4.address, - proxyPort: server.port, - sslEnabled: tls, - securityContext: certificates?.clientContext(), - operationTimeout: operationTimeout, - ); - addTearDown(() => client.close().catchError((_) {})); - await client.connect(); - await client.connectTo('localhost', 443); - return (client, await accepted.future.timeout(_deadline)); -} - -void main() { - test('peer EOF drains a pending underlying socket flush', () async { - final (client, peer) = await _connect(); - final eof = client.inputStream.drain(); - // 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.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.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 { - final (client, peer) = await _connect(tls: tls); + test('peer EOF drains a pending write (TLS=$tls)', () async { + final (client, peer) = await connectTunnel(tls: tls); final eof = client.inputStream.drain(); - final source = StreamController>(); - 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.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 { - final (client, peer) = await _connect(tls: tls); + final (client, peer) = await connectTunnel(tls: tls); final eof = client.inputStream.drain(); final source = StreamController>(); addTearDown(source.close); @@ -151,11 +82,30 @@ void main() { await peer.done.future.timeout(_deadline); expect(peer.bytes.takeBytes(), [65, 66]); }); + test('closeOnPeerEof: false keeps writes open after peer EOF (TLS=$tls)', + () async { + final (client, peer) = + await connectTunnel(tls: tls, closeOnPeerEof: false); + final eof = client.inputStream.drain(); + await peer.socket.close(); + await eof.timeout(_deadline); + + expect(client.state, ConnectionState.connected); + await client.write('A').timeout(_deadline); + await client.outputStream + .addStream(Stream.value([66])) + .timeout(_deadline); + client.outputStream.add([67]); + await client.close().timeout(_deadline); + await peer.done.future.timeout(_deadline); + expect(peer.bytes.takeBytes(), [65, 66, 67]); + expect(client.state, ConnectionState.disconnected); + }); } test('an output sink first used after peer EOF rejects new uploads', () async { - final (client, peer) = await _connect(); + final (client, peer) = await connectTunnel(); final eof = client.inputStream.drain(); await peer.socket.close(); await eof.timeout(_deadline); @@ -167,24 +117,21 @@ void main() { expect(peer.bytes.length, 0); }); - test('reset after peer EOF cancels a paused underlying upload source', - () async { - final (client, peer) = await _connect(); + test('reset after peer EOF cancels a paused outputStream source', () async { + final (client, peer) = await connectTunnel(); final eof = client.inputStream.drain(); final source = StreamController>(); 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())); + final uploading = client.outputStream.addStream(source.stream); + final failed = + expectLater(uploading, throwsA(isA())); source.add([65]); await peer.firstChunk.future.timeout(_deadline); peer.input.pause(); - source.add(List.filled(512 * 1024, 66)); + source.add(List.filled(8 * 1024 * 1024, 66)); await Future.delayed(const Duration(milliseconds: 20)); expect(source.isPaused, isTrue); await peer.socket.close(); @@ -194,31 +141,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())); 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 _connect( - operationTimeout: const Duration(milliseconds: 100), - ); - final eof = client.inputStream.drain(); - final source = StreamController>(); - 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())); - 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(); + final source = StreamController>(); + 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())); + await peer.done.future.timeout(_deadline); + expect(client.state, ConnectionState.disconnected); + expect(source.hasListener, isFalse); + }); } diff --git a/test/socks_socket_test.dart b/test/socks_socket_test.dart index 6d71c0c..eb803d1 100644 --- a/test/socks_socket_test.dart +++ b/test/socks_socket_test.dart @@ -859,6 +859,23 @@ void main() { await socket.close(); }); + test('authentication receives a fresh handshake timeout', () async { + server + ..requireAuth = true + ..responseDelay = const Duration(milliseconds: 150); + await server.start(); + + final socket = await SOCKSSocket.create( + proxyHost: InternetAddress.loopbackIPv4.address, + proxyPort: server.port, + isolationToken: 'slow-auth', + handshakeTimeout: const Duration(milliseconds: 250), + ); + await socket.connect().timeout(const Duration(seconds: 2)); + expect(socket.state, ConnectionState.connecting); + await socket.close(); + }); + test('reconnect with different isolationToken rotates circuit', () async { server.requireAuth = true; await server.start(); diff --git a/test/teardown_race_test.dart b/test/teardown_race_test.dart new file mode 100644 index 0000000..d31ac69 --- /dev/null +++ b/test/teardown_race_test.dart @@ -0,0 +1,198 @@ +import 'dart:async'; + +import 'package:socks_socket/socks_socket.dart'; +import 'package:test/test.dart'; + +import 'helpers/tunnel_peer.dart'; + +/// close() and destroy() racing a reconnect, a TLS handshake, or each other. +void main() { + final teardowns = Function(SOCKSSocket)>{ + 'destroy()': (client) async => client.destroy(), + 'close()': (client) => client.close(), + }; + + for (final tls in [false, true]) { + for (final MapEntry(key: name, value: teardown) in teardowns.entries) { + group('$name during reconnect (TLS=$tls)', () { + test('right after it starts', () async { + final server = await TunnelServer.start(tls: tls); + final (client, peer) = await server.connect(); + final reconnecting = client.reconnect(); + final failed = expectLater( + reconnecting, throwsA(isA())); + final tearingDown = teardown(client); + await failed.timeout(tunnelDeadline); + await tearingDown.timeout(tunnelDeadline); + expect(client.state, SocksSocketState.disconnected); + await expectLater(client.write('A'), throwsStateError); + await peer.done.future.timeout(tunnelDeadline); + expect(server.connections, 1); + await client.close().timeout(tunnelDeadline); + }); + + test('while the proxy answers the greeting', () async { + final server = await TunnelServer.start(tls: tls); + final (client, peer) = await server.connect(); + await client.write('before'); + await peer.firstChunk.future.timeout(tunnelDeadline); + final hold = server.holdGreeting = TunnelHold(); + final reconnecting = client.reconnect(); + final failed = expectLater( + reconnecting, throwsA(isA())); + await hold.reached.future.timeout(tunnelDeadline); + expect(server.connections, 2); + final tearingDown = teardown(client); + await failed.timeout(tunnelDeadline); + await tearingDown.timeout(tunnelDeadline); + expect(client.state, SocksSocketState.disconnected); + hold.release.complete(); + // The abandoned handshake must not connect the socket after all. + await Future.delayed(const Duration(milliseconds: 200)); + expect(client.state, SocksSocketState.disconnected); + await expectLater(client.write('after'), throwsStateError); + await client.close().timeout(tunnelDeadline); + }); + + test('from the old inputStream ending', () async { + // reconnect() creates the connect task a few microtasks after the + // old connection's close completes, and the old inputStream's done + // event is delivered in that gap. A teardown from there finds no + // task to cancel yet, so sweep a few microtasks after the event. + const microtasks = 4; + final server = await TunnelServer.start(tls: tls); + final clients = [ + for (var i = 0; i < microtasks; i++) + await server.connect(handshakeTimeout: const Duration(seconds: 2)) + ]; + server.holdGreeting = TunnelHold(); + for (var delay = 0; delay < microtasks; delay++) { + final (client, _) = clients[delay]; + final stopwatch = Stopwatch(); + final tornDown = Completer(); + client.inputStream.listen(null, onDone: () async { + for (var i = 0; i < delay; i++) { + await Future.value(); + } + stopwatch.start(); + tornDown.complete(teardown(client)); + }); + await expectLater( + client.reconnect(), throwsA(isA())) + .timeout(tunnelDeadline); + expect(stopwatch.elapsed, lessThan(const Duration(seconds: 1)), + reason: 'teardown $delay microtasks after the old input ended'); + await tornDown.future.timeout(tunnelDeadline); + expect(client.state, SocksSocketState.disconnected); + } + }); + }); + } + + test('reconnect() after cancel() connects again (TLS=$tls)', () async { + final server = await TunnelServer.start(tls: tls); + final hold = server.holdConnect = TunnelHold(); + final client = await server.createClient(); + await client.connect(); + final connecting = client.connectTo('localhost', 443); + final failed = + expectLater(connecting, throwsA(isA())); + await hold.reached.future.timeout(tunnelDeadline); + await client.cancel(); + await failed.timeout(tunnelDeadline); + // The abandoned handshake stays parked; only new connections proceed. + server.holdConnect = null; + await client.reconnect().timeout(tunnelDeadline); + expect(client.state, SocksSocketState.connected); + final peer = await server.nextPeer(); + await client.write('again'); + await peer.firstChunk.future.timeout(tunnelDeadline); + expect(peer.bytes.takeBytes(), 'again'.codeUnits); + }); + + test('reconnect() after destroy() connects again (TLS=$tls)', () async { + final server = await TunnelServer.start(tls: tls); + final (client, _) = await server.connect(); + client.destroy(); + await client.reconnect().timeout(tunnelDeadline); + expect(client.state, SocksSocketState.connected); + final peer = await server.nextPeer(); + await client.write('again'); + await peer.firstChunk.future.timeout(tunnelDeadline); + expect(peer.bytes.takeBytes(), 'again'.codeUnits); + }); + } + + test('a second reconnect() is rejected while one is in progress', () async { + final server = await TunnelServer.start(); + final (client, _) = await server.connect(); + final reconnecting = client.reconnect(); + await expectLater(client.reconnect(), throwsStateError); + await reconnecting.timeout(tunnelDeadline); + expect(client.state, SocksSocketState.connected); + }); + + for (final MapEntry(key: name, value: teardown) in teardowns.entries) { + test('$name ends a TLS handshake at once', () async { + final server = await TunnelServer.start(tls: true); + // Part of the server's reply sits in the TLS filter, so the SDK ignores + // the destroyed transport and waits for the handshake deadline. + final hold = server.holdTls = TunnelHold(after: 64); + final client = await server.createClient( + handshakeTimeout: const Duration(seconds: 5)); + await client.connect(); + final connecting = client.connectTo('localhost', 443); + final failed = + expectLater(connecting, throwsA(isA())); + await hold.reached.future.timeout(tunnelDeadline); + final stopwatch = Stopwatch()..start(); + final tearingDown = teardown(client); + await failed.timeout(tunnelDeadline); + expect(stopwatch.elapsed, lessThan(const Duration(seconds: 1))); + await tearingDown.timeout(tunnelDeadline); + expect(client.state, SocksSocketState.disconnected); + hold.release.complete(); + await client.close().timeout(tunnelDeadline); + }); + } + + for (final tls in [false, true]) { + group('close() after destroy() (TLS=$tls)', () { + final payload = List.filled(8 * 1024 * 1024, 1); + + test('completes when destroy() failed an explicit close in flight', + () async { + final (client, peer) = await connectTunnel(tls: tls); + peer.input.pause(); + final uploading = client.outputStream.addStream(Stream.value(payload)); + final uploadFailed = + expectLater(uploading, throwsA(isA())); + await Future.delayed(const Duration(milliseconds: 200)); + final closing = client.close(); + final closeFailed = + expectLater(closing, throwsA(isA())); + client.destroy(); + await Future.wait([uploadFailed, closeFailed]).timeout(tunnelDeadline); + await client.close().timeout(tunnelDeadline); + expect(client.state, SocksSocketState.disconnected); + }); + + test('completes when destroy() failed a peer-EOF close in flight', + () async { + final (client, peer) = await connectTunnel(tls: tls); + final eof = client.inputStream.drain(); + peer.input.pause(); + final uploading = client.outputStream.addStream(Stream.value(payload)); + final uploadFailed = + expectLater(uploading, throwsA(isA())); + await Future.delayed(const Duration(milliseconds: 200)); + await peer.socket.close(); + await eof.timeout(tunnelDeadline); + client.destroy(); + await uploadFailed.timeout(tunnelDeadline); + await client.close().timeout(tunnelDeadline); + expect(client.state, SocksSocketState.disconnected); + }); + }); + } +} diff --git a/test/manual_tls_test.dart b/test/tls_configuration_test.dart similarity index 62% rename from test/manual_tls_test.dart rename to test/tls_configuration_test.dart index e05b433..a64ef1c 100644 --- a/test/manual_tls_test.dart +++ b/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'); }); }