CORDA-311-post PR merged fixes (corda#2106)

* SSH server integration

Author: m4ksio

core/src/main/kotlin/net/corda/core/flows/FlowLogic.kt

@@ -348,13 +348,25 @@ abstract class FlowLogic<out T> {
         }
     }
 
+    /**
+     * Returns a pair of the current progress step index (as integer) in steps tree of current [progressTracker], and an observable
+     * of its upcoming changes.
+     *
+     * @return Returns null if this flow has no progress tracker.
+     */
     fun trackStepsTreeIndex(): DataFeed<Int, Int>? {
         // TODO this is not threadsafe, needs an atomic get-step-and-subscribe
         return progressTracker?.let {
             DataFeed(it.stepsTreeIndex, it.stepsTreeIndexChanges)
         }
     }
 
+    /**
+     * Returns a pair of the current steps tree of current [progressTracker] as pairs of zero-based depth and stringified step
+     * label and observable of upcoming changes to the structure.
+     *
+     * @return Returns null if this flow has no progress tracker.
+     */
     fun trackStepsTree(): DataFeed<List<Pair<Int,String>>, List<Pair<Int,String>>>? {
         // TODO this is not threadsafe, needs an atomic get-step-and-subscribe
         return progressTracker?.let {

core/src/main/kotlin/net/corda/core/messaging/FlowHandle.kt

@@ -8,13 +8,17 @@ import rx.Observable
 
 /**
  * [FlowHandle] is a serialisable handle for the started flow, parameterised by the type of the flow's return value.
- *
- * @property id The started state machine's ID.
- * @property returnValue A [CordaFuture] of the flow's return value.
  */
 @DoNotImplement
 interface FlowHandle<A> : AutoCloseable {
+    /**
+     * The started state machine's ID.
+     */
     val id: StateMachineRunId
+
+    /**
+     * A [CordaFuture] of the flow's return value.
+     */
     val returnValue: CordaFuture<A>
 
     /**
@@ -25,15 +29,23 @@ interface FlowHandle<A> : AutoCloseable {
 
 /**
  * [FlowProgressHandle] is a serialisable handle for the started flow, parameterised by the type of the flow's return value.
- *
- * @property progress The stream of progress tracker events.
  */
 interface FlowProgressHandle<A> : FlowHandle<A> {
+    /**
+     * The stream of progress tracker events.
+     */
     val progress: Observable<String>
 
+    /**
+     * [DataFeed] of current step in the steps tree, see [ProgressTracker]
+     */
     val stepsTreeIndexFeed: DataFeed<Int, Int>?
 
+    /**
+     * [DataFeed] of current steps tree, see [ProgressTracker]
+     */
     val stepsTreeFeed: DataFeed<List<Pair<Int, String>>, List<Pair<Int, String>>>?
+
     /**
      * Use this function for flows whose returnValue and progress are not going to be used or tracked, so as to free up
      * server resources.

core/src/main/kotlin/net/corda/core/utilities/ProgressTracker.kt

@@ -99,6 +99,7 @@ class ProgressTracker(vararg steps: Step) {
             field = value
         }
 
+    /** The zero-bases index of the current step in a [allStepsLabels] list */
     var stepsTreeIndex: Int = -1
         private set(value) {
             field = value
@@ -226,6 +227,10 @@ class ProgressTracker(vararg steps: Step) {
      */
     val allSteps: List<Pair<Int, Step>> get() = _allStepsCache
 
+    /**
+     * A list of all steps label in this ProgressTracker and the children, with the indent level provided starting at zero.
+     * Note that UNSTARTED is never counted, and DONE is only counted at the calling level.
+     */
     val allStepsLabels: List<Pair<Int, String>> get() = _allStepsLabels()
 
     private var curChangeSubscription: Subscription? = null
@@ -245,8 +250,14 @@ class ProgressTracker(vararg steps: Step) {
      */
     val changes: Observable<Change> get() = _changes
 
+    /**
+     * An observable stream of changes to the [allStepsLabels]
+     */
     val stepsTreeChanges: Observable<List<Pair<Int,String>>> get() = _stepsTreeChanges
 
+    /**
+     * An observable stream of changes to the [stepsTreeIndex]
+     */
     val stepsTreeIndexChanges: Observable<Int> get() = _stepsTreeIndexChanges
 
     /** Returns true if the progress tracker has ended, either by reaching the [DONE] step or prematurely with an error */

core/src/test/kotlin/net/corda/core/utilities/ProgressTrackerTest.kt

@@ -98,7 +98,7 @@ class ProgressTrackerTest {
 
         val allSteps = pt.allSteps
 
-        //capture notifications
+        // Capture notifications.
         val stepsIndexNotifications = LinkedList<Int>()
         pt.stepsTreeIndexChanges.subscribe {
             stepsIndexNotifications += it
@@ -113,7 +113,7 @@ class ProgressTrackerTest {
             assertEquals(step, allSteps[pt.stepsTreeIndex].second)
         }
 
-        //travel tree
+        // Travel tree.
         pt.currentStep = SimpleSteps.ONE
         assertCurrentStepsTree(0, SimpleSteps.ONE)
 
@@ -126,7 +126,7 @@ class ProgressTrackerTest {
         pt.currentStep = SimpleSteps.THREE
         assertCurrentStepsTree(5, SimpleSteps.THREE)
 
-        //assert no structure changes and proper steps propagation
+        // Assert no structure changes and proper steps propagation.
         assertThat(stepsIndexNotifications).containsExactlyElementsOf(listOf(0, 1, 3, 5))
         assertThat(stepsTreeNotification).isEmpty()
     }
@@ -135,13 +135,13 @@ class ProgressTrackerTest {
     fun `structure changes are pushed down when progress trackers are added`() {
         pt.setChildProgressTracker(SimpleSteps.TWO, pt2)
 
-        //capture notifications
+        // Capture notifications.
         val stepsIndexNotifications = LinkedList<Int>()
         pt.stepsTreeIndexChanges.subscribe {
             stepsIndexNotifications += it
         }
 
-        //put current state as a first change for simplicity when asserting
+        // Put current state as a first change for simplicity when asserting.
         val stepsTreeNotification = mutableListOf(pt.allStepsLabels)
         println(pt.allStepsLabels)
         pt.stepsTreeChanges.subscribe {
@@ -164,7 +164,7 @@ class ProgressTrackerTest {
 
         assertCurrentStepsTree(9, SimpleSteps.FOUR)
 
-        //assert no structure changes and proper steps propagation
+        // Assert no structure changes and proper steps propagation.
         assertThat(stepsIndexNotifications).containsExactlyElementsOf(listOf(1, 6, 9))
         assertThat(stepsTreeNotification).hasSize(2) // 1 change + 1 our initial state
     }
@@ -173,13 +173,13 @@ class ProgressTrackerTest {
     fun `structure changes are pushed down when progress trackers are removed`() {
         pt.setChildProgressTracker(SimpleSteps.TWO, pt2)
 
-        //capture notifications
+        // Capture notifications.
         val stepsIndexNotifications = LinkedList<Int>()
         pt.stepsTreeIndexChanges.subscribe {
             stepsIndexNotifications += it
         }
 
-        //put current state as a first change for simplicity when asserting
+        // Put current state as a first change for simplicity when asserting.
         val stepsTreeNotification = mutableListOf(pt.allStepsLabels)
         pt.stepsTreeChanges.subscribe {
             stepsTreeNotification += it
@@ -199,9 +199,9 @@ class ProgressTrackerTest {
 
         assertCurrentStepsTree(2, BabySteps.UNOS)
 
-        //assert no structure changes and proper steps propagation
+        // Assert no structure changes and proper steps propagation.
         assertThat(stepsIndexNotifications).containsExactlyElementsOf(listOf(1, 4, 2))
-        assertThat(stepsTreeNotification).hasSize(2) // 1 change + 1 our initial state
+        assertThat(stepsTreeNotification).hasSize(2) // 1 change + 1 our initial state.
     }
 
     @Test

docs/source/shell.rst

@@ -35,9 +35,9 @@ Shell can also be accessible via SSH. By default SSH server is *disabled*. To en
 
 Authentication and authorization
 --------------------------------
-SSH require user to login first - using the same users as RPC system. In fact, shell serves as a proxy to RPC and communicates
-with node using RPC calls. This also means that RPC permissions are enforced. No permissions are required to allow the connection
-and login in.
+SSH requires users to login first - using the same users as RPC system. In fact, the shell serves as a proxy to RPC and communicates
+with the node using RPC calls. This also means that RPC permissions are enforced. No permissions are required to allow the connection
+and log in.
 Watching flows (``flow watch``) requires ``InvokeRpc.stateMachinesFeed`` while starting flows requires
 ``InvokeRpc.startTrackedFlowDynamic`` and ``InvokeRpc.registeredFlows`` in addition to a permission for a particular flow.
 
@@ -51,7 +51,7 @@ errors.
 Connecting
 ----------
 
-Linux and MacOS computers usually come with SSH client preinstalled. On Windows it usually require extra download.
+Linux and MacOS computers usually come with SSH client preinstalled. On Windows it usually requires extra download.
 Usual connection syntax is ``ssh user@host -p 2222`` - where ``user`` is a RPC username, and ``-p`` specifies a port parameters -
 it's the same as setup in ``node.conf`` file. ``host`` should point to a node hostname, usually ``localhost`` if connecting and
 running node on the same computer. Password will be asked after establishing connection.

node/src/integration-test/kotlin/net/corda/node/SSHServerTest.kt

@@ -139,8 +139,8 @@ class SSHServerTest {
 
             val response = String(Streams.readAll(channel.inputStream))
 
-            //There are ANSI control characters involved, so we want to avoid direct byte to byte matching
-            assertThat(response.lines()).filteredOn( { it.contains("✓") && it.contains("Done")}).hasSize(1)
+            // There are ANSI control characters involved, so we want to avoid direct byte to byte matching.
+            assertThat(response.lines()).filteredOn( { it.contains("Done")}).hasSize(1)
         }
     }
 

node/src/main/java/net/corda/node/shell/FlowShellCommand.java

@@ -4,15 +4,14 @@
 
 import net.corda.core.messaging.CordaRPCOps;
 import net.corda.node.utilities.ANSIProgressRenderer;
-import net.corda.node.utilities.CRaSHNSIProgressRenderer;
+import net.corda.node.utilities.CRaSHANSIProgressRenderer;
 import org.crsh.cli.*;
 import org.crsh.command.*;
 import org.crsh.text.*;
 import org.crsh.text.ui.TableElement;
 
 import java.util.*;
 
-import static net.corda.node.services.messaging.RPCServerKt.CURRENT_RPC_CONTEXT;
 import static net.corda.node.shell.InteractiveShell.*;
 
 @Man(
@@ -49,7 +48,7 @@ static void startFlow(@Usage("The class name of the flow to run, or an unambiguo
             return;
         }
         String inp = input == null ? "" : String.join(" ", input).trim();
-        runFlowByNameFragment(name, inp, out, rpcOps, ansiProgressRenderer != null ? ansiProgressRenderer : new CRaSHNSIProgressRenderer(out) );
+        runFlowByNameFragment(name, inp, out, rpcOps, ansiProgressRenderer != null ? ansiProgressRenderer : new CRaSHANSIProgressRenderer(out) );
     }
 
     @Command

node/src/main/java/net/corda/node/shell/RunShellCommand.java

@@ -30,7 +30,7 @@ public Object main(
             return null;
         }
 
-        return InteractiveShell.runRPCFromString(command, out, context);
+        return InteractiveShell.runRPCFromString(command, out, context, ops());
     }
 
     private void emitHelp(InvocationContext<Map> context, StringToMethodCallParser<CordaRPCOps> parser) {

node/src/main/java/net/corda/node/shell/StartShellCommand.java

@@ -3,7 +3,7 @@
 // A simple forwarder to the "flow start" command, for easier typing.
 
 import net.corda.node.utilities.ANSIProgressRenderer;
-import net.corda.node.utilities.CRaSHNSIProgressRenderer;
+import net.corda.node.utilities.CRaSHANSIProgressRenderer;
 import org.crsh.cli.*;
 
 import java.util.*;
@@ -14,6 +14,6 @@ public class StartShellCommand extends InteractiveShellCommand {
     public void main(@Usage("The class name of the flow to run, or an unambiguous substring") @Argument String name,
                      @Usage("The data to pass as input") @Argument(unquote = false) List<String> input) {
         ANSIProgressRenderer ansiProgressRenderer = ansiProgressRenderer();
-        FlowShellCommand.startFlow(name, input, out, ops(), ansiProgressRenderer != null ? ansiProgressRenderer : new CRaSHNSIProgressRenderer(out));
+        FlowShellCommand.startFlow(name, input, out, ops(), ansiProgressRenderer != null ? ansiProgressRenderer : new CRaSHANSIProgressRenderer(out));
     }
 }

node/src/main/kotlin/net/corda/node/shell/CordaAuthenticationPlugin.kt

@@ -25,7 +25,7 @@ class CordaAuthenticationPlugin(val rpcOps:CordaRPCOps, val userService:RPCUserS
 
         if (user != null && user.password == credential) {
             val actor = Actor(Actor.Id(username), userService.id, nodeLegalName)
-            return CordaSSHAuthInfo(true, RPCOpsWithContext(rpcOps, InvocationContext.rpc(actor), RpcPermissions(user.permissions)))
+            return CordaSSHAuthInfo(true, makeRPCOpsWithContext(rpcOps, InvocationContext.rpc(actor), RpcPermissions(user.permissions)))
         }
 
         return AuthInfo.UNSUCCESSFUL;

node/src/main/kotlin/net/corda/node/shell/InteractiveShell.kt

@@ -101,10 +101,10 @@ object InteractiveShell {
         this.nodeLegalName = configuration.myLegalName
         this.database = database
         val dir = configuration.baseDirectory
-        val runSshDeamon = configuration.sshd != null
+        val runSshDaemon = configuration.sshd != null
 
         val config = Properties()
-        if (runSshDeamon) {
+        if (runSshDaemon) {
             val sshKeysDir = dir / "sshkey"
             sshKeysDir.toFile().mkdirs()
 
@@ -120,7 +120,7 @@ object InteractiveShell {
         ExternalResolver.INSTANCE.addCommand("start", "An alias for 'flow start'", StartShellCommand::class.java)
         shell = ShellLifecycle(dir).start(config)
 
-        if (runSshDeamon) {
+        if (runSshDaemon) {
             Node.printBasicNodeInfo("SSH server listening on port", configuration.sshd!!.port.toString())
         }
     }
@@ -182,7 +182,7 @@ object InteractiveShell {
             context.refresh()
             this.config = config
             start(context)
-            return context.getPlugin(ShellFactory::class.java).create(null, CordaSSHAuthInfo(false, RPCOpsWithContext(rpcOps, net.corda.core.context.InvocationContext.shell(), RpcPermissions.ALL), StdoutANSIProgressRenderer))
+            return context.getPlugin(ShellFactory::class.java).create(null, CordaSSHAuthInfo(false, makeRPCOpsWithContext(rpcOps, net.corda.core.context.InvocationContext.shell(), RpcPermissions.ALL), StdoutANSIProgressRenderer))
         }
     }
 
@@ -236,7 +236,7 @@ object InteractiveShell {
         try {
             // Show the progress tracker on the console until the flow completes or is interrupted with a
             // Ctrl-C keypress.
-            val stateObservable = runFlowFromString({ clazz,args -> rpcOps.startTrackedFlowDynamic (clazz, *args) }, inputData, clazz)
+            val stateObservable = runFlowFromString({ clazz, args -> rpcOps.startTrackedFlowDynamic(clazz, *args) }, inputData, clazz)
 
             val latch = CountDownLatch(1)
             ansiProgressRenderer.render(stateObservable, { latch.countDown() })
@@ -247,7 +247,6 @@ object InteractiveShell {
             } catch (e: InterruptedException) {
                 // TODO: When the flow framework allows us to kill flows mid-flight, do so here.
             }
-
         } catch (e: NoApplicableConstructor) {
             output.println("No matching constructor found:", Color.red)
             e.errors.forEach { output.println("- $it", Color.red) }
@@ -326,7 +325,9 @@ object InteractiveShell {
         val (stateMachines, stateMachineUpdates) = proxy.stateMachinesFeed()
         val currentStateMachines = stateMachines.map { StateMachineUpdate.Added(it) }
         val subscriber = FlowWatchPrintingSubscriber(out)
-        stateMachineUpdates.startWith(currentStateMachines).subscribe(subscriber)
+        database.transaction {
+            stateMachineUpdates.startWith(currentStateMachines).subscribe(subscriber)
+        }
         var result: Any? = subscriber.future
         if (result is Future<*>) {
             if (!result.isDone) {
@@ -348,7 +349,7 @@ object InteractiveShell {
     }
 
     @JvmStatic
-    fun runRPCFromString(input: List<String>, out: RenderPrintWriter, context: InvocationContext<out Any>): Any? {
+    fun runRPCFromString(input: List<String>, out: RenderPrintWriter, context: InvocationContext<out Any>, cordaRPCOps: CordaRPCOps): Any? {
         val parser = StringToMethodCallParser(CordaRPCOps::class.java, context.attributes["mapper"] as ObjectMapper)
 
         val cmd = input.joinToString(" ").trim { it <= ' ' }
@@ -363,7 +364,7 @@ object InteractiveShell {
         var result: Any? = null
         try {
             InputStreamSerializer.invokeContext = context
-            val call = database.transaction { parser.parse(context.attributes["ops"] as CordaRPCOps, cmd) }
+            val call = database.transaction { parser.parse(cordaRPCOps, cmd) }
             result = call.call()
             if (result != null && result !is kotlin.Unit && result !is Void) {
                 result = printAndFollowRPCResponse(result, out)