Create header files for discussion of possible flow audit api.

Fix compile error

Address PR comments

Change from a general interface to a restricted set of audit event types.

Fixup after rebase

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

@@ -168,10 +168,32 @@ abstract class FlowLogic<out T> {
         return result
     }
 
+    /**
+     * Flows can call this method to ensure that the active FlowInitiator is authorised for a particular action.
+     * This provides fine grained control over application level permissions, when RPC control over starting the flow is insufficient,
+     * or the permission is runtime dependent upon the choices made inside long lived flow code.
+     * For example some users may have restricted limits on how much cash they can transfer, or whether they can change certain fields.
+     * An audit event is always recorded whenever this method is used.
+     * If the permission is not granted for the FlowInitiator a FlowException is thrown.
+     * @param permissionName is a string representing the desired permission. Each flow is given a distinct namespace for these permissions.
+     * @param extraAuditData in the audit log for this permission check these extra key value pairs will be recorded.
+     */
+    @Throws(FlowException::class)
+    fun checkFlowPermission(permissionName: String, extraAuditData: Map<String,String>) = stateMachine.checkFlowPermission(permissionName, extraAuditData)
+
+
+    /**
+     * Flows can call this method to record application level flow audit events
+     * @param eventType is a string representing the type of event. Each flow is given a distinct namespace for these names.
+     * @param comment a general human readable summary of the event.
+     * @param extraAuditData in the audit log for this permission check these extra key value pairs will be recorded.
+     */
+    fun recordAuditEvent(eventType: String, comment: String, extraAuditData: Map<String,String>) = stateMachine.recordAuditEvent(eventType, comment, extraAuditData)
+
     /**
      * Override this to provide a [ProgressTracker]. If one is provided and stepped, the framework will do something
-     * helpful with the progress reports. If this flow is invoked as a subflow of another, then the
-     * tracker will be made a child of the current step in the parent. If it's null, this flow doesn't track
+     * helpful with the progress reports e.g record to the audit service. If this flow is invoked as a subflow of another,
+     * then the tracker will be made a child of the current step in the parent. If it's null, this flow doesn't track
      * progress.
      *
      * Note that this has to return a tracker before the flow is invoked. You can't change your mind half way

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

@@ -10,6 +10,7 @@ import net.corda.core.serialization.CordaSerializable
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.UntrustworthyData
 import org.slf4j.Logger
+import java.security.Principal
 import java.util.*
 
 /**
@@ -18,14 +19,23 @@ import java.util.*
  * or via the Corda Shell [FlowInitiator.Shell].
  */
 @CordaSerializable
-sealed class FlowInitiator {
+sealed class FlowInitiator : Principal {
     /** Started using [net.corda.core.messaging.CordaRPCOps.startFlowDynamic]. */
-    data class RPC(val username: String) : FlowInitiator()
+    data class RPC(val username: String) : FlowInitiator() {
+        override fun getName(): String = username
+    }
     /** Started when we get new session initiation request. */
-    data class Peer(val party: Party) : FlowInitiator()
+    data class Peer(val party: Party) : FlowInitiator() {
+        override fun getName(): String = party.name.toString()
+    }
     /** Started as scheduled activity. */
-    data class Scheduled(val scheduledState: ScheduledStateRef) : FlowInitiator()
-    object Shell : FlowInitiator() // TODO When proper ssh access enabled, add username/use RPC?
+    data class Scheduled(val scheduledState: ScheduledStateRef) : FlowInitiator() {
+        override fun getName(): String = "Scheduler"
+    }
+    // TODO When proper ssh access enabled, add username/use RPC?
+    object Shell : FlowInitiator() {
+        override fun getName(): String = "Shell User"
+    }
 }
 
 /**
@@ -59,6 +69,10 @@ interface FlowStateMachine<R> {
     @Suspendable
     fun waitForLedgerCommit(hash: SecureHash, sessionFlow: FlowLogic<*>): SignedTransaction
 
+    fun checkFlowPermission(permissionName: String, extraAuditData: Map<String,String>)
+
+    fun recordAuditEvent(eventType: String, comment: String, extraAuditData: Map<String,String>)
+
     val serviceHub: ServiceHub
     val logger: Logger
     val id: StateMachineRunId

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

@@ -55,6 +55,12 @@ class ProgressTracker(vararg steps: Step) {
     open class Step(open val label: String) {
         open val changes: Observable<Change> get() = Observable.empty()
         open fun childProgressTracker(): ProgressTracker? = null
+        /**
+         * A flow may populate this property with flow specific context data.
+         * The extra data will be recorded to the audit logs when the flow progresses.
+         * Even if empty the basic details (i.e. label) of the step will be recorded for audit purposes.
+         */
+        open val extraAuditData: Map<String, String> get() = emptyMap()
     }
 
     // Sentinel objects. Overrides equals() to survive process restarts and serialization.

node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt

@@ -132,6 +132,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
         override val myInfo: NodeInfo get() = info
         override val schemaService: SchemaService get() = schemas
         override val transactionVerifierService: TransactionVerifierService get() = txVerifierService
+        override val auditService: AuditService get() = auditService
 
         // Internal only
         override val monitoringService: MonitoringService = MonitoringService(MetricRegistry())
@@ -177,6 +178,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
     lateinit var scheduler: NodeSchedulerService
     lateinit var flowLogicFactory: FlowLogicRefFactoryInternal
     lateinit var schemas: SchemaService
+    lateinit var auditService: AuditService
     val customServices: ArrayList<Any> = ArrayList()
     protected val runOnStop: ArrayList<Runnable> = ArrayList()
     lateinit var database: Database
@@ -295,6 +297,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
         schemas = makeSchemaService()
         vault = makeVaultService(configuration.dataSourceProperties)
         txVerifierService = makeTransactionVerifierService()
+        auditService = DummyAuditService()
 
         info = makeInfo()
         identity = makeIdentityService()

node/src/main/kotlin/net/corda/node/services/RPCUserService.kt

@@ -15,6 +15,7 @@ interface RPCUserService {
 
 // TODO Store passwords as salted hashes
 // TODO Or ditch this and consider something like Apache Shiro
+// TODO Need access to permission checks from inside flows and at other point during audit checking.
 class RPCUserServiceImpl(override val users: List<User>) : RPCUserService {
     override fun getUser(username: String): User? = users.find { it.username == username }
 }

node/src/main/kotlin/net/corda/node/services/api/AuditService.kt

@@ -0,0 +1,138 @@
+package net.corda.node.services.api
+
+import net.corda.core.flows.FlowLogic
+import net.corda.core.flows.StateMachineRunId
+import net.corda.core.serialization.SingletonSerializeAsToken
+import net.corda.core.utilities.ProgressTracker
+import java.security.Principal
+import java.time.Instant
+
+/**
+ * Minimum event specific data for any audit event to be logged. It is expected that the underlying audit service
+ * will enrich this to include details of the node, so that in clustered configurations the source node can be identified.
+ */
+sealed class AuditEvent {
+    /**
+     * The UTC time point at which the audit event happened.
+     */
+    abstract val timestamp: Instant
+    /**
+     * The responsible individual, node, or subsystem to which the audit event can be mapped.
+     */
+    abstract val principal: Principal
+    /**
+     * A human readable description of audit event including any permission check results.
+     */
+    abstract val description: String
+    /**
+     * Further tagged details that should be recorded along with the common data of the audit event.
+     * Examples of this might be trade identifiers, system error codes, or source IP addresses, which could be useful
+     * when searching the historic audit data for trails of evidence.
+     */
+    abstract val contextData: Map<String, String>
+}
+
+/**
+ * Sealed data class to mark system related events as a distinct category.
+ */
+data class SystemAuditEvent(override val timestamp: Instant,
+                            override val principal: Principal,
+                            override val description: String,
+                            override val contextData: Map<String, String>) : AuditEvent()
+
+/**
+ * Interface to mandate flow identification properties
+ */
+interface FlowAuditInfo {
+    /**
+     * The concrete type of FlowLogic being referenced.
+     * TODO This should be replaced with the fully versioned name/signature of the flow.
+     */
+    val flowType: Class<out FlowLogic<*>>
+    /**
+     * The stable identifier of the flow as stored with Checkpoints.
+     */
+    val flowId: StateMachineRunId
+}
+
+/**
+ * Sealed data class to record custom application specified flow event.
+ */
+data class FlowAppAuditEvent(
+        override val timestamp: Instant,
+        override val principal: Principal,
+        override val description: String,
+        override val contextData: Map<String, String>,
+        override val flowType: Class<out FlowLogic<*>>,
+        override val flowId: StateMachineRunId,
+        val auditEventType: String) : AuditEvent(), FlowAuditInfo
+
+/**
+ * Sealed data class to record the initiation of a new flow.
+ * The flow parameters should be captured to the context data.
+ */
+data class FlowStartEvent(
+        override val timestamp: Instant,
+        override val principal: Principal,
+        override val description: String,
+        override val contextData: Map<String, String>,
+        override val flowType: Class<out FlowLogic<*>>,
+        override val flowId: StateMachineRunId) : AuditEvent(), FlowAuditInfo
+
+/**
+ * Sealed data class to record ProgressTracker Step object whenever a change is signalled.
+ * The API for ProgressTracker has been extended so that the Step can contain some extra context data,
+ * which is copied into the contextData Map.
+ */
+data class FlowProgressAuditEvent(
+        override val timestamp: Instant,
+        override val principal: Principal,
+        override val description: String,
+        override val flowType: Class<out FlowLogic<*>>,
+        override val flowId: StateMachineRunId,
+        val flowProgress: ProgressTracker.Step) : AuditEvent(), FlowAuditInfo {
+    override val contextData: Map<String, String> get() = flowProgress.extraAuditData
+}
+
+/**
+ * Sealed data class to record any FlowExceptions, or other unexpected terminations of a Flow.
+ */
+data class FlowErrorAuditEvent(override val timestamp: Instant,
+                               override val principal: Principal,
+                               override val description: String,
+                               override val contextData: Map<String, String>,
+                               override val flowType: Class<out FlowLogic<*>>,
+                               override val flowId: StateMachineRunId,
+                               val error: Throwable) : AuditEvent(), FlowAuditInfo
+
+/**
+ * Sealed data class to record  checks on per flow permissions and the verdict of these checks
+ * If the permission is denied i.e. permissionGranted is false, then it is expected that the flow will be terminated immediately
+ * after recording the FlowPermissionAuditEvent. This may cause an extra FlowErrorAuditEvent to be recorded too.
+ */
+data class FlowPermissionAuditEvent(override val timestamp: Instant,
+                                    override val principal: Principal,
+                                    override val description: String,
+                                    override val contextData: Map<String, String>,
+                                    override val flowType: Class<out FlowLogic<*>>,
+                                    override val flowId: StateMachineRunId,
+                                    val permissionRequested: String,
+                                    val permissionGranted: Boolean) : AuditEvent(), FlowAuditInfo
+/**
+ * Minimal interface for recording audit information within the system. The AuditService is assumed to be available only
+ * to trusted internal components via ServiceHubInternal.
+ */
+interface AuditService {
+    fun recordAuditEvent(event: AuditEvent)
+}
+
+/**
+ * Empty do nothing AuditService as placeholder.
+ * TODO Write a full implementation that expands all the audit events to the database.
+ */
+class DummyAuditService : AuditService, SingletonSerializeAsToken() {
+    override fun recordAuditEvent(event: AuditEvent) {
+        //TODO Implement transformation of the audit events to formal audit data
+    }
+}
+

node/src/main/kotlin/net/corda/node/services/api/ServiceHubInternal.kt

@@ -68,6 +68,7 @@ abstract class ServiceHubInternal : PluginServiceHub {
     abstract val schemaService: SchemaService
     abstract override val networkMapCache: NetworkMapCacheInternal
     abstract val schedulerService: SchedulerService
+    abstract val auditService: AuditService
 
     abstract val networkService: MessagingService
 

node/src/main/kotlin/net/corda/node/services/statemachine/FlowStateMachineImpl.kt

@@ -17,6 +17,8 @@ import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.UntrustworthyData
 import net.corda.core.utilities.debug
 import net.corda.core.utilities.trace
+import net.corda.node.services.api.FlowAppAuditEvent
+import net.corda.node.services.api.FlowPermissionAuditEvent
 import net.corda.node.services.api.ServiceHubInternal
 import net.corda.node.utilities.StrandLocalTransactionManager
 import net.corda.node.utilities.createTransaction
@@ -31,6 +33,8 @@ import java.sql.SQLException
 import java.util.*
 import java.util.concurrent.TimeUnit
 
+class FlowPermissionException(message: String) : FlowException(message)
+
 class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
                               val logic: FlowLogic<R>,
                               scheduler: FiberScheduler,
@@ -221,6 +225,37 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
         throw IllegalStateException("We were resumed after waiting for $hash but it wasn't found in our local storage")
     }
 
+    // TODO Dummy implementation of access to application specific permission controls and audit logging
+    override fun checkFlowPermission(permissionName: String, extraAuditData: Map<String, String>) {
+        val permissionGranted = true // TODO define permission control service on ServiceHubInternal and actually check authorization.
+        val checkPermissionEvent = FlowPermissionAuditEvent(
+            serviceHub.clock.instant(),
+            flowInitiator,
+            "Flow Permission Required: $permissionName",
+            extraAuditData,
+            logic.javaClass,
+            id,
+            permissionName,
+            permissionGranted)
+        serviceHub.auditService.recordAuditEvent(checkPermissionEvent)
+        if (!permissionGranted) {
+            throw FlowPermissionException("User $flowInitiator not permissioned for $permissionName on flow $id")
+        }
+    }
+
+    // TODO Dummy implementation of access to application specific audit logging
+    override fun recordAuditEvent(eventType: String, comment: String, extraAuditData: Map<String,String>) {
+        val flowAuditEvent = FlowAppAuditEvent(
+                    serviceHub.clock.instant(),
+                    flowInitiator,
+                    comment,
+                    extraAuditData,
+                    logic.javaClass,
+                id,
+                eventType)
+        serviceHub.auditService.recordAuditEvent(flowAuditEvent)
+    }
+
     /**
      * This method will suspend the state machine and wait for incoming session init response from other party.
      */

node/src/test/kotlin/net/corda/node/InteractiveShellTest.kt

@@ -96,5 +96,13 @@ class InteractiveShellTest {
             get() = throw UnsupportedOperationException()
         override val flowInitiator: FlowInitiator
             get() = throw UnsupportedOperationException()
+
+        override fun checkFlowPermission(permissionName: String, extraAuditData: Map<String, String>) {
+            // Do nothing
+        }
+
+        override fun recordAuditEvent(eventType: String, comment: String, extraAuditData: Map<String, String>) {
+            // Do nothing
+        }
     }
 }

node/src/test/kotlin/net/corda/node/services/MockServiceHubInternal.kt

@@ -60,7 +60,7 @@ open class MockServiceHubInternal(
         get() = flowFactory ?: throw UnsupportedOperationException()
     override val schemaService: SchemaService
         get() = schemas ?: throw UnsupportedOperationException()
-
+    override val auditService: AuditService = DummyAuditService()
     // We isolate the storage service with writable TXes so that it can't be accessed except via recordTransactions()
     private val txStorageService: TxWritableStorageService
         get() = storage ?: throw UnsupportedOperationException()