Ziti socket timeout stall

ziti/build.gradle.kts

@@ -135,6 +135,7 @@ testing {
         val integrationTest by registering(JvmTestSuite::class) {
             dependencies {
                 implementation(libs.kotlin.test)
+                implementation(libs.kotlin.coroutines.lib)
                 implementation(libs.kotlin.coroutines.test)
                 implementation(libs.slf4j.simple)
                 implementation(project(":management-api"))
@@ -172,7 +173,7 @@ tasks.register("start-quickstart") {
         val pb = ProcessBuilder().apply {
             command(
                 zitiCLI.toString(),
-                "edge", "quickstart", "--home", quickstartHome.asFile.absolutePath)
+                "edge", "quickstart", "--verbose", "--home", quickstartHome.asFile.absolutePath)
             redirectOutput(qsLog)
             redirectError(errLog)
         }

ziti/src/integrationTest/kotlin/org/openziti/api/ControllerTests.kt

@@ -43,12 +43,7 @@ class ControllerTests: BaseTest() {
         Ziti.setApplicationInfo(info.displayName, appVersion)
 
         idName = "id-${info.displayName}-${System.nanoTime()}"
-        val token = createIdentity(idName)
-
-        val enrollment = Ziti.createEnrollment(token)
-        assertEquals(enrollment.getMethod(), Enrollment.Method.ott)
-
-        cfg = enrollment.enroll()
+        cfg = createIdentity(idName)
     }
 
     @Test

ziti/src/integrationTest/kotlin/org/openziti/integ/BaseTest.kt

@@ -16,8 +16,12 @@
 
 package org.openziti.integ
 
+import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.TestInfo
+import org.openziti.Enrollment
+import org.openziti.IdentityConfig
+import org.openziti.Ziti
 
 abstract class BaseTest {
     protected lateinit var info: TestInfo
@@ -27,4 +31,12 @@ abstract class BaseTest {
         info = testInfo
     }
 
+    fun createIdentity(name: String = "id-${info.displayName}-${System.nanoTime()}"): IdentityConfig {
+        val token = ManagementHelper.createIdentity(name)
+
+        val enrollment = Ziti.createEnrollment(token)
+        Assertions.assertEquals(enrollment.getMethod(), Enrollment.Method.ott)
+
+        return enrollment.enroll()
+    }
 }

ziti/src/integrationTest/kotlin/org/openziti/integ/ManagementHelper.kt

@@ -18,15 +18,19 @@ package org.openziti.integ
 
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.asExecutor
+import okhttp3.internal.notifyAll
+import org.openziti.api.InterceptConfig
+import org.openziti.api.InterceptV1Cfg
 import org.openziti.management.ApiClient
+import org.openziti.management.JSON
 import org.openziti.management.api.AuthenticationApi
+import org.openziti.management.api.ConfigApi
 import org.openziti.management.api.EnrollmentApi
 import org.openziti.management.api.IdentityApi
 import org.openziti.management.api.InformationalApi
-import org.openziti.management.model.Authenticate
-import org.openziti.management.model.EnrollmentCreate
-import org.openziti.management.model.IdentityCreate
-import org.openziti.management.model.IdentityType
+import org.openziti.management.api.ServiceApi
+import org.openziti.management.api.ServicePolicyApi
+import org.openziti.management.model.*
 import org.openziti.util.fingerprint
 import org.openziti.util.parsePKCS7
 import java.net.InetAddress
@@ -69,8 +73,10 @@ internal object ManagementHelper {
         parsePKCS7(pkcs7).makeSSL()
     }
 
-    internal val identityApi by lazy { IdentityApi(api) }
-    internal val enrollmentApi by lazy { EnrollmentApi(api) }
+    private val identityApi by lazy { IdentityApi(api) }
+    private val enrollmentApi by lazy { EnrollmentApi(api) }
+    private val serviceApi by lazy { ServiceApi(api) }
+    private val spApi by lazy { ServicePolicyApi(api) }
 
     private object TrustAll : X509TrustManager {
         override fun checkClientTrusted(p0: Array<out X509Certificate>?, p1: String?) {}
@@ -143,4 +149,48 @@ internal object ManagementHelper {
     internal fun <T> CompletableFuture<T>.waitFor(timeout: Duration = Duration.of(5, ChronoUnit.SECONDS)): T =
         this.get(timeout.toMillis(), TimeUnit.MILLISECONDS)
 
+    internal fun createService(
+        srvName: String = "service-${System.nanoTime()}",
+        dialRoles: List<String> = listOf("#all"),
+        bindRoles: List<String> = listOf("#all"),
+        configs: Map<String, Any>,
+    ): String {
+
+        val cfgIds = mutableListOf<String>()
+        configs.forEach { (type, v) ->
+            val typeId = ConfigApi(api).listConfigTypes(1,0, """name = "$type" """).waitFor().data.first().id
+            val data = JSON.getDefault().mapper.convertValue(v, Map::class.java) as Map<String, *>
+            val id = ConfigApi(api).createConfig(ConfigCreate().apply {
+                name("$srvName-$type")
+                configTypeId(typeId)
+                data(data)
+            }).waitFor().data!!.id!!
+            cfgIds.add(id)
+        }
+
+        val srvId = serviceApi.createService(ServiceCreate().apply {
+            name = srvName
+            encryptionRequired = true
+            configs(cfgIds)
+        }).waitFor().data!!.id
+
+        // dial policy
+        spApi.createServicePolicy(ServicePolicyCreate().apply{
+            name("$srvName-dial")
+            type(DialBind.DIAL)
+            serviceRoles(listOf("@$srvId"))
+            identityRoles(dialRoles)
+            semantic(Semantic.ANY_OF)
+        }).waitFor()
+
+        // dial policy
+        spApi.createServicePolicy(ServicePolicyCreate().apply{
+            name("$srvName-bind")
+            type(DialBind.BIND)
+            serviceRoles(listOf("@$srvId"))
+            identityRoles(bindRoles)
+            semantic(Semantic.ANY_OF)
+        }).waitFor()
+        return srvName
+    }
 }

ziti/src/integrationTest/kotlin/org/openziti/net/ConnectionTests.kt

@@ -0,0 +1,219 @@
+/*
+ * Copyright (c) 2018-2025 NetFoundry Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.openziti.net
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
+import org.junit.jupiter.api.*
+import org.openziti.IdentityConfig
+import org.openziti.Ziti
+import org.openziti.ZitiAddress
+import org.openziti.ZitiContext
+import org.openziti.api.DNSName
+import org.openziti.api.InterceptConfig
+import org.openziti.api.InterceptV1Cfg
+import org.openziti.api.PortRange
+import org.openziti.edge.model.DialBind
+import org.openziti.integ.BaseTest
+import org.openziti.integ.ManagementHelper
+import org.openziti.net.nio.acceptSuspend
+import org.openziti.net.nio.readSuspend
+import org.openziti.net.nio.writeSuspend
+import java.net.ConnectException
+import java.net.InetSocketAddress
+import java.net.SocketTimeoutException
+import java.nio.ByteBuffer
+import java.nio.channels.InterruptedByTimeoutException
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeUnit
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
+
+class ConnectionTests: BaseTest() {
+
+    private val hostname = "test${System.nanoTime()}.ziti"
+    private val port = 5000
+    private lateinit var service: String
+    private lateinit var cfg: IdentityConfig
+    private lateinit var ztx: ZitiContext
+
+    @BeforeEach
+    fun before() {
+        service = ManagementHelper.createService(
+            configs = mapOf(
+                InterceptV1Cfg to InterceptConfig(
+                    protocols = setOf(Protocol.TCP),
+                    addresses = setOf(DNSName(hostname)),
+                    portRanges = sortedSetOf(PortRange(port, port)),
+                )
+            )
+        )
+        cfg = createIdentity()
+        ztx = Ziti.newContext(cfg)
+    }
+
+    @AfterEach
+    fun after() {
+        ztx.destroy()
+    }
+
+    @Test
+    fun `test dial without terminator`() = runTest(timeout = 10.seconds) {
+        val srv = assertDoesNotThrow {
+            ztx.serviceUpdates().filter { it.service.name == service }.first().service
+        }
+
+        assertFalse(srv.config.isEmpty())
+        assertThrows<ConnectException> {
+            ztx.dial(service)
+        }.run {
+            assert(message!!.contains("has no terminators"))
+        }
+
+        assertThrows<ExecutionException> {
+            val ch = ztx.open()
+            ch.connect(ZitiAddress.Dial(service)).get()
+        }.run {
+            assert(cause!!.message!!.contains("has no terminators"))
+        }
+        assertThrows<ExecutionException> {
+            val ch = ztx.open()
+            ch.connect(InetSocketAddress.createUnresolved(hostname, port)).get()
+        }.run {
+            assert(cause!!.message!!.contains("has no terminators"))
+        }
+
+        assertThrows<ConnectException> {
+            ztx.connect(hostname, port)
+        }.run {
+            assertTrue(message!!.contains("has no terminators"))
+        }
+    }
+
+    @Test
+    fun `test bind-connect-read-timeout`() = runTest(timeout = 10.seconds) {
+        val greeting = "Hello from Ziti".toByteArray()
+        val s = assertDoesNotThrow {
+            ztx.serviceUpdates().filter { it.service.name == service }.first().service
+        }
+        assertTrue(s.permissions.contains(DialBind.DIAL))
+        assertTrue(s.permissions.contains(DialBind.BIND))
+
+        ztx.openServer().use { srv ->
+
+            srv.bind(ZitiAddress.Bind(service))
+
+            // wait for binding -- test dispatcher skips delays
+            withContext(Dispatchers.Default) {
+                val zrv = srv as ZitiServerSocketChannel
+                while (zrv.state != ZitiServerSocketChannel.State.bound) {
+                    delay(50)
+                }
+            }
+
+            launch(Dispatchers.IO) {
+                val c = srv.acceptSuspend()
+                c.writeSuspend(ByteBuffer.wrap(greeting))
+                delay(1000)
+                c.close()
+            }
+
+            ztx.open().use { clt ->
+                clt.connect(ZitiAddress.Dial(service)).get(1, TimeUnit.SECONDS)
+
+                val buf = ByteBuffer.allocate(1024)
+
+                // read 1: return greeting
+                val read1 = clt.readSuspend(buf, 100, TimeUnit.MILLISECONDS)
+                assertEquals(read1, greeting.size)
+                val readMsg = ByteArray(read1)
+                buf.flip().get(readMsg)
+                assertContentEquals(greeting, readMsg)
+                assertFalse(buf.hasRemaining())
+
+                buf.clear()
+                // read 2: should time out
+                assertThrows<InterruptedByTimeoutException> {
+                    clt.readSuspend(buf, 600, TimeUnit.MILLISECONDS)
+                }
+
+                buf.clear()
+                // read 3: should get EOF
+                assertEquals(-1, clt.readSuspend(buf, 600, TimeUnit.MILLISECONDS))
+            }
+        }
+    }
+
+    @Test
+    fun `test socket-connect-read-timeout`() = runTest(timeout = 10.seconds) {
+        val greeting = "Hello from Ziti".toByteArray()
+        val s = assertDoesNotThrow {
+            ztx.serviceUpdates().filter { it.service.name == service }.first().service
+        }
+        assertTrue(s.permissions.contains(DialBind.DIAL))
+        assertTrue(s.permissions.contains(DialBind.BIND))
+
+        ztx.openServer().use { srv ->
+
+            srv.bind(ZitiAddress.Bind(service))
+
+            // wait for binding -- test dispatcher skips delays
+            withContext(Dispatchers.Default) {
+                val zrv = srv as ZitiServerSocketChannel
+                while (zrv.state != ZitiServerSocketChannel.State.bound) {
+                    delay(50)
+                }
+            }
+
+            launch(Dispatchers.IO) {
+                val c = srv.acceptSuspend()
+                c.writeSuspend(ByteBuffer.wrap(greeting))
+                val b = ByteBuffer.allocate(1024)
+                c.readSuspend(b)
+            }
+
+            ztx.connect(hostname, port).use { clt ->
+                assertTrue(clt.isConnected)
+                val buf = ByteArray(1024)
+                clt.soTimeout = 500
+                val input = clt.getInputStream()
+
+                // read 1: return greeting
+                val read1 = input.read(buf)
+                assertEquals(read1, greeting.size)
+                val readMsg = buf.sliceArray(0..<read1)
+                assertContentEquals(greeting, readMsg)
+
+                // other reads would get a timeout
+                for (i in 0 .. 10) {
+                    assertThrows<SocketTimeoutException> {
+                        input.read(buf)
+                    }
+                }
+            }
+        }
+    }
+
+}

ziti/src/main/kotlin/org/openziti/api/Controller.kt

@@ -142,7 +142,7 @@ internal class Controller(endpoint: String, sslContext: SSLContext):
 
         return pagingApiRequest {
                 limit, offset -> serviceApi.listServices(limit,  offset,
-            null, null, null, null)
+            null, listOf("all"), null, null)
         }
     }
 
@@ -277,7 +277,7 @@ internal class Controller(endpoint: String, sslContext: SSLContext):
             .os(info.os)
             .osRelease(info.osRelease)
             .osVersion(info.osVersion)
-        configTypes = listOf(InterceptV1Cfg, ClientV1Cfg)
+        configTypes = listOf("#all")
     }
 
     internal inner class ReqInterceptor(val session: ApiSession? = null): Consumer<HttpRequest.Builder> {