disable transport test

packages/simplexmq/example/simplexmq_example.dart

@@ -1 +1 @@
-
+void main() {}

packages/simplexmq/lib/simplexmq.dart

@@ -4,4 +4,4 @@
 library simplexmq;
 
 export 'src/protocol.dart';
-export 'src/transport.dart' show Transport;
+export 'src/transport.dart' show Transport, SMPTransportClient;

packages/simplexmq/lib/src/buffer.dart

@@ -130,3 +130,25 @@ Uint8List? decode64(Uint8List b64) {
 
   return bytes;
 }
+
+Uint8List encodeInt32(int n) {
+  final data = Uint8List(4);
+  ByteData.sublistView(data).setInt32(0, n);
+  return data;
+}
+
+Uint8List encodeInt16(int n) {
+  final data = Uint8List(2);
+  ByteData.sublistView(data).setInt16(0, n);
+  return data;
+}
+
+extension EqualUint8List on Uint8List {
+  bool equal(Uint8List b) {
+    if (length != b.length) return false;
+    for (int i = 0; i < length; i++) {
+      if (this[i] != b[i]) return false;
+    }
+    return true;
+  }
+}

packages/simplexmq/lib/src/crypto.dart

@@ -6,9 +6,12 @@ import 'package:pointycastle/asymmetric/oaep.dart';
 import 'package:pointycastle/asymmetric/rsa.dart';
 import 'package:pointycastle/block/aes_fast.dart';
 import 'package:pointycastle/block/modes/gcm.dart';
+import 'package:pointycastle/digests/sha256.dart';
 import 'package:pointycastle/key_generators/api.dart';
 import 'package:pointycastle/key_generators/rsa_key_generator.dart';
 import 'package:pointycastle/random/fortuna_random.dart';
+import 'package:pointycastle/signers/pss_signer.dart';
+import 'buffer.dart';
 
 class AESKey {
   final Uint8List _key;
@@ -19,24 +22,17 @@ class AESKey {
 
   static AESKey decode(Uint8List rawKey) => AESKey._make(rawKey);
 
-  Uint8List encode() => _key;
+  Uint8List get bytes => _key;
 }
 
-Uint8List randomIV() {
-  return pseudoRandomBytes(16);
-}
+Uint8List randomIV() => pseudoRandomBytes(16);
 
-Uint8List secureRandomBytes(int len) {
-  return _randomBytes(len, Random.secure());
-}
+Uint8List secureRandomBytes(int len) => _randomBytes(len, Random.secure());
 
 final sessionSeed = Random.secure();
 
-Uint8List pseudoRandomBytes(int len) {
-  return _randomBytes(len, sessionSeed);
-}
+Uint8List pseudoRandomBytes(int len) => _randomBytes(len, sessionSeed);
 
-// len should be divisible by 4
 Uint8List _randomBytes(int len, Random seedSource) {
   final bytes = Uint8List(len);
   for (int i = 0; i < len; i++) {
@@ -45,7 +41,6 @@ Uint8List _randomBytes(int len, Random seedSource) {
   return bytes;
 }
 
-final empty = Uint8List(0);
 final paddingByte = '#'.codeUnitAt(0);
 
 Uint8List encryptAES(AESKey key, Uint8List iv, int padTo, Uint8List data) {
@@ -56,18 +51,15 @@ Uint8List encryptAES(AESKey key, Uint8List iv, int padTo, Uint8List data) {
   return _makeGCMCipher(key, iv, true).process(padded);
 }
 
-Uint8List decryptAES(AESKey key, Uint8List iv, Uint8List encryptedAndTag) {
-  return _makeGCMCipher(key, iv, false).process(encryptedAndTag);
-}
+Uint8List decryptAES(AESKey key, Uint8List iv, Uint8List encryptedAndTag) =>
+    _makeGCMCipher(key, iv, false).process(encryptedAndTag);
 
-GCMBlockCipher _makeGCMCipher(AESKey key, Uint8List iv, bool encrypt) {
-  return GCMBlockCipher(AESFastEngine())
-    ..init(encrypt, AEADParameters(KeyParameter(key._key), 128, iv, empty));
-}
+GCMBlockCipher _makeGCMCipher(AESKey key, Uint8List iv, bool encrypt) =>
+    GCMBlockCipher(AESFastEngine())
+      ..init(encrypt, AEADParameters(KeyParameter(key._key), 128, iv, empty));
 
-FortunaRandom _secureFortunaRandom() {
-  return FortunaRandom()..seed(KeyParameter(secureRandomBytes(32)));
-}
+FortunaRandom _secureFortunaRandom() =>
+    FortunaRandom()..seed(KeyParameter(secureRandomBytes(32)));
 
 AsymmetricKeyPair<RSAPublicKey, RSAPrivateKey> generateRSAkeyPair(
     [int bitLength = 2048]) {
@@ -80,14 +72,30 @@ AsymmetricKeyPair<RSAPublicKey, RSAPrivateKey> generateRSAkeyPair(
       pair.publicKey as RSAPublicKey, pair.privateKey as RSAPrivateKey);
 }
 
-Uint8List encryptOAEP(RSAPublicKey key, Uint8List data) {
-  final oaep = OAEPEncoding(RSAEngine())
-    ..init(true, PublicKeyParameter<RSAPublicKey>(key));
-  return oaep.process(data);
+Uint8List encryptOAEP(RSAPublicKey key, Uint8List data) =>
+    _oaep(true, PublicKeyParameter<RSAPublicKey>(key)).process(data);
+
+Uint8List decryptOAEP(RSAPrivateKey key, Uint8List data) =>
+    _oaep(false, PrivateKeyParameter<RSAPrivateKey>(key)).process(data);
+
+OAEPEncoding _oaep(bool encrypt, AsymmetricKeyParameter keyParam) =>
+    OAEPEncoding.withSHA256(RSAEngine())..init(encrypt, keyParam);
+
+Uint8List signPSS(RSAPrivateKey privateKey, Uint8List data) =>
+    _pss(true, PrivateKeyParameter<RSAPrivateKey>(privateKey))
+        .generateSignature(data)
+        .bytes;
+
+bool verifyPSS(RSAPublicKey publicKey, Uint8List data, Uint8List sig) {
+  try {
+    return _pss(false, PublicKeyParameter<RSAPublicKey>(publicKey))
+        .verifySignature(data, PSSSignature(sig));
+  } on ArgumentError {
+    return false;
+  }
 }
 
-Uint8List decryptOAEP(RSAPrivateKey key, Uint8List data) {
-  final oaep = OAEPEncoding(RSAEngine())
-    ..init(false, PrivateKeyParameter<RSAPrivateKey>(key));
-  return oaep.process(data);
-}
+PSSSigner _pss(bool sign, AsymmetricKeyParameter keyParam) => PSSSigner(
+    RSAEngine(), SHA256Digest(), SHA256Digest())
+  ..init(sign,
+      ParametersWithSaltConfiguration(keyParam, _secureFortunaRandom(), 32));

packages/simplexmq/lib/src/parser.dart

@@ -13,9 +13,11 @@ final charUpperA = cc('A');
 final charUpperZ = cc('Z');
 final charPlus = cc('+');
 final charSlash = cc('/');
+final charDot = cc('.');
 
 class Parser {
   final Uint8List _s;
+  final List<int> _positions = [];
   int _pos = 0;
   bool _fail = false;
   Parser(this._s);
@@ -34,6 +36,20 @@ class Parser {
     return res;
   }
 
+  T? tryParse<T>(T? Function(Parser p) parse) {
+    if (_fail || _pos >= _s.length) {
+      _fail = true;
+      return null;
+    }
+    _positions.add(_pos);
+    final res = parse(this);
+    final prevPos = _positions.removeLast();
+    if (res == null) {
+      _pos = prevPos;
+      _fail = false;
+    }
+  }
+
   // takes a required number of bytes
   Uint8List? take(int len) => _run(() {
         final end = _pos + len;
@@ -70,6 +86,14 @@ class Parser {
         return true;
       });
 
+  // takes the passed char
+  bool? char(int c) => _run(() {
+        if (_s[_pos] == c) {
+          _pos++;
+          return true;
+        }
+      });
+
   // takes space
   bool? space() => _run(() {
         if (_s[_pos] == charSpace) {

packages/simplexmq/lib/src/protocol.dart

@@ -174,7 +174,7 @@ class ERR extends BrokerCommand {
       : cmdErr = err == ErrorType.CMD
             ? throw ArgumentError('CMD error should be created with ERR.CMD')
             : null;
-  ERR.cmd(this.cmdErr) : err = ErrorType.CMD;
+  ERR.cmd(CmdErrorType this.cmdErr) : err = ErrorType.CMD;
   @override
   Uint8List serialize() {
     final _err = errorTags[err]!;

packages/simplexmq/lib/src/transport.dart

@@ -1,17 +1,310 @@
 import 'dart:async';
 import 'dart:typed_data';
+import 'package:pointycastle/asymmetric/api.dart';
+import 'package:pointycastle/digests/sha256.dart';
+import 'buffer.dart';
+import 'crypto.dart';
+import 'parser.dart';
+import 'protocol.dart';
+import 'rsa_keys.dart';
 
 abstract class Transport {
   Future<Uint8List> read(int n);
   Future<void> write(Uint8List data);
+  Future<void> close();
 }
 
-Stream<Uint8List> blockStream(Transport t, int blockSize) async* {
-  try {
-    while (true) {
-      yield await t.read(blockSize);
+class SMPServer {
+  final String host;
+  final int? port;
+  final Uint8List? keyHash;
+  SMPServer(this.host, [this.port, this.keyHash]);
+}
+
+class SessionKey {
+  final AESKey aesKey;
+  final Uint8List baseIV;
+  int _counter = 0;
+  SessionKey._(this.aesKey, this.baseIV);
+  static SessionKey create() => SessionKey._(AESKey.random(), randomIV());
+  Uint8List serialize() => concat(aesKey.bytes, baseIV);
+}
+
+class ServerHeader {
+  final int blockSize;
+  final int keySize;
+  ServerHeader(this.blockSize, this.keySize);
+}
+
+class ServerHandshake {
+  final RSAPublicKey publicKey;
+  final int blockSize;
+  ServerHandshake(this.publicKey, this.blockSize);
+}
+
+typedef SMPVersion = List<int>;
+
+SMPVersion _currentSMPVersion = const [0, 4, 1, 0];
+const int _serverHeaderSize = 8;
+const int _binaryRsaTransport = 0;
+const int _transportBlockSize = 4096;
+const int _maxTransportBlockSize = 65536;
+
+class _Request {
+  final Uint8List queueId;
+  final Completer<BrokerResponse> completer = Completer();
+  _Request(this.queueId);
+}
+
+class Either<L, R> {
+  final L? left;
+  final R? right;
+  Either.left(L this.left) : right = null;
+  Either.right(R this.right) : left = null;
+}
+
+enum ClientErrorType { SMPServerError, SMPResponseError, SMPUnexpectedResponse }
+
+class BrokerResponse {
+  final BrokerCommand? command;
+  final ClientErrorType? errorType;
+  final ERR? error;
+  BrokerResponse(BrokerCommand this.command)
+      : errorType = null,
+        error = null;
+  BrokerResponse.error(ClientErrorType this.errorType, this.error)
+      : command = null;
+}
+
+class ClientTransmission {
+  final String corrId;
+  final Uint8List queueId;
+  final ClientCommand command;
+  ClientTransmission(this.corrId, this.queueId, this.command);
+
+  Uint8List serialize() =>
+      unwordsN([encodeAscii(corrId), encode64(queueId), command.serialize()]);
+}
+
+class BrokerTransmission {
+  final String corrId;
+  final Uint8List queueId;
+  final BrokerCommand? command;
+  final ERR? error;
+  BrokerTransmission(this.corrId, this.queueId, BrokerCommand this.command)
+      : error = null;
+  BrokerTransmission.error(this.corrId, this.queueId, ERR this.error)
+      : command = null;
+}
+
+final badBlock = BrokerTransmission.error('', empty, ERR(ErrorType.BLOCK));
+
+class SMPTransportClient {
+  final Transport _conn;
+  final _sndKey = SessionKey.create();
+  final _rcvKey = SessionKey.create();
+  final int blockSize;
+  int _corrId = 0;
+  bool _messageStreamCreated = false;
+  final Map<String, _Request> _sentCommands = {};
+  SMPTransportClient._(this._conn, this.blockSize);
+
+  static Future<SMPTransportClient> connect(Transport conn,
+      {Uint8List? keyHash, int? blockSize}) {
+    return _clientHandshake(conn, keyHash, blockSize);
+  }
+
+  Future<void> close() {
+    return _conn.close();
+  }
+
+  Future<BrokerResponse> sendSMPCommand(
+      RSAPrivateKey? key, Uint8List queueId, ClientCommand cmd) async {
+    final corrId = (++_corrId).toString();
+    final t = ClientTransmission(corrId, queueId, cmd).serialize();
+    final sig = key == null ? empty : encode64(signPSS(key, t));
+    final data = unwordsN([sig, t, empty]);
+    final r = _sentCommands[corrId] = _Request(queueId);
+    await _writeEncrypted(data);
+    return r.completer.future;
+  }
+
+  Stream<BrokerTransmission> messageStream() {
+    if (_messageStreamCreated) {
+      throw Exception('message stream already created');
     }
-  } catch (e) {
-    return;
+    _messageStreamCreated = true;
+    return _messageStream();
+  }
+
+  Stream<BrokerTransmission> _messageStream() async* {
+    try {
+      while (true) {
+        final block = await _readEncrypted();
+        final t = _parseBrokerTransmission(block);
+        if (t.corrId == '') {
+          yield t;
+        } else {
+          final r = _sentCommands.remove(t.corrId);
+          if (r == null) {
+            yield t;
+          } else {
+            final cmd = t.command;
+            r.completer.complete(r.queueId.equal(t.queueId)
+                ? cmd == null
+                    ? BrokerResponse.error(
+                        ClientErrorType.SMPResponseError, t.error)
+                    : cmd is ERR
+                        ? BrokerResponse.error(
+                            ClientErrorType.SMPServerError, cmd)
+                        : BrokerResponse(cmd)
+                : BrokerResponse.error(
+                    ClientErrorType.SMPUnexpectedResponse, null));
+          }
+        }
+      }
+    } catch (e) {
+      return;
+    }
+  }
+
+  static BrokerTransmission _parseBrokerTransmission(Uint8List s) {
+    final p = Parser(s);
+    p.space();
+    final cId = p.word();
+    p.space();
+    final queueId = p.tryParse((p) => p.base64()) ?? empty;
+    p.space();
+    if (p.fail || cId == null) return badBlock;
+    final corrId = decodeAscii(cId);
+    final command = smpCommandP(p);
+    if (command == null) {
+      return BrokerTransmission.error(
+          corrId, queueId, ERR.cmd(CmdErrorType.SYNTAX));
+    }
+    if (command is! BrokerCommand) {
+      return BrokerTransmission.error(
+          corrId, queueId, ERR.cmd(CmdErrorType.PROHIBITED));
+    }
+    final qErr = _tQueueError(queueId, command);
+    if (qErr != null) {
+      return BrokerTransmission.error(corrId, queueId, ERR.cmd(qErr));
+    }
+    return BrokerTransmission(corrId, queueId, command);
+  }
+
+  static CmdErrorType? _tQueueError(Uint8List queueId, BrokerCommand cmd) {
+    if (cmd is IDS || cmd is PONG) {
+      if (queueId.isNotEmpty) return CmdErrorType.HAS_AUTH;
+    } else if (cmd is! ERR && queueId.isEmpty) {
+      return CmdErrorType.NO_QUEUE;
+    }
+  }
+
+  static Future<SMPTransportClient> _clientHandshake(
+      Transport conn, Uint8List? keyHash, int? blkSize) async {
+    final srv = await _getHeaderAndPublicKey_1_2(conn, keyHash);
+    final t = SMPTransportClient._(conn, blkSize ?? srv.blockSize);
+    await t._sendEncryptedKeys_4(srv.publicKey);
+    _checkVersion(await t._getWelcome_6());
+    return t;
+  }
+
+  static Future<ServerHandshake> _getHeaderAndPublicKey_1_2(
+      Transport conn, Uint8List? keyHash) async {
+    final srvHeader = parseServerHeader(await conn.read(_serverHeaderSize));
+    final blkSize = srvHeader.blockSize;
+    if (blkSize < _transportBlockSize || blkSize > _maxTransportBlockSize) {
+      throw Exception('smp handshake header error: bad block size $blkSize');
+    }
+    final rawKey = await conn.read(srvHeader.keySize);
+    if (keyHash != null) validateKeyHash_2(rawKey, keyHash);
+    final serverKey = decodeRsaPubKey(rawKey);
+    return ServerHandshake(serverKey, blkSize);
+  }
+
+  static void validateKeyHash_2(Uint8List rawKey, Uint8List keyHash) {
+    if (keyHash.equal(SHA256Digest().process(rawKey))) return;
+    throw Exception('smp handshake error: bad key hash');
+  }
+
+  static ServerHeader parseServerHeader(Uint8List a) {
+    if (a.length != 8) {
+      throw Exception('smp handshake error: bad header size ${a.length}');
+    }
+    final v = ByteData.sublistView(a);
+    final blockSize = v.getUint32(0);
+    final transportMode = v.getUint16(4);
+    if (transportMode != _binaryRsaTransport) {
+      throw Exception('smp handshake error: bad transport mode $transportMode');
+    }
+    final keySize = v.getUint16(6);
+    return ServerHeader(blockSize, keySize);
+  }
+
+  Future<void> _sendEncryptedKeys_4(RSAPublicKey serverKey) async {
+    final header = encryptOAEP(serverKey, _clientHeader());
+    await _conn.write(header);
+  }
+
+  Uint8List _clientHeader() => concatN([
+        encodeInt32(blockSize),
+        encodeInt16(_binaryRsaTransport),
+        _sndKey.serialize(),
+        _rcvKey.serialize()
+      ]);
+
+  Future<SMPVersion> _getWelcome_6() async =>
+      _parseSMPVersion(await _readEncrypted());
+
+  static SMPVersion _parseSMPVersion(Uint8List block) {
+    final p = Parser(block);
+    final SMPVersion version = [0, 0, 0, 0];
+    void setVer(int i, int? v) {
+      if (v == null || p.fail) {
+        throw Exception('smp handshake error: bad version format');
+      }
+      version[i] = v;
+    }
+
+    for (var i = 0; i < 3; i++) {
+      final v = p.decimal();
+      p.char(charDot);
+      setVer(i, v);
+    }
+    final v = p.decimal();
+    p.space();
+    setVer(3, v);
+    return version;
+  }
+
+  static void _checkVersion(SMPVersion srvVersion) {
+    final s0 = srvVersion[0];
+    final c0 = _currentSMPVersion[0];
+    if (s0 > c0 || (s0 == c0 && srvVersion[1] > _currentSMPVersion[1])) {
+      throw Exception('smp handshake error: incompatible server version');
+    }
+  }
+
+  Future<Uint8List> _readEncrypted() async {
+    final block = await _conn.read(blockSize);
+    final iv = _nextIV(_rcvKey);
+    return decryptAES(_rcvKey.aesKey, iv, block);
+  }
+
+  Future<void> _writeEncrypted(Uint8List data) {
+    final iv = _nextIV(_sndKey);
+    final block = encryptAES(_sndKey.aesKey, iv, blockSize - 16, data);
+    return _conn.write(block);
+  }
+
+  static Uint8List _nextIV(SessionKey sk) {
+    final c = encodeInt32(sk._counter++);
+    final start = sk.baseIV.sublist(0, 4);
+    final rest = sk.baseIV.sublist(4);
+    for (int i = 0; i < 4; i++) {
+      start[i] ^= c[i];
+    }
+    return concat(start, rest);
   }
 }

packages/simplexmq/pubspec.yaml

@@ -8,7 +8,12 @@ environment:
 
 dependencies:
   asn1lib: ^1.0.2
-  pointycastle: ^3.3.4
+  pointycastle:
+    git:
+      url: https://github.com/simplex-chat/pc-dart.git
+      ref: nullsafety
+
+  # pointycastle: ^3.3.4
 
 dev_dependencies:
   lints: ^1.0.0

packages/simplexmq_io/example/simplexmq_io_example.dart

@@ -1,6 +1,12 @@
-// import 'package:simplexmq_io/simplexmq_io.dart';
+import 'dart:io';
 
-// void main() {
-//   var awesome = Awesome();
-//   print('awesome: ${awesome.isAwesome}');
-// }
+void main() {
+  ServerSocket.bind('localhost', 8080)
+      .then((server) => server.listen((Socket socket) {
+            print('New client connection');
+            socket.listen((List<int> data) {
+              String result = String.fromCharCodes(data);
+              print(result);
+            });
+          }));
+}

packages/simplexmq_io/lib/src/socket.dart

@@ -20,22 +20,22 @@ class SocketTransport implements Transport {
   final int _bufferSize;
   Uint8List _buffer = Uint8List(0);
   final ListQueue<_STReaders> _readers = ListQueue(16);
-  SocketTransport._new(this._socket, this._timeout, this._bufferSize);
+  SocketTransport._(this._socket, this._timeout, this._bufferSize);
 
   static Future<SocketTransport> connect(String host, int port,
-      {Duration timeout = const Duration(seconds: 1),
+      {Duration timeout = const Duration(seconds: 4),
       int bufferSize = 16384}) async {
     final socket = await Socket.connect(host, port, timeout: timeout);
-    final t = SocketTransport._new(socket, timeout, bufferSize);
+    final t = SocketTransport._(socket, timeout, bufferSize);
     // ignore: cancel_subscriptions
     final subscription = socket.listen(t._onData,
-        onError: (Object e) => t._finalize, onDone: t._finalize);
+        onError: (Object e) => t.close, onDone: t.close);
     t._subscription = subscription;
     return t;
   }
 
   @override
-  Future<Uint8List> read(int n) async {
+  Future<Uint8List> read(int n) {
     if (_readers.isEmpty && _buffer.length >= n) {
       final data = _buffer.sublist(0, n);
       _buffer = _buffer.sublist(n);
@@ -71,17 +71,14 @@ class SocketTransport implements Transport {
     }
   }
 
-  void _finalize() {
-    _subscription.cancel();
+  /// Close the client transport
+  @override
+  Future<void> close() async {
+    await _subscription.cancel();
     _socket.destroy();
     while (_readers.isNotEmpty) {
       final r = _readers.removeFirst();
       r.completer.completeError(Exception('socket closed'));
     }
   }
-
-  /// Allow closing the client transport.
-  void close() {
-    _finalize();
-  }
 }

packages/simplexmq_io/pubspec.yaml

@@ -11,7 +11,7 @@ dependencies:
     git:
       url: git://github.com/simplex-chat/simplex-chat
       path: packages/simplexmq
-      ref: ep/socket-transport
+      ref: ep/smp-client
 
 dependency_overrides:
   simplexmq:

packages/simplexmq_io/test/simplexmq_io_test.dart

@@ -12,7 +12,7 @@ void main() {
   group('transport', () {
     Future<ServerSocket> startServer(
         void Function(Socket client) handleConnection) async {
-      var server = await ServerSocket.bind(InternetAddress.anyIPv4, 0);
+      var server = await ServerSocket.bind(localhost, 8080);
       server.listen(handleConnection);
       return server;
     }
@@ -27,11 +27,11 @@ void main() {
           },
         );
       });
-      var transport = await SocketTransport.connect(localhost, server.port);
+      var transport = await SocketTransport.connect(localhost, 8080);
       await transport.write(Uint8List.fromList([1, 2, 3]));
 
       expect(await completer.future, [1, 2, 3]);
-      transport.close();
+      await transport.close();
       await server.close();
     });
 
@@ -41,7 +41,7 @@ void main() {
       });
       var transport = await SocketTransport.connect(localhost, server.port);
       expect(await transport.read(3), [1, 2, 3]);
-      transport.close();
+      await transport.close();
       await server.close();
     });
   });

packages/simplexmq_io/test/transport_test.dart

@@ -0,0 +1,38 @@
+import 'package:simplexmq/simplexmq.dart';
+import 'package:simplexmq/src/buffer.dart';
+import 'package:simplexmq/src/crypto.dart';
+import 'package:simplexmq/src/rsa_keys.dart';
+import 'package:simplexmq_io/simplexmq_io.dart';
+import 'package:test/test.dart';
+
+final keyHash =
+    decode64(encodeAscii('pH7bg7B6vB3uJ1poKmClTAqr7yYWnAtapnIDN7ypKxU='));
+void main() {
+  group('SMP transport (expects SMP server on localhost:5223)', () {
+    test('establish connection', () async {
+      final conn = await SocketTransport.connect('localhost', 5223);
+      final smp = await SMPTransportClient.connect(conn, keyHash: keyHash);
+      expect(smp is SMPTransportClient, true);
+    });
+
+    test('should create SMP queue and send message', () async {
+      final conn1 = await SocketTransport.connect('localhost', 5223);
+      final alice = await SMPTransportClient.connect(conn1, keyHash: keyHash);
+      final aliceMessages = alice.messageStream();
+      final aliceKeys = generateRSAkeyPair();
+      final rcvKeyBytes = encodeRsaPubKey(aliceKeys.publicKey);
+      // input stream is not processed without the call to listen()
+      aliceMessages.listen((_) {});
+
+      // final conn2 = await SocketTransport.connect('localhost', 5223);
+      // final bob = await SMPTransportClient.connect(conn2, keyHash: keyHash);
+      // final bobKeys = generateRSAkeyPair();
+      // final sndKeyStr = encode64(encodeRsaPubKey(bobKeys.publicKey));
+
+      final resp = await alice.sendSMPCommand(
+          aliceKeys.privateKey, empty, NEW(rcvKeyBytes));
+      expect(resp.command is IDS, true);
+    });
+    // });
+  }, skip: 'requires SMP server on port 5223');
+}