Add WebSocket support and admin key rotation

redsolver · redsolver · commit 9a2b8f78989d · 2023-06-01T16:51:31.000+02:00
diff --git a/bin/s5_server.dart b/bin/s5_server.dart
@@ -85,8 +85,8 @@ port = 5050
 
 [p2p.peers]
 initial = [
-  'tcp://z2DWuWNZcdSyZLpXFK2uCU3haaWMXrDAgxzv17sDEMHstZb@199.247.20.119:4444', # s5.garden
-  'tcp://z2DWuPbL5pweybXnEB618pMnV58ECj2VPDNfVGm3tFqBvjF@116.203.139.40:4444', # s5.ninja
+  'wss://z2DWuWNZcdSyZLpXFK2uCU3haaWMXrDAgxzv17sDEMHstZb@s5.garden/s5/p2p',
+  'wss://z2DWuPbL5pweybXnEB618pMnV58ECj2VPDNfVGm3tFqBvjF@s5.ninja/s5/p2p',
 ]
 ''';
 
@@ -111,7 +111,7 @@ bind = '0.0.0.0'
 
 [p2p.peers]
 initial = [
-  'tcp://z2DWuWNZcdSyZLpXFK2uCU3haaWMXrDAgxzv17sDEMHstZb@199.247.20.119:4444', # s5.garden
-  'tcp://z2DWuPbL5pweybXnEB618pMnV58ECj2VPDNfVGm3tFqBvjF@116.203.139.40:4444', # s5.ninja
+  'wss://z2DWuWNZcdSyZLpXFK2uCU3haaWMXrDAgxzv17sDEMHstZb@s5.garden/s5/p2p',
+  'wss://z2DWuPbL5pweybXnEB618pMnV58ECj2VPDNfVGm3tFqBvjF@s5.ninja/s5/p2p',
 ]
 ''';
diff --git a/lib/http_api/admin.dart b/lib/http_api/admin.dart
@@ -17,7 +17,11 @@ class AdminAPI {
     final adminApiKey = base58BitcoinEncode(
       await node.crypto.hashBlake3(
         Uint8List.fromList(
-          node.p2p.nodeKeyPair.extractBytes() + encodeEndian(32, 32),
+          node.p2p.nodeKeyPair.extractBytes() +
+              encodeEndian(
+                node.config['http']?['api']?['admin']?['keyRotation'] ?? 0,
+                32,
+              ),
         ),
       ),
     );
diff --git a/lib/http_api/http_api.dart b/lib/http_api/http_api.dart
@@ -11,7 +11,7 @@ import 'package:http/http.dart';
 import 'package:lib5/constants.dart';
 import 'package:lib5/lib5.dart';
 import 'package:lib5/util.dart';
-import 'package:messagepack/messagepack.dart';
+import 'package:s5_msgpack/s5_msgpack.dart';
 import 'package:mime/mime.dart';
 import 'package:path/path.dart';
 import 'package:s5_server/accounts/account.dart';
@@ -21,6 +21,7 @@ import 'package:s5_server/download/uri_provider.dart';
 import 'package:s5_server/http_api/admin.dart';
 import 'package:s5_server/node.dart';
 import 'package:s5_server/service/accounts.dart';
+import 'package:s5_server/service/p2p.dart';
 import 'package:s5_server/util/uid.dart';
 import 'package:s5_server/util/multipart.dart';
 import 'serve_chunked_file.dart';
@@ -886,7 +887,24 @@ class HttpAPIServer {
         final String? domain = node.config['http']?['api']?['domain'];
 
         if (domain != null) {
-          if (uri.host == 'docs.$domain') {
+          if (uri.host == domain) {
+            if (request.uri.path == '/s5/p2p') {
+              final socket = await WebSocketTransformer.upgrade(request);
+
+              final p = WebSocketPeer(
+                socket,
+                connectionUris: [],
+              );
+
+              runZonedGuarded(
+                () => node.p2p.onNewPeer(p, verifyId: false),
+                (e, st) {
+                  node.logger.catched(e, st);
+                },
+              );
+              return;
+            }
+          } else if (uri.host == 'docs.$domain') {
             cid = CID.decode('zrjD7xwmgP8U6hquPUtSRcZP1J1LvksSwTq4CPZ2ck96FHu');
           } else if (!('.${uri.host}').endsWith('.$domain')) {
             final cidStr = await node.resolveName(uri.host);
diff --git a/lib/node.dart b/lib/node.dart
@@ -7,7 +7,7 @@ import 'dart:typed_data';
 import 'package:cancellation_token/cancellation_token.dart';
 import 'package:hive/hive.dart';
 import 'package:http/http.dart';
-import 'package:messagepack/messagepack.dart';
+import 'package:s5_msgpack/s5_msgpack.dart';
 import 'package:mime/mime.dart';
 import 'package:path/path.dart';
 import 'package:pool/pool.dart';
diff --git a/lib/service/accounts.dart b/lib/service/accounts.dart
@@ -9,7 +9,7 @@ import 'package:lib5/util.dart';
 import 'package:s5_server/accounts/account.dart';
 import 'package:s5_server/logger/base.dart';
 import 'package:s5_server/service/sql.dart';
-import 'package:messagepack/messagepack.dart';
+import 'package:s5_msgpack/s5_msgpack.dart';
 
 class AuthResponse {
   final Account? account;
@@ -381,7 +381,8 @@ class AccountsService {
       final auth = await checkAuth(req, 'account/api/account/pins');
       if (auth.denied) return res.unauthorized(auth);
 
-      final cursor = await getObjectPinsCursorForAccount(account: auth.account!);
+      final cursor =
+          await getObjectPinsCursorForAccount(account: auth.account!);
       final pins = await getObjectPinsForAccount(
         account: auth.account!,
         afterCursor: int.parse(
diff --git a/lib/service/p2p.dart b/lib/service/p2p.dart
@@ -7,23 +7,122 @@ import 'package:hive/hive.dart';
 import 'package:lib5/constants.dart';
 import 'package:lib5/lib5.dart';
 import 'package:lib5/util.dart';
-import 'package:messagepack/messagepack.dart';
+import 'package:s5_msgpack/s5_msgpack.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
 import 'package:s5_server/db/hive_key_value_db.dart';
 import 'package:tint/tint.dart';
 
 import 'package:s5_server/logger/base.dart';
 import 'package:s5_server/model/signed_message.dart';
 import 'package:s5_server/node.dart';
 
-class Peer {
+abstract class Peer {
   late final NodeID id;
+  final List<Uri> connectionUris;
+  bool isConnected = false;
+  late final Uint8List challenge;
+
+  Peer({required this.connectionUris});
+
+  void sendMessage(Uint8List message);
+
+  void listenForMessages(
+    Function callback, {
+    dynamic onDone,
+    Function? onError,
+    required Logger logger,
+  });
 
+  String renderLocationUri();
+}
+
+class WebSocketChannelPeer extends Peer {
+  final WebSocketChannel _socket;
+  final String locationUri;
+
+  WebSocketChannelPeer(
+    this._socket, {
+    required this.locationUri,
+    required super.connectionUris,
+  });
+
+  @override
+  void sendMessage(Uint8List message) {
+    _socket.sink.add(message);
+  }
+
+  @override
+  void listenForMessages(
+    Function callback, {
+    dynamic onDone,
+    Function? onError,
+    required Logger logger,
+  }) {
+    final sub = _socket.stream.listen(
+      (event) async {
+        await callback(event);
+      },
+      onDone: onDone,
+      onError: onError,
+      cancelOnError: false,
+    );
+  }
+
+  @override
+  String renderLocationUri() {
+    return locationUri;
+  }
+}
+
+class WebSocketPeer extends Peer {
+  final WebSocket _socket;
+
+  WebSocketPeer(
+    this._socket, {
+    required super.connectionUris,
+  });
+
+  @override
+  void sendMessage(Uint8List message) {
+    _socket.add(message);
+  }
+
+  @override
+  void listenForMessages(
+    Function callback, {
+    dynamic onDone,
+    Function? onError,
+    required Logger logger,
+  }) {
+    final sub = _socket.listen(
+      (event) async {
+        await callback(event);
+      },
+      onDone: onDone,
+      onError: onError,
+      cancelOnError: false,
+    );
+  }
+
+  @override
+  String renderLocationUri() {
+    return 'WebSocket client';
+  }
+}
+
+class TcpPeer extends Peer {
   final Socket _socket;
+  TcpPeer(
+    this._socket, {
+    required super.connectionUris,
+  });
 
+  @override
   void sendMessage(Uint8List message) {
     _socket.add(encodeEndian(message.length, 4) + message);
   }
 
+  @override
   void listenForMessages(
     Function callback, {
     dynamic onDone,
@@ -57,17 +156,7 @@ class Peer {
     );
   }
 
-  final List<Uri> connectionUris;
-
-  bool isConnected = false;
-
-  late final Uint8List challenge;
-
-  Peer(
-    this._socket, {
-    required this.connectionUris,
-  });
-
+  @override
   String renderLocationUri() {
     return connectionUris.isEmpty
         ? _socket.remoteAddress.address
@@ -109,7 +198,7 @@ class P2PService {
       final socket = await ServerSocket.bind('0.0.0.0', networkSelf['port']);
       socket.listen(
         (peerSocket) {
-          final p = Peer(
+          final p = TcpPeer(
             peerSocket,
             connectionUris: [],
           );
@@ -130,9 +219,17 @@ class P2PService {
       selfConnectionUris.add(
         Uri.parse('tcp://${networkSelf['ip']}:${networkSelf['port']}'),
       );
+    }
 
-      logger.info('connection uris: $selfConnectionUris');
+    final String? domain = node.config['http']?['api']?['domain'];
+    if (domain != null && node.config['p2p']?['self']?['disabled'] != true) {
+      selfConnectionUris.add(
+        Uri.parse('wss://$domain/s5/p2p'),
+      );
     }
+
+    logger.info('connection uris: $selfConnectionUris');
+
     final initialPeers = node.config['p2p']?['peers']?['initial'] ?? [];
 
     for (final p in initialPeers) {
@@ -538,38 +635,63 @@ class P2PService {
   }
 
   void connectToNode(List<Uri> connectionUris) async {
-    final connectionUri = connectionUris.first;
-    final protocol = connectionUri.scheme;
-    if (protocol != 'tcp') {
-      throw 'Protocol $protocol not supported';
+    final connectionUri = connectionUris.firstWhere(
+      (uri) => ['ws', 'wss'].contains(uri.scheme),
+      orElse: () => connectionUris.firstWhere(
+        (uri) => uri.scheme == 'tcp',
+        orElse: () => Uri(scheme: 'unsupported'),
+      ),
+    );
+    if (connectionUri.scheme == 'unsupported') {
+      throw 'None of the available connection URIs are supported ($connectionUris)';
     }
+
+    final protocol = connectionUri.scheme;
+
     if (connectionUri.userInfo.isEmpty) {
       throw 'Connection URI does not contain node id';
     }
     final id = NodeID.decode(connectionUri.userInfo);
 
     reconnectDelay[id] = reconnectDelay[id] ?? 1;
 
-    final ip = connectionUri.host;
-    final port = connectionUri.port;
-
     if (id == localNodeId) {
       return;
     }
     bool retried = false;
     runZonedGuarded(
       () async {
         logger.verbose('[connect] $connectionUri');
+        if (protocol == 'tcp') {
+          final ip = connectionUri.host;
+          final port = connectionUri.port;
+          final socket = await Socket.connect(ip, port);
+
+          await onNewPeer(
+            TcpPeer(
+              socket,
+              connectionUris: [connectionUri],
+            )..id = id,
+            verifyId: true,
+          );
+        } else {
+          final locationUri = connectionUri.replace(
+            userInfo: '',
+          );
 
-        final socket = await Socket.connect(ip, port);
+          final channel = WebSocketChannel.connect(
+            locationUri,
+          );
 
-        await onNewPeer(
-          Peer(
-            socket,
-            connectionUris: [connectionUri],
-          )..id = id,
-          verifyId: true,
-        );
+          await onNewPeer(
+            WebSocketChannelPeer(
+              channel,
+              locationUri: locationUri.toString(),
+              connectionUris: [connectionUri],
+            )..id = id,
+            verifyId: true,
+          );
+        }
       },
       (e, st) async {
         if (retried) return;
diff --git a/lib/service/registry.dart b/lib/service/registry.dart
@@ -6,7 +6,7 @@ import 'package:lib5/constants.dart';
 import 'package:lib5/lib5.dart';
 import 'package:lib5/registry.dart';
 import 'package:lib5/util.dart';
-import 'package:messagepack/messagepack.dart';
+import 'package:s5_msgpack/s5_msgpack.dart';
 import 'package:s5_server/db/hive_key_value_db.dart';
 
 import 'package:s5_server/node.dart';
