Merge pull request #741 from atsign-foundation/gkc/feat/progress-and-…

…logger-listeners

feat: provide a way to listen to progress and logger messages

packages/dart/noports_core/lib/src/common/streaming_logging_handler.dart

@@ -0,0 +1,22 @@
+import 'dart:async';
+
+import 'package:at_utils/at_logger.dart';
+import 'package:logging/logging.dart';
+
+class StreamingLoggingHandler implements LoggingHandler {
+  final LoggingHandler _wrappedLoggingHandler;
+  final StreamController<String> _logSC = StreamController.broadcast();
+
+  StreamingLoggingHandler(this._wrappedLoggingHandler);
+
+  @override
+  void call(LogRecord record) {
+    _wrappedLoggingHandler.call(record);
+    _logSC.add('${record.level.name}'
+        '|${record.time}'
+        '|${record.loggerName}'
+        '|${record.message}');
+  }
+
+  Stream<String> get stream => _logSC.stream;
+}

packages/dart/noports_core/lib/src/sshnp/impl/sshnp_dart_pure_impl.dart

@@ -11,6 +11,7 @@ class SshnpDartPureImpl extends SshnpCore
     required super.atClient,
     required super.params,
     required AtSshKeyPair? identityKeyPair,
+    required super.logStream,
   }) {
     this.identityKeyPair = identityKeyPair;
     _sshnpdChannel = SshnpdDefaultChannel(
@@ -52,7 +53,9 @@ class SshnpDartPureImpl extends SshnpCore
     /// Ensure that sshnp is initialized
     await callInitialization();
 
-    logger.info('Sending request to sshnpd');
+    var msg = 'Sending session request to the device daemon';
+    logger.info(msg);
+    sendProgress(msg);
 
     /// Send an ssh request to sshnpd
     await notify(
@@ -81,14 +84,17 @@ class SshnpDartPureImpl extends SshnpCore
     );
 
     /// Wait for a response from sshnpd
+    sendProgress('Waiting for response from the device daemon');
     var acked = await sshnpdChannel.waitForDaemonResponse();
     if (acked != SshnpdAck.acknowledged) {
-      throw SshnpError('sshnpd did not acknowledge the request');
+      throw SshnpError('No response from the device daemon');
+    } else {
+      sendProgress('Received response from the device daemon');
     }
 
     if (sshnpdChannel.ephemeralPrivateKey == null) {
       throw SshnpError(
-        'Expected an ephemeral private key from sshnpd, but it was not set',
+        'Expected an ephemeral private key from device daemon, but it was not set',
       );
     }
 
@@ -102,13 +108,15 @@ class SshnpDartPureImpl extends SshnpCore
     await keyUtil.addKeyPair(keyPair: ephemeralKeyPair);
 
     /// Start srv
+    sendProgress('Creating connection to socket rendezvous');
     SSHSocket? sshSocket = await srvdChannel.runSrv(
       directSsh: true,
       sessionAESKeyString: sshnpdChannel.sessionAESKeyString,
       sessionIVString: sshnpdChannel.sessionIVString,
     );
 
     /// Start the initial tunnel
+    sendProgress('Starting tunnel session');
     tunnelSshClient = await startInitialTunnelSession(
       ephemeralKeyPairIdentifier: ephemeralKeyPair.identifier,
       sshSocket: sshSocket,
@@ -142,9 +150,11 @@ class SshnpDartPureImpl extends SshnpCore
           'Cannot execute runShell, tunnel has not yet been created');
     }
 
+    sendProgress('Starting user session');
     SSHClient userSession =
         await startUserSession(tunnelSession: tunnelSshClient!);
 
+    sendProgress('Starting remote shell');
     SSHSession shell = await userSession.shell();
 
     return SSHSessionAsSshnpRemoteProcess(shell);

packages/dart/noports_core/lib/src/sshnp/impl/sshnp_openssh_local_impl.dart

@@ -14,6 +14,7 @@ class SshnpOpensshLocalImpl extends SshnpCore
   SshnpOpensshLocalImpl({
     required super.atClient,
     required super.params,
+    required super.logStream,
   }) {
     _sshnpdChannel = SshnpdDefaultChannel(
       atClient: atClient,
@@ -49,7 +50,9 @@ class SshnpOpensshLocalImpl extends SshnpCore
     /// Ensure that sshnp is initialized
     await callInitialization();
 
-    logger.info('Sending request to sshnpd');
+    var msg = 'Sending session request to the device daemon';
+    logger.info(msg);
+    sendProgress(msg);
 
     /// Send an ssh request to sshnpd
     await notify(
@@ -78,9 +81,12 @@ class SshnpOpensshLocalImpl extends SshnpCore
     );
 
     /// Wait for a response from sshnpd
+    sendProgress('Waiting for response from the device daemon');
     var acked = await sshnpdChannel.waitForDaemonResponse();
     if (acked != SshnpdAck.acknowledged) {
-      throw SshnpError('sshnpd did not acknowledge the request');
+      throw SshnpError('No response from the device daemon');
+    } else {
+      sendProgress('Received response from the device daemon');
     }
 
     if (sshnpdChannel.ephemeralPrivateKey == null) {
@@ -95,6 +101,7 @@ class SshnpOpensshLocalImpl extends SshnpCore
     await server.close();
 
     /// Start srv
+    sendProgress('Creating connection to socket rendezvous');
     await srvdChannel.runSrv(
       directSsh: true,
       localRvPort: localRvPort,
@@ -113,6 +120,7 @@ class SshnpOpensshLocalImpl extends SshnpCore
     await keyUtil.addKeyPair(keyPair: ephemeralKeyPair);
 
     /// Start the initial tunnel
+    sendProgress('Starting tunnel session');
     Process? bean = await startInitialTunnelSession(
       ephemeralKeyPairIdentifier: ephemeralKeyPair.identifier,
       localRvPort: localRvPort,

packages/dart/noports_core/lib/src/sshnp/impl/sshnp_unsigned_impl.dart

@@ -10,6 +10,7 @@ class SshnpUnsignedImpl extends SshnpCore
   SshnpUnsignedImpl({
     required super.atClient,
     required super.params,
+    required super.logStream,
   }) {
     if (Platform.isWindows) {
       throw SshnpError(
@@ -90,7 +91,7 @@ class SshnpUnsignedImpl extends SshnpCore
         ..sharedBy = params.clientAtSign
         ..sharedWith = params.sshnpdAtSign
         ..metadata = (Metadata()..ttl = 10000),
-      '$localPort ${srvdChannel.clientPort} ${keyUtil.username} ${srvdChannel.host} $sessionId',
+      '$localPort ${srvdChannel.daemonPort} ${keyUtil.username} ${srvdChannel.host} $sessionId',
       checkForFinalDeliveryStatus: false,
       waitForFinalDeliveryStatus: false,
     );

packages/dart/noports_core/lib/src/sshnp/sshnp.dart

@@ -1,32 +1,44 @@
 import 'dart:async';
 
 import 'package:at_client/at_client.dart' hide StringBuffer;
+import 'package:at_utils/at_logger.dart';
+import 'package:noports_core/src/common/streaming_logging_handler.dart';
 import 'package:noports_core/sshnp_foundation.dart';
 
 abstract interface class SshnpRemoteProcess {
   Future<void> get done;
+
   Stream<List<int>> get stderr;
+
   StreamSink<List<int>> get stdin;
+
   Stream<List<int>> get stdout;
 }
 
 abstract interface class Sshnp {
+  static final StreamingLoggingHandler _slh =
+      StreamingLoggingHandler(AtSignLogger.defaultLoggingHandler);
+
   /// Legacy v3.x.x client
   @Deprecated(
       'Legacy unsigned client - only for connecting with ^3.0.0 daemons')
   factory Sshnp.unsigned({
     required AtClient atClient,
     required SshnpParams params,
   }) {
-    return SshnpUnsignedImpl(atClient: atClient, params: params);
+    AtSignLogger.defaultLoggingHandler = _slh;
+    return SshnpUnsignedImpl(
+        atClient: atClient, params: params, logStream: _slh.stream);
   }
 
   /// Think of this as the "default" client - calls openssh
   factory Sshnp.openssh({
     required AtClient atClient,
     required SshnpParams params,
   }) {
-    return SshnpOpensshLocalImpl(atClient: atClient, params: params);
+    AtSignLogger.defaultLoggingHandler = _slh;
+    return SshnpOpensshLocalImpl(
+        atClient: atClient, params: params, logStream: _slh.stream);
   }
 
   /// Uses a dartssh2 ssh client - requires that you pass in the identity keypair
@@ -35,8 +47,12 @@ abstract interface class Sshnp {
     required SshnpParams params,
     required AtSshKeyPair? identityKeyPair,
   }) {
+    AtSignLogger.defaultLoggingHandler = _slh;
     var sshnp = SshnpDartPureImpl(
-        atClient: atClient, params: params, identityKeyPair: identityKeyPair);
+        atClient: atClient,
+        params: params,
+        identityKeyPair: identityKeyPair,
+        logStream: _slh.stream);
     if (identityKeyPair != null) {
       sshnp.keyUtil.addKeyPair(keyPair: identityKeyPair);
     }
@@ -72,4 +88,11 @@ abstract interface class Sshnp {
   /// - Iterable<String> of atSigns of sshnpd that did not respond
   /// - Map<String, dynamic> where the keys are all atSigns included in the maps, and the values being their device info
   Future<SshnpDeviceList> listDevices();
+
+  /// Yields a string every time something interesting happens with regards to
+  /// progress towards establishing the ssh connection.
+  Stream<String>? get progressStream;
+
+  /// Yields every log message that is written to [stderr]
+  Stream<String>? get logStream;
 }

packages/dart/noports_core/lib/src/sshnp/sshnp_core.dart

@@ -61,9 +61,27 @@ abstract class SshnpCore
   @protected
   SshnpdChannel get sshnpdChannel;
 
+  final StreamController<String> _progressStreamController =
+      StreamController<String>.broadcast();
+
+  /// Yields a string every time something interesting happens with regards to
+  /// progress towards establishing the ssh connection.
+  @override
+  Stream<String>? get progressStream => _progressStreamController.stream;
+
+  /// Yields every log message that is written to [stderr]
+  @override
+  final Stream<String>? logStream;
+
+  /// Subclasses should use this method to generate progress messages
+  sendProgress(String message) {
+    _progressStreamController.add(message);
+  }
+
   SshnpCore({
     required this.atClient,
     required this.params,
+    this.logStream,
   })  : sessionId = Uuid().v4(),
         namespace = '${params.device}.${DefaultArgs.namespace}',
         localPort = params.localPort {
@@ -80,6 +98,7 @@ abstract class SshnpCore
   @mustCallSuper
   Future<void> initialize() async {
     if (!isSafeToInitialize) return;
+
     logger.info('Initializing SshnpCore');
 
     /// Start the sshnpd payload handler
@@ -88,11 +107,14 @@ abstract class SshnpCore
     if (params.discoverDaemonFeatures) {
       late Map<String, dynamic> pingResponse;
       try {
+        sendProgress('Pinging daemon to discover features');
         pingResponse =
             await sshnpdChannel.ping().timeout(Duration(seconds: 10));
       } catch (e) {
-        logger.severe(
-            'No ping response from ${params.device}${params.sshnpdAtSign}');
+        var msg =
+            'No ping response from ${params.device}${params.sshnpdAtSign}';
+        sendProgress(msg);
+        logger.severe(msg);
         rethrow;
       }
 
@@ -112,16 +134,22 @@ abstract class SshnpCore
     }
 
     /// Set the remote username to use for the ssh session
+    sendProgress('Resolving remote username for user session');
     remoteUsername = await sshnpdChannel.resolveRemoteUsername();
 
     /// Set the username to use for the initial ssh tunnel
+    sendProgress('Resolving remote username for tunnel session');
     tunnelUsername = await sshnpdChannel.resolveTunnelUsername(
         remoteUsername: remoteUsername);
 
     /// Shares the public key if required
+    if (params.sendSshPublicKey) {
+      sendProgress('Sharing ssh public key');
+    }
     await sshnpdChannel.sharePublicKeyIfRequired(identityKeyPair);
 
     /// Retrieve the srvd host and port pair
+    sendProgress('Fetching host and port from srvd');
     await srvdChannel.callInitialization();
   }
 

packages/dart/noports_core/lib/src/sshnp/util/srvd_channel/srvd_channel.dart

@@ -122,7 +122,7 @@ abstract class SrvdChannel<T> with AsyncInitialization, AtClientBindings {
       if (params.host.startsWith('@')) {
         srv = srvGenerator(
           host,
-          daemonPort!, // everything was backwards back then
+          clientPort,
           localPort: params.localSshdPort,
           bindLocalPort: false,
         );
@@ -212,7 +212,7 @@ abstract class SrvdChannel<T> with AsyncInitialization, AtClientBindings {
       counter++;
       if (counter > 150) {
         logger.warning('Timed out waiting for srvd response');
-        throw ('Connection timeout to srvd $host service\nhint: make sure host is valid and online');
+        throw ('Connection timeout to srvd $host service');
       }
     }
   }

packages/dart/noports_core/lib/src/sshnp/util/ssh_session_handler/dart_ssh_session_handler.dart

@@ -162,6 +162,7 @@ class SshClientHelper {
       // Ensure we are connected and authenticated correctly
       logger.info('awaiting SSHClient.ping');
       await client.ping().catchError((e) => throw e);
+      logger.info('SSHClient.ping complete');
     } catch (e, s) {
       throw SshnpError(
         'Failed to authenticate as $username@$host:$port : $e',

packages/dart/noports_core/test/sshnp/sshnp_core_test.dart

@@ -42,6 +42,7 @@ void main() {
       when(() => mockParams.localPort).thenReturn(0);
       when(() => mockParams.verbose).thenReturn(verbose);
       when(() => mockParams.discoverDaemonFeatures).thenReturn(false);
+      when(() => mockParams.sendSshPublicKey).thenReturn(false);
       when(() => mockAtClient.getPreferences()).thenReturn(null);
       when(() => mockAtClient.setPreferences(any())).thenReturn(null);
     }

packages/dart/noports_core/test/sshnp/util/srvd_channel/srvd_channel_test.dart

@@ -111,6 +111,7 @@ void main() {
       when(() => mockParams.authenticateClientToRvd).thenReturn(true);
       when(() => mockParams.encryptRvdTraffic).thenReturn(true);
       when(() => mockParams.discoverDaemonFeatures).thenReturn(false);
+      when(() => mockParams.sendSshPublicKey).thenReturn(false);
 
       when(subscribeInvocation)
           .thenAnswer((_) => notificationStreamController.stream);

packages/dart/noports_core/test/sshnp/util/sshnp_ssh_key_handler/sshnp_local_ssh_key_handler_test.dart

@@ -33,6 +33,7 @@ void main() {
       when(() => mockParams.localPort).thenReturn(0);
       when(() => mockParams.verbose).thenReturn(false);
       when(() => mockParams.discoverDaemonFeatures).thenReturn(false);
+      when(() => mockParams.sendSshPublicKey).thenReturn(false);
       when(() => mockAtClient.getPreferences()).thenReturn(null);
       when(() => mockAtClient.setPreferences(any())).thenReturn(null);
     }

packages/dart/noports_core/test/sshnp/util/sshnpd_channel/sshnpd_default_channel_test.dart

@@ -71,6 +71,7 @@ void main() {
       when(() => mockParams.authenticateClientToRvd).thenReturn(true);
       when(() => mockParams.encryptRvdTraffic).thenReturn(true);
       when(() => mockParams.discoverDaemonFeatures).thenReturn(false);
+      when(() => mockParams.sendSshPublicKey).thenReturn(false);
       when(subscribeInvocation)
           .thenAnswer((_) => notificationStreamController.stream);
     }