From f6f1cb4cb065f0e8b1160955a377ea25696a0fda Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Mon, 12 Oct 2026 15:14:47 -0700 Subject: [PATCH] fix(mobile): bound one-shot relay attempts and release failed auth Signed-off-by: Tom Brow --- mobile/lib/shared/relay/relay_socket.dart | 8 + .../lib/shared/relay/signed_event_relay.dart | 29 ++- .../relay/relay_socket_liveness_test.dart | 38 ++++ .../shared/relay/signed_event_relay_test.dart | 212 ++++++++++++++++++ 4 files changed, 282 insertions(+), 5 deletions(-) create mode 100644 mobile/test/shared/relay/signed_event_relay_test.dart diff --git a/mobile/lib/shared/relay/relay_socket.dart b/mobile/lib/shared/relay/relay_socket.dart index 5b23279e81..ca619a2862 100644 --- a/mobile/lib/shared/relay/relay_socket.dart +++ b/mobile/lib/shared/relay/relay_socket.dart @@ -39,6 +39,7 @@ class RelaySocket { final String _wsUrl; final String? _nsec; + final Duration? _connectTimeout; final void Function(List message) _onMessage; final void Function() _onConnected; final void Function(Object? error) _onDisconnected; @@ -58,8 +59,10 @@ class RelaySocket { required void Function(List message) onMessage, required void Function() onConnected, required void Function(Object? error) onDisconnected, + Duration? connectTimeout, }) : _wsUrl = wsUrl, _nsec = nsec, + _connectTimeout = connectTimeout, _onMessage = onMessage, _onConnected = onConnected, _onDisconnected = onDisconnected; @@ -73,9 +76,13 @@ class RelaySocket { _channel = IOWebSocketChannel.connect( Uri.parse(_wsUrl), pingInterval: debugPingInterval, + connectTimeout: _connectTimeout, ); await _channel!.ready; } catch (e) { + // A failed handshake also queues the error on the channel stream. No + // normal subscription exists yet; cancel it so sink.close can complete. + await _channel?.stream.listen((_) {}, onError: (Object _) {}).cancel(); _state = SocketState.disconnected; _onDisconnected(e); return; @@ -133,6 +140,7 @@ class RelaySocket { /// Gracefully close the connection. Future disconnect() async { + _failAuth(Exception('Relay disconnected during authentication')); _resetConnection(); final channel = _channel; _channel = null; diff --git a/mobile/lib/shared/relay/signed_event_relay.dart b/mobile/lib/shared/relay/signed_event_relay.dart index b0106639eb..958a4abcac 100644 --- a/mobile/lib/shared/relay/signed_event_relay.dart +++ b/mobile/lib/shared/relay/signed_event_relay.dart @@ -64,7 +64,8 @@ class SignedEventRelay { /// Publishes one signed event over a short-lived authenticated NIP-42 socket. /// /// This is used for community-removal tombstones because the community being -/// removed is not necessarily the app's active relay session. +/// removed is not necessarily the app's active relay session. The deadline +/// covers connection, authentication, and acknowledgement. Future submitSignedEventOnce({ required String wsUrl, required String nsec, @@ -73,6 +74,7 @@ Future submitSignedEventOnce({ required List> tags, int? createdAt, Duration timeout = const Duration(seconds: 12), + RelaySocketFactory socketFactory = _oneShotSocket, }) async { final privateKey = nostr.Nip19.decode(payload: nsec).data; if (privateKey.isEmpty) throw const FormatException('Invalid nsec'); @@ -87,7 +89,7 @@ Future submitSignedEventOnce({ final event = NostrEvent.fromJson(signed.toMap()); final result = Completer(); late final RelaySocket socket; - socket = RelaySocket( + socket = socketFactory( wsUrl: wsUrl, nsec: nsec, onMessage: (message) { @@ -122,11 +124,28 @@ Future submitSignedEventOnce({ } }, ); - final resultFuture = result.future.timeout(timeout); try { - await socket.connect(); - return await resultFuture; + await Future.wait([ + socket.connect(), + result.future.then((_) {}), + ], eagerError: true).timeout(timeout); + return await result.future; } finally { await socket.disconnect(); } } + +RelaySocket _oneShotSocket({ + required String wsUrl, + required String? nsec, + required void Function(List) onMessage, + required void Function() onConnected, + required void Function(Object? error) onDisconnected, +}) => RelaySocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + connectTimeout: const Duration(seconds: 8), +); diff --git a/mobile/test/shared/relay/relay_socket_liveness_test.dart b/mobile/test/shared/relay/relay_socket_liveness_test.dart index 835ebc2b61..51a80673c0 100644 --- a/mobile/test/shared/relay/relay_socket_liveness_test.dart +++ b/mobile/test/shared/relay/relay_socket_liveness_test.dart @@ -5,6 +5,7 @@ import 'dart:io'; import 'package:buzz/shared/relay/relay_socket.dart'; import 'package:crypto/crypto.dart'; import 'package:flutter_test/flutter_test.dart'; +import 'package:nostr/nostr.dart' as nostr; /// A server that completes the WS handshake then never speaks again: no pongs, /// no close frame. Only a client-side ping timeout can notice. @@ -51,6 +52,43 @@ void main() { RelaySocket.debugPingInterval = RelaySocket.pingInterval; }); + test('disconnect completes pending authentication', () async { + RelaySocket.debugPingInterval = RelaySocket.pingInterval; + final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + final authReceived = Completer(); + WebSocket? peer; + server.listen((request) async { + final connection = await WebSocketTransformer.upgrade(request); + peer = connection; + connection.listen((raw) { + final frame = jsonDecode(raw as String) as List; + if (frame.first == 'AUTH' && !authReceived.isCompleted) { + authReceived.complete(); + } + }); + connection.add(jsonEncode(['AUTH', 'fixture-challenge'])); + }); + final socket = RelaySocket( + wsUrl: 'ws://127.0.0.1:${server.port}', + nsec: nostr.Keys.generate().nsec, + onMessage: (_) {}, + onConnected: () => fail('The peer never acknowledges authentication'), + onDisconnected: (_) {}, + ); + final connecting = socket.connect(); + try { + await authReceived.future; + expect(socket.state, SocketState.authenticating); + await socket.disconnect(); + await connecting.timeout(const Duration(seconds: 1)); + expect(socket.state, SocketState.disconnected); + } finally { + socket.dispose(); + await peer?.close(); + await server.close(force: true); + } + }); + test('detects a peer that stops answering pings', () async { final server = await _silentAfterHandshakeServer(); diff --git a/mobile/test/shared/relay/signed_event_relay_test.dart b/mobile/test/shared/relay/signed_event_relay_test.dart new file mode 100644 index 0000000000..2b0b82ed53 --- /dev/null +++ b/mobile/test/shared/relay/signed_event_relay_test.dart @@ -0,0 +1,212 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:buzz/shared/relay/nostr_models.dart'; +import 'package:buzz/shared/relay/relay_socket.dart'; +import 'package:buzz/shared/relay/signed_event_relay.dart'; +import 'package:fake_async/fake_async.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:nostr/nostr.dart' as nostr; + +class _ControlledSocket extends RelaySocket { + final Completer? connectionGate; + final void Function() connected; + final void Function(List) message; + bool disconnected = false; + bool eventSent = false; + bool acknowledge = false; + + _ControlledSocket({ + required super.wsUrl, + required super.nsec, + required super.onMessage, + required super.onConnected, + required super.onDisconnected, + this.connectionGate, + }) : connected = onConnected, + message = onMessage; + + @override + Future connect() async { + if (connectionGate != null) await connectionGate!.future; + if (!disconnected) connected(); + } + + @override + void send(List payload) { + eventSent = true; + if (acknowledge) { + final event = payload[1] as Map; + message(['OK', event['id'], true, 'saved']); + } + } + + @override + Future disconnect() async { + disconnected = true; + if (connectionGate != null && !connectionGate!.isCompleted) { + connectionGate!.complete(); + } + } +} + +void main() { + final key = nostr.Keys.generate(); + const timeout = Duration(seconds: 12); + + test('connection timeout closes a stalled WebSocket handshake', () async { + final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + server.listen((request) {}); + try { + await expectLater( + submitSignedEventOnce( + wsUrl: 'ws://127.0.0.1:${server.port}', + nsec: key.nsec, + kind: 30350, + content: 'inactive fixture', + tags: const [], + socketFactory: + ({ + required wsUrl, + required nsec, + required onMessage, + required onConnected, + required onDisconnected, + }) => RelaySocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + connectTimeout: const Duration(milliseconds: 50), + ), + ).timeout(const Duration(seconds: 2)), + throwsA( + isA().having( + (error) => error.duration, + 'connection deadline', + const Duration(milliseconds: 50), + ), + ), + ); + } finally { + await server.close(force: true); + } + }); + + test('one-shot deadline includes a stalled connection', () { + fakeAsync((clock) { + final gate = Completer(); + late _ControlledSocket socket; + final errors = []; + final publication = submitSignedEventOnce( + wsUrl: 'wss://relay.example', + nsec: key.nsec, + kind: 30350, + content: 'inactive fixture', + tags: const [], + timeout: timeout, + socketFactory: + ({ + required wsUrl, + required nsec, + required onMessage, + required onConnected, + required onDisconnected, + }) => socket = _ControlledSocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + connectionGate: gate, + ), + ); + unawaited( + publication.then((_) {}, onError: (Object e) => errors.add(e)), + ); + clock.flushMicrotasks(); + try { + clock.elapse(timeout); + clock.flushMicrotasks(); + expect(errors, [isA()]); + expect(socket.disconnected, isTrue); + expect(socket.eventSent, isFalse); + } finally { + if (!gate.isCompleted) gate.complete(); + clock.flushMicrotasks(); + } + }); + }); + + test('one-shot deadline bounds waiting for relay acknowledgement', () { + fakeAsync((clock) { + late _ControlledSocket socket; + final errors = []; + unawaited( + submitSignedEventOnce( + wsUrl: 'wss://relay.example', + nsec: key.nsec, + kind: 30350, + content: 'inactive fixture', + tags: const [], + socketFactory: + ({ + required wsUrl, + required nsec, + required onMessage, + required onConnected, + required onDisconnected, + }) => socket = _ControlledSocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + ), + ).then((_) {}, onError: (Object e) => errors.add(e)), + ); + clock.flushMicrotasks(); + expect(socket.eventSent, isTrue); + clock.elapse(timeout); + expect(errors, [isA()]); + expect(socket.disconnected, isTrue); + }); + }); + + test('accepted one-shot publication closes its temporary socket', () { + fakeAsync((clock) { + late _ControlledSocket socket; + final accepted = []; + unawaited( + submitSignedEventOnce( + wsUrl: 'wss://relay.example', + nsec: key.nsec, + kind: 30350, + content: 'inactive fixture', + tags: const [], + socketFactory: + ({ + required wsUrl, + required nsec, + required onMessage, + required onConnected, + required onDisconnected, + }) { + socket = _ControlledSocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + )..acknowledge = true; + return socket; + }, + ).then(accepted.add), + ); + clock.flushMicrotasks(); + expect(accepted.single.content, 'saved'); + expect(socket.disconnected, isTrue); + }); + }); +}