From 2db6669dd450e0cfc0a91efd018d76a3d42c45e8 Mon Sep 17 00:00:00 2001 From: jm Date: Wed, 14 Oct 2026 17:27:17 +0000 Subject: [PATCH] Verify mobile relay events on receipt Co-authored-by: Codex Signed-off-by: jm --- .../shared/relay/relay_event_verifier.dart | 236 +++++++++ mobile/lib/shared/relay/relay_session.dart | 137 +++--- .../relay/relay_session_event_admission.dart | 240 +++++++++ .../relay_event_ingress_integration_test.dart | 191 ++++++++ .../test/shared/relay/relay_session_test.dart | 457 +++++++++++++++++- 5 files changed, 1180 insertions(+), 81 deletions(-) create mode 100644 mobile/lib/shared/relay/relay_event_verifier.dart create mode 100644 mobile/lib/shared/relay/relay_session_event_admission.dart create mode 100644 mobile/test/shared/relay/relay_event_ingress_integration_test.dart diff --git a/mobile/lib/shared/relay/relay_event_verifier.dart b/mobile/lib/shared/relay/relay_event_verifier.dart new file mode 100644 index 0000000000..452f18342a --- /dev/null +++ b/mobile/lib/shared/relay/relay_event_verifier.dart @@ -0,0 +1,236 @@ +part of 'relay_session.dart'; + +const _maxPendingRelayEvents = 2048; +const _relayEventVerificationBatchSize = 64; +// The relay caps frames at 512 KiB and event content at 256 KiB. This is a +// conservative upper bound for JSON-encoded bytes: a Dart UTF-16 code unit can +// expand to six bytes when JSON escaping is required. +const _maxRelayEventEncodedByteUpperBound = 4 * 1024 * 1024; +const _maxPendingRelayEventEncodedByteUpperBound = 32 * 1024 * 1024; +const _maxRelayEventNestingDepth = 64; + +/// Serializes relay-event verification through one bounded background worker. +class _RelayEventVerifier { + _RelayEventVerifier({ + required Future?>> Function(List) + verifyBatch, + }) : _verifyBatch = verifyBatch; + + final Future?>> Function(List) + _verifyBatch; + final Queue<_PendingRelayEventVerification> _pending = Queue(); + List<_PendingRelayEventVerification> _active = const []; + Timer? _drainTimer; + var _pendingCount = 0; + var _pendingEncodedByteUpperBound = 0; + var _draining = false; + var _closed = false; + + Future verify(Object? eventJson) async { + final verified = await verifyAll([eventJson]); + return verified.single; + } + + Future> verifyAll(List eventJsons) { + if (_closed) { + return Future.error(StateError('Relay event verifier is closed')); + } + final encodedByteUpperBounds = []; + var addedCount = 0; + var addedEncodedByteUpperBound = 0; + for (final eventJson in eventJsons) { + final encodedByteUpperBound = _relayEventEncodedByteUpperBound(eventJson); + encodedByteUpperBounds.add(encodedByteUpperBound); + if (encodedByteUpperBound != null) { + addedCount++; + addedEncodedByteUpperBound += encodedByteUpperBound; + } + } + if (addedCount > _maxPendingRelayEvents - _pendingCount || + addedEncodedByteUpperBound > + _maxPendingRelayEventEncodedByteUpperBound - + _pendingEncodedByteUpperBound) { + return Future.error(StateError('Relay event verification queue is full')); + } + + final futures = >[]; + for (var i = 0; i < eventJsons.length; i++) { + final encodedByteUpperBound = encodedByteUpperBounds[i]; + if (encodedByteUpperBound == null) { + futures.add(Future.value()); + continue; + } + final pending = _PendingRelayEventVerification( + eventJsons[i], + encodedByteUpperBound, + ); + _pending.add(pending); + _pendingCount++; + _pendingEncodedByteUpperBound += encodedByteUpperBound; + futures.add(pending.completer.future); + } + _scheduleDrain(); + return Future.wait(futures); + } + + void close() { + if (_closed) return; + _closed = true; + _drainTimer?.cancel(); + _drainTimer = null; + final error = StateError('Relay event verifier is closed'); + for (final pending in [..._active, ..._pending]) { + _completeError(pending, error, StackTrace.current); + } + _pending.clear(); + } + + void _scheduleDrain() { + if (_closed || _draining || _drainTimer != null || _pending.isEmpty) { + return; + } + // A zero-duration timer coalesces adjacent WebSocket frames into batches. + _drainTimer = Timer(Duration.zero, () { + _drainTimer = null; + unawaited(_drain()); + }); + } + + Future _drain() async { + if (_closed || _draining) return; + _draining = true; + try { + while (!_closed && _pending.isNotEmpty) { + final batch = <_PendingRelayEventVerification>[]; + while (batch.length < _relayEventVerificationBatchSize && + _pending.isNotEmpty) { + batch.add(_pending.removeFirst()); + } + _active = batch; + + try { + final verified = await _verifyBatch([ + for (final pending in batch) pending.eventJson, + ]); + if (verified.length != batch.length) { + throw StateError( + 'Relay event verifier returned ${verified.length} results ' + 'for ${batch.length} events', + ); + } + for (var i = 0; i < batch.length; i++) { + final eventJson = verified[i]; + _complete( + batch[i], + eventJson == null ? null : NostrEvent.fromJson(eventJson), + ); + } + } catch (error, stackTrace) { + for (final pending in batch) { + _completeError(pending, error, stackTrace); + } + } finally { + _active = const []; + } + } + } finally { + _draining = false; + _scheduleDrain(); + } + } + + void _complete(_PendingRelayEventVerification pending, NostrEvent? event) { + if (pending.completer.isCompleted) return; + _pendingCount--; + _pendingEncodedByteUpperBound -= pending.encodedByteUpperBound; + pending.completer.complete(event); + } + + void _completeError( + _PendingRelayEventVerification pending, + Object error, + StackTrace stackTrace, + ) { + if (pending.completer.isCompleted) return; + _pendingCount--; + _pendingEncodedByteUpperBound -= pending.encodedByteUpperBound; + pending.completer.completeError(error, stackTrace); + } +} + +class _PendingRelayEventVerification { + _PendingRelayEventVerification(this.eventJson, this.encodedByteUpperBound); + + final Object? eventJson; + final int encodedByteUpperBound; + final Completer completer = Completer(); +} + +int? _relayEventEncodedByteUpperBound(Object? value) { + return _encodedByteUpperBound(value, _maxRelayEventEncodedByteUpperBound, 0); +} + +int? _encodedByteUpperBound(Object? value, int remaining, int depth) { + if (remaining < 0 || depth > _maxRelayEventNestingDepth) return null; + if (value == null || value is bool || value is num) { + return remaining >= 32 ? 32 : null; + } + if (value is String) { + final bytes = 2 + value.length * 6; + return bytes <= remaining ? bytes : null; + } + if (value is List) { + var total = 2; + for (final element in value) { + final elementBytes = _encodedByteUpperBound( + element, + remaining - total - 1, + depth + 1, + ); + if (elementBytes == null) return null; + total += elementBytes + 1; + } + return total <= remaining ? total : null; + } + if (value is Map) { + var total = 2; + for (final entry in value.entries) { + if (entry.key is! String) return null; + final keyBytes = _encodedByteUpperBound( + entry.key, + remaining - total - 1, + depth + 1, + ); + if (keyBytes == null) return null; + total += keyBytes + 1; + final valueBytes = _encodedByteUpperBound( + entry.value, + remaining - total - 1, + depth + 1, + ); + if (valueBytes == null) return null; + total += valueBytes + 1; + } + return total <= remaining ? total : null; + } + return null; +} + +Future?>> _verifyRelayEventBatchInBackground( + List eventJsons, +) => compute(_verifyRelayEventBatch, eventJsons); + +List?> _verifyRelayEventBatch(List eventJsons) { + return [for (final eventJson in eventJsons) _verifyRelayEvent(eventJson)]; +} + +Map? _verifyRelayEvent(Object? eventJson) { + try { + // Event.fromJson verifies the canonical seven-field ID and BIP-340 + // signature. Its Schnorr boundary strictly decodes the complete key, + // message, and signature widths before invoking bip340 verification. + return nostr.Event.fromJson(jsonEncode(eventJson)).toMap(); + } catch (_) { + return null; + } +} diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index ece4b9266b..bc6f59f85a 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:collection'; import 'dart:convert'; import 'dart:math'; @@ -24,11 +25,18 @@ import 'relay_socket.dart'; export 'relay_session_types.dart'; part 'relay_session_auth.dart'; +part 'relay_event_verifier.dart'; +part 'relay_session_event_admission.dart'; class _HistorySubscription { final List events = []; final Completer> completer; final Timer timeout; + Future verificationTail = Future.value(); + int pendingVerifications = 0; + int receivedEventFrames = 0; + Object? verificationError; + StackTrace? verificationStackTrace; _HistorySubscription({required this.completer, required this.timeout}); } @@ -42,6 +50,11 @@ class _LiveSubscription { int? lastSeenCreatedAt; int closedRetryAttempt = 0; Timer? closedRetryTimer; + Future verificationTail = Future.value(); + int pendingVerifications = 0; + int receivedEventFrames = 0; + int validEventFrames = 0; + int verificationEpoch = 0; _LiveSubscription({ required this.filter, @@ -73,6 +86,12 @@ class _BufferedEvent { } class RelaySessionNotifier extends Notifier { + static const _maxQueryResponseBytes = 8 * 1024 * 1024; + + /// Creates a relay session. + /// + /// [verifyEventBatch] is a test seam for controlling asynchronous event + /// admission. Production callers must use the default background verifier. RelaySessionNotifier({ http.Client? httpClient, http.Client Function()? httpClientFactory, @@ -81,6 +100,8 @@ class RelaySessionNotifier extends Notifier { RelayRateLimitGate? rateLimitGate, RelayTimerFactory retryTimerFactory = Timer.new, Future Function(Duration) replayDelay = Future.delayed, + Future?>> Function(List)? + verifyEventBatch, }) : _httpQueryClient = RelayHttpQueryClient( client: httpClient, clientFactory: httpClientFactory, @@ -89,7 +110,9 @@ class RelaySessionNotifier extends Notifier { _now = now ?? DateTime.now, _rateLimitGate = rateLimitGate ?? RelayRateLimitGate(), _retryTimerFactory = retryTimerFactory, - _replayDelay = replayDelay; + _replayDelay = replayDelay, + _verifyEventBatch = + verifyEventBatch ?? _verifyRelayEventBatchInBackground; final RelayHttpQueryClient _httpQueryClient; final RelaySocketFactory _socketFactory; @@ -97,6 +120,8 @@ class RelaySessionNotifier extends Notifier { final RelayRateLimitGate _rateLimitGate; final RelayTimerFactory _retryTimerFactory; final Future Function(Duration) _replayDelay; + final Future?>> Function(List) + _verifyEventBatch; static const _baseReconnectDelayMs = 1000; static const _maxReconnectDelayMs = 30000; @@ -130,6 +155,10 @@ class RelaySessionNotifier extends Notifier { final Map Function()> _beforePauseCallbacks = {}; bool _socketConnected = false; bool _closedRetryReplayScheduled = false; + _RelayEventVerifier? _eventVerifier; + + _RelayEventVerifier get _verifier => + _eventVerifier ??= _RelayEventVerifier(verifyBatch: _verifyEventBatch); @override SessionState build() { @@ -190,22 +219,30 @@ class RelaySessionNotifier extends Notifier { _activateRateLimitGateFromHttpError(response.body); throw RelayException(response.statusCode, response.body); } + if (response.bodyBytes.length > _maxQueryResponseBytes) { + throw const FormatException('relay query response exceeds size limit'); + } final decoded = jsonDecode(response.body); if (decoded is! List) { throw const FormatException('relay returned malformed query response'); } - try { - return [ - for (final eventJson in decoded) - if (eventJson is Map) - NostrEvent.fromJson(eventJson) - else - throw const FormatException('relay returned malformed query event'), - ]; - } catch (error) { - if (error is FormatException) rethrow; - throw FormatException('relay returned malformed query event: $error'); + final verified = await _verifier.verifyAll(decoded.cast()); + if (_disposed || _ageRestricted || generation != _contextGeneration) { + throw StateError('Relay query belongs to a retired session'); } + final events = [ + for (final event in verified) + if (event != null) event, + ]; + // A mixed response can still contribute its independently verified events. + // A non-empty response with no valid events is not authoritative absence: + // surface it as a failure so callers cannot wipe caches with forged data. + if (decoded.isNotEmpty && events.isEmpty) { + throw const FormatException( + 'relay query returned no valid signed events', + ); + } + return events; } void _activateRateLimitGateFromHttpError(String body) { @@ -373,7 +410,10 @@ class RelaySessionNotifier extends Notifier { void debugDispose() => _dispose(); @visibleForTesting - void debugSupersedeConnection() => _connectionGeneration++; + void debugSupersedeConnection() { + _connectionGeneration++; + _retireLiveVerification(); + } @visibleForTesting void debugHandleDisconnected([Object? error]) { @@ -417,6 +457,7 @@ class RelaySessionNotifier extends Notifier { /// Force a reconnect (e.g., returning from background). Future reconnect() async { _socketConnected = false; + _retireLiveVerification(); await _socket?.disconnect(); _reconnectDelayMs = _baseReconnectDelayMs; final config = ref.read(relayConfigProvider); @@ -452,6 +493,7 @@ class RelaySessionNotifier extends Notifier { void _pauseNow() { _paused = true; _socketConnected = false; + _retireLiveVerification(); _reconnectTimer?.cancel(); _cancelAllHistory(Exception('App moved to background')); _rejectAllPending(Exception('App moved to background')); @@ -487,6 +529,7 @@ class RelaySessionNotifier extends Notifier { if (_disposed || _ageRestricted) return; final generation = ++_connectionGeneration; + _retireLiveVerification(); state = SessionState( status: _hasConnectedOnce ? SessionStatus.reconnecting @@ -521,6 +564,7 @@ class RelaySessionNotifier extends Notifier { void _handleDisconnected(int generation, Object? error) { if (_disposed || generation != _connectionGeneration) return; _socketConnected = false; + _retireLiveVerification(); _cancelAllHistory(error); _rejectAllPending(error); _resetAllClosedRetries(); @@ -633,7 +677,8 @@ class RelaySessionNotifier extends Notifier { void _handleMessage(List data) { if (data.isEmpty) return; - final type = data[0] as String; + final type = data[0]; + if (type is! String) return; switch (type) { case 'EVENT': @@ -647,66 +692,6 @@ class RelaySessionNotifier extends Notifier { } } - void _handleEvent(List data) { - if (data.length < 3) return; - final subId = data[1] as String; - final eventJson = data[2] as Map; - final event = NostrEvent.fromJson(eventJson); - - // History subscriptions accumulate immediately. - final historySub = _historySubscriptions[subId]; - if (historySub != null) { - historySub.events.add(event); - return; - } - - // Live subscriptions get batched. An EVENT proves the stream is active, - // but not that a retry replay is complete; only EOSE is that boundary. - final liveSub = _liveSubscriptions[subId]; - if (liveSub != null) { - _resetClosedRetry(liveSub); - // Track last seen timestamp for reconnect replay. - if (liveSub.lastSeenCreatedAt == null || - event.createdAt > liveSub.lastSeenCreatedAt!) { - liveSub.lastSeenCreatedAt = event.createdAt; - } - _eventBuffer.add(_BufferedEvent(subId, event)); - _scheduleFlush(); - } - } - - void _handleEose(List data) { - if (data.length < 2) return; - final subId = data[1] as String; - - // History subscription: resolve with collected events. - final historySub = _historySubscriptions.remove(subId); - if (historySub != null) { - historySub.timeout.cancel(); - if (!historySub.completer.isCompleted) { - historySub.completer.complete(historySub.events); - } - _sendClose(subId); - return; - } - - // Live subscription: flush replay callbacks before signaling ready. This - // ordering matters for retry replays, whose original ready completer has - // already been released. - final liveSub = _liveSubscriptions[subId]; - if (liveSub != null) { - _resetClosedRetry(liveSub); - _flushBufferedEventsNow(); - liveSub.onStatusChanged?.call(RelaySubscriptionStatus.ready); - } - if (liveSub != null && - liveSub.readyCompleter != null && - !liveSub.readyCompleter!.isCompleted) { - liveSub.readyCompleter!.complete(); - liveSub.readyCompleter = null; - } - } - void _handleClosed(List data) { if (data.length < 2) return; final subId = data[1] as String; @@ -925,6 +910,7 @@ class RelaySessionNotifier extends Notifier { void _removeLiveSubscription(String subId, _LiveSubscription subscription) { if (_liveSubscriptions[subId] != subscription) return; _liveSubscriptions.remove(subId); + _retireLiveSubscriptionVerification(subscription); _pendingClosedRetries.remove(subId); subscription.closedRetryTimer?.cancel(); subscription.closedRetryTimer = null; @@ -994,6 +980,9 @@ class RelaySessionNotifier extends Notifier { subscription.closedRetryTimer = null; } _recentDeliveryKeys.clear(); + _eventBuffer.clear(); + _eventVerifier?.close(); + _eventVerifier = null; _socket?.dispose(); _socket = null; _httpQueryClient.close(); diff --git a/mobile/lib/shared/relay/relay_session_event_admission.dart b/mobile/lib/shared/relay/relay_session_event_admission.dart new file mode 100644 index 0000000000..e790e335e6 --- /dev/null +++ b/mobile/lib/shared/relay/relay_session_event_admission.dart @@ -0,0 +1,240 @@ +part of 'relay_session.dart'; + +extension _RelaySessionEventAdmission on RelaySessionNotifier { + void _handleEvent(List data) { + if (data.length < 3 || data[1] is! String) return; + final subId = data[1] as String; + final connectionGeneration = _connectionGeneration; + final contextGeneration = _contextGeneration; + + final historySub = _historySubscriptions[subId]; + if (historySub != null) { + historySub.receivedEventFrames++; + historySub.pendingVerifications++; + final verification = _verifier.verify(data[2]); + historySub.verificationTail = historySub.verificationTail.then((_) async { + try { + final event = await verification; + if (_historySubscriptions[subId] != historySub || + connectionGeneration != _connectionGeneration || + contextGeneration != _contextGeneration) { + return; + } + if (event != null) historySub.events.add(event); + } catch (error, stackTrace) { + if (_historySubscriptions[subId] == historySub && + connectionGeneration == _connectionGeneration && + contextGeneration == _contextGeneration) { + historySub.verificationError ??= error; + historySub.verificationStackTrace ??= stackTrace; + } + } finally { + historySub.pendingVerifications--; + } + }); + return; + } + + final liveSub = _liveSubscriptions[subId]; + if (liveSub != null) { + final verificationEpoch = liveSub.verificationEpoch; + liveSub.receivedEventFrames++; + liveSub.pendingVerifications++; + final verification = _verifier.verify(data[2]); + liveSub.verificationTail = liveSub.verificationTail.then((_) async { + try { + final event = await verification; + if (_liveSubscriptions[subId] != liveSub || + connectionGeneration != _connectionGeneration || + contextGeneration != _contextGeneration || + verificationEpoch != liveSub.verificationEpoch || + event == null) { + return; + } + + // Only a verified event can prove activity or advance replay state. + _resetClosedRetry(liveSub); + liveSub.validEventFrames++; + if (liveSub.lastSeenCreatedAt == null || + event.createdAt > liveSub.lastSeenCreatedAt!) { + liveSub.lastSeenCreatedAt = event.createdAt; + } + _eventBuffer.add(_BufferedEvent(subId, event)); + _scheduleFlush(); + } catch (error, stackTrace) { + if (_liveSubscriptions[subId] == liveSub && + connectionGeneration == _connectionGeneration && + contextGeneration == _contextGeneration && + verificationEpoch == liveSub.verificationEpoch) { + _failLiveVerification(subId, liveSub, error, stackTrace); + } + } finally { + if (verificationEpoch == liveSub.verificationEpoch) { + liveSub.pendingVerifications--; + } + } + }); + } + } + + void _handleEose(List data) { + if (data.length < 2 || data[1] is! String) return; + final subId = data[1] as String; + + final historySub = _historySubscriptions[subId]; + if (historySub != null) { + if (historySub.pendingVerifications == 0) { + _completeHistoryEose(subId, historySub); + } else { + unawaited(_completeHistoryEoseAfterVerification(subId, historySub)); + } + return; + } + + final liveSub = _liveSubscriptions[subId]; + if (liveSub != null) { + final connectionGeneration = _connectionGeneration; + final contextGeneration = _contextGeneration; + final verificationEpoch = liveSub.verificationEpoch; + if (liveSub.pendingVerifications == 0) { + _completeLiveEose( + subId, + liveSub, + connectionGeneration, + contextGeneration, + verificationEpoch, + ); + } else { + unawaited( + _completeLiveEoseAfterVerification( + subId, + liveSub, + connectionGeneration, + contextGeneration, + verificationEpoch, + ), + ); + } + } + } + + Future _completeHistoryEoseAfterVerification( + String subId, + _HistorySubscription subscription, + ) async { + await subscription.verificationTail; + _completeHistoryEose(subId, subscription); + } + + void _completeHistoryEose(String subId, _HistorySubscription subscription) { + if (_historySubscriptions[subId] != subscription) return; + _historySubscriptions.remove(subId); + subscription.timeout.cancel(); + final verificationError = subscription.verificationError; + if (!subscription.completer.isCompleted) { + if (verificationError != null) { + subscription.completer.completeError( + verificationError, + subscription.verificationStackTrace, + ); + } else if (subscription.receivedEventFrames > 0 && + subscription.events.isEmpty) { + subscription.completer.completeError( + const FormatException( + 'relay history returned no valid signed events', + ), + ); + } else { + subscription.completer.complete(subscription.events); + } + } + _sendClose(subId); + } + + Future _completeLiveEoseAfterVerification( + String subId, + _LiveSubscription subscription, + int connectionGeneration, + int contextGeneration, + int verificationEpoch, + ) async { + await subscription.verificationTail; + _completeLiveEose( + subId, + subscription, + connectionGeneration, + contextGeneration, + verificationEpoch, + ); + } + + void _completeLiveEose( + String subId, + _LiveSubscription subscription, + int connectionGeneration, + int contextGeneration, + int verificationEpoch, + ) { + if (_liveSubscriptions[subId] != subscription || + connectionGeneration != _connectionGeneration || + contextGeneration != _contextGeneration || + verificationEpoch != subscription.verificationEpoch) { + return; + } + if (subscription.receivedEventFrames > 0 && + subscription.validEventFrames == 0) { + _failLiveVerification( + subId, + subscription, + const FormatException( + 'relay subscription returned no valid signed events', + ), + StackTrace.current, + ); + return; + } + + subscription.receivedEventFrames = 0; + subscription.validEventFrames = 0; + // Flush replay callbacks before signaling ready. This ordering matters for + // retry replays, whose original ready completer has already been released. + _resetClosedRetry(subscription); + _flushBufferedEventsNow(); + subscription.onStatusChanged?.call(RelaySubscriptionStatus.ready); + final readyCompleter = subscription.readyCompleter; + if (readyCompleter != null && !readyCompleter.isCompleted) { + readyCompleter.complete(); + subscription.readyCompleter = null; + } + } + + void _failLiveVerification( + String subId, + _LiveSubscription subscription, + Object error, + StackTrace stackTrace, + ) { + if (_liveSubscriptions[subId] != subscription) return; + final readyCompleter = subscription.readyCompleter; + if (readyCompleter != null && !readyCompleter.isCompleted) { + readyCompleter.completeError(error, stackTrace); + } else { + subscription.onClosed?.call('event verification failed: $error'); + } + _removeLiveSubscription(subId, subscription); + } + + void _retireLiveVerification() { + for (final subscription in _liveSubscriptions.values) { + _retireLiveSubscriptionVerification(subscription); + } + } + + void _retireLiveSubscriptionVerification(_LiveSubscription subscription) { + subscription.verificationEpoch++; + subscription.verificationTail = Future.value(); + subscription.pendingVerifications = 0; + subscription.receivedEventFrames = 0; + subscription.validEventFrames = 0; + } +} diff --git a/mobile/test/shared/relay/relay_event_ingress_integration_test.dart b/mobile/test/shared/relay/relay_event_ingress_integration_test.dart new file mode 100644 index 0000000000..c7eb2c47db --- /dev/null +++ b/mobile/test/shared/relay/relay_event_ingress_integration_test.dart @@ -0,0 +1,191 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:nostr/nostr.dart' as nostr; + +import 'package:buzz/shared/auth/auth_provider.dart'; +import 'package:buzz/shared/relay/nostr_models.dart'; +import 'package:buzz/shared/relay/relay_provider.dart'; +import 'package:buzz/shared/relay/relay_session.dart'; + +void main() { + test('real HTTP and WebSocket ingress admits only verified events', () async { + final queryEvent = _signedEvent('query'); + final historyEvent = _signedEvent('history'); + final liveEvent = _signedEvent('live'); + final relay = await _AdversarialRelay.start( + queryEvents: [_forgedEvent('forged query'), queryEvent.toJson()], + historyEvents: [_forgedEvent('forged history'), historyEvent.toJson()], + ); + addTearDown(relay.close); + + final keys = nostr.Keys.generate(); + final session = RelaySessionNotifier(); + final container = ProviderContainer( + overrides: [ + authProvider.overrideWith(() => _AuthenticatedAuthNotifier()), + relayConfigProvider.overrideWith( + () => _TestRelayConfigNotifier(relay.origin, keys.nsec), + ), + relaySessionProvider.overrideWith(() => session), + ], + ); + addTearDown(container.dispose); + await container.read(authProvider.future); + final listener = container.listen(relaySessionProvider, (_, _) {}); + addTearDown(listener.close); + await _waitUntil( + () => + container.read(relaySessionProvider).status == + SessionStatus.connected, + ); + + expect(await session.queryRelay(const []), [queryEvent]); + expect(await session.fetchHistory(_filter), [historyEvent]); + + final delivered = []; + final unsubscribe = await session.subscribe(_filter, delivered.add); + addTearDown(unsubscribe); + await relay.sendLive([_forgedEvent('forged live'), liveEvent.toJson()]); + await _waitUntil(() => delivered.isNotEmpty); + + expect(delivered, [liveEvent]); + }); +} + +class _AdversarialRelay { + _AdversarialRelay._(this._server, this.queryEvents, this.historyEvents); + + final HttpServer _server; + final List queryEvents; + final List historyEvents; + final List _sockets = []; + final Completer _liveSubscription = Completer(); + + String get origin => 'http://${_server.address.address}:${_server.port}'; + + static Future<_AdversarialRelay> start({ + required List queryEvents, + required List historyEvents, + }) async { + final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + final relay = _AdversarialRelay._(server, queryEvents, historyEvents); + server.listen(relay._handleRequest); + return relay; + } + + Future _handleRequest(HttpRequest request) async { + if (WebSocketTransformer.isUpgradeRequest(request)) { + final socket = await WebSocketTransformer.upgrade(request); + _sockets.add(socket); + socket.add(jsonEncode(['AUTH', 'local-test-challenge'])); + socket.listen((raw) => _handleSocketMessage(socket, raw)); + return; + } + if (request.method == 'POST' && request.uri.path == '/query') { + await request.drain(); + request.response + ..statusCode = HttpStatus.ok + ..headers.contentType = ContentType.json + ..write(jsonEncode(queryEvents)); + await request.response.close(); + return; + } + request.response.statusCode = HttpStatus.notFound; + await request.response.close(); + } + + void _handleSocketMessage(WebSocket socket, Object? raw) { + if (raw is! String) return; + final decoded = jsonDecode(raw); + if (decoded is! List || decoded.isEmpty) return; + switch (decoded.first) { + case 'AUTH': + final event = decoded[1] as Map; + socket.add(jsonEncode(['OK', event['id'], true, 'authenticated'])); + case 'REQ': + final subId = decoded[1] as String; + if (subId.startsWith('h-')) { + for (final event in historyEvents) { + socket.add(jsonEncode(['EVENT', subId, event])); + } + socket.add(jsonEncode(['EOSE', subId])); + } else { + if (!_liveSubscription.isCompleted) { + _liveSubscription.complete(subId); + } + socket.add(jsonEncode(['EOSE', subId])); + } + } + } + + Future sendLive(List events) async { + final subId = await _liveSubscription.future; + final socket = _sockets.single; + for (final event in events) { + socket.add(jsonEncode(['EVENT', subId, event])); + } + } + + Future close() async { + for (final socket in _sockets) { + await socket.close(); + } + await _server.close(force: true); + } +} + +class _AuthenticatedAuthNotifier extends AuthNotifier { + @override + Future build() async => + const AuthState(status: AuthStatus.authenticated); +} + +class _TestRelayConfigNotifier extends RelayConfigNotifier { + _TestRelayConfigNotifier(this._origin, this._nsec); + + final String _origin; + final String _nsec; + + @override + RelayConfig build() => RelayConfig(baseUrl: _origin, nsec: _nsec); +} + +const _channelId = '11111111-1111-4111-8111-111111111111'; +const _filter = NostrFilter( + kinds: EventKind.channelEventKinds, + tags: { + '#h': [_channelId], + }, + limit: 50, +); + +NostrEvent _signedEvent(String content) { + final event = nostr.Event.from( + kind: EventKind.streamMessageV2, + content: content, + tags: const [ + ['h', _channelId], + ], + secretKey: '1'.padLeft(64, '0'), + ); + return NostrEvent.fromJson(event.toMap()); +} + +Map _forgedEvent(String content) => { + ..._signedEvent(content).toJson(), + 'sig': List.filled(64, '00').join(), +}; + +Future _waitUntil(bool Function() predicate) async { + final deadline = DateTime.now().add(const Duration(seconds: 3)); + while (!predicate()) { + if (DateTime.now().isAfter(deadline)) { + fail('Timed out waiting for local relay workflow'); + } + await Future.delayed(const Duration(milliseconds: 5)); + } +} diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 50bdb5201d..3951b83e61 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -234,6 +234,137 @@ void main() { ); }); + test('queryRelay accepts a valid signed event', () async { + final valid = _signedEvent(); + final harness = _eventQueryHarness([valid.toJson()]); + addTearDown(harness.container.dispose); + + final events = await harness.session.queryRelay(const []); + + expect(events, [valid]); + }); + + test( + 'queryRelay rejects a canonical id with a forged zero signature', + () async { + final forged = { + ..._signedEvent().toJson(), + 'sig': _repeatedHex('00', 64), + }; + final harness = _eventQueryHarness([forged]); + addTearDown(harness.container.dispose); + + await expectLater( + harness.session.queryRelay(const []), + throwsA(isA()), + ); + }, + ); + + final alteredSignedEvents = >{ + 'altered content': { + ..._signedEvent().toJson(), + 'content': 'attacker content', + }, + 'altered id': {..._signedEvent().toJson(), 'id': _repeatedHex('00', 32)}, + 'short id': {..._signedEvent().toJson(), 'id': '00'}, + 'overlong id': {..._signedEvent().toJson(), 'id': _repeatedHex('00', 33)}, + 'nonhex id': {..._signedEvent().toJson(), 'id': _repeatedHex('gg', 32)}, + 'short pubkey': {..._signedEvent().toJson(), 'pubkey': '00'}, + 'overlong pubkey': { + ..._signedEvent().toJson(), + 'pubkey': _repeatedHex('00', 33), + }, + 'nonhex pubkey': { + ..._signedEvent().toJson(), + 'pubkey': _repeatedHex('gg', 32), + }, + 'short signature': {..._signedEvent().toJson(), 'sig': '00'}, + 'overlong signature': { + ..._signedEvent().toJson(), + 'sig': _repeatedHex('00', 65), + }, + 'nonhex signature': { + ..._signedEvent().toJson(), + 'sig': _repeatedHex('gg', 64), + }, + }; + for (final entry in alteredSignedEvents.entries) { + test('queryRelay rejects ${entry.key}', () async { + final harness = _eventQueryHarness([entry.value]); + addTearDown(harness.container.dispose); + + await expectLater( + harness.session.queryRelay(const []), + throwsA(isA()), + ); + }); + } + + test('queryRelay keeps valid events from a mixed invalid batch', () async { + final valid = _signedEvent(content: 'valid after forged'); + final forged = { + ..._signedEvent(content: 'forged').toJson(), + 'sig': _repeatedHex('00', 64), + }; + final harness = _eventQueryHarness([forged, valid.toJson()]); + addTearDown(harness.container.dispose); + + final events = await harness.session.queryRelay(const []); + + expect(events, [valid]); + }); + + test('queryRelay rejects an oversized event before verification', () async { + final oversized = { + ..._signedEvent().toJson(), + 'content': List.filled(700000, 'x').join(), + }; + final harness = _eventQueryHarness([oversized]); + addTearDown(harness.container.dispose); + + await expectLater( + harness.session.queryRelay(const []), + throwsA(isA()), + ); + }); + + test('queryRelay propagates verification queue exhaustion', () async { + final large = { + ..._signedEvent().toJson(), + 'content': List.filled(600000, 'x').join(), + }; + final harness = _eventQueryHarness(List.filled(10, large)); + addTearDown(harness.container.dispose); + + await expectLater( + harness.session.queryRelay(const []), + throwsA(isA()), + ); + }); + + test('queryRelay rejects a response above its byte limit', () async { + final harness = _queryHarness( + gate: RelayRateLimitGate(), + client: http_testing.MockClient( + (_) async => + http.Response(List.filled(8 * 1024 * 1024 + 1, 'x').join(), 200), + ), + ); + addTearDown(harness.container.dispose); + + await expectLater( + harness.session.queryRelay(const []), + throwsA( + isA().having( + (error) => error.message, + 'message', + contains('size limit'), + ), + ), + ); + }); + test('queryRelay rotates the client after a timeout', () async { final clients = <_ControlledHttpClient>[]; final session = RelaySessionNotifier( @@ -508,6 +639,33 @@ void main() { }, ); + test('websocket history keeps valid events from a mixed batch', () async { + final session = RelaySessionNotifier(); + final history = session.fetchHistory(_channelFilter); + final valid = _signedEvent(content: 'valid history'); + final forged = { + ..._signedEvent(content: 'forged history').toJson(), + 'sig': _repeatedHex('00', 64), + }; + + session.debugHandleMessage(['EVENT', 'h-1', forged]); + session.debugHandleMessage(['EVENT', 'h-1', valid.toJson()]); + session.debugHandleMessage(['EOSE', 'h-1']); + + expect(await history, [valid]); + }); + + test('websocket history rejects an all-invalid response', () async { + final session = RelaySessionNotifier(); + final history = session.fetchHistory(_channelFilter); + final forged = {..._signedEvent().toJson(), 'sig': _repeatedHex('00', 64)}; + + session.debugHandleMessage(['EVENT', 'h-1', forged]); + session.debugHandleMessage(['EOSE', 'h-1']); + + await expectLater(history, throwsA(isA())); + }); + test('background disconnect rejects in-flight history', () async { final session = RelaySessionNotifier(); final container = ProviderContainer( @@ -782,16 +940,16 @@ void main() { session.debugHandleMessage(['EOSE', 'l-2']); final unsubscribeSecond = await secondSubscribe; - final event = _event(); + final event = _signedEvent(); session.debugHandleMessage(['EVENT', 'l-1', event.toJson()]); session.debugHandleMessage(['EVENT', 'l-2', event.toJson()]); - session.debugFlushEventBuffer(); + await _waitUntil(() => firstEvents.length == 1 && secondEvents.length == 1); expect(firstEvents.map((event) => event.id), [event.id]); expect(secondEvents.map((event) => event.id), [event.id]); session.debugHandleMessage(['EVENT', 'l-1', event.toJson()]); - session.debugFlushEventBuffer(); + await Future.delayed(const Duration(milliseconds: 30)); expect(firstEvents.map((event) => event.id), [event.id]); expect(secondEvents.map((event) => event.id), [event.id]); @@ -816,7 +974,7 @@ void main() { filter, (_) => deliveryPhases.add(queryHasBegun), ); - final replayEvent = _event(); + final replayEvent = _signedEvent(); session.debugHandleMessage(['EVENT', 'l-1', replayEvent.toJson()]); session.debugHandleMessage(['EOSE', 'l-1']); @@ -830,6 +988,256 @@ void main() { unsubscribe(); }); + test( + 'live delivery continues with a valid event after an invalid event', + () async { + final session = RelaySessionNotifier(); + final delivered = []; + final subscribe = session.subscribe(_channelFilter, delivered.add); + session.debugHandleMessage(['EOSE', 'l-1']); + final unsubscribe = await subscribe; + final valid = _signedEvent(content: 'valid live'); + final forged = { + ..._signedEvent(content: 'forged live').toJson(), + 'sig': _repeatedHex('00', 64), + }; + + session.debugHandleMessage(['EVENT', 'l-1', forged]); + session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]); + + await _waitUntil(() => delivered.isNotEmpty); + expect(delivered, [valid]); + unsubscribe(); + }, + ); + + test('EOSE waits for preceding event verification and delivery', () async { + final verification = Completer?>>(); + var verificationStarted = false; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationStarted = true; + return verification.future; + }, + ); + final delivered = []; + var ready = false; + final subscribe = session.subscribe(_channelFilter, delivered.add) + ..then((_) => ready = true); + final valid = _signedEvent(); + + session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]); + session.debugHandleMessage(['EOSE', 'l-1']); + await _waitUntil(() => verificationStarted); + expect(delivered, isEmpty); + expect(ready, isFalse); + + verification.complete([valid.toJson()]); + final unsubscribe = await subscribe; + expect(delivered, [valid]); + expect(ready, isTrue); + unsubscribe(); + }); + + test('EOSE does not wait for a later event verification', () async { + final verifications = ?>>>[]; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + final verification = Completer?>>(); + verifications.add(verification); + return verification.future; + }, + ); + final delivered = []; + final subscribe = session.subscribe(_channelFilter, delivered.add); + final beforeEose = _signedEvent(content: 'before EOSE'); + final afterEose = _signedEvent(content: 'after EOSE'); + + session.debugHandleMessage(['EVENT', 'l-1', beforeEose.toJson()]); + session.debugHandleMessage(['EOSE', 'l-1']); + await _waitUntil(() => verifications.length == 1); + session.debugHandleMessage(['EVENT', 'l-1', afterEose.toJson()]); + verifications.first.complete([beforeEose.toJson()]); + + final unsubscribe = await subscribe.timeout( + const Duration(milliseconds: 200), + ); + expect(delivered, [beforeEose]); + await _waitUntil(() => verifications.length == 2); + unsubscribe(); + verifications.last.complete([afterEose.toJson()]); + await Future.delayed(const Duration(milliseconds: 30)); + expect(delivered, [beforeEose]); + }); + + test('disconnect immediately retires pending live verification', () async { + final verification = Completer?>>(); + var verificationStarted = false; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationStarted = true; + return verification.future; + }, + ); + final container = ProviderContainer( + overrides: [ + authProvider.overrideWith(() => _PendingAuthNotifier()), + relaySessionProvider.overrideWith(() => session), + relayConfigProvider.overrideWith( + () => _FakeRelayConfigNotifier( + baseUrl: 'https://relay.example', + nsec: nostr.Keys.generate().nsec, + ), + ), + ], + ); + addTearDown(container.dispose); + container.read(relaySessionProvider); + final delivered = []; + final subscribe = session.subscribe(_channelFilter, delivered.add); + session.debugHandleMessage(['EOSE', 'l-1']); + final unsubscribe = await subscribe; + final valid = _signedEvent(); + + session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]); + await _waitUntil(() => verificationStarted); + session.debugHandleDisconnected(); + verification.complete([valid.toJson()]); + await Future.delayed(const Duration(milliseconds: 30)); + session.debugFlushEventBuffer(); + + expect(delivered, isEmpty); + unsubscribe(); + }); + + test('unsubscribe immediately retires pending live verification', () async { + final verification = Completer?>>(); + var verificationStarted = false; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationStarted = true; + return verification.future; + }, + ); + final delivered = []; + final subscribe = session.subscribe(_channelFilter, delivered.add); + session.debugHandleMessage(['EOSE', 'l-1']); + final unsubscribe = await subscribe; + final valid = _signedEvent(); + + session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]); + await _waitUntil(() => verificationStarted); + unsubscribe(); + verification.complete([valid.toJson()]); + await Future.delayed(const Duration(milliseconds: 30)); + session.debugFlushEventBuffer(); + + expect(delivered, isEmpty); + }); + + test('terminal CLOSED immediately retires pending verification', () async { + final verification = Completer?>>(); + var verificationStarted = false; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationStarted = true; + return verification.future; + }, + ); + final delivered = []; + final subscribe = session.subscribe(_channelFilter, delivered.add); + session.debugHandleMessage(['EOSE', 'l-1']); + await subscribe; + final valid = _signedEvent(); + + session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]); + await _waitUntil(() => verificationStarted); + session.debugHandleMessage(['CLOSED', 'l-1', 'restricted: removed']); + verification.complete([valid.toJson()]); + await Future.delayed(const Duration(milliseconds: 30)); + session.debugFlushEventBuffer(); + + expect(delivered, isEmpty); + }); + + test('local OK acknowledgment bypasses pending verification', () async { + final verification = Completer?>>(); + var verificationStarted = false; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationStarted = true; + return verification.future; + }, + ); + final subscribe = session.subscribe(_channelFilter, (_) {}); + session.debugHandleMessage(['EOSE', 'l-1']); + final unsubscribe = await subscribe; + session.debugHandleMessage(['EVENT', 'l-1', _signedEvent().toJson()]); + await _waitUntil(() => verificationStarted); + session.debugAttachSocketForTest(_RecordingRelaySocket()); + + final publish = session.publish(_event(id: 'local-event')); + session.debugHandleMessage(['OK', 'local-event', true, 'accepted']); + + expect( + (await publish.timeout(const Duration(milliseconds: 100))).id, + 'local-event', + ); + unsubscribe(); + verification.complete([_signedEvent().toJson()]); + }); + + test('session teardown fences a pending verification result', () async { + final verification = Completer?>>(); + var verificationStarted = false; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationStarted = true; + return verification.future; + }, + ); + final delivered = []; + final subscribe = session.subscribe(_channelFilter, delivered.add); + session.debugHandleMessage(['EOSE', 'l-1']); + await subscribe; + final valid = _signedEvent(); + + session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]); + await _waitUntil(() => verificationStarted); + session.debugDispose(); + verification.complete([valid.toJson()]); + await Future.delayed(const Duration(milliseconds: 30)); + session.debugFlushEventBuffer(); + + expect(delivered, isEmpty); + }); + + test('connection replacement fences a pending verification result', () async { + final verification = Completer?>>(); + var verificationStarted = false; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationStarted = true; + return verification.future; + }, + ); + final delivered = []; + final subscribe = session.subscribe(_channelFilter, delivered.add); + session.debugHandleMessage(['EOSE', 'l-1']); + final unsubscribe = await subscribe; + final valid = _signedEvent(); + + session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]); + await _waitUntil(() => verificationStarted); + session.debugSupersedeConnection(); + verification.complete([valid.toJson()]); + await Future.delayed(const Duration(milliseconds: 30)); + session.debugFlushEventBuffer(); + + expect(delivered, isEmpty); + unsubscribe(); + }); + test('terminal CLOSED fails a live subscribe before ready', () async { final session = RelaySessionNotifier(); const filter = NostrFilter(kinds: [EventKind.agentObserverFrame], limit: 0); @@ -919,7 +1327,7 @@ void main() { session.debugHandleMessage([ 'EVENT', 'l-1', - _event(createdAt: 30).toJson(), + _signedEvent(createdAt: 30).toJson(), ]); expect(statuses, [ RelaySubscriptionStatus.ready, @@ -927,6 +1335,7 @@ void main() { ]); session.debugHandleMessage(['EOSE', 'l-1']); + await _waitUntil(() => statuses.length == 3); expect(statuses, [ RelaySubscriptionStatus.ready, RelaySubscriptionStatus.retrying, @@ -1004,7 +1413,8 @@ void main() { }, ); session.debugAttachSocketForTest(socket); - final subscribe = session.subscribe(_channelFilter, (_) {}); + final delivered = []; + final subscribe = session.subscribe(_channelFilter, delivered.add); session.debugHandleMessage(['EOSE', 'l-1']); final unsubscribe = await subscribe; @@ -1017,8 +1427,9 @@ void main() { session.debugHandleMessage([ 'EVENT', 'l-1', - _event(createdAt: 30).toJson(), + _signedEvent(createdAt: 30).toJson(), ]); + await _waitUntil(() => delivered.isNotEmpty); session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']); expect(timers.last.duration, const Duration(seconds: 1)); unsubscribe(); @@ -1772,6 +2183,13 @@ _QueryHarness _queryHarness({ return _QueryHarness(container: container, session: session); } +_QueryHarness _eventQueryHarness(List events) => _queryHarness( + gate: RelayRateLimitGate(), + client: http_testing.MockClient( + (_) async => http.Response(jsonEncode(events), 200), + ), +); + class _FakeAuthNotifier extends AuthNotifier { int signOutCount = 0; @@ -1846,6 +2264,31 @@ NostrEvent _event({int createdAt = 20, String id = 'event-1'}) { ); } +NostrEvent _signedEvent({int createdAt = 20, String content = 'hello'}) { + final event = nostr.Event.from( + kind: EventKind.streamMessageV2, + content: content, + tags: const [ + ['h', _channelId], + ], + createdAt: createdAt, + secretKey: '1'.padLeft(64, '0'), + ); + return NostrEvent.fromJson(event.toMap()); +} + +String _repeatedHex(String pair, int bytes) => List.filled(bytes, pair).join(); + +Future _waitUntil(bool Function() predicate) async { + final deadline = DateTime.now().add(const Duration(seconds: 2)); + while (!predicate()) { + if (DateTime.now().isAfter(deadline)) { + fail('Timed out waiting for asynchronous relay work'); + } + await Future.delayed(const Duration(milliseconds: 1)); + } +} + const _visibleChannelId = '99999999-9999-4999-8999-999999999999'; const _channelFilter = NostrFilter( kinds: EventKind.channelEventKinds,