diff --git a/.ci/dev/nightly-regression/Jenkinsfile b/.ci/dev/nightly-regression/Jenkinsfile
index 2fb82b3b7f..9102d5e39d 100644
--- a/.ci/dev/nightly-regression/Jenkinsfile
+++ b/.ci/dev/nightly-regression/Jenkinsfile
@@ -46,6 +46,7 @@ pipeline {
         CORDA_ARTIFACTORY_USERNAME = "${env.ARTIFACTORY_CREDENTIALS_USR}"
         CORDA_USE_CACHE = "corda-remotes"
         JAVA_HOME = "/usr/lib/jvm/java-17-amazon-corretto"
+        JAVA_8_HOME = "/usr/lib/jvm/java-1.8.0-amazon-corretto"
     }
 
     stages {
diff --git a/.ci/dev/regression/Jenkinsfile b/.ci/dev/regression/Jenkinsfile
index d66d647198..9d299ea1a2 100644
--- a/.ci/dev/regression/Jenkinsfile
+++ b/.ci/dev/regression/Jenkinsfile
@@ -71,6 +71,7 @@ pipeline {
         SNYK_TOKEN = credentials('c4-os-snyk-api-token-secret') //Jenkins credential type: Secret text
         C4_OS_SNYK_ORG_ID = credentials('corda4-os-snyk-org-id')
         JAVA_HOME = "/usr/lib/jvm/java-17-amazon-corretto"
+        JAVA_8_HOME = "/usr/lib/jvm/java-1.8.0-amazon-corretto"
     }
 
     stages {
diff --git a/Jenkinsfile b/Jenkinsfile
index 8cdbb77c9e..b5013bbb77 100644
--- a/Jenkinsfile
+++ b/Jenkinsfile
@@ -54,6 +54,7 @@ pipeline {
         CORDA_GRADLE_SCAN_KEY = credentials('gradle-build-scans-key')
         CORDA_USE_CACHE = "corda-remotes"
         JAVA_HOME="/usr/lib/jvm/java-17-amazon-corretto"
+        JAVA_8_HOME = "/usr/lib/jvm/java-1.8.0-amazon-corretto"
     }
 
     stages {
diff --git a/client/rpc/src/smoke-test/java/net/corda/java/rpc/StandaloneCordaRPCJavaClientTest.java b/client/rpc/src/smoke-test/java/net/corda/java/rpc/StandaloneCordaRPCJavaClientTest.java
index 2c17ac72e3..bb06d45679 100644
--- a/client/rpc/src/smoke-test/java/net/corda/java/rpc/StandaloneCordaRPCJavaClientTest.java
+++ b/client/rpc/src/smoke-test/java/net/corda/java/rpc/StandaloneCordaRPCJavaClientTest.java
@@ -1,6 +1,5 @@
 package net.corda.java.rpc;
 
-import net.corda.client.rpc.CordaRPCConnection;
 import net.corda.core.contracts.Amount;
 import net.corda.core.identity.CordaX500Name;
 import net.corda.core.identity.Party;
@@ -10,88 +9,48 @@ import net.corda.core.utilities.OpaqueBytes;
 import net.corda.finance.flows.AbstractCashFlow;
 import net.corda.finance.flows.CashIssueFlow;
 import net.corda.nodeapi.internal.config.User;
-import net.corda.smoketesting.NodeConfig;
+import net.corda.smoketesting.NodeParams;
 import net.corda.smoketesting.NodeProcess;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.*;
+import java.util.Currency;
+import java.util.HashSet;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Stream;
 
 import static java.util.Collections.singletonList;
 import static kotlin.test.AssertionsKt.assertEquals;
-import static kotlin.test.AssertionsKt.fail;
 import static net.corda.finance.workflows.GetBalances.getCashBalance;
+import static net.corda.kotlin.rpc.StandaloneCordaRPClientTest.gatherCordapps;
 
 public class StandaloneCordaRPCJavaClientTest {
+    private final User superUser = new User("superUser", "test", new HashSet<>(singletonList("ALL")));
 
-    public static void copyCordapps(NodeProcess.Factory factory, NodeConfig notaryConfig) {
-        Path cordappsDir = (factory.baseDirectory(notaryConfig).resolve(NodeProcess.CORDAPPS_DIR_NAME));
-        try {
-            Files.createDirectories(cordappsDir);
-        } catch (IOException ex) {
-            fail("Failed to create directories");
-        }
-        try (Stream<Path> paths = Files.walk(Paths.get("build", "resources", "smokeTest"))) {
-            paths.filter(path -> path.toFile().getName().startsWith("cordapp")).forEach(file -> {
-                try {
-                    Files.copy(file, cordappsDir.resolve(file.getFileName()));
-                } catch (IOException ex) {
-                    fail("Failed to copy cordapp jar");
-                }
-            });
-        } catch (IOException e) {
-            fail("Failed to walk files");
-        }
-    }
+    private final AtomicInteger port = new AtomicInteger(15000);
+    private final NodeProcess.Factory factory = new NodeProcess.Factory();
 
-    private List<String> perms = singletonList("ALL");
-    private Set<String> permSet = new HashSet<>(perms);
-    private User superUser = new User("superUser", "test", permSet);
-
-    private AtomicInteger port = new AtomicInteger(15000);
-
-    private NodeProcess notary;
     private CordaRPCOps rpcProxy;
-    private CordaRPCConnection connection;
     private Party notaryNodeIdentity;
 
-    private NodeConfig notaryConfig = new NodeConfig(
-            new CordaX500Name("Notary Service", "Zurich", "CH"),
-            port.getAndIncrement(),
-            port.getAndIncrement(),
-            port.getAndIncrement(),
-            true,
-            singletonList(superUser),
-            true
-    );
-
     @Before
     public void setUp() {
-        NodeProcess.Factory factory = new NodeProcess.Factory();
-        copyCordapps(factory, notaryConfig);
-        notary = factory.create(notaryConfig);
-        connection = notary.connect(superUser);
-        rpcProxy = connection.getProxy();
+        NodeProcess notary = factory.createNotaries(new NodeParams(
+                new CordaX500Name("Notary Service", "Zurich", "CH"),
+                port.getAndIncrement(),
+                port.getAndIncrement(),
+                port.getAndIncrement(),
+                singletonList(superUser),
+                gatherCordapps()
+        )).get(0);
+        rpcProxy = notary.connect(superUser).getProxy();
         notaryNodeIdentity = rpcProxy.nodeInfo().getLegalIdentities().get(0);
     }
 
     @After
     public void done() {
-        try {
-            connection.close();
-        } finally {
-            if (notary != null) {
-                notary.close();
-            }
-        }
+        factory.close();
     }
 
     @Test
diff --git a/client/rpc/src/smoke-test/kotlin/net/corda/kotlin/rpc/StandaloneCordaRPClientTest.kt b/client/rpc/src/smoke-test/kotlin/net/corda/kotlin/rpc/StandaloneCordaRPClientTest.kt
index f63d67a467..9c08d7a48d 100644
--- a/client/rpc/src/smoke-test/kotlin/net/corda/kotlin/rpc/StandaloneCordaRPClientTest.kt
+++ b/client/rpc/src/smoke-test/kotlin/net/corda/kotlin/rpc/StandaloneCordaRPClientTest.kt
@@ -27,6 +27,7 @@ import net.corda.core.utilities.getOrThrow
 import net.corda.core.utilities.minutes
 import net.corda.core.utilities.seconds
 import net.corda.finance.DOLLARS
+import net.corda.finance.GBP
 import net.corda.finance.POUNDS
 import net.corda.finance.SWISS_FRANCS
 import net.corda.finance.USD
@@ -35,14 +36,15 @@ import net.corda.finance.flows.CashIssueFlow
 import net.corda.finance.flows.CashPaymentFlow
 import net.corda.finance.workflows.getCashBalance
 import net.corda.finance.workflows.getCashBalances
-import net.corda.java.rpc.StandaloneCordaRPCJavaClientTest
 import net.corda.nodeapi.internal.config.User
 import net.corda.sleeping.SleepingFlow
-import net.corda.smoketesting.NodeConfig
+import net.corda.smoketesting.NodeParams
 import net.corda.smoketesting.NodeProcess
 import org.hamcrest.text.MatchesPattern
 import org.junit.After
+import org.junit.AfterClass
 import org.junit.Before
+import org.junit.BeforeClass
 import org.junit.Ignore
 import org.junit.Rule
 import org.junit.Test
@@ -50,17 +52,19 @@ import org.junit.rules.ExpectedException
 import java.io.FilterInputStream
 import java.io.InputStream
 import java.io.OutputStream.nullOutputStream
-import java.util.Currency
+import java.nio.file.Path
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.regex.Pattern
+import kotlin.io.path.Path
+import kotlin.io.path.listDirectoryEntries
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
 import kotlin.test.assertNotEquals
 import kotlin.test.assertTrue
 
 class StandaloneCordaRPClientTest {
-    private companion object {
+    companion object {
         private val log = contextLogger()
         val superUser = User("superUser", "test", permissions = setOf("ALL"))
         val nonUser = User("nonUser", "test", permissions = emptySet())
@@ -69,46 +73,57 @@ class StandaloneCordaRPClientTest {
         val port = AtomicInteger(15200)
         const val ATTACHMENT_SIZE = 2116
         val timeout = 60.seconds
+
+        private val factory = NodeProcess.Factory()
+
+        private lateinit var notary: NodeProcess
+
+        private val notaryConfig = NodeParams(
+                legalName = CordaX500Name(organisation = "Notary Service", locality = "Zurich", country = "CH"),
+                p2pPort = port.andIncrement,
+                rpcPort = port.andIncrement,
+                rpcAdminPort = port.andIncrement,
+                users = listOf(superUser, nonUser, rpcUser, flowUser),
+                cordappJars = gatherCordapps()
+        )
+
+        @BeforeClass
+        @JvmStatic
+        fun startNotary() {
+            notary = factory.createNotaries(notaryConfig)[0]
+        }
+
+        @AfterClass
+        @JvmStatic
+        fun close() {
+            factory.close()
+        }
+
+        @JvmStatic
+        fun gatherCordapps(): List<Path> {
+            return Path("build", "resources", "smokeTest").listDirectoryEntries("cordapp*.jar")
+        }
     }
 
-    private lateinit var factory: NodeProcess.Factory
-    private lateinit var notary: NodeProcess
-    private lateinit var rpcProxy: CordaRPCOps
     private lateinit var connection: CordaRPCConnection
-    private lateinit var notaryNode: NodeInfo
+    private lateinit var rpcProxy: CordaRPCOps
     private lateinit var notaryNodeIdentity: Party
 
-    private val notaryConfig = NodeConfig(
-            legalName = CordaX500Name(organisation = "Notary Service", locality = "Zurich", country = "CH"),
-            p2pPort = port.andIncrement,
-            rpcPort = port.andIncrement,
-            rpcAdminPort = port.andIncrement,
-            isNotary = true,
-            users = listOf(superUser, nonUser, rpcUser, flowUser)
-    )
-
     @get:Rule
     val exception: ExpectedException = ExpectedException.none()
 
     @Before
     fun setUp() {
-        factory = NodeProcess.Factory()
-        StandaloneCordaRPCJavaClientTest.copyCordapps(factory, notaryConfig)
-        notary = factory.create(notaryConfig)
         connection = notary.connect(superUser)
         rpcProxy = connection.proxy
-        notaryNode = fetchNotaryIdentity()
         notaryNodeIdentity = rpcProxy.nodeInfo().legalIdentitiesAndCerts.first().party
     }
 
     @After
-    fun done() {
-        connection.use {
-            notary.close()
-        }
+    fun closeConnection() {
+        connection.close()
     }
 
-
     @Test(timeout=300_000)
 	fun `test attachments`() {
         val attachment = InputStreamAndHash.createInMemoryTestZip(ATTACHMENT_SIZE, 1)
@@ -168,8 +183,7 @@ class StandaloneCordaRPClientTest {
 
     @Test(timeout=300_000)
 	fun `test state machines`() {
-        val (stateMachines, updates) = rpcProxy.stateMachinesFeed()
-        assertEquals(0, stateMachines.size)
+        val (_, updates) = rpcProxy.stateMachinesFeed()
 
         val updateLatch = CountDownLatch(1)
         val updateCount = AtomicInteger(0)
@@ -190,8 +204,9 @@ class StandaloneCordaRPClientTest {
 
     @Test(timeout=300_000)
 	fun `test vault track by`() {
-        val (vault, vaultUpdates) = rpcProxy.vaultTrackBy<Cash.State>(paging = PageSpecification(DEFAULT_PAGE_NUM))
-        assertEquals(0, vault.totalStatesAvailable)
+        val initialGbpBalance = rpcProxy.getCashBalance(GBP)
+
+        val (_, vaultUpdates) = rpcProxy.vaultTrackBy<Cash.State>(paging = PageSpecification(DEFAULT_PAGE_NUM))
 
         val updateLatch = CountDownLatch(1)
         vaultUpdates.subscribe { update ->
@@ -207,34 +222,35 @@ class StandaloneCordaRPClientTest {
         // Check that this cash exists in the vault
         val cashBalance = rpcProxy.getCashBalances()
         log.info("Cash Balances: $cashBalance")
-        assertEquals(1, cashBalance.size)
-        assertEquals(629.POUNDS, cashBalance[Currency.getInstance("GBP")])
+        assertEquals(629.POUNDS, cashBalance[GBP]!! - initialGbpBalance)
     }
 
     @Test(timeout=300_000)
 	fun `test vault query by`() {
-        // Now issue some cash
-        rpcProxy.startFlow(::CashIssueFlow, 629.POUNDS, OpaqueBytes.of(0), notaryNodeIdentity)
-                .returnValue.getOrThrow(timeout)
-
         val criteria = QueryCriteria.VaultQueryCriteria(status = Vault.StateStatus.ALL)
         val paging = PageSpecification(DEFAULT_PAGE_NUM, 10)
         val sorting = Sort(setOf(Sort.SortColumn(SortAttribute.Standard(Sort.VaultStateAttribute.RECORDED_TIME), Sort.Direction.DESC)))
 
+        val initialStateCount = rpcProxy.vaultQueryBy<Cash.State>(criteria, paging, sorting).totalStatesAvailable
+        val initialGbpBalance = rpcProxy.getCashBalance(GBP)
+
+        // Now issue some cash
+        rpcProxy.startFlow(::CashIssueFlow, 629.POUNDS, OpaqueBytes.of(0), notaryNodeIdentity)
+                .returnValue.getOrThrow(timeout)
+
         val queryResults = rpcProxy.vaultQueryBy<Cash.State>(criteria, paging, sorting)
-        assertEquals(1, queryResults.totalStatesAvailable)
+        assertEquals(1, queryResults.totalStatesAvailable - initialStateCount)
         assertEquals(queryResults.states.first().state.data.amount.quantity, 629.POUNDS.quantity)
 
         rpcProxy.startFlow(::CashPaymentFlow, 100.POUNDS, notaryNodeIdentity, true, notaryNodeIdentity).returnValue.getOrThrow()
 
         val moreResults = rpcProxy.vaultQueryBy<Cash.State>(criteria, paging, sorting)
-        assertEquals(3, moreResults.totalStatesAvailable)   // 629 - 100 + 100
+        assertEquals(3, moreResults.totalStatesAvailable - initialStateCount)   // 629 - 100 + 100
 
         // Check that this cash exists in the vault
         val cashBalances = rpcProxy.getCashBalances()
         log.info("Cash Balances: $cashBalances")
-        assertEquals(1, cashBalances.size)
-        assertEquals(629.POUNDS, cashBalances[Currency.getInstance("GBP")])
+        assertEquals(629.POUNDS, cashBalances[GBP]!! - initialGbpBalance)
     }
 
     @Test(timeout=300_000)
diff --git a/core-tests/build.gradle b/core-tests/build.gradle
index 8d9150999c..4f414ee2ea 100644
--- a/core-tests/build.gradle
+++ b/core-tests/build.gradle
@@ -9,11 +9,13 @@ configurations {
     integrationTestImplementation.extendsFrom testImplementation
     integrationTestRuntimeOnly.extendsFrom testRuntimeOnly
 
-    smokeTestCompile.extendsFrom compile
+    smokeTestImplementation.extendsFrom implementation
     smokeTestRuntimeOnly.extendsFrom runtimeOnly
-}
 
-evaluationDependsOn(':node:capsule')
+    testArtifacts.extendsFrom testRuntimeOnlyClasspath
+
+    corda4_11
+}
 
 sourceSets {
     integrationTest {
@@ -42,9 +44,17 @@ sourceSets {
 
 processSmokeTestResources {
     // Bring in the fully built corda.jar for use by NodeFactory in the smoke tests
-    from(project(":node:capsule").tasks['buildCordaJAR']) {
+    from(tasks.getByPath(":node:capsule:buildCordaJAR")) {
         rename 'corda-(.*)', 'corda.jar'
     }
+    from(tasks.getByPath(":finance:workflows:jar")) {
+        rename 'corda-finance-workflows-.*.jar', 'corda-finance-workflows.jar'
+    }
+    from(tasks.getByPath(":finance:contracts:jar")) {
+        rename 'corda-finance-contracts-.*.jar', 'corda-finance-contracts.jar'
+    }
+    from(tasks.getByPath(":testing:cordapps:4.11-workflows:jar"))
+    from(configurations.corda4_11)
 }
 
 dependencies {
@@ -69,7 +79,6 @@ dependencies {
     testImplementation project(":test-utils")
     testImplementation project(path: ':core', configuration: 'testArtifacts')
 
-
     // Guava: Google test library (collections test suite)
     testImplementation "com.google.guava:guava-testlib:$guava_version"
     testImplementation "com.google.jimfs:jimfs:1.1"
@@ -98,7 +107,12 @@ dependencies {
     smokeTestImplementation project(":core")
     smokeTestImplementation project(":node-api")
     smokeTestImplementation project(":client:rpc")
-
+    smokeTestImplementation project(':smoke-test-utils')
+    smokeTestImplementation project(':core-test-utils')
+    smokeTestImplementation project(":finance:contracts")
+    smokeTestImplementation project(":finance:workflows")
+    smokeTestImplementation project(":testing:cordapps:4.11-workflows")
+    smokeTestImplementation "org.assertj:assertj-core:${assertj_version}"
     smokeTestImplementation "org.bouncycastle:bcprov-jdk18on:${bouncycastle_version}"
     smokeTestImplementation "co.paralleluniverse:quasar-core:$quasar_version"
     smokeTestImplementation "org.junit.jupiter:junit-jupiter-api:${junit_jupiter_version}"
@@ -109,15 +123,12 @@ dependencies {
     smokeTestRuntimeOnly "org.junit.platform:junit-platform-launcher:${junit_platform_version}"
     smokeTestRuntimeOnly "org.slf4j:slf4j-simple:$slf4j_version"
 
-    smokeTestCompile project(':smoke-test-utils')
-    smokeTestCompile "org.assertj:assertj-core:${assertj_version}"
-
     // used by FinalityFlowTests
     testImplementation project(':testing:cordapps:cashobservers')
-}
 
-configurations {
-    testArtifacts.extendsFrom testRuntimeOnlyClasspath
+    corda4_11 "net.corda:corda-finance-contracts:4.11"
+    corda4_11 "net.corda:corda-finance-workflows:4.11"
+    corda4_11 "net.corda:corda:4.11"
 }
 
 tasks.withType(Test).configureEach {
@@ -125,22 +136,24 @@ tasks.withType(Test).configureEach {
     forkEvery = 10
 }
 
-task testJar(type: Jar) {
+tasks.register('testJar', Jar) {
     classifier "tests"
     from sourceSets.test.output
 }
 
-task integrationTest(type: Test) {
+tasks.register('integrationTest', Test) {
     testClassesDirs = sourceSets.integrationTest.output.classesDirs
     classpath = sourceSets.integrationTest.runtimeClasspath
 }
 
-task smokeTestJar(type: Jar) {
+tasks.register('smokeTestJar', Jar) {
     classifier 'smokeTests'
-    from sourceSets.smokeTest.output
+    from(sourceSets.smokeTest.output) {
+        exclude("*.jar")
+    }
 }
 
-task smokeTest(type: Test) {
+tasks.register('smokeTest', Test) {
     dependsOn smokeTestJar
     testClassesDirs = sourceSets.smokeTest.output.classesDirs
     classpath = sourceSets.smokeTest.runtimeClasspath
diff --git a/core-tests/src/smoke-test/kotlin/net/corda/coretests/NodeVersioningTest.kt b/core-tests/src/smoke-test/kotlin/net/corda/coretests/NodeVersioningTest.kt
index c9f5d9f15d..e290b19644 100644
--- a/core-tests/src/smoke-test/kotlin/net/corda/coretests/NodeVersioningTest.kt
+++ b/core-tests/src/smoke-test/kotlin/net/corda/coretests/NodeVersioningTest.kt
@@ -5,11 +5,10 @@ import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.StartableByRPC
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.internal.PLATFORM_VERSION
-import net.corda.core.internal.copyToDirectory
 import net.corda.core.messaging.startFlow
 import net.corda.core.utilities.getOrThrow
 import net.corda.nodeapi.internal.config.User
-import net.corda.smoketesting.NodeConfig
+import net.corda.smoketesting.NodeParams
 import net.corda.smoketesting.NodeProcess
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.After
@@ -18,8 +17,6 @@ import org.junit.Test
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.jar.JarFile
 import kotlin.io.path.Path
-import kotlin.io.path.createDirectories
-import kotlin.io.path.div
 import kotlin.io.path.listDirectoryEntries
 
 class NodeVersioningTest {
@@ -30,56 +27,39 @@ class NodeVersioningTest {
 
     private val factory = NodeProcess.Factory()
 
-    private val notaryConfig = NodeConfig(
-            legalName = CordaX500Name(organisation = "Notary Service", locality = "Zurich", country = "CH"),
-            p2pPort = port.andIncrement,
-            rpcPort = port.andIncrement,
-            rpcAdminPort = port.andIncrement,
-            isNotary = true,
-            users = listOf(superUser)
-    )
-
-    private val aliceConfig = NodeConfig(
-            legalName = CordaX500Name(organisation = "Alice Corp", locality = "Madrid", country = "ES"),
-            p2pPort = port.andIncrement,
-            rpcPort = port.andIncrement,
-            rpcAdminPort = port.andIncrement,
-            isNotary = false,
-            users = listOf(superUser)
-    )
-
-    private var notary: NodeProcess? = null
+    private lateinit var notary: NodeProcess
 
     @Before
-    fun setUp() {
-        notary = factory.create(notaryConfig)
+    fun startNotary() {
+        notary = factory.createNotaries(NodeParams(
+                legalName = CordaX500Name(organisation = "Notary Service", locality = "Zurich", country = "CH"),
+                p2pPort = port.andIncrement,
+                rpcPort = port.andIncrement,
+                rpcAdminPort = port.andIncrement,
+                users = listOf(superUser),
+                // Find the jar file for the smoke tests of this module
+                cordappJars = Path("build", "libs").listDirectoryEntries("*-smokeTests*")
+        ))[0]
     }
 
     @After
     fun done() {
-        notary?.close()
+        factory.close()
     }
 
     @Test(timeout=300_000)
 	fun `platform version in manifest file`() {
-        val manifest = JarFile(factory.cordaJar.toFile()).manifest
+        val manifest = JarFile(NodeProcess.Factory.getCordaJar().toFile()).manifest
         assertThat(manifest.mainAttributes.getValue("Corda-Platform-Version").toInt()).isEqualTo(PLATFORM_VERSION)
     }
 
     @Test(timeout=300_000)
 	fun `platform version from RPC`() {
-        val cordappsDir = (factory.baseDirectory(aliceConfig) / NodeProcess.CORDAPPS_DIR_NAME).createDirectories()
-        // Find the jar file for the smoke tests of this module
-        val selfCordapp = Path("build", "libs").listDirectoryEntries("*-smokeTests*").single()
-        selfCordapp.copyToDirectory(cordappsDir)
-
-        factory.create(aliceConfig).use { alice ->
-            alice.connect(superUser).use {
-                val rpc = it.proxy
-                assertThat(rpc.protocolVersion).isEqualTo(PLATFORM_VERSION)
-                assertThat(rpc.nodeInfo().platformVersion).isEqualTo(PLATFORM_VERSION)
-                assertThat(rpc.startFlow(NodeVersioningTest::GetPlatformVersionFlow).returnValue.getOrThrow()).isEqualTo(PLATFORM_VERSION)
-            }
+        notary.connect(superUser).use {
+            val rpc = it.proxy
+            assertThat(rpc.protocolVersion).isEqualTo(PLATFORM_VERSION)
+            assertThat(rpc.nodeInfo().platformVersion).isEqualTo(PLATFORM_VERSION)
+            assertThat(rpc.startFlow(NodeVersioningTest::GetPlatformVersionFlow).returnValue.getOrThrow()).isEqualTo(PLATFORM_VERSION)
         }
     }
 
diff --git a/core-tests/src/smoke-test/kotlin/net/corda/coretests/cordapp/CordappSmokeTest.kt b/core-tests/src/smoke-test/kotlin/net/corda/coretests/cordapp/CordappSmokeTest.kt
index 12147bafd8..d4b2fbda44 100644
--- a/core-tests/src/smoke-test/kotlin/net/corda/coretests/cordapp/CordappSmokeTest.kt
+++ b/core-tests/src/smoke-test/kotlin/net/corda/coretests/cordapp/CordappSmokeTest.kt
@@ -12,7 +12,6 @@ import net.corda.core.flows.StartableByRPC
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.identity.Party
 import net.corda.core.identity.PartyAndCertificate
-import net.corda.core.internal.copyToDirectory
 import net.corda.core.messaging.startFlow
 import net.corda.core.node.NodeInfo
 import net.corda.core.serialization.serialize
@@ -26,9 +25,8 @@ import net.corda.nodeapi.internal.config.User
 import net.corda.nodeapi.internal.createDevNodeCa
 import net.corda.nodeapi.internal.crypto.CertificateType
 import net.corda.nodeapi.internal.crypto.X509Utilities
-import net.corda.smoketesting.NodeConfig
+import net.corda.smoketesting.NodeParams
 import net.corda.smoketesting.NodeProcess
-import net.corda.smoketesting.NodeProcess.Companion.CORDAPPS_DIR_NAME
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.After
 import org.junit.Before
@@ -41,8 +39,8 @@ import java.util.concurrent.atomic.AtomicInteger
 import kotlin.io.path.Path
 import kotlin.io.path.createDirectories
 import kotlin.io.path.div
+import kotlin.io.path.listDirectoryEntries
 import kotlin.io.path.name
-import kotlin.io.path.useDirectoryEntries
 import kotlin.io.path.writeBytes
 
 class CordappSmokeTest {
@@ -53,43 +51,35 @@ class CordappSmokeTest {
 
     private val factory = NodeProcess.Factory()
 
-    private val notaryConfig = NodeConfig(
-            legalName = CordaX500Name(organisation = "Notary Service", locality = "Zurich", country = "CH"),
-            p2pPort = port.andIncrement,
-            rpcPort = port.andIncrement,
-            rpcAdminPort = port.andIncrement,
-            isNotary = true,
-            users = listOf(superUser)
-    )
-
-    private val aliceConfig = NodeConfig(
+    private val aliceConfig = NodeParams(
             legalName = CordaX500Name(organisation = "Alice Corp", locality = "Madrid", country = "ES"),
             p2pPort = port.andIncrement,
             rpcPort = port.andIncrement,
             rpcAdminPort = port.andIncrement,
-            isNotary = false,
-            users = listOf(superUser)
+            users = listOf(superUser),
+            // Find the jar file for the smoke tests of this module
+            cordappJars = Path("build", "libs").listDirectoryEntries("*-smokeTests*")
     )
 
-    private lateinit var notary: NodeProcess
-
     @Before
-    fun setUp() {
-        notary = factory.create(notaryConfig)
+    fun startNotary() {
+        factory.createNotaries(NodeParams(
+                legalName = CordaX500Name(organisation = "Notary Service", locality = "Zurich", country = "CH"),
+                p2pPort = port.andIncrement,
+                rpcPort = port.andIncrement,
+                rpcAdminPort = port.andIncrement,
+                users = listOf(superUser)
+        ))
     }
 
     @After
     fun done() {
-        notary.close()
+        factory.close()
     }
 
     @Test(timeout=300_000)
 	fun `FlowContent appName returns the filename of the CorDapp jar`() {
         val baseDir = factory.baseDirectory(aliceConfig)
-        val cordappsDir = (baseDir / CORDAPPS_DIR_NAME).createDirectories()
-        // Find the jar file for the smoke tests of this module
-        val selfCordapp = Path("build", "libs").useDirectoryEntries { it.single { "-smokeTests" in it.toString() } }
-        selfCordapp.copyToDirectory(cordappsDir)
 
         // The `nodeReadyFuture` in the persistent network map cache will not complete unless there is at least one other
         // node in the network. We work around this limitation by putting another node info file in the additional-node-info
@@ -98,24 +88,17 @@ class CordappSmokeTest {
         val additionalNodeInfoDir = (baseDir / "additional-node-infos").createDirectories()
         createDummyNodeInfo(additionalNodeInfoDir)
 
-        factory.create(aliceConfig).use { alice ->
-            alice.connect(superUser).use { connectionToAlice ->
-                val aliceIdentity = connectionToAlice.proxy.nodeInfo().legalIdentitiesAndCerts.first().party
-                val future = connectionToAlice.proxy.startFlow(CordappSmokeTest::GatherContextsFlow, aliceIdentity).returnValue
-                val (sessionInitContext, sessionConfirmContext) = future.getOrThrow()
-                val selfCordappName = selfCordapp.name.removeSuffix(".jar")
-                assertThat(sessionInitContext.appName).isEqualTo(selfCordappName)
-                assertThat(sessionConfirmContext.appName).isEqualTo(selfCordappName)
-            }
+        val alice = factory.createNode(aliceConfig)
+        alice.connect(superUser).use { connectionToAlice ->
+            val aliceIdentity = connectionToAlice.proxy.nodeInfo().legalIdentitiesAndCerts.first().party
+            val future = connectionToAlice.proxy.startFlow(CordappSmokeTest::GatherContextsFlow, aliceIdentity).returnValue
+            val (sessionInitContext, sessionConfirmContext) = future.getOrThrow()
+            val selfCordappName = aliceConfig.cordappJars[0].name.removeSuffix(".jar")
+            assertThat(sessionInitContext.appName).isEqualTo(selfCordappName)
+            assertThat(sessionConfirmContext.appName).isEqualTo(selfCordappName)
         }
     }
 
-    @Test(timeout=300_000)
-	fun `empty cordapps directory`() {
-        (factory.baseDirectory(aliceConfig) / CORDAPPS_DIR_NAME).createDirectories()
-        factory.create(aliceConfig).close()
-    }
-
     @InitiatingFlow
     @StartableByRPC
     class GatherContextsFlow(private val otherParty: Party) : FlowLogic<Pair<FlowInfo, FlowInfo>>() {
diff --git a/core-tests/src/smoke-test/kotlin/net/corda/coretests/verification/ExternalVerificationTests.kt b/core-tests/src/smoke-test/kotlin/net/corda/coretests/verification/ExternalVerificationTests.kt
new file mode 100644
index 0000000000..c62690e14e
--- /dev/null
+++ b/core-tests/src/smoke-test/kotlin/net/corda/coretests/verification/ExternalVerificationTests.kt
@@ -0,0 +1,247 @@
+package net.corda.coretests.verification
+
+import co.paralleluniverse.strands.concurrent.CountDownLatch
+import net.corda.client.rpc.CordaRPCClientConfiguration
+import net.corda.client.rpc.notUsed
+import net.corda.core.contracts.Amount
+import net.corda.core.crypto.SecureHash
+import net.corda.core.flows.UnexpectedFlowEndException
+import net.corda.core.identity.CordaX500Name
+import net.corda.core.identity.Party
+import net.corda.core.internal.PlatformVersionSwitches.MIGRATE_ATTACHMENT_TO_SIGNATURE_CONSTRAINTS
+import net.corda.core.internal.toPath
+import net.corda.core.messaging.CordaRPCOps
+import net.corda.core.messaging.startFlow
+import net.corda.core.node.NodeInfo
+import net.corda.core.utilities.OpaqueBytes
+import net.corda.core.utilities.getOrThrow
+import net.corda.finance.DOLLARS
+import net.corda.finance.flows.AbstractCashFlow
+import net.corda.finance.flows.CashIssueFlow
+import net.corda.finance.flows.CashPaymentFlow
+import net.corda.nodeapi.internal.config.User
+import net.corda.smoketesting.NodeParams
+import net.corda.smoketesting.NodeProcess
+import net.corda.testing.common.internal.testNetworkParameters
+import net.corda.testing.cordapps.workflows411.IssueAndChangeNotaryFlow
+import net.corda.testing.core.DUMMY_NOTARY_NAME
+import net.corda.testing.core.internal.JarSignatureTestUtils.unsignJar
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatExceptionOfType
+import org.junit.AfterClass
+import org.junit.BeforeClass
+import org.junit.Test
+import java.net.InetAddress
+import java.nio.file.Path
+import java.util.Currency
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.io.path.Path
+import kotlin.io.path.copyTo
+import kotlin.io.path.div
+import kotlin.io.path.listDirectoryEntries
+import kotlin.io.path.name
+import kotlin.io.path.readText
+
+class ExternalVerificationSignedCordappsTest {
+    private companion object {
+        private val factory = NodeProcess.Factory(testNetworkParameters(minimumPlatformVersion = MIGRATE_ATTACHMENT_TO_SIGNATURE_CONSTRAINTS))
+
+        private lateinit var notaries: List<NodeProcess>
+        private lateinit var oldNode: NodeProcess
+        private lateinit var newNode: NodeProcess
+
+        @BeforeClass
+        @JvmStatic
+        fun startNodes() {
+            // The 4.11 finance CorDapp jars
+            val oldCordapps = listOf("contracts", "workflows").map { smokeTestResource("corda-finance-$it-4.11.jar") }
+            // The current version finance CorDapp jars
+            val newCordapps = listOf("contracts", "workflows").map { smokeTestResource("corda-finance-$it.jar") }
+
+            notaries = factory.createNotaries(
+                    nodeParams(DUMMY_NOTARY_NAME, oldCordapps),
+                    nodeParams(CordaX500Name("Notary Service 2", "Zurich", "CH"), newCordapps)
+            )
+            oldNode = factory.createNode(nodeParams(
+                    CordaX500Name("Old", "Delhi", "IN"),
+                    oldCordapps + listOf(smokeTestResource("4.11-workflows-cordapp.jar")),
+                    CordaRPCClientConfiguration(minimumServerProtocolVersion = 13),
+                    version = "4.11"
+            ))
+            newNode = factory.createNode(nodeParams(CordaX500Name("New", "York", "US"), newCordapps))
+        }
+
+        @AfterClass
+        @JvmStatic
+        fun close() {
+            factory.close()
+        }
+    }
+
+    @Test(timeout=300_000)
+    fun `transaction containing 4_11 contract sent to new node`() {
+        assertCashIssuanceAndPayment(issuer = oldNode, recipient = newNode)
+    }
+
+    @Test(timeout=300_000)
+    fun `notary change transaction`() {
+        val oldRpc = oldNode.connect(superUser).proxy
+        val oldNodeInfo = oldRpc.nodeInfo()
+        val notaryIdentities = oldRpc.notaryIdentities()
+        for (notary in notaries) {
+            notary.connect(superUser).use { it.proxy.waitForVisibility(oldNodeInfo) }
+        }
+        oldRpc.startFlow(::IssueAndChangeNotaryFlow, notaryIdentities[0], notaryIdentities[1]).returnValue.getOrThrow()
+    }
+
+    private fun assertCashIssuanceAndPayment(issuer: NodeProcess, recipient: NodeProcess) {
+        val issuerRpc = issuer.connect(superUser).proxy
+        val recipientRpc = recipient.connect(superUser).proxy
+        val recipientNodeInfo = recipientRpc.nodeInfo()
+        val notaryIdentity = issuerRpc.notaryIdentities()[0]
+
+        val (issuanceTx) = issuerRpc.startFlow(
+                ::CashIssueFlow,
+                10.DOLLARS,
+                OpaqueBytes.of(0x01),
+                notaryIdentity
+        ).returnValue.getOrThrow()
+
+        issuerRpc.waitForVisibility(recipientNodeInfo)
+        recipientRpc.waitForVisibility(issuerRpc.nodeInfo())
+
+        val (paymentTx) = issuerRpc.startFlow(
+                ::CashPaymentFlow,
+                10.DOLLARS,
+                recipientNodeInfo.legalIdentities[0],
+                false,
+        ).returnValue.getOrThrow()
+
+        notaries[0].assertTransactionsWereVerifiedExternally(issuanceTx.id, paymentTx.id)
+        recipient.assertTransactionsWereVerifiedExternally(issuanceTx.id, paymentTx.id)
+    }
+}
+
+class ExternalVerificationUnsignedCordappsTest {
+    private companion object {
+        private val factory = NodeProcess.Factory(testNetworkParameters(minimumPlatformVersion = MIGRATE_ATTACHMENT_TO_SIGNATURE_CONSTRAINTS))
+
+        private lateinit var notary: NodeProcess
+        private lateinit var oldNode: NodeProcess
+        private lateinit var newNode: NodeProcess
+
+        @BeforeClass
+        @JvmStatic
+        fun startNodes() {
+            // The 4.11 finance CorDapp jars
+            val oldCordapps = listOf(unsignedResourceJar("corda-finance-contracts-4.11.jar"), smokeTestResource("corda-finance-workflows-4.11.jar"))
+            // The current version finance CorDapp jars
+            val newCordapps = listOf(unsignedResourceJar("corda-finance-contracts.jar"), smokeTestResource("corda-finance-workflows.jar"))
+
+            notary = factory.createNotaries(nodeParams(DUMMY_NOTARY_NAME, oldCordapps))[0]
+            oldNode = factory.createNode(nodeParams(
+                    CordaX500Name("Old", "Delhi", "IN"),
+                    oldCordapps,
+                    CordaRPCClientConfiguration(minimumServerProtocolVersion = 13),
+                    version = "4.11"
+            ))
+            newNode = factory.createNode(nodeParams(CordaX500Name("New", "York", "US"), newCordapps))
+        }
+
+        @AfterClass
+        @JvmStatic
+        fun close() {
+            factory.close()
+        }
+
+        private fun unsignedResourceJar(name: String): Path {
+            val signedJar = smokeTestResource(name)
+            val copy = signedJar.copyTo(Path("${signedJar.toString().substringBeforeLast(".")}-UNSIGNED.jar"), overwrite = true)
+            copy.unsignJar()
+            return copy
+        }
+    }
+
+    @Test(timeout = 300_000)
+    fun `transactions can fail verification in external verifier`() {
+        val issuerRpc = oldNode.connect(superUser).proxy
+        val recipientRpc = newNode.connect(superUser).proxy
+        val recipientNodeInfo = recipientRpc.nodeInfo()
+        val notaryIdentity = issuerRpc.notaryIdentities()[0]
+
+        val (issuanceTx) = issuerRpc.startFlow(
+                ::CashIssueFlow,
+                10.DOLLARS,
+                OpaqueBytes.of(0x01),
+                notaryIdentity
+        ).returnValue.getOrThrow()
+
+        issuerRpc.waitForVisibility(recipientNodeInfo)
+        recipientRpc.waitForVisibility(issuerRpc.nodeInfo())
+
+        assertThatExceptionOfType(UnexpectedFlowEndException::class.java).isThrownBy {
+            issuerRpc.startFlow<AbstractCashFlow.Result, Amount<Currency>, Party, Boolean, CashPaymentFlow>(
+                    ::CashPaymentFlow,
+                    10.DOLLARS,
+                    recipientNodeInfo.legalIdentities[0],
+                    false,
+            ).returnValue.getOrThrow()
+        }
+
+        assertThat(newNode.externalVerifierLogs()).contains("$issuanceTx failed to verify")
+    }
+}
+
+private val superUser = User("superUser", "test", permissions = setOf("ALL"))
+private val portCounter = AtomicInteger(15100)
+
+private fun smokeTestResource(name: String): Path = ExternalVerificationSignedCordappsTest::class.java.getResource("/$name")!!.toPath()
+
+private fun nodeParams(
+        legalName: CordaX500Name,
+        cordappJars: List<Path> = emptyList(),
+        clientRpcConfig: CordaRPCClientConfiguration = CordaRPCClientConfiguration.DEFAULT,
+        version: String? = null
+): NodeParams {
+    return NodeParams(
+            legalName = legalName,
+            p2pPort = portCounter.andIncrement,
+            rpcPort = portCounter.andIncrement,
+            rpcAdminPort = portCounter.andIncrement,
+            users = listOf(superUser),
+            cordappJars = cordappJars,
+            clientRpcConfig = clientRpcConfig,
+            version = version
+    )
+}
+
+private fun CordaRPCOps.waitForVisibility(other: NodeInfo) {
+    val (snapshot, updates) = networkMapFeed()
+    if (other in snapshot) {
+        updates.notUsed()
+    } else {
+        val found = CountDownLatch(1)
+        val subscription = updates.subscribe {
+            if (it.node == other) {
+                found.countDown()
+            }
+        }
+        found.await()
+        subscription.unsubscribe()
+    }
+}
+
+private fun NodeProcess.assertTransactionsWereVerifiedExternally(vararg txIds: SecureHash) {
+    val verifierLogContent = externalVerifierLogs()
+    for (txId in txIds) {
+        assertThat(verifierLogContent).contains("SignedTransaction(id=$txId) verified")
+    }
+}
+
+private fun NodeProcess.externalVerifierLogs(): String {
+    val verifierLogs = (nodeDir / "logs")
+            .listDirectoryEntries()
+            .filter { it.name == "verifier-${InetAddress.getLocalHost().hostName}.log" }
+    assertThat(verifierLogs).describedAs("External verifier was not started").hasSize(1)
+    return verifierLogs[0].readText()
+}
diff --git a/core/src/main/kotlin/net/corda/core/contracts/AttachmentConstraint.kt b/core/src/main/kotlin/net/corda/core/contracts/AttachmentConstraint.kt
index e7efccdde9..f989f1f2fa 100644
--- a/core/src/main/kotlin/net/corda/core/contracts/AttachmentConstraint.kt
+++ b/core/src/main/kotlin/net/corda/core/contracts/AttachmentConstraint.kt
@@ -11,6 +11,7 @@ import net.corda.core.internal.utilities.Internable
 import net.corda.core.internal.utilities.PrivateInterner
 import net.corda.core.serialization.CordaSerializable
 import net.corda.core.transactions.TransactionBuilder
+import net.corda.core.utilities.debug
 import net.corda.core.utilities.loggerFor
 import java.lang.annotation.Inherited
 import java.security.PublicKey
@@ -70,7 +71,7 @@ object WhitelistedByZoneAttachmentConstraint : AttachmentConstraint {
     override fun isSatisfiedBy(attachment: Attachment): Boolean {
         return if (attachment is AttachmentWithContext) {
             val whitelist = attachment.whitelistedContractImplementations
-            log.debug("Checking ${attachment.contract} is in CZ whitelist $whitelist")
+            log.debug { "Checking ${attachment.contract} is in CZ whitelist $whitelist" }
             attachment.id in (whitelist[attachment.contract] ?: emptyList())
         } else {
             log.warn("CZ whitelisted constraint check failed: ${attachment.id} not in CZ whitelist")
@@ -111,8 +112,8 @@ object AutomaticPlaceholderConstraint : AttachmentConstraint {
  */
 data class SignatureAttachmentConstraint(val key: PublicKey) : AttachmentConstraint {
     override fun isSatisfiedBy(attachment: Attachment): Boolean {
-        log.debug("Checking signature constraints: verifying $key in contract attachment signer keys: ${attachment.signerKeys}")
-        return if (!key.isFulfilledBy(attachment.signerKeys.map { it })) {
+        log.debug { "Checking signature constraints: verifying $key in contract attachment signer keys: ${attachment.signerKeys}" }
+        return if (!key.isFulfilledBy(attachment.signerKeys)) {
             log.warn("Untrusted signing key: expected $key. but contract attachment contains ${attachment.signerKeys}")
             false
         } else true
diff --git a/core/src/main/kotlin/net/corda/core/internal/InternalAttachment.kt b/core/src/main/kotlin/net/corda/core/internal/InternalAttachment.kt
new file mode 100644
index 0000000000..8214064bb2
--- /dev/null
+++ b/core/src/main/kotlin/net/corda/core/internal/InternalAttachment.kt
@@ -0,0 +1,28 @@
+package net.corda.core.internal
+
+import net.corda.core.contracts.Attachment
+import net.corda.core.contracts.ContractAttachment
+
+interface InternalAttachment : Attachment {
+    /**
+     * The version of the Kotlin metadata, if this attachment has one. See `kotlinx.metadata.jvm.JvmMetadataVersion` for more information on
+     * how this maps to the Kotlin language version.
+     */
+    val kotlinMetadataVersion: String?
+}
+
+/**
+ * Because [ContractAttachment] is public API, we can't make it implement [InternalAttachment] without also leaking it out.
+ *
+ * @see InternalAttachment.kotlinMetadataVersion
+ */
+val Attachment.kotlinMetadataVersion: String? get() {
+    var attachment = this
+    while (true) {
+        when (attachment) {
+            is InternalAttachment -> return attachment.kotlinMetadataVersion
+            is ContractAttachment -> attachment = attachment.attachment
+            else -> return null
+        }
+    }
+}
diff --git a/core/src/main/kotlin/net/corda/core/internal/InternalUtils.kt b/core/src/main/kotlin/net/corda/core/internal/InternalUtils.kt
index 490994e8dd..785af70002 100644
--- a/core/src/main/kotlin/net/corda/core/internal/InternalUtils.kt
+++ b/core/src/main/kotlin/net/corda/core/internal/InternalUtils.kt
@@ -16,6 +16,7 @@ import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.UntrustworthyData
 import net.corda.core.utilities.seconds
 import org.slf4j.Logger
+import org.slf4j.event.Level
 import rx.Observable
 import rx.Observer
 import rx.observers.Subscribers
@@ -619,5 +620,15 @@ fun Logger.warnOnce(warning: String) {
     }
 }
 
+val Logger.level: Level
+    get() = when {
+        isTraceEnabled -> Level.TRACE
+        isDebugEnabled -> Level.DEBUG
+        isInfoEnabled -> Level.INFO
+        isWarnEnabled -> Level.WARN
+        isErrorEnabled -> Level.ERROR
+        else -> throw IllegalStateException("Unknown logging level")
+    }
+
 const val JAVA_1_2_CLASS_FILE_FORMAT_MAJOR_VERSION = 46
 const val JAVA_17_CLASS_FILE_FORMAT_MAJOR_VERSION = 61
diff --git a/core/src/main/kotlin/net/corda/core/internal/NetworkParametersServiceInternal.kt b/core/src/main/kotlin/net/corda/core/internal/NetworkParametersStorage.kt
similarity index 100%
rename from core/src/main/kotlin/net/corda/core/internal/NetworkParametersServiceInternal.kt
rename to core/src/main/kotlin/net/corda/core/internal/NetworkParametersStorage.kt
diff --git a/core/src/main/kotlin/net/corda/core/internal/TransactionUtils.kt b/core/src/main/kotlin/net/corda/core/internal/TransactionUtils.kt
index a6aa9c2ac4..e07b50d020 100644
--- a/core/src/main/kotlin/net/corda/core/internal/TransactionUtils.kt
+++ b/core/src/main/kotlin/net/corda/core/internal/TransactionUtils.kt
@@ -268,5 +268,9 @@ internal fun checkNotaryWhitelisted(ftx: FullTransaction) {
     }
 }
 
-
-
+val CoreTransaction.attachmentIds: List<SecureHash>
+    get() = when (this) {
+        is WireTransaction -> attachments
+        is ContractUpgradeWireTransaction -> listOf(legacyContractAttachmentId, upgradedContractAttachmentId)
+        else -> emptyList()
+    }
diff --git a/core/src/main/kotlin/net/corda/core/internal/verification/ExternalVerifierHandle.kt b/core/src/main/kotlin/net/corda/core/internal/verification/ExternalVerifierHandle.kt
new file mode 100644
index 0000000000..e9f1e92e88
--- /dev/null
+++ b/core/src/main/kotlin/net/corda/core/internal/verification/ExternalVerifierHandle.kt
@@ -0,0 +1,7 @@
+package net.corda.core.internal.verification
+
+import net.corda.core.transactions.SignedTransaction
+
+interface ExternalVerifierHandle : AutoCloseable {
+    fun verifyTransaction(stx: SignedTransaction, checkSufficientSignatures: Boolean)
+}
diff --git a/core/src/main/kotlin/net/corda/core/internal/verification/VerificationService.kt b/core/src/main/kotlin/net/corda/core/internal/verification/NodeVerificationSupport.kt
similarity index 89%
rename from core/src/main/kotlin/net/corda/core/internal/verification/VerificationService.kt
rename to core/src/main/kotlin/net/corda/core/internal/verification/NodeVerificationSupport.kt
index ebf90f3e94..de76e987c3 100644
--- a/core/src/main/kotlin/net/corda/core/internal/verification/VerificationService.kt
+++ b/core/src/main/kotlin/net/corda/core/internal/verification/NodeVerificationSupport.kt
@@ -9,6 +9,7 @@ import net.corda.core.identity.Party
 import net.corda.core.internal.AttachmentTrustCalculator
 import net.corda.core.internal.SerializedTransactionState
 import net.corda.core.internal.TRUSTED_UPLOADERS
+import net.corda.core.internal.cordapp.CordappProviderInternal
 import net.corda.core.internal.entries
 import net.corda.core.internal.getRequiredTransaction
 import net.corda.core.node.NetworkParameters
@@ -36,26 +37,33 @@ import java.security.PublicKey
 /**
  * Implements [VerificationSupport] in terms of node-based services.
  */
-interface VerificationService : VerificationSupport {
-    val transactionStorage: TransactionStorage
+interface NodeVerificationSupport : VerificationSupport {
+    val networkParameters: NetworkParameters
+
+    val validatedTransactions: TransactionStorage
 
     val identityService: IdentityService
 
-    val attachmentStorage: AttachmentStorage
+    val attachments: AttachmentStorage
 
     val networkParametersService: NetworkParametersService
 
+    val cordappProvider: CordappProviderInternal
+
     val attachmentTrustCalculator: AttachmentTrustCalculator
 
-    val attachmentFixups: AttachmentFixups
+    val externalVerifierHandle: ExternalVerifierHandle
+
+    override val appClassLoader: ClassLoader
+        get() = cordappProvider.appClassLoader
 
     // TODO Bulk party lookup?
     override fun getParties(keys: Collection<PublicKey>): List<Party?> = keys.map(identityService::partyFromKey)
 
-    override fun getAttachment(id: SecureHash): Attachment? = attachmentStorage.openAttachment(id)
+    override fun getAttachment(id: SecureHash): Attachment? = attachments.openAttachment(id)
 
     override fun getNetworkParameters(id: SecureHash?): NetworkParameters? {
-        return networkParametersService.lookup(id ?: networkParametersService.defaultHash)
+        return if (id != null) networkParametersService.lookup(id) else networkParameters
     }
 
     /**
@@ -65,7 +73,7 @@ interface VerificationService : VerificationSupport {
      * correct classloader independent of the node's classpath.
      */
     override fun getSerializedState(stateRef: StateRef): SerializedTransactionState {
-        val coreTransaction = transactionStorage.getRequiredTransaction(stateRef.txhash).coreTransaction
+        val coreTransaction = validatedTransactions.getRequiredTransaction(stateRef.txhash).coreTransaction
         return when (coreTransaction) {
             is WireTransaction -> getRegularOutput(coreTransaction, stateRef.index)
             is ContractUpgradeWireTransaction -> getContractUpdateOutput(coreTransaction, stateRef.index)
@@ -127,14 +135,14 @@ interface VerificationService : VerificationSupport {
      */
     // TODO Should throw when the class is found in multiple contract attachments (not different versions).
     override fun getTrustedClassAttachment(className: String): Attachment? {
-        val allTrusted = attachmentStorage.queryAttachments(
+        val allTrusted = attachments.queryAttachments(
                 AttachmentsQueryCriteria().withUploader(Builder.`in`(TRUSTED_UPLOADERS)),
                 AttachmentSort(listOf(AttachmentSortColumn(AttachmentSortAttribute.VERSION, Sort.Direction.DESC)))
         )
 
         // TODO - add caching if performance is affected.
         for (attId in allTrusted) {
-            val attch = attachmentStorage.openAttachment(attId)!!
+            val attch = attachments.openAttachment(attId)!!
             if (attch.hasFile("$className.class")) return attch
         }
         return null
@@ -145,6 +153,6 @@ interface VerificationService : VerificationSupport {
     override fun isAttachmentTrusted(attachment: Attachment): Boolean = attachmentTrustCalculator.calculate(attachment)
 
     override fun fixupAttachmentIds(attachmentIds: Collection<SecureHash>): Set<SecureHash> {
-        return attachmentFixups.fixupAttachmentIds(attachmentIds)
+        return cordappProvider.attachmentFixups.fixupAttachmentIds(attachmentIds)
     }
 }
diff --git a/core/src/main/kotlin/net/corda/core/internal/verification/VerifyingServiceHub.kt b/core/src/main/kotlin/net/corda/core/internal/verification/VerifyingServiceHub.kt
index 6bcec51d16..78c5e790cf 100644
--- a/core/src/main/kotlin/net/corda/core/internal/verification/VerifyingServiceHub.kt
+++ b/core/src/main/kotlin/net/corda/core/internal/verification/VerifyingServiceHub.kt
@@ -8,30 +8,16 @@ import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.StateRef
 import net.corda.core.contracts.TransactionState
 import net.corda.core.crypto.SecureHash
-import net.corda.core.internal.cordapp.CordappProviderInternal
 import net.corda.core.internal.getRequiredTransaction
 import net.corda.core.node.ServiceHub
 import net.corda.core.node.ServicesForResolution
-import net.corda.core.node.services.AttachmentStorage
-import net.corda.core.node.services.TransactionStorage
 import net.corda.core.serialization.deserialize
 import net.corda.core.transactions.ContractUpgradeWireTransaction
 import net.corda.core.transactions.NotaryChangeWireTransaction
-import net.corda.core.transactions.SignedTransaction
 import net.corda.core.transactions.WireTransaction
 
 @Suppress("TooManyFunctions", "ThrowsCount")
-interface VerifyingServiceHub : ServiceHub, VerificationService {
-    override val cordappProvider: CordappProviderInternal
-
-    override val transactionStorage: TransactionStorage get() = validatedTransactions
-
-    override val attachmentStorage: AttachmentStorage get() = attachments
-
-    override val appClassLoader: ClassLoader get() = cordappProvider.appClassLoader
-
-    override val attachmentFixups: AttachmentFixups get() = cordappProvider.attachmentFixups
-
+interface VerifyingServiceHub : ServiceHub, NodeVerificationSupport {
     override fun loadContractAttachment(stateRef: StateRef): Attachment {
         // We may need to recursively chase transactions if there are notary changes.
         return loadContractAttachment(stateRef, null)
@@ -72,18 +58,6 @@ interface VerifyingServiceHub : ServiceHub, VerificationService {
     fun <T : ContractState, C : MutableCollection<StateAndRef<T>>> loadStatesInternal(input: Iterable<StateRef>, output: C): C {
         return input.mapTo(output, ::toStateAndRef)
     }
-
-    /**
-     * Try to verify the given transaction on the external verifier, assuming it is available. It is not required to verify externally even
-     * if the verifier is available.
-     *
-     * The default implementation is to only do internal verification.
-     *
-     * @return true if the transaction should (also) be verified internally, regardless of whether it was verified externally.
-     */
-    fun tryExternalVerification(stx: SignedTransaction, checkSufficientSignatures: Boolean): Boolean {
-        return true
-    }
 }
 
 fun ServicesForResolution.toVerifyingServiceHub(): VerifyingServiceHub {
diff --git a/core/src/main/kotlin/net/corda/core/transactions/SignedTransaction.kt b/core/src/main/kotlin/net/corda/core/transactions/SignedTransaction.kt
index c825d74256..726788bb1b 100644
--- a/core/src/main/kotlin/net/corda/core/transactions/SignedTransaction.kt
+++ b/core/src/main/kotlin/net/corda/core/transactions/SignedTransaction.kt
@@ -18,8 +18,11 @@ import net.corda.core.crypto.toStringShort
 import net.corda.core.identity.Party
 import net.corda.core.internal.TransactionDeserialisationException
 import net.corda.core.internal.VisibleForTesting
+import net.corda.core.internal.attachmentIds
 import net.corda.core.internal.equivalent
 import net.corda.core.internal.isUploaderTrusted
+import net.corda.core.internal.kotlinMetadataVersion
+import net.corda.core.internal.verification.NodeVerificationSupport
 import net.corda.core.internal.verification.VerificationSupport
 import net.corda.core.internal.verification.toVerifyingServiceHub
 import net.corda.core.node.ServiceHub
@@ -31,6 +34,7 @@ import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.internal.MissingSerializerException
 import net.corda.core.serialization.serialize
 import net.corda.core.utilities.contextLogger
+import net.corda.core.utilities.debug
 import java.io.NotSerializableException
 import java.security.KeyPair
 import java.security.PublicKey
@@ -155,9 +159,10 @@ data class SignedTransaction(val txBits: SerializedBytes<CoreTransaction>,
     @JvmOverloads
     @Throws(SignatureException::class, AttachmentResolutionException::class, TransactionResolutionException::class)
     fun toLedgerTransaction(services: ServiceHub, checkSufficientSignatures: Boolean = true): LedgerTransaction {
+        val verifyingServiceHub = services.toVerifyingServiceHub()
         // We need parameters check here, because finality flow calls stx.toLedgerTransaction() and then verify.
-        resolveAndCheckNetworkParameters(services)
-        return toLedgerTransactionInternal(services.toVerifyingServiceHub(), checkSufficientSignatures)
+        resolveAndCheckNetworkParameters(verifyingServiceHub)
+        return toLedgerTransactionInternal(verifyingServiceHub, checkSufficientSignatures)
     }
 
     @JvmSynthetic
@@ -191,16 +196,58 @@ data class SignedTransaction(val txBits: SerializedBytes<CoreTransaction>,
     @JvmOverloads
     @Throws(SignatureException::class, AttachmentResolutionException::class, TransactionResolutionException::class, TransactionVerificationException::class)
     fun verify(services: ServiceHub, checkSufficientSignatures: Boolean = true) {
-        resolveAndCheckNetworkParameters(services)
-        val verifyingServiceHub = services.toVerifyingServiceHub()
-        if (verifyingServiceHub.tryExternalVerification(this, checkSufficientSignatures)) {
-            verifyInternal(verifyingServiceHub, checkSufficientSignatures)
+        verifyInternal(services.toVerifyingServiceHub(), checkSufficientSignatures)
+    }
+
+    /**
+     * Internal version of the public [verify] which takes in a [NodeVerificationSupport] instead of the heavier [ServiceHub].
+     *
+     * Depending on the contract attachments, this method will either verify this transaction in-process or send it to the external verifier
+     * for out-of-process verification.
+     */
+    @CordaInternal
+    @JvmSynthetic
+    fun verifyInternal(verificationSupport: NodeVerificationSupport, checkSufficientSignatures: Boolean = true) {
+        resolveAndCheckNetworkParameters(verificationSupport)
+        val verificationType = determineVerificationType(verificationSupport)
+        log.debug { "Transaction $id has verification type $verificationType" }
+        if (verificationType == VerificationType.IN_PROCESS || verificationType == VerificationType.BOTH) {
+            verifyInProcess(verificationSupport, checkSufficientSignatures)
+        }
+        if (verificationType == VerificationType.EXTERNAL || verificationType == VerificationType.BOTH) {
+            verificationSupport.externalVerifierHandle.verifyTransaction(this, checkSufficientSignatures)
         }
     }
 
+    private fun determineVerificationType(verificationSupport: VerificationSupport): VerificationType {
+        var old = false
+        var new = false
+        for (attachmentId in coreTransaction.attachmentIds) {
+            val (major, minor) = verificationSupport.getAttachment(attachmentId)?.kotlinMetadataVersion?.split(".") ?: continue
+            // Metadata version 1.1 maps to language versions 1.0 to 1.3
+            if (major == "1" && minor == "1") {
+                old = true
+            } else {
+                new = true
+            }
+        }
+        return when {
+            old && new -> VerificationType.BOTH
+            old -> VerificationType.EXTERNAL
+            else -> VerificationType.IN_PROCESS
+        }
+    }
+
+    private enum class VerificationType {
+        IN_PROCESS, EXTERNAL, BOTH
+    }
+
+    /**
+     * Verifies this transaction in-process. This assumes the current process has the correct classpath for all the contracts.
+     */
     @CordaInternal
     @JvmSynthetic
-    fun verifyInternal(verificationSupport: VerificationSupport, checkSufficientSignatures: Boolean) {
+    fun verifyInProcess(verificationSupport: VerificationSupport, checkSufficientSignatures: Boolean) {
         when (coreTransaction) {
             is NotaryChangeWireTransaction -> verifyNotaryChangeTransaction(verificationSupport, checkSufficientSignatures)
             is ContractUpgradeWireTransaction -> verifyContractUpgradeTransaction(verificationSupport, checkSufficientSignatures)
@@ -209,7 +256,7 @@ data class SignedTransaction(val txBits: SerializedBytes<CoreTransaction>,
     }
 
     @Suppress("ThrowsCount")
-    private fun resolveAndCheckNetworkParameters(services: ServiceHub) {
+    private fun resolveAndCheckNetworkParameters(services: NodeVerificationSupport) {
         val hashOrDefault = networkParametersHash ?: services.networkParametersService.defaultHash
         val txNetworkParameters = services.networkParametersService.lookup(hashOrDefault)
                 ?: throw TransactionResolutionException(id)
diff --git a/node/build.gradle b/node/build.gradle
index 4759f663f9..c177dc6ec0 100644
--- a/node/build.gradle
+++ b/node/build.gradle
@@ -242,6 +242,7 @@ dependencies {
 
     // Adding native SSL library to allow using native SSL with Artemis and AMQP
     implementation "io.netty:netty-tcnative-boringssl-static:$tcnative_version"
+    implementation 'org.jetbrains.kotlinx:kotlinx-metadata-jvm:0.8.0'
 
     // Byteman for runtime (termination) rules injection on the running node
     // Submission tool allowing to install rules on running nodes
diff --git a/node/src/integration-test/kotlin/net/corda/node/verification/ExternalVerificationTest.kt b/node/src/integration-test/kotlin/net/corda/node/verification/ExternalVerificationTest.kt
deleted file mode 100644
index 091564fafb..0000000000
--- a/node/src/integration-test/kotlin/net/corda/node/verification/ExternalVerificationTest.kt
+++ /dev/null
@@ -1,286 +0,0 @@
-package net.corda.node.verification
-
-import co.paralleluniverse.fibers.Suspendable
-import com.typesafe.config.ConfigFactory
-import net.corda.core.contracts.CommandData
-import net.corda.core.contracts.Contract
-import net.corda.core.contracts.ContractState
-import net.corda.core.contracts.StateAndRef
-import net.corda.core.contracts.TransactionVerificationException.ContractRejection
-import net.corda.core.contracts.TypeOnlyCommandData
-import net.corda.core.crypto.SecureHash
-import net.corda.core.flows.FinalityFlow
-import net.corda.core.flows.FlowLogic
-import net.corda.core.flows.FlowSession
-import net.corda.core.flows.InitiatedBy
-import net.corda.core.flows.InitiatingFlow
-import net.corda.core.flows.NotaryChangeFlow
-import net.corda.core.flows.ReceiveFinalityFlow
-import net.corda.core.flows.StartableByRPC
-import net.corda.core.flows.UnexpectedFlowEndException
-import net.corda.core.identity.AbstractParty
-import net.corda.core.identity.CordaX500Name
-import net.corda.core.identity.Party
-import net.corda.core.internal.concurrent.map
-import net.corda.core.internal.concurrent.transpose
-import net.corda.core.messaging.startFlow
-import net.corda.core.node.NodeInfo
-import net.corda.core.transactions.LedgerTransaction
-import net.corda.core.transactions.NotaryChangeWireTransaction
-import net.corda.core.transactions.TransactionBuilder
-import net.corda.core.utilities.OpaqueBytes
-import net.corda.core.utilities.getOrThrow
-import net.corda.finance.DOLLARS
-import net.corda.finance.contracts.asset.Cash
-import net.corda.finance.flows.CashIssueFlow
-import net.corda.finance.flows.CashPaymentFlow
-import net.corda.testing.core.ALICE_NAME
-import net.corda.testing.core.BOB_NAME
-import net.corda.testing.core.BOC_NAME
-import net.corda.testing.core.CHARLIE_NAME
-import net.corda.testing.core.DUMMY_NOTARY_NAME
-import net.corda.testing.core.singleIdentity
-import net.corda.testing.driver.NodeHandle
-import net.corda.testing.driver.NodeParameters
-import net.corda.testing.node.NotarySpec
-import net.corda.testing.node.internal.FINANCE_CORDAPPS
-import net.corda.testing.node.internal.cordappWithPackages
-import net.corda.testing.node.internal.enclosedCordapp
-import net.corda.testing.node.internal.internalDriver
-import org.assertj.core.api.Assertions.assertThat
-import org.assertj.core.api.Assertions.assertThatExceptionOfType
-import org.junit.Test
-import java.io.File
-import java.net.InetAddress
-import kotlin.io.path.div
-import kotlin.io.path.listDirectoryEntries
-import kotlin.io.path.name
-import kotlin.io.path.readText
-
-class ExternalVerificationTest {
-    @Test(timeout=300_000)
-    fun `regular transactions are verified in external verifier`() {
-        internalDriver(
-                systemProperties = mapOf("net.corda.node.verification.external" to "true"),
-                cordappsForAllNodes = FINANCE_CORDAPPS,
-                notarySpecs = listOf(NotarySpec(DUMMY_NOTARY_NAME, validating = true, startInProcess = false))
-        ) {
-            val (notary, alice, bob) = listOf(
-                    defaultNotaryNode,
-                    startNode(NodeParameters(providedName = ALICE_NAME)),
-                    startNode(NodeParameters(providedName = BOB_NAME))
-            ).transpose().getOrThrow()
-
-            val (issuanceTx) = alice.rpc.startFlow(
-                    ::CashIssueFlow,
-                    10.DOLLARS,
-                    OpaqueBytes.of(0x01),
-                    defaultNotaryIdentity
-            ).returnValue.getOrThrow()
-
-            val (paymentTx) = alice.rpc.startFlow(
-                    ::CashPaymentFlow,
-                    10.DOLLARS,
-                    bob.nodeInfo.singleIdentity(),
-                    false,
-            ).returnValue.getOrThrow()
-
-            notary.assertTransactionsWereVerifiedExternally(issuanceTx.id, paymentTx.id)
-            bob.assertTransactionsWereVerifiedExternally(issuanceTx.id, paymentTx.id)
-        }
-    }
-
-    @Test(timeout=300_000)
-    fun `external verifier is unable to verify contracts which use new Kotlin APIs`() {
-        check(!IntArray::maxOrNull.isInline)
-
-        internalDriver(
-                systemProperties = mapOf("net.corda.node.verification.external" to "true"),
-                cordappsForAllNodes = listOf(cordappWithPackages("net.corda.node.verification"))
-        ) {
-            val (alice, bob) = listOf(
-                    startNode(NodeParameters(providedName = ALICE_NAME)),
-                    startNode(NodeParameters(providedName = BOB_NAME)),
-            ).transpose().getOrThrow()
-
-            assertThatExceptionOfType(UnexpectedFlowEndException::class.java).isThrownBy {
-                alice.rpc.startFlow(::NewKotlinApiFlow, bob.nodeInfo).returnValue.getOrThrow()
-            }
-
-            assertThat(bob.externalVerifierLogs()).contains("""
-                java.lang.NoSuchMethodError: 'java.lang.Integer kotlin.collections.ArraysKt.maxOrNull(int[])'
-                	at net.corda.node.verification.ExternalVerificationTest${'$'}NewKotlinApiContract.verify(ExternalVerificationTest.kt:
-            """.trimIndent())
-        }
-    }
-
-    @Test(timeout=300_000)
-    fun `regular transactions can fail verification in external verifier`() {
-        internalDriver(
-                systemProperties = mapOf("net.corda.node.verification.external" to "true"),
-                cordappsForAllNodes = listOf(cordappWithPackages("net.corda.node.verification", "com.typesafe.config"))
-        ) {
-            val (alice, bob, charlie) = listOf(
-                    startNode(NodeParameters(providedName = ALICE_NAME)),
-                    startNode(NodeParameters(providedName = BOB_NAME)),
-                    startNode(NodeParameters(providedName = CHARLIE_NAME))
-            ).transpose().getOrThrow()
-
-            // Create a transaction from Alice to Bob, where Charlie is specified as the contract verification trigger
-            val firstState = alice.rpc.startFlow(::FailExternallyFlow, null, charlie.nodeInfo, bob.nodeInfo).returnValue.getOrThrow()
-            // When the transaction chain tries to moves onto Charlie, it will trigger the failure
-            assertThatExceptionOfType(ContractRejection::class.java)
-                    .isThrownBy { bob.rpc.startFlow(::FailExternallyFlow, firstState, charlie.nodeInfo, charlie.nodeInfo).returnValue.getOrThrow() }
-                    .withMessageContaining("Fail in external verifier: ${firstState.ref.txhash}")
-
-            // Make sure Charlie tried to verify the first transaction externally
-            assertThat(charlie.externalVerifierLogs()).contains("Fail in external verifier: ${firstState.ref.txhash}")
-        }
-    }
-
-    @Test(timeout=300_000)
-    fun `notary change transactions are verified in external verifier`() {
-        internalDriver(
-                systemProperties = mapOf("net.corda.node.verification.external" to "true"),
-                cordappsForAllNodes = FINANCE_CORDAPPS + enclosedCordapp(),
-                notarySpecs = listOf(DUMMY_NOTARY_NAME, BOC_NAME).map { NotarySpec(it, validating = true, startInProcess = false) }
-        ) {
-            val (notary1, notary2) = notaryHandles.map { handle -> handle.nodeHandles.map { it[0] } }.transpose().getOrThrow()
-            val alice = startNode(NodeParameters(providedName = ALICE_NAME)).getOrThrow()
-
-            val txId = alice.rpc.startFlow(
-                    ::IssueAndChangeNotaryFlow,
-                    notary1.nodeInfo.singleIdentity(),
-                    notary2.nodeInfo.singleIdentity()
-            ).returnValue.getOrThrow()
-
-            notary1.assertTransactionsWereVerifiedExternally(txId)
-            alice.assertTransactionsWereVerifiedExternally(txId)
-        }
-    }
-
-    private fun NodeHandle.assertTransactionsWereVerifiedExternally(vararg txIds: SecureHash) {
-        val verifierLogContent = externalVerifierLogs()
-        for (txId in txIds) {
-            assertThat(verifierLogContent).contains("SignedTransaction(id=$txId) verified")
-        }
-    }
-
-    private fun NodeHandle.externalVerifierLogs(): String {
-        val verifierLogs = (baseDirectory / "logs")
-                .listDirectoryEntries()
-                .filter { it.name == "verifier-${InetAddress.getLocalHost().hostName}.log" }
-        assertThat(verifierLogs).describedAs("External verifier was not started").hasSize(1)
-        return verifierLogs[0].readText()
-    }
-
-
-    class FailExternallyContract : Contract {
-        override fun verify(tx: LedgerTransaction) {
-            val command = tx.commandsOfType<Command>().single()
-            if (insideExternalVerifier()) {
-                // The current directory for the external verifier is the node's base directory
-                val localName = CordaX500Name.parse(ConfigFactory.parseFile(File("node.conf")).getString("myLegalName"))
-                check(localName != command.value.failForParty.name) { "Fail in external verifier: ${tx.id}" }
-            }
-        }
-
-        private fun insideExternalVerifier(): Boolean {
-            return StackWalker.getInstance().walk { frames ->
-                frames.anyMatch { it.className.startsWith("net.corda.verifier.") }
-            }
-        }
-
-        data class State(override val party: Party) : TestState
-        data class Command(val failForParty: Party) : CommandData
-    }
-
-
-    @StartableByRPC
-    @InitiatingFlow
-    class FailExternallyFlow(inputState: StateAndRef<FailExternallyContract.State>?,
-                             private val failForParty: NodeInfo,
-                             recipient: NodeInfo) : TestFlow<FailExternallyContract.State>(inputState, recipient) {
-        override fun newOutput() = FailExternallyContract.State(serviceHub.myInfo.legalIdentities[0])
-        override fun newCommand() = FailExternallyContract.Command(failForParty.legalIdentities[0])
-
-        @Suppress("unused")
-        @InitiatedBy(FailExternallyFlow::class)
-        class ReceiverFlow(otherSide: FlowSession) : TestReceiverFlow(otherSide)
-    }
-
-
-    class NewKotlinApiContract : Contract {
-        override fun verify(tx: LedgerTransaction) {
-            check(tx.commandsOfType<Command>().isNotEmpty())
-            // New post-1.2 API which is non-inlined
-            intArrayOf().maxOrNull()
-        }
-
-        data class State(override val party: Party) : TestState
-        object Command : TypeOnlyCommandData()
-    }
-
-
-    @StartableByRPC
-    @InitiatingFlow
-    class NewKotlinApiFlow(recipient: NodeInfo) : TestFlow<NewKotlinApiContract.State>(null, recipient) {
-        override fun newOutput() = NewKotlinApiContract.State(serviceHub.myInfo.legalIdentities[0])
-        override fun newCommand() = NewKotlinApiContract.Command
-
-        @Suppress("unused")
-        @InitiatedBy(NewKotlinApiFlow::class)
-        class ReceiverFlow(otherSide: FlowSession) : TestReceiverFlow(otherSide)
-    }
-
-
-    @StartableByRPC
-    class IssueAndChangeNotaryFlow(private val oldNotary: Party, private val newNotary: Party) : FlowLogic<SecureHash>() {
-        @Suspendable
-        override fun call(): SecureHash {
-            subFlow(CashIssueFlow(10.DOLLARS, OpaqueBytes.of(0x01), oldNotary))
-            val oldState = serviceHub.vaultService.queryBy(Cash.State::class.java).states.single()
-            assertThat(oldState.state.notary).isEqualTo(oldNotary)
-            val newState = subFlow(NotaryChangeFlow(oldState, newNotary))
-            assertThat(newState.state.notary).isEqualTo(newNotary)
-            val notaryChangeTx = serviceHub.validatedTransactions.getTransaction(newState.ref.txhash)
-            assertThat(notaryChangeTx?.coreTransaction).isInstanceOf(NotaryChangeWireTransaction::class.java)
-            return notaryChangeTx!!.id
-        }
-    }
-
-
-    abstract class TestFlow<T : TestState>(
-            private val inputState: StateAndRef<T>?,
-            private val recipient: NodeInfo
-    ) : FlowLogic<StateAndRef<T>>() {
-        @Suspendable
-        override fun call(): StateAndRef<T> {
-            val myParty = serviceHub.myInfo.legalIdentities[0]
-            val txBuilder = TransactionBuilder(serviceHub.networkMapCache.notaryIdentities[0])
-            inputState?.let(txBuilder::addInputState)
-            txBuilder.addOutputState(newOutput())
-            txBuilder.addCommand(newCommand(), myParty.owningKey)
-            val initialTx = serviceHub.signInitialTransaction(txBuilder)
-            val sessions = arrayListOf(initiateFlow(recipient.legalIdentities[0]))
-            inputState?.let { sessions += initiateFlow(it.state.data.party) }
-            val notarisedTx = subFlow(FinalityFlow(initialTx, sessions))
-            return notarisedTx.toLedgerTransaction(serviceHub).outRef(0)
-        }
-
-        protected abstract fun newOutput(): T
-        protected abstract fun newCommand(): CommandData
-    }
-
-    abstract class TestReceiverFlow(private val otherSide: FlowSession) : FlowLogic<Unit>() {
-        @Suspendable
-        override fun call() {
-            subFlow(ReceiveFinalityFlow(otherSide))
-        }
-    }
-
-    interface TestState : ContractState {
-        val party: Party
-        override val participants: List<AbstractParty> get() = listOf(party)
-    }
-}
diff --git a/node/src/integration-test/kotlin/net/corda/node/verification/ExternalVerifierTest.kt b/node/src/integration-test/kotlin/net/corda/node/verification/ExternalVerifierTest.kt
new file mode 100644
index 0000000000..2d58b3a20c
--- /dev/null
+++ b/node/src/integration-test/kotlin/net/corda/node/verification/ExternalVerifierTest.kt
@@ -0,0 +1,29 @@
+package net.corda.node.verification
+
+import io.github.classgraph.ClassGraph
+import net.corda.core.internal.pooledScan
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.Test
+
+class ExternalVerifierTest {
+    @Test(timeout=300_000)
+    fun `external verifier does not have newer Kotlin`() {
+        val kotlinClasses = ClassGraph()
+                .overrideClasspath(javaClass.getResource("external-verifier.jar")!!)
+                .enableAnnotationInfo()
+                .pooledScan()
+                .use { result ->
+                    result.getClassesWithAnnotation(Metadata::class.java).associateBy({ it.name }, {
+                        val annotationInfo = it.getAnnotationInfo(Metadata::class.java)
+                        val metadataVersion = annotationInfo.parameterValues.get("mv").value as IntArray
+                        "${metadataVersion[0]}.${metadataVersion[1]}"
+                    })
+                }
+
+        // First make sure we're capturing the right data
+        assertThat(kotlinClasses).containsKeys("net.corda.verifier.ExternalVerifier")
+        // Kotlin metadata version 1.1 maps to language versions 1.0 to 1.3
+        val newerKotlinClasses = kotlinClasses.filterValues { metadataVersion -> metadataVersion != "1.1" }
+        assertThat(newerKotlinClasses).isEmpty()
+    }
+}
diff --git a/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt b/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
index 272338ced5..4f98774738 100644
--- a/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
+++ b/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
@@ -46,7 +46,6 @@ import net.corda.core.internal.telemetry.SimpleLogTelemetryComponent
 import net.corda.core.internal.telemetry.TelemetryComponent
 import net.corda.core.internal.telemetry.TelemetryServiceImpl
 import net.corda.core.internal.uncheckedCast
-import net.corda.core.internal.verification.VerifyingServiceHub
 import net.corda.core.messaging.ClientRpcSslOptions
 import net.corda.core.messaging.CordaRPCOps
 import net.corda.core.messaging.RPCOps
@@ -68,7 +67,6 @@ import net.corda.core.serialization.SingletonSerializeAsToken
 import net.corda.core.serialization.internal.AttachmentsClassLoaderCache
 import net.corda.core.serialization.internal.AttachmentsClassLoaderCacheImpl
 import net.corda.core.toFuture
-import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.NetworkHostAndPort
 import net.corda.core.utilities.days
 import net.corda.core.utilities.millis
@@ -144,6 +142,7 @@ import net.corda.node.utilities.AffinityExecutor
 import net.corda.node.utilities.BindableNamedCacheFactory
 import net.corda.node.utilities.NamedThreadFactory
 import net.corda.node.utilities.NotaryLoader
+import net.corda.node.verification.ExternalVerifierHandleImpl
 import net.corda.nodeapi.internal.NodeInfoAndSigned
 import net.corda.nodeapi.internal.NodeStatus
 import net.corda.nodeapi.internal.SignedNodeInfo
@@ -1152,14 +1151,6 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
         return NodeVaultService(platformClock, keyManagementService, services, database, schemaService, cordappLoader.appClassLoader)
     }
 
-    /**
-     * Dy default only internal verification is done.
-     * @see VerifyingServiceHub.tryExternalVerification
-     */
-    protected open fun tryExternalVerification(stx: SignedTransaction, checkSufficientSignatures: Boolean): Boolean {
-        return true
-    }
-
     // JDK 11: switch to directly instantiating jolokia server (rather than indirectly via dynamically self attaching Java Agents,
     // which is no longer supported from JDK 9 onwards (https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8180425).
     // No longer need to use https://github.com/electronicarts/ea-agent-loader either (which is also deprecated)
@@ -1175,6 +1166,7 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
     inner class ServiceHubImpl : SingletonSerializeAsToken(), ServiceHubInternal, NetworkParameterUpdateListener {
         override val rpcFlows = ArrayList<Class<out FlowLogic<*>>>()
         override val stateMachineRecordedTransactionMapping = DBTransactionMappingStorage(database)
+        override val externalVerifierHandle = ExternalVerifierHandleImpl(this, configuration.baseDirectory).also { runOnStop += it::close }
         override val identityService: IdentityService get() = this@AbstractNode.identityService
         override val keyManagementService: KeyManagementService get() = this@AbstractNode.keyManagementService
         override val schemaService: SchemaService get() = this@AbstractNode.schemaService
@@ -1298,10 +1290,6 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
         override fun onNewNetworkParameters(networkParameters: NetworkParameters) {
             this.networkParameters = networkParameters
         }
-
-        override fun tryExternalVerification(stx: SignedTransaction, checkSufficientSignatures: Boolean): Boolean {
-            return this@AbstractNode.tryExternalVerification(stx, checkSufficientSignatures)
-        }
     }
 }
 
diff --git a/node/src/main/kotlin/net/corda/node/internal/Node.kt b/node/src/main/kotlin/net/corda/node/internal/Node.kt
index a02cb95dec..60b32a9b10 100644
--- a/node/src/main/kotlin/net/corda/node/internal/Node.kt
+++ b/node/src/main/kotlin/net/corda/node/internal/Node.kt
@@ -25,7 +25,6 @@ import net.corda.core.node.NodeInfo
 import net.corda.core.node.ServiceHub
 import net.corda.core.serialization.internal.SerializationEnvironment
 import net.corda.core.serialization.internal.nodeSerializationEnv
-import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.NetworkHostAndPort
 import net.corda.core.utilities.contextLogger
 import net.corda.node.CordaClock
@@ -61,7 +60,6 @@ import net.corda.node.utilities.BindableNamedCacheFactory
 import net.corda.node.utilities.DefaultNamedCacheFactory
 import net.corda.node.utilities.DemoClock
 import net.corda.node.utilities.errorAndTerminate
-import net.corda.node.verification.ExternalVerifierHandle
 import net.corda.nodeapi.internal.ArtemisMessagingClient
 import net.corda.nodeapi.internal.ShutdownHook
 import net.corda.nodeapi.internal.addShutdownHook
@@ -201,8 +199,6 @@ open class Node(configuration: NodeConfiguration,
 
     protected open val journalBufferTimeout : Int? = null
 
-    private val externalVerifierHandle = ExternalVerifierHandle(services).also { runOnStop += it::close }
-
     private var shutdownHook: ShutdownHook? = null
 
     // DISCUSSION
@@ -588,17 +584,6 @@ open class Node(configuration: NodeConfiguration,
         )
     }
 
-    override fun tryExternalVerification(stx: SignedTransaction, checkSufficientSignatures: Boolean): Boolean {
-        // TODO Determine from transaction whether it should be verified externally
-        // TODO If both old and new attachments are present then return true so that internal verification is also done.
-        return if (java.lang.Boolean.getBoolean("net.corda.node.verification.external")) {
-            externalVerifierHandle.verifyTransaction(stx, checkSufficientSignatures)
-            false
-        } else {
-            true
-        }
-    }
-
     /** Starts a blocking event loop for message dispatch. */
     fun run() {
         internalRpcMessagingClient?.start(rpcBroker!!.serverControl)
diff --git a/node/src/main/kotlin/net/corda/node/services/persistence/NodeAttachmentService.kt b/node/src/main/kotlin/net/corda/node/services/persistence/NodeAttachmentService.kt
index aa998245f7..7e4c06e5d8 100644
--- a/node/src/main/kotlin/net/corda/node/services/persistence/NodeAttachmentService.kt
+++ b/node/src/main/kotlin/net/corda/node/services/persistence/NodeAttachmentService.kt
@@ -6,6 +6,8 @@ import com.google.common.hash.HashCode
 import com.google.common.hash.Hashing
 import com.google.common.hash.HashingInputStream
 import com.google.common.io.CountingInputStream
+import kotlinx.metadata.jvm.KotlinModuleMetadata
+import kotlinx.metadata.jvm.UnstableMetadataApi
 import net.corda.core.CordaRuntimeException
 import net.corda.core.contracts.Attachment
 import net.corda.core.contracts.ContractAttachment
@@ -16,6 +18,7 @@ import net.corda.core.internal.AbstractAttachment
 import net.corda.core.internal.DEPLOYED_CORDAPP_UPLOADER
 import net.corda.core.internal.FetchAttachmentsFlow
 import net.corda.core.internal.JarSignatureCollector
+import net.corda.core.internal.InternalAttachment
 import net.corda.core.internal.NamedCacheFactory
 import net.corda.core.internal.P2P_UPLOADER
 import net.corda.core.internal.RPC_UPLOADER
@@ -25,6 +28,7 @@ import net.corda.core.internal.Version
 import net.corda.core.internal.VisibleForTesting
 import net.corda.core.internal.cordapp.CordappImpl.Companion.CORDAPP_CONTRACT_VERSION
 import net.corda.core.internal.cordapp.CordappImpl.Companion.DEFAULT_CORDAPP_VERSION
+import net.corda.core.internal.entries
 import net.corda.core.internal.isUploaderTrusted
 import net.corda.core.internal.readFully
 import net.corda.core.internal.utilities.ZipBombDetector
@@ -53,6 +57,7 @@ import java.io.ByteArrayInputStream
 import java.io.FilterInputStream
 import java.io.IOException
 import java.io.InputStream
+import java.nio.file.FileAlreadyExistsException
 import java.nio.file.Paths
 import java.security.PublicKey
 import java.time.Instant
@@ -109,7 +114,7 @@ class NodeAttachmentService @JvmOverloads constructor(
             // Can be null for not-signed JARs.
             val allManifestEntries = jar.manifest?.entries?.keys?.toMutableList()
             val extraFilesNotFoundInEntries = mutableListOf<JarEntry>()
-            val manifestHasEntries= allManifestEntries != null && allManifestEntries.isNotEmpty()
+            val manifestHasEntries = !allManifestEntries.isNullOrEmpty()
 
             while (true) {
                 val cursor = jar.nextJarEntry ?: break
@@ -225,7 +230,7 @@ class NodeAttachmentService @JvmOverloads constructor(
 
         // This is invoked by [InputStreamSerializer], which does NOT close the stream afterwards.
         @Throws(IOException::class)
-        override fun read(b: ByteArray?, off: Int, len: Int): Int {
+        override fun read(b: ByteArray, off: Int, len: Int): Int {
             return super.read(b, off, len).apply {
                 if (this == -1) {
                     validate()
@@ -256,12 +261,13 @@ class NodeAttachmentService @JvmOverloads constructor(
     }
 
     private class AttachmentImpl(
-        override val id: SecureHash,
-        dataLoader: () -> ByteArray,
-        private val checkOnLoad: Boolean,
-        uploader: String?,
-        override val signerKeys: List<PublicKey>
-    ) : AbstractAttachment(dataLoader, uploader), SerializeAsToken {
+            override val id: SecureHash,
+            dataLoader: () -> ByteArray,
+            private val checkOnLoad: Boolean,
+            uploader: String?,
+            override val signerKeys: List<PublicKey>,
+            override val kotlinMetadataVersion: String?
+    ) : AbstractAttachment(dataLoader, uploader), InternalAttachment, SerializeAsToken {
 
         override fun open(): InputStream {
             val stream = super.open()
@@ -270,22 +276,24 @@ class NodeAttachmentService @JvmOverloads constructor(
         }
 
         private class Token(
-            private val id: SecureHash,
-            private val checkOnLoad: Boolean,
-            private val uploader: String?,
-            private val signerKeys: List<PublicKey>
+                private val id: SecureHash,
+                private val checkOnLoad: Boolean,
+                private val uploader: String?,
+                private val signerKeys: List<PublicKey>,
+                private val kotlinMetadataVersion: String?
         ) : SerializationToken {
             override fun fromToken(context: SerializeAsTokenContext) = AttachmentImpl(
-                id,
-                context.attachmentDataLoader(id),
-                checkOnLoad,
-                uploader,
-                signerKeys
+                    id,
+                    context.attachmentDataLoader(id),
+                    checkOnLoad,
+                    uploader,
+                    signerKeys,
+                    kotlinMetadataVersion
             )
         }
 
         override fun toToken(context: SerializeAsTokenContext) =
-            Token(id, checkOnLoad, uploader, signerKeys)
+            Token(id, checkOnLoad, uploader, signerKeys, kotlinMetadataVersion)
     }
 
     private val attachmentContentCache = NonInvalidatingWeightBasedCache(
@@ -303,16 +311,27 @@ class NodeAttachmentService @JvmOverloads constructor(
         }
     }
 
+    @OptIn(UnstableMetadataApi::class)
     private fun createAttachmentFromDatabase(attachment: DBAttachment): Attachment {
+        // TODO Cache this as a column in the database
+        val jis = JarInputStream(attachment.content.inputStream())
+        val kotlinMetadataVersions = jis.entries()
+                .filter { it.name.endsWith(".kotlin_module") }
+                .map { KotlinModuleMetadata.read(jis.readAllBytes()).version }
+                .toSortedSet()
+        if (kotlinMetadataVersions.size > 1) {
+            log.warn("Attachment ${attachment.attId} seems to be comprised of multiple Kotlin versions: $kotlinMetadataVersions")
+        }
         val attachmentImpl = AttachmentImpl(
-            id = SecureHash.create(attachment.attId),
-            dataLoader = { attachment.content },
-            checkOnLoad = checkAttachmentsOnLoad,
-            uploader = attachment.uploader,
-            signerKeys = attachment.signers?.toList() ?: emptyList()
+                id = SecureHash.create(attachment.attId),
+                dataLoader = { attachment.content },
+                checkOnLoad = checkAttachmentsOnLoad,
+                uploader = attachment.uploader,
+                signerKeys = attachment.signers?.toList() ?: emptyList(),
+                kotlinMetadataVersion = kotlinMetadataVersions.takeIf { it.isNotEmpty() }?.last()?.toString()
         )
         val contracts = attachment.contractClassNames
-        return if (contracts != null && contracts.isNotEmpty()) {
+        return if (!contracts.isNullOrEmpty()) {
             ContractAttachment.create(
                 attachment = attachmentImpl,
                 contract = contracts.first(),
@@ -336,7 +355,7 @@ class NodeAttachmentService @JvmOverloads constructor(
         return null
     }
 
-    @Suppress("OverridingDeprecatedMember")
+    @Suppress("OVERRIDE_DEPRECATION")
     override fun importAttachment(jar: InputStream): AttachmentId {
         return import(jar, UNKNOWN_UPLOADER, null)
     }
@@ -360,7 +379,7 @@ class NodeAttachmentService @JvmOverloads constructor(
     override fun privilegedImportOrGetAttachment(jar: InputStream, uploader: String, filename: String?): AttachmentId {
         return try {
             import(jar, uploader, filename)
-        } catch (faee: java.nio.file.FileAlreadyExistsException) {
+        } catch (faee: FileAlreadyExistsException) {
             AttachmentId.create(faee.message!!)
         }
     }
@@ -447,18 +466,14 @@ class NodeAttachmentService @JvmOverloads constructor(
 
     private fun getVersion(attachmentBytes: ByteArray) =
             JarInputStream(attachmentBytes.inputStream()).use {
-                try {
-                    it.manifest?.mainAttributes?.getValue(CORDAPP_CONTRACT_VERSION)?.toInt() ?: DEFAULT_CORDAPP_VERSION
-                } catch (e: NumberFormatException) {
-                    DEFAULT_CORDAPP_VERSION
-                }
+                it.manifest?.mainAttributes?.getValue(CORDAPP_CONTRACT_VERSION)?.toIntOrNull() ?: DEFAULT_CORDAPP_VERSION
             }
 
-    @Suppress("OverridingDeprecatedMember")
+    @Suppress("OVERRIDE_DEPRECATION")
     override fun importOrGetAttachment(jar: InputStream): AttachmentId {
         return try {
             import(jar, UNKNOWN_UPLOADER, null)
-        } catch (faee: java.nio.file.FileAlreadyExistsException) {
+        } catch (faee: FileAlreadyExistsException) {
             AttachmentId.create(faee.message!!)
         }
     }
diff --git a/node/src/main/kotlin/net/corda/node/verification/ExternalVerifierHandle.kt b/node/src/main/kotlin/net/corda/node/verification/ExternalVerifierHandleImpl.kt
similarity index 86%
rename from node/src/main/kotlin/net/corda/node/verification/ExternalVerifierHandle.kt
rename to node/src/main/kotlin/net/corda/node/verification/ExternalVerifierHandleImpl.kt
index aa82ef45d9..0108e78dde 100644
--- a/node/src/main/kotlin/net/corda/node/verification/ExternalVerifierHandle.kt
+++ b/node/src/main/kotlin/net/corda/node/verification/ExternalVerifierHandleImpl.kt
@@ -3,14 +3,16 @@ package net.corda.node.verification
 import net.corda.core.contracts.Attachment
 import net.corda.core.internal.AbstractAttachment
 import net.corda.core.internal.copyTo
+import net.corda.core.internal.level
 import net.corda.core.internal.mapToSet
 import net.corda.core.internal.readFully
+import net.corda.core.internal.verification.ExternalVerifierHandle
+import net.corda.core.internal.verification.NodeVerificationSupport
 import net.corda.core.serialization.serialize
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.Try
 import net.corda.core.utilities.contextLogger
 import net.corda.core.utilities.debug
-import net.corda.node.services.api.ServiceHubInternal
 import net.corda.serialization.internal.GeneratedAttachment
 import net.corda.serialization.internal.amqp.AbstractAMQPSerializationScheme.Companion.customSerializers
 import net.corda.serialization.internal.amqp.AbstractAMQPSerializationScheme.Companion.serializationWhitelists
@@ -49,7 +51,10 @@ import kotlin.io.path.div
 /**
  * Handle to the node's external verifier. The verifier process is started lazily on the first verification request.
  */
-class ExternalVerifierHandle(private val serviceHub: ServiceHubInternal) : AutoCloseable {
+class ExternalVerifierHandleImpl(
+        private val verificationSupport: NodeVerificationSupport,
+        private val baseDirectory: Path
+) : ExternalVerifierHandle {
     companion object {
         private val log = contextLogger()
 
@@ -69,12 +74,12 @@ class ExternalVerifierHandle(private val serviceHub: ServiceHubInternal) : AutoC
     @Volatile
     private var connection: Connection? = null
 
-    fun verifyTransaction(stx: SignedTransaction, checkSufficientSignatures: Boolean) {
+    override fun verifyTransaction(stx: SignedTransaction, checkSufficientSignatures: Boolean) {
         log.info("Verify $stx externally, checkSufficientSignatures=$checkSufficientSignatures")
         // By definition input states are unique, and so it makes sense to eagerly send them across with the transaction.
         // Reference states are not, but for now we'll send them anyway and assume they aren't used often. If this assumption is not
         // correct, and there's a benefit, then we can send them lazily.
-        val stxInputsAndReferences = (stx.inputs + stx.references).associateWith(serviceHub::getSerializedState)
+        val stxInputsAndReferences = (stx.inputs + stx.references).associateWith(verificationSupport::getSerializedState)
         val request = VerificationRequest(stx, stxInputsAndReferences, checkSufficientSignatures)
 
         // To keep things simple the verifier only supports one verification request at a time.
@@ -140,11 +145,11 @@ class ExternalVerifierHandle(private val serviceHub: ServiceHubInternal) : AutoC
 
     private fun processVerifierRequest(request: VerifierRequest, connection: Connection) {
         val result = when (request) {
-            is GetParties -> PartiesResult(serviceHub.getParties(request.keys))
-            is GetAttachment -> AttachmentResult(prepare(serviceHub.attachments.openAttachment(request.id)))
-            is GetAttachments -> AttachmentsResult(serviceHub.getAttachments(request.ids).map(::prepare))
-            is GetNetworkParameters -> NetworkParametersResult(serviceHub.getNetworkParameters(request.id))
-            is GetTrustedClassAttachment -> TrustedClassAttachmentResult(serviceHub.getTrustedClassAttachment(request.className)?.id)
+            is GetParties -> PartiesResult(verificationSupport.getParties(request.keys))
+            is GetAttachment -> AttachmentResult(prepare(verificationSupport.getAttachment(request.id)))
+            is GetAttachments -> AttachmentsResult(verificationSupport.getAttachments(request.ids).map(::prepare))
+            is GetNetworkParameters -> NetworkParametersResult(verificationSupport.getNetworkParameters(request.id))
+            is GetTrustedClassAttachment -> TrustedClassAttachmentResult(verificationSupport.getTrustedClassAttachment(request.className)?.id)
         }
         log.debug { "Sending response to external verifier: $result" }
         connection.toVerifier.writeCordaSerializable(result)
@@ -152,7 +157,7 @@ class ExternalVerifierHandle(private val serviceHub: ServiceHubInternal) : AutoC
 
     private fun prepare(attachment: Attachment?): AttachmentWithTrust? {
         if (attachment == null) return null
-        val isTrusted = serviceHub.isAttachmentTrusted(attachment)
+        val isTrusted = verificationSupport.isAttachmentTrusted(attachment)
         val attachmentForSer = when (attachment) {
             // The Attachment retrieved from the database is not serialisable, so we have to convert it into one
             is AbstractAttachment -> GeneratedAttachment(attachment.open().readFully(), attachment.uploader)
@@ -188,20 +193,20 @@ class ExternalVerifierHandle(private val serviceHub: ServiceHubInternal) : AutoC
                     "-jar",
                     "$verifierJar",
                     "${server.localPort}",
-                    System.getProperty("log4j2.level")?.lowercase() ?: "info"
+                    log.level.name.lowercase()
             )
             log.debug { "Verifier command: $command" }
-            val logsDirectory = (serviceHub.configuration.baseDirectory / "logs").createDirectories()
+            val logsDirectory = (baseDirectory / "logs").createDirectories()
             verifierProcess = ProcessBuilder(command)
                     .redirectOutput(Redirect.appendTo((logsDirectory / "verifier-stdout.log").toFile()))
                     .redirectError(Redirect.appendTo((logsDirectory / "verifier-stderr.log").toFile()))
-                    .directory(serviceHub.configuration.baseDirectory.toFile())
+                    .directory(baseDirectory.toFile())
                     .start()
             log.info("External verifier process started; PID ${verifierProcess.pid()}")
 
             verifierProcess.onExit().whenComplete { _, _ ->
                 if (connection != null) {
-                    log.error("The external verifier has unexpectedly terminated with error code ${verifierProcess.exitValue()}. " +
+                    log.warn("The external verifier has unexpectedly terminated with error code ${verifierProcess.exitValue()}. " +
                             "Please check verifier logs for more details.")
                 }
                 // Allow a new process to be started on the next verification request
@@ -212,12 +217,12 @@ class ExternalVerifierHandle(private val serviceHub: ServiceHubInternal) : AutoC
             toVerifier = DataOutputStream(socket.outputStream)
             fromVerifier = DataInputStream(socket.inputStream)
 
-            val cordapps = serviceHub.cordappProvider.cordapps
+            val cordapps = verificationSupport.cordappProvider.cordapps
             val initialisation = Initialisation(
                     customSerializerClassNames = cordapps.customSerializers.mapToSet { it.javaClass.name },
                     serializationWhitelistClassNames = cordapps.serializationWhitelists.mapToSet { it.javaClass.name },
                     System.getProperty("experimental.corda.customSerializationScheme"), // See Node#initialiseSerialization
-                    serializedCurrentNetworkParameters = serviceHub.networkParameters.serialize()
+                    serializedCurrentNetworkParameters = verificationSupport.networkParameters.serialize()
             )
             toVerifier.writeCordaSerializable(initialisation)
         }
diff --git a/settings.gradle b/settings.gradle
index 67e26a11be..6193fd6ef2 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -110,6 +110,7 @@ include 'testing:cordapps:dbfailure:dbfworkflows'
 include 'testing:cordapps:missingmigration'
 include 'testing:cordapps:sleeping'
 include 'testing:cordapps:cashobservers'
+include 'testing:cordapps:4.11-workflows'
 
 // Common libraries - start
 include 'common-validation'
diff --git a/testing/cordapps/4.11-workflows/build.gradle b/testing/cordapps/4.11-workflows/build.gradle
new file mode 100644
index 0000000000..86b1ff21e5
--- /dev/null
+++ b/testing/cordapps/4.11-workflows/build.gradle
@@ -0,0 +1,17 @@
+apply plugin: 'corda.kotlin-1.2'
+
+dependencies {
+    compileOnly "net.corda:corda-core:4.11.+"
+    compileOnly "net.corda:corda-finance-contracts:4.11.+"
+    compileOnly "net.corda:corda-finance-workflows:4.11.+"
+}
+
+jar {
+    archiveBaseName = "4.11-workflows-cordapp"
+    archiveVersion = ""
+    manifest {
+        // This JAR is part of Corda's testing framework.
+        // Driver will not include it as part of an out-of-process node.
+        attributes('Corda-Testing': true)
+    }
+}
diff --git a/testing/cordapps/4.11-workflows/src/main/kotlin/net/corda/testing/cordapps/workflows411/IssueAndChangeNotaryFlow.kt b/testing/cordapps/4.11-workflows/src/main/kotlin/net/corda/testing/cordapps/workflows411/IssueAndChangeNotaryFlow.kt
new file mode 100644
index 0000000000..ccbcf5b3ee
--- /dev/null
+++ b/testing/cordapps/4.11-workflows/src/main/kotlin/net/corda/testing/cordapps/workflows411/IssueAndChangeNotaryFlow.kt
@@ -0,0 +1,30 @@
+package net.corda.testing.cordapps.workflows411
+
+import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.crypto.SecureHash
+import net.corda.core.flows.FlowLogic
+import net.corda.core.flows.NotaryChangeFlow
+import net.corda.core.flows.StartableByRPC
+import net.corda.core.identity.Party
+import net.corda.core.transactions.NotaryChangeWireTransaction
+import net.corda.core.utilities.OpaqueBytes
+import net.corda.finance.DOLLARS
+import net.corda.finance.contracts.asset.Cash
+import net.corda.finance.flows.CashIssueFlow
+
+// We need a separate flow as NotaryChangeFlow is not StartableByRPC
+@StartableByRPC
+class IssueAndChangeNotaryFlow(private val oldNotary: Party, private val newNotary: Party) : FlowLogic<SecureHash>() {
+    @Suppress("MagicNumber")
+    @Suspendable
+    override fun call(): SecureHash {
+        subFlow(CashIssueFlow(10.DOLLARS, OpaqueBytes.of(0x01), oldNotary))
+        val oldState = serviceHub.vaultService.queryBy(Cash.State::class.java).states.single()
+        check(oldState.state.notary == oldNotary) { oldState.state.notary }
+        val newState = subFlow(NotaryChangeFlow(oldState, newNotary))
+        check(newState.state.notary == newNotary) { newState.state.notary }
+        val notaryChangeTx = checkNotNull(serviceHub.validatedTransactions.getTransaction(newState.ref.txhash))
+        check(notaryChangeTx.coreTransaction is NotaryChangeWireTransaction) { notaryChangeTx.coreTransaction }
+        return notaryChangeTx.id
+    }
+}
diff --git a/testing/core-test-utils/src/main/kotlin/net/corda/testing/core/internal/JarSignatureTestUtils.kt b/testing/core-test-utils/src/main/kotlin/net/corda/testing/core/internal/JarSignatureTestUtils.kt
index b8e48ecbe0..a3e2d78d12 100644
--- a/testing/core-test-utils/src/main/kotlin/net/corda/testing/core/internal/JarSignatureTestUtils.kt
+++ b/testing/core-test-utils/src/main/kotlin/net/corda/testing/core/internal/JarSignatureTestUtils.kt
@@ -7,6 +7,7 @@ import net.corda.nodeapi.internal.crypto.loadKeyStore
 import java.io.Closeable
 import java.io.FileInputStream
 import java.io.FileOutputStream
+import java.nio.file.FileSystems
 import java.nio.file.Files
 import java.nio.file.NoSuchFileException
 import java.nio.file.Path
@@ -17,7 +18,11 @@ import java.util.jar.Attributes
 import java.util.jar.JarInputStream
 import java.util.jar.JarOutputStream
 import java.util.jar.Manifest
+import kotlin.io.path.deleteExisting
 import kotlin.io.path.div
+import kotlin.io.path.inputStream
+import kotlin.io.path.listDirectoryEntries
+import kotlin.io.path.outputStream
 import kotlin.test.assertEquals
 
 /**
@@ -36,7 +41,6 @@ object JarSignatureTestUtils {
     private fun Path.executeProcess(vararg command: String) {
         val shredder = (this / "_shredder").toFile() // No need to delete after each test.
         assertEquals(0, ProcessBuilder()
-                .inheritIO()
                 .redirectOutput(shredder)
                 .redirectError(shredder)
                 .directory(this.toFile())
@@ -69,6 +73,16 @@ object JarSignatureTestUtils {
         return ks.getCertificate(alias).publicKey
     }
 
+    fun Path.unsignJar() {
+        FileSystems.newFileSystem(this).use { zipFs ->
+            zipFs.getPath("META-INF").listDirectoryEntries("*.{SF,DSA,RSA,EC}").forEach(Path::deleteExisting)
+            val manifestFile = zipFs.getPath("META-INF", "MANIFEST.MF")
+            val manifest = manifestFile.inputStream().use(::Manifest)
+            manifest.entries.clear()  // Remove all the hash information of the jar contents
+            manifestFile.outputStream().use(manifest::write)
+        }
+    }
+
     fun Path.getPublicKey(alias: String, storeName: String, storePassword: String) : PublicKey {
         val ks = loadKeyStore(this.resolve(storeName), storePassword)
         return ks.getCertificate(alias).publicKey
diff --git a/testing/node-driver/src/main/kotlin/net/corda/testing/node/MockServices.kt b/testing/node-driver/src/main/kotlin/net/corda/testing/node/MockServices.kt
index b29d633680..6348ce0873 100644
--- a/testing/node-driver/src/main/kotlin/net/corda/testing/node/MockServices.kt
+++ b/testing/node-driver/src/main/kotlin/net/corda/testing/node/MockServices.kt
@@ -10,11 +10,12 @@ import net.corda.core.contracts.StateRef
 import net.corda.core.contracts.TransactionState
 import net.corda.core.cordapp.CordappProvider
 import net.corda.core.crypto.SecureHash
+import net.corda.core.crypto.sha256
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.StateMachineRunId
 import net.corda.core.identity.CordaX500Name
-import net.corda.core.identity.Party
 import net.corda.core.identity.PartyAndCertificate
+import net.corda.core.internal.AbstractAttachment
 import net.corda.core.internal.PLATFORM_VERSION
 import net.corda.core.internal.VisibleForTesting
 import net.corda.core.internal.cordapp.CordappProviderInternal
@@ -23,6 +24,7 @@ import net.corda.core.internal.mapToSet
 import net.corda.core.internal.requireSupportedHashType
 import net.corda.core.internal.telemetry.TelemetryComponent
 import net.corda.core.internal.telemetry.TelemetryServiceImpl
+import net.corda.core.internal.verification.ExternalVerifierHandle
 import net.corda.core.internal.verification.VerifyingServiceHub
 import net.corda.core.messaging.DataFeed
 import net.corda.core.messaging.FlowHandle
@@ -302,22 +304,19 @@ open class MockServices private constructor(
         // Because Kotlin is dumb and makes not publicly visible objects public, thus changing the public API.
         private val mockStateMachineRecordedTransactionMappingStorage = MockStateMachineRecordedTransactionMappingStorage()
 
-        private val dummyAttachment by lazy {
-            val inputStream = ByteArrayOutputStream().apply {
-                ZipOutputStream(this).use {
-                    with(it) {
-                        putNextEntry(ZipEntry(JarFile.MANIFEST_NAME))
-                    }
-                }
-            }.toByteArray().inputStream()
-            val attachment = object : Attachment {
-                override val id get() = throw UnsupportedOperationException()
-                override fun open() = inputStream
-                override val signerKeys get() = throw UnsupportedOperationException()
-                override val signers: List<Party> get() = throw UnsupportedOperationException()
-                override val size: Int = 512
+        private val dummyAttachment: Attachment by lazy {
+            object : AbstractAttachment(
+                    {
+                        val baos = ByteArrayOutputStream()
+                        ZipOutputStream(baos).use { zip ->
+                            zip.putNextEntry(ZipEntry(JarFile.MANIFEST_NAME))
+                        }
+                        baos.toByteArray()
+                    },
+                    null
+            ) {
+                override val id: SecureHash by lazy(attachmentData::sha256)
             }
-            attachment
         }
     }
 
@@ -576,6 +575,9 @@ open class MockServices private constructor(
         override fun loadState(stateRef: StateRef): TransactionState<*> = mockServices.loadState(stateRef)
 
         override fun loadStates(stateRefs: Set<StateRef>): Set<StateAndRef<ContractState>> = mockServices.loadStates(stateRefs)
+
+        override val externalVerifierHandle: ExternalVerifierHandle
+            get() = throw UnsupportedOperationException("External verification is not supported by MockServices")
     }
 
 
diff --git a/testing/smoke-test-utils/build.gradle b/testing/smoke-test-utils/build.gradle
index 73a6c27af8..a0f72cfce6 100644
--- a/testing/smoke-test-utils/build.gradle
+++ b/testing/smoke-test-utils/build.gradle
@@ -9,6 +9,7 @@ dependencies {
     implementation project(':test-common')
     implementation project(':client:rpc')
 
+    implementation "com.google.guava:guava:$guava_version"
     implementation "com.typesafe:config:$typesafe_config_version"
     implementation "org.slf4j:slf4j-api:$slf4j_version"
 }
diff --git a/testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeConfig.kt b/testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeParams.kt
similarity index 79%
rename from testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeConfig.kt
rename to testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeParams.kt
index 120faa5314..efe68ad2f1 100644
--- a/testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeConfig.kt
+++ b/testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeParams.kt
@@ -1,22 +1,25 @@
 package net.corda.smoketesting
 
-import com.typesafe.config.Config
 import com.typesafe.config.ConfigFactory.empty
 import com.typesafe.config.ConfigRenderOptions
 import com.typesafe.config.ConfigValue
 import com.typesafe.config.ConfigValueFactory
+import net.corda.client.rpc.CordaRPCClientConfiguration
 import net.corda.core.identity.CordaX500Name
 import net.corda.nodeapi.internal.config.User
 import net.corda.nodeapi.internal.config.toConfig
+import java.nio.file.Path
 
-class NodeConfig(
+class NodeParams @JvmOverloads constructor(
         val legalName: CordaX500Name,
         val p2pPort: Int,
         val rpcPort: Int,
         val rpcAdminPort: Int,
-        val isNotary: Boolean,
         val users: List<User>,
-        val devMode: Boolean = true
+        val cordappJars: List<Path> = emptyList(),
+        val clientRpcConfig: CordaRPCClientConfiguration = CordaRPCClientConfiguration.DEFAULT,
+        val devMode: Boolean = true,
+        val version: String? = null
 ) {
     companion object {
         val renderOptions: ConfigRenderOptions = ConfigRenderOptions.defaults().setOriginComments(false)
@@ -24,12 +27,7 @@ class NodeConfig(
 
     val commonName: String get() = legalName.organisation
 
-    /*
-     * The configuration object depends upon the networkMap,
-     * which is mutable.
-     */
-    //TODO Make use of Any.toConfig
-    private fun toFileConfig(): Config {
+    fun createNodeConfig(isNotary: Boolean): String {
         val config = empty()
                 .withValue("myLegalName", valueFor(legalName.toString()))
                 .withValue("p2pAddress", addressValueFor(p2pPort))
@@ -44,11 +42,9 @@ class NodeConfig(
             config.withValue("notary", ConfigValueFactory.fromMap(mapOf("validating" to true)))
         } else {
             config
-        }
+        }.root().render(renderOptions)
     }
 
-    fun toText(): String = toFileConfig().root().render(renderOptions)
-
     private fun <T> valueFor(any: T): ConfigValue? = ConfigValueFactory.fromAnyRef(any)
 
     private fun addressValueFor(port: Int) = valueFor("localhost:$port")
diff --git a/testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeProcess.kt b/testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeProcess.kt
index 216db2e120..41f06b28ca 100644
--- a/testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeProcess.kt
+++ b/testing/smoke-test-utils/src/main/kotlin/net/corda/smoketesting/NodeProcess.kt
@@ -1,16 +1,20 @@
 package net.corda.smoketesting
 
+import com.google.common.collect.Lists
 import net.corda.client.rpc.CordaRPCClient
 import net.corda.client.rpc.CordaRPCConnection
-import net.corda.core.identity.Party
+import net.corda.core.internal.PLATFORM_VERSION
+import net.corda.core.internal.copyToDirectory
 import net.corda.core.internal.deleteRecursively
 import net.corda.core.internal.toPath
+import net.corda.core.node.NetworkParameters
 import net.corda.core.node.NotaryInfo
 import net.corda.core.utilities.NetworkHostAndPort
 import net.corda.core.utilities.contextLogger
 import net.corda.nodeapi.internal.DevIdentityGenerator
 import net.corda.nodeapi.internal.config.User
 import net.corda.nodeapi.internal.network.NetworkParametersCopier
+import net.corda.nodeapi.internal.network.NodeInfoFilesCopier
 import net.corda.nodeapi.internal.rpc.client.AMQPClientSerializationScheme
 import net.corda.testing.common.internal.asContextEnv
 import net.corda.testing.common.internal.checkNotOnClasspath
@@ -20,15 +24,18 @@ import java.nio.file.Paths
 import java.time.Instant
 import java.time.ZoneId.systemDefault
 import java.time.format.DateTimeFormatter
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit.SECONDS
+import kotlin.io.path.Path
 import kotlin.io.path.createDirectories
+import kotlin.io.path.createDirectory
 import kotlin.io.path.div
 import kotlin.io.path.writeText
 
 class NodeProcess(
-        private val config: NodeConfig,
-        private val nodeDir: Path,
+        private val config: NodeParams,
+        val nodeDir: Path,
         private val node: Process,
         private val client: CordaRPCClient
 ) : AutoCloseable {
@@ -43,6 +50,7 @@ class NodeProcess(
     }
 
     override fun close() {
+        if (!node.isAlive) return
         log.info("Stopping node '${config.commonName}'")
         node.destroy()
         if (!node.waitFor(60, SECONDS)) {
@@ -56,65 +64,94 @@ class NodeProcess(
 
     // TODO All use of this factory have duplicate code which is either bundling the calling module or a 3rd party module
     // as a CorDapp for the nodes.
-    class Factory(private val buildDirectory: Path = Paths.get("build")) {
-        val cordaJar: Path by lazy {
-            val cordaJarUrl = requireNotNull(this::class.java.getResource("/corda.jar")) {
-                "corda.jar could not be found in classpath"
-            }
-            cordaJarUrl.toPath()
-        }
+    class Factory(
+            private val baseNetworkParameters: NetworkParameters = testNetworkParameters(minimumPlatformVersion = PLATFORM_VERSION),
+            private val buildDirectory: Path = Paths.get("build")
+    ) : AutoCloseable {
+        companion object {
+            private val formatter = DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss.SSS").withZone(systemDefault())
+            private val cordaJars = ConcurrentHashMap<String, CordaJar>()
 
-        private companion object {
-            val javaPath: Path = Paths.get(System.getProperty("java.home"), "bin", "java")
-            val formatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss.SSS").withZone(systemDefault())
             init {
                 checkNotOnClasspath("net.corda.node.Corda") {
                     "Smoke test has the node in its classpath. Please remove the offending dependency."
                 }
             }
+
+            @Suppress("MagicNumber")
+            private fun getCordaJarInfo(version: String): CordaJar {
+                return cordaJars.computeIfAbsent(version) {
+                    val (javaHome, versionSuffix) = if (version.isEmpty()) {
+                        System.getProperty("java.home") to ""
+                    } else {
+                        val javaHome = if (version.split(".")[1].toInt() > 11) {
+                            System.getProperty("java.home")
+                        } else {
+                            // 4.11 and below need JDK 8 to run
+                            checkNotNull(System.getenv("JAVA_8_HOME")) { "Please set JAVA_8_HOME env variable to home directory of JDK 8" }
+                        }
+                        javaHome to "-$version"
+                    }
+                    val cordaJar = this::class.java.getResource("/corda$versionSuffix.jar")!!.toPath()
+                    CordaJar(cordaJar, Path(javaHome, "bin", "java"))
+                }
+            }
+
+            fun getCordaJar(version: String? = null): Path = getCordaJarInfo(version ?: "").jarPath
         }
 
+        private val nodesDirectory: Path = (buildDirectory / "smoke-testing" / formatter.format(Instant.now())).createDirectories()
+        private val nodeInfoFilesCopier = NodeInfoFilesCopier()
+        private var nodes: MutableList<NodeProcess>? = ArrayList()
         private lateinit var networkParametersCopier: NetworkParametersCopier
 
-        private val nodesDirectory = (buildDirectory / formatter.format(Instant.now())).createDirectories()
+        fun baseDirectory(config: NodeParams): Path = nodesDirectory / config.commonName
 
-        private var notaryParty: Party? = null
+        fun createNotaries(first: NodeParams, vararg rest: NodeParams): List<NodeProcess> {
+            check(!::networkParametersCopier.isInitialized) { "Notaries have already been created" }
 
-        private fun createNetworkParameters(notaryInfo: NotaryInfo, nodeDir: Path) {
-            try {
-                networkParametersCopier = NetworkParametersCopier(testNetworkParameters(notaries = listOf(notaryInfo)))
+            val notariesParams = Lists.asList(first, rest)
+            val notaryInfos = notariesParams.map { notaryParams ->
+                val nodeDir = baseDirectory(notaryParams).createDirectories()
+                val notaryParty = DevIdentityGenerator.installKeyStoreWithNodeIdentity(nodeDir, notaryParams.legalName)
+                NotaryInfo(notaryParty, true)
+            }
+            val networkParameters = baseNetworkParameters.copy(notaries = notaryInfos)
+            networkParametersCopier = try {
+                NetworkParametersCopier(networkParameters)
             } catch (_: IllegalStateException) {
                 // Assuming serialization env not in context.
                 AMQPClientSerializationScheme.createSerializationEnv().asContextEnv {
-                    networkParametersCopier = NetworkParametersCopier(testNetworkParameters(notaries = listOf(notaryInfo)))
+                    NetworkParametersCopier(networkParameters)
                 }
             }
-            networkParametersCopier.install(nodeDir)
+
+            return notariesParams.map { createNode(it, isNotary = true) }
         }
 
-        fun baseDirectory(config: NodeConfig): Path = nodesDirectory / config.commonName
+        fun createNode(params: NodeParams): NodeProcess = createNode(params, isNotary = false)
 
-        fun create(config: NodeConfig): NodeProcess {
-            val nodeDir = baseDirectory(config).createDirectories()
+        private fun createNode(params: NodeParams, isNotary: Boolean): NodeProcess {
+            check(::networkParametersCopier.isInitialized) { "Notary not created. Please call `creatNotaries` first." }
+
+            val nodeDir = baseDirectory(params).createDirectories()
             log.info("Node directory: {}", nodeDir)
-            if (config.isNotary) {
-                require(notaryParty == null) { "Only one notary can be created." }
-                notaryParty = DevIdentityGenerator.installKeyStoreWithNodeIdentity(nodeDir, config.legalName)
-            } else {
-                require(notaryParty != null) { "Notary not created. Please call `create` with a notary config first." }
-            }
+            val cordappsDir = (nodeDir / CORDAPPS_DIR_NAME).createDirectory()
+            params.cordappJars.forEach { it.copyToDirectory(cordappsDir) }
+            (nodeDir / "node.conf").writeText(params.createNodeConfig(isNotary))
+            networkParametersCopier.install(nodeDir)
+            nodeInfoFilesCopier.addConfig(nodeDir)
 
-            (nodeDir / "node.conf").writeText(config.toText())
-            createNetworkParameters(NotaryInfo(notaryParty!!, true), nodeDir)
-
-            createSchema(nodeDir)
-            val process = startNode(nodeDir)
-            val client = CordaRPCClient(NetworkHostAndPort("localhost", config.rpcPort))
-            waitForNode(process, config, client)
-            return NodeProcess(config, nodeDir, process, client)
+            createSchema(nodeDir, params.version)
+            val process = startNode(nodeDir, params.version)
+            val client = CordaRPCClient(NetworkHostAndPort("localhost", params.rpcPort), params.clientRpcConfig)
+            waitForNode(process, params, client)
+            val nodeProcess = NodeProcess(params, nodeDir, process, client)
+            nodes!! += nodeProcess
+            return nodeProcess
         }
 
-        private fun waitForNode(process: Process, config: NodeConfig, client: CordaRPCClient) {
+        private fun waitForNode(process: Process, config: NodeParams, client: CordaRPCClient) {
             val executor = Executors.newSingleThreadScheduledExecutor()
             try {
                 executor.scheduleWithFixedDelay({
@@ -129,7 +166,7 @@ class NodeProcess(
                         // Cancel the "setup" task now that we've created the RPC client.
                         executor.shutdown()
                     } catch (e: Exception) {
-                        log.warn("Node '{}' not ready yet (Error: {})", config.commonName, e.message)
+                        log.debug("Node '{}' not ready yet (Error: {})", config.commonName, e.message)
                     }
                 }, 5, 1, SECONDS)
 
@@ -147,10 +184,10 @@ class NodeProcess(
         class SchemaCreationFailedError(nodeDir: Path) : Exception("Creating node schema failed for $nodeDir")
 
 
-        private fun createSchema(nodeDir: Path){
-            val process = startNode(nodeDir, "run-migration-scripts", "--core-schemas", "--app-schemas")
+        private fun createSchema(nodeDir: Path, version: String?) {
+            val process = startNode(nodeDir, version, "run-migration-scripts", "--core-schemas", "--app-schemas")
             if (!process.waitFor(schemaCreationTimeOutSeconds, SECONDS)) {
-                process.destroy()
+                process.destroyForcibly()
                 throw SchemaCreationTimedOutError(nodeDir)
             }
             if (process.exitValue() != 0) {
@@ -158,8 +195,9 @@ class NodeProcess(
             }
         }
 
-        private fun startNode(nodeDir: Path, vararg extraArgs: String): Process {
-            val command = arrayListOf(javaPath.toString(), "-Dcapsule.log=verbose", "-jar", cordaJar.toString())
+        private fun startNode(nodeDir: Path, version: String?, vararg extraArgs: String): Process {
+            val cordaJar = getCordaJarInfo(version ?: "")
+            val command = arrayListOf("${cordaJar.javaPath}", "-Dcapsule.log=verbose", "-jar", "${cordaJar.jarPath}", "--logging-level=debug")
             command += extraArgs
             val now = formatter.format(Instant.now())
             val builder = ProcessBuilder()
@@ -171,7 +209,17 @@ class NodeProcess(
                     "CAPSULE_CACHE_DIR" to (buildDirectory / "capsule").toString()
             ))
 
-            return builder.start()
+            val process = builder.start()
+            Runtime.getRuntime().addShutdownHook(Thread(process::destroyForcibly))
+            return process
         }
+
+        override fun close() {
+            nodes?.parallelStream()?.forEach { it.close() }
+            nodes = null
+            nodeInfoFilesCopier.close()
+        }
+
+        private data class CordaJar(val jarPath: Path, val javaPath: Path)
     }
 }
diff --git a/testing/test-utils/src/main/kotlin/net/corda/testing/dsl/TestDSL.kt b/testing/test-utils/src/main/kotlin/net/corda/testing/dsl/TestDSL.kt
index eb52d9d614..4199288630 100644
--- a/testing/test-utils/src/main/kotlin/net/corda/testing/dsl/TestDSL.kt
+++ b/testing/test-utils/src/main/kotlin/net/corda/testing/dsl/TestDSL.kt
@@ -13,6 +13,7 @@ import net.corda.core.identity.Party
 import net.corda.core.internal.*
 import net.corda.core.internal.cordapp.CordappProviderInternal
 import net.corda.core.internal.notary.NotaryService
+import net.corda.core.internal.verification.ExternalVerifierHandle
 import net.corda.core.node.ServiceHub
 import net.corda.core.node.ServicesForResolution
 import net.corda.core.node.StatesToRecord
@@ -139,6 +140,9 @@ data class TestTransactionDSLInterpreter private constructor(
             return ledgerInterpreter.services.loadContractAttachment(stateRef)
         }
 
+        override val externalVerifierHandle: ExternalVerifierHandle
+            get() = throw UnsupportedOperationException("External verification is not supported by TestTransactionDSLInterpreter")
+
         override fun recordUnnotarisedTransaction(txn: SignedTransaction) {}
 
         override fun removeUnnotarisedTransaction(id: SecureHash) {}
diff --git a/testing/test-utils/src/main/kotlin/net/corda/testing/services/MockAttachmentStorage.kt b/testing/test-utils/src/main/kotlin/net/corda/testing/services/MockAttachmentStorage.kt
index c220bcfb1b..7b589f5602 100644
--- a/testing/test-utils/src/main/kotlin/net/corda/testing/services/MockAttachmentStorage.kt
+++ b/testing/test-utils/src/main/kotlin/net/corda/testing/services/MockAttachmentStorage.kt
@@ -12,12 +12,16 @@ import net.corda.core.internal.cordapp.CordappImpl.Companion.DEFAULT_CORDAPP_VER
 import net.corda.core.internal.readFully
 import net.corda.core.node.services.AttachmentId
 import net.corda.core.node.services.AttachmentStorage
-import net.corda.core.node.services.vault.*
+import net.corda.core.node.services.vault.AttachmentQueryCriteria
+import net.corda.core.node.services.vault.AttachmentSort
+import net.corda.core.node.services.vault.Builder
+import net.corda.core.node.services.vault.ColumnPredicate
+import net.corda.core.node.services.vault.Sort
 import net.corda.core.serialization.SingletonSerializeAsToken
 import net.corda.nodeapi.internal.withContractsInJar
 import java.io.InputStream
+import java.nio.file.FileAlreadyExistsException
 import java.security.PublicKey
-import java.util.*
 import java.util.jar.Attributes
 import java.util.jar.JarInputStream
 
@@ -33,7 +37,7 @@ class MockAttachmentStorage : AttachmentStorage, SingletonSerializeAsToken() {
     /** A map of the currently stored files by their [SecureHash] */
     val files: Map<SecureHash, Pair<Attachment, ByteArray>> get() = _files
 
-    @Suppress("OverridingDeprecatedMember")
+    @Suppress("OVERRIDE_DEPRECATION")
     override fun importAttachment(jar: InputStream): AttachmentId = importAttachment(jar, UNKNOWN_UPLOADER, null)
 
     override fun importAttachment(jar: InputStream, uploader: String, filename: String?): AttachmentId {
@@ -78,11 +82,11 @@ class MockAttachmentStorage : AttachmentStorage, SingletonSerializeAsToken() {
 
     override fun hasAttachment(attachmentId: AttachmentId) = files.containsKey(attachmentId)
 
-    @Suppress("OverridingDeprecatedMember")
+    @Suppress("OVERRIDE_DEPRECATION")
     override fun importOrGetAttachment(jar: InputStream): AttachmentId {
         return try {
             importAttachment(jar, UNKNOWN_UPLOADER, null)
-        } catch (e: java.nio.file.FileAlreadyExistsException) {
+        } catch (e: FileAlreadyExistsException) {
             AttachmentId.create(e.message!!)
         }
     }
@@ -109,7 +113,7 @@ class MockAttachmentStorage : AttachmentStorage, SingletonSerializeAsToken() {
             val baseAttachment = MockAttachment({ bytes }, sha256, signers, uploader)
             val version = try { Integer.parseInt(baseAttachment.openAsJAR().manifest?.mainAttributes?.getValue(Attributes.Name.IMPLEMENTATION_VERSION)) } catch (e: Exception) { DEFAULT_CORDAPP_VERSION }
             val attachment =
-                    if (contractClassNames == null || contractClassNames.isEmpty()) baseAttachment
+                    if (contractClassNames.isNullOrEmpty()) baseAttachment
                     else {
                         contractClassNames.map {contractClassName ->
                             val contractClassMetadata = ContractAttachmentMetadata(contractClassName, version, signers.isNotEmpty(), signers, uploader)
diff --git a/verifier/src/main/kotlin/net/corda/verifier/ExternalVerifier.kt b/verifier/src/main/kotlin/net/corda/verifier/ExternalVerifier.kt
index 902d03d0be..140788746c 100644
--- a/verifier/src/main/kotlin/net/corda/verifier/ExternalVerifier.kt
+++ b/verifier/src/main/kotlin/net/corda/verifier/ExternalVerifier.kt
@@ -124,9 +124,8 @@ class ExternalVerifier(
     }
 
     private fun createAppClassLoader(): ClassLoader {
-        val cordappJarUrls = (baseDirectory / "cordapps").listDirectoryEntries()
+        val cordappJarUrls = (baseDirectory / "cordapps").listDirectoryEntries("*.jar")
                 .stream()
-                .filter { it.toString().endsWith(".jar") }
                 .map { it.toUri().toURL() }
                 .toTypedArray()
         log.debug { "CorDapps: ${cordappJarUrls?.joinToString()}" }
@@ -136,7 +135,7 @@ class ExternalVerifier(
     private fun verifyTransaction(request: VerificationRequest) {
         val verificationContext = ExternalVerificationContext(appClassLoader, attachmentsClassLoaderCache, this, request.stxInputsAndReferences)
         val result: Try<Unit> = try {
-            request.stx.verifyInternal(verificationContext, request.checkSufficientSignatures)
+            request.stx.verifyInProcess(verificationContext, request.checkSufficientSignatures)
             log.info("${request.stx} verified")
             Try.Success(Unit)
         } catch (t: Throwable) {
diff --git a/verifier/src/main/kotlin/net/corda/verifier/Main.kt b/verifier/src/main/kotlin/net/corda/verifier/Main.kt
index 7507d01d5a..970498c48f 100644
--- a/verifier/src/main/kotlin/net/corda/verifier/Main.kt
+++ b/verifier/src/main/kotlin/net/corda/verifier/Main.kt
@@ -10,15 +10,14 @@ import kotlin.io.path.div
 import kotlin.system.exitProcess
 
 object Main {
-    private val log = loggerFor<Main>()
-
     @JvmStatic
     fun main(args: Array<String>) {
         val port = args[0].toInt()
-        val loggingLevel = args[0]
+        val loggingLevel = args[1]
         val baseDirectory = Path.of("").toAbsolutePath()
 
         initLogging(baseDirectory, loggingLevel)
+        val log = loggerFor<Main>()
 
         log.info("External verifier started; PID ${ProcessHandle.current().pid()}")
         log.info("Node base directory: $baseDirectory")