diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509ClientIntegration.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509ClientIntegration.kt index 6f8718901..00ef211ae 100644 --- a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509ClientIntegration.kt +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509ClientIntegration.kt @@ -12,7 +12,11 @@ import com.openai.errors.OpenAIIoException import com.openai.errors.OpenAIRetryableException import com.openai.errors.UnexpectedStatusCodeException import java.io.IOException +import java.security.cert.CertPathBuilderException +import java.security.cert.CertPathValidatorException +import java.security.cert.CertificateException import java.time.Duration +import java.util.IdentityHashMap import java.util.Locale import java.util.concurrent.CompletableFuture import java.util.concurrent.CompletionException @@ -23,6 +27,7 @@ import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicReference +import javax.net.ssl.SSLException internal const val X509_API_BASE_URL = "https://mtls.api.openai.com/v1" @@ -30,6 +35,7 @@ internal class X509ClientConfiguration private constructor( private val identity: X509WorkloadIdentity, private val bindTransport: (Timeout) -> BoundX509Transport, + private val authenticatorNanoTime: () -> Long, private val installTransport: (ClientOptions.Builder, OkHttpClient, HttpRequestAttemptAuthenticator) -> ClientOptions, ) { @@ -39,9 +45,12 @@ private constructor( identity: X509WorkloadIdentity, bindTransport: (Timeout) -> BoundX509Transport, ) = - X509ClientConfiguration(identity, bindTransport) { options, client, authenticator -> - options.buildWithFixedBearerTransport(client, authenticator) - } + X509ClientConfiguration( + identity, + bindTransport, + System::nanoTime, + ClientOptions.Builder::buildWithFixedBearerTransport, + ) @JvmSynthetic internal fun createForTest( @@ -51,7 +60,20 @@ private constructor( ( ClientOptions.Builder, OkHttpClient, HttpRequestAttemptAuthenticator, ) -> ClientOptions, - ) = X509ClientConfiguration(identity, bindTransport, installTransport) + ) = X509ClientConfiguration(identity, bindTransport, System::nanoTime, installTransport) + + @JvmSynthetic + internal fun createWithNanoTimeForTest( + identity: X509WorkloadIdentity, + bindTransport: (Timeout) -> BoundX509Transport, + nanoTime: () -> Long, + ) = + X509ClientConfiguration( + identity, + bindTransport, + nanoTime, + ClientOptions.Builder::buildWithFixedBearerTransport, + ) } @JvmSynthetic @@ -64,7 +86,7 @@ private constructor( val transport = bindTransport(clientOptions.timeout()) val authenticator = try { - X509AttemptAuthenticator(identity, transport.exchangeClient) + X509AttemptAuthenticator(identity, transport.exchangeClient, authenticatorNanoTime) } catch (error: Throwable) { closeAfterFailure(error, transport::close) throw error @@ -100,10 +122,11 @@ private class X509AttemptAuthenticator( constructor( identity: X509WorkloadIdentity, exchangeClient: OkHttpClient, + nanoTime: () -> Long = System::nanoTime, ) : this( X509TokenExchange(identity, exchangeClient)::executeAsync, exchangeClient::close, - System::nanoTime, + nanoTime, {}, {}, {}, @@ -541,8 +564,9 @@ private class X509AttemptAuthenticator( fun unchecked(error: Throwable): RuntimeException = unwrap(error).let { if (it is RuntimeException) it else OpenAIIoException(cause = it) } - fun isTransient(error: Throwable?): Boolean = - when (val cause = unwrap(error)) { + fun isTransient(error: Throwable?): Boolean { + if (hasPermanentTlsFailure(error)) return false + return when (val cause = unwrap(error)) { is IOException, is OpenAIIoException, is OpenAIRetryableException -> true @@ -555,6 +579,24 @@ private class X509AttemptAuthenticator( } else -> false } + } + + fun hasPermanentTlsFailure(error: Throwable?): Boolean { + val seen = IdentityHashMap() + var cause = error + while (cause != null && seen.put(cause, Unit) == null) { + if ( + cause is SSLException || + cause is CertificateException || + cause is CertPathBuilderException || + cause is CertPathValidatorException + ) { + return true + } + cause = cause.cause + } + return false + } val FORBIDDEN_HEADERS = setOf( diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/OpenAIOkHttpClientX509Test.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/OpenAIOkHttpClientX509Test.kt index 4a9c7e1e4..71c4e5930 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/OpenAIOkHttpClientX509Test.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/OpenAIOkHttpClientX509Test.kt @@ -8,11 +8,14 @@ import com.openai.credential.BearerTokenCredential import com.openai.errors.OpenAIIoException import com.openai.models.files.FileListParams import java.net.Proxy +import java.security.cert.CertificateException import java.security.cert.X509Certificate import java.time.Duration import java.util.concurrent.ExecutionException import java.util.concurrent.Executors import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicLong +import javax.net.ssl.SSLException import okhttp3.mockwebserver.MockResponse import okhttp3.mockwebserver.RecordedRequest import okhttp3.mockwebserver.SocketPolicy @@ -252,6 +255,54 @@ internal class OpenAIOkHttpClientX509Test { } } + @Test + fun publicSyncAndAsyncClientsRejectCachedBearerAfterIssuerTlsFailure() { + listOf(false, true).forEach { async -> + Fixture().use { fixture -> + val now = AtomicLong() + fixture.authPeer.enqueue( + fixture + .exchangeResponse("cachedtoken") + .setSocketPolicy(SocketPolicy.DISCONNECT_AT_END) + ) + fixture.enqueueApiSuccess() + fixture.enqueueApiSuccess() + + if (async) { + val client = fixture.asyncBuilder(now::get).maxRetries(0).build() + try { + client.files().list().get(10, TimeUnit.SECONDS) + now.set(Duration.ofSeconds(3_000).toNanos()) + fixture.replaceIssuerWithUntrustedCertificate() + + val failure = + runCatching { client.files().list().get(10, TimeUnit.SECONDS) } + .exceptionOrNull() + assertThat(failure).isInstanceOf(ExecutionException::class.java) + assertTlsFailure(requireNotNull(failure)) + } finally { + client.close() + } + } else { + val client = fixture.syncBuilder(now::get).maxRetries(0).build() + try { + client.files().list() + now.set(Duration.ofSeconds(3_000).toNanos()) + fixture.replaceIssuerWithUntrustedCertificate() + + val failure = runCatching { client.files().list() }.exceptionOrNull() + assertThat(failure).isInstanceOf(OpenAIIoException::class.java) + assertTlsFailure(requireNotNull(failure)) + } finally { + client.close() + } + } + + assertThat(fixture.apiPeer.server.requestCount).isEqualTo(2) + } + } + } + @Test fun onlyPublicJavaConstructionPathIsX509BuilderFactory() { listOf(OpenAIOkHttpClient.Builder::class.java, OpenAIOkHttpClientAsync.Builder::class.java) @@ -644,6 +695,12 @@ internal class OpenAIOkHttpClientX509Test { assertThat(requireNotNull(request.handshake).peerCertificates.first()).isEqualTo(expected) } + private fun assertTlsFailure(failure: Throwable) { + assertThat(generateSequence(failure) { it.cause }.toList()).anyMatch { + it is SSLException || it is CertificateException + } + } + private class Fixture : AutoCloseable { val identity = X509TestIdentity.create("SDK X.509 identity") val authPeer = X509TestPeer(AUTH_HOST, identity.root.certificate) @@ -682,6 +739,24 @@ internal class OpenAIOkHttpClientX509Test { apiPeer.proxy, ) + fun syncBuilder(nanoTime: () -> Long): OpenAIOkHttpClient.Builder = + OpenAIOkHttpClient.Builder.x509(configuration(nanoTime)) + + fun asyncBuilder(nanoTime: () -> Long): OpenAIOkHttpClientAsync.Builder = + OpenAIOkHttpClientAsync.Builder.x509(configuration(nanoTime)) + + fun replaceIssuerWithUntrustedCertificate() { + authPeer.replaceWithUntrustedCertificate() + authPeer.enqueue(exchangeResponse(ACCESS_TOKEN)) + } + + private fun configuration(nanoTime: () -> Long) = + X509ClientConfiguration.createWithNanoTimeForTest( + workloadIdentity, + { timeout -> transport.bindForTest(timeout, authPeer.proxy, apiPeer.proxy) }, + nanoTime, + ) + fun enqueueSuccess() { authPeer.enqueue( MockResponse().setHeader("Content-Type", "application/json").setBody(TOKEN_RESPONSE) diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509AttemptAuthenticatorTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509AttemptAuthenticatorTest.kt index c4f4b5081..86b587fec 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509AttemptAuthenticatorTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509AttemptAuthenticatorTest.kt @@ -13,6 +13,9 @@ import com.openai.core.http.RetryingHttpClient import com.openai.errors.OpenAIIoException import com.openai.errors.OpenAIRetryableException import java.io.ByteArrayInputStream +import java.io.IOException +import java.security.cert.CertPathBuilderException +import java.security.cert.CertificateException import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.concurrent.CountDownLatch @@ -23,6 +26,10 @@ import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicReference +import javax.net.ssl.SSLException +import javax.net.ssl.SSLHandshakeException +import javax.net.ssl.SSLPeerUnverifiedException +import javax.net.ssl.SSLProtocolException import org.assertj.core.api.Assertions.assertThat import org.assertj.core.api.Assertions.assertThatThrownBy import org.junit.jupiter.api.Test @@ -445,6 +452,75 @@ internal class X509AttemptAuthenticatorTest { authenticator.close() } + @Test + fun syncAndAsyncPermanentTlsRefreshFailureCannotFallbackToCachedBearer() { + val tlsFailures = + listOf<() -> Throwable>( + { + SSLHandshakeException("untrusted issuer certificate").apply { + initCause(CertificateException("certificate path rejected")) + } + }, + { CertPathBuilderException("issuer certificate path could not be built") }, + { SSLPeerUnverifiedException("issuer hostname mismatch") }, + { SSLProtocolException("issuer TLS protocol failure") }, + { SSLException("issuer TLS failure") }, + ) + + tlsFailures.forEach { tlsFailure -> + listOf(false, true).forEach { async -> + val now = AtomicLong() + val permanentFailure = + OpenAIIoException( + "issuer exchange failed", + IOException("transport wrapper", tlsFailure()), + ) + val exchanges = + ArrayDeque( + listOf( + CompletableFuture.completedFuture( + X509AccessToken("cachedtoken", Duration.ofMillis(500)) + ), + CompletableFuture().apply { + completeExceptionally(permanentFailure) + }, + ) + ) + val authenticator = + x509AttemptAuthenticatorForTest(nanoTime = now::get) { exchanges.removeFirst() } + + try { + if (async) { + authenticator + .authenticateAsync(request(), Duration.ofSeconds(5)) + .get(5, TimeUnit.SECONDS) + } else { + authenticator.authenticate(request(), Duration.ofSeconds(5)) + } + now.set(Duration.ofMillis(425).toNanos()) + + if (async) { + assertThatThrownBy { + authenticator + .authenticateAsync(request(), Duration.ofSeconds(5)) + .get(5, TimeUnit.SECONDS) + } + .isInstanceOf(ExecutionException::class.java) + .hasCause(permanentFailure) + } else { + assertThatThrownBy { + authenticator.authenticate(request(), Duration.ofSeconds(5)) + } + .isSameAs(permanentFailure) + } + assertThat(exchanges).isEmpty() + } finally { + authenticator.close() + } + } + } + } + @Test fun delayedExchangeCannotExtendTokenLifetime() { val now = AtomicLong() diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509WireTestInfrastructure.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509WireTestInfrastructure.kt index 57ee080d8..d5fa3ebfe 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509WireTestInfrastructure.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509WireTestInfrastructure.kt @@ -78,12 +78,13 @@ internal class X509TestPeer(val authority: String, trustedClientRoot: X509Certif init(arrayOf(serverIdentity.keyManager), arrayOf(recordingTrustManager), SecureRandom()) } - val server = + var server = MockWebServer().apply { useHttps(sslContext.socketFactory, true) requireClientAuth() start() } + private set val proxy: Proxy get() = server.toProxyAddress() @@ -104,6 +105,38 @@ internal class X509TestPeer(val authority: String, trustedClientRoot: X509Certif "No request received by $authority within $timeout" } + fun replaceWithUntrustedCertificate() { + val port = server.port + server.close() + val untrustedRoot = + HeldCertificate.Builder() + .commonName("$authority untrusted root") + .certificateAuthority(1) + .build() + val untrustedLeaf = + HeldCertificate.Builder() + .commonName(authority) + .addSubjectAlternativeName(authority) + .signedBy(untrustedRoot) + .build() + val untrustedIdentity = + HandshakeCertificates.Builder().heldCertificate(untrustedLeaf).build() + val untrustedContext = + SSLContext.getInstance("TLS").apply { + init( + arrayOf(untrustedIdentity.keyManager), + arrayOf(recordingTrustManager), + SecureRandom(), + ) + } + server = + MockWebServer().apply { + useHttps(untrustedContext.socketFactory, true) + requireClientAuth() + start(port) + } + } + override fun close() { server.close() } diff --git a/openai-java-core/src/main/kotlin/com/openai/core/http/PipelineRequestBody.kt b/openai-java-core/src/main/kotlin/com/openai/core/http/PipelineRequestBody.kt index e5002c66b..f982e4fc7 100644 --- a/openai-java-core/src/main/kotlin/com/openai/core/http/PipelineRequestBody.kt +++ b/openai-java-core/src/main/kotlin/com/openai/core/http/PipelineRequestBody.kt @@ -14,6 +14,13 @@ internal fun HttpRequest.withPipelineOwnedBody(): HttpRequest { return toBuilder().body(CloseOncePipelineRequestBody(current)).build() } +/** Gives one transport attempt a close-isolated view of the pipeline-owned request body. */ +@JvmSynthetic +internal fun HttpRequest.forPipelineAttempt(): HttpRequest { + val current = body as? PipelineOwnedRequestBody ?: return this + return toBuilder().body(PipelineAttemptRequestBody(current)).build() +} + /** Best-effort terminal cleanup for an authenticated request body. */ @JvmSynthetic internal fun HttpRequest.closePipelineBody(failure: Throwable? = null) { @@ -50,3 +57,16 @@ private class CloseOncePipelineRequestBody(private val delegate: HttpRequestBody if (closed.compareAndSet(false, true)) delegate.close() } } + +private class PipelineAttemptRequestBody(private val delegate: PipelineOwnedRequestBody) : + HttpRequestBody { + override fun writeTo(outputStream: OutputStream) = delegate.writeTo(outputStream) + + override fun contentType(): String? = delegate.contentType() + + override fun contentLength(): Long = delegate.contentLength() + + override fun repeatable(): Boolean = delegate.repeatable() + + override fun close() = Unit +} diff --git a/openai-java-core/src/main/kotlin/com/openai/core/http/RetryingHttpClientOrchestrator.kt b/openai-java-core/src/main/kotlin/com/openai/core/http/RetryingHttpClientOrchestrator.kt index d8a6eb064..f8d9ce15e 100644 --- a/openai-java-core/src/main/kotlin/com/openai/core/http/RetryingHttpClientOrchestrator.kt +++ b/openai-java-core/src/main/kotlin/com/openai/core/http/RetryingHttpClientOrchestrator.kt @@ -145,9 +145,10 @@ internal class RetryingHttpClientOrchestrator( val options = deadline?.let { remainingOptions(requestOptions, it) } ?: requestOptions val authenticatedRequest = authenticated.request() + val attemptRequest = authenticatedRequest.forPipelineAttempt() val response = try { - httpClient.execute(authenticatedRequest, options) + httpClient.execute(attemptRequest, options) } catch (error: Throwable) { if ( !isRetryable(authenticatedRequest) || @@ -366,10 +367,11 @@ internal class RetryingHttpClientOrchestrator( fun dispatch(authenticated: AuthenticatedHttpRequest, options: RequestOptions) { val authenticatedRequest = authenticated.request() + val attemptRequest = authenticatedRequest.forPipelineAttempt() beforeAsyncApiDispatch() val call = try { - startStage { httpClient.executeAsync(authenticatedRequest, options) } ?: return + startStage { httpClient.executeAsync(attemptRequest, options) } ?: return } catch (error: Throwable) { retry(error = error, requestRetryable = isRetryable(authenticatedRequest)) return diff --git a/openai-java-core/src/test/kotlin/com/openai/core/X509PublicRequestBodyLifecycleTest.kt b/openai-java-core/src/test/kotlin/com/openai/core/X509PublicRequestBodyLifecycleTest.kt new file mode 100644 index 000000000..c69f5ed2b --- /dev/null +++ b/openai-java-core/src/test/kotlin/com/openai/core/X509PublicRequestBodyLifecycleTest.kt @@ -0,0 +1,90 @@ +package com.openai.core + +import com.openai.client.OpenAIClientAsyncImpl +import com.openai.client.OpenAIClientImpl +import com.openai.core.http.CachingAuthenticator +import com.openai.core.http.HttpClient +import com.openai.core.http.HttpRequest +import com.openai.core.http.HttpRequestBody +import com.openai.core.http.HttpResponse +import com.openai.core.http.response +import com.openai.models.responses.ResponseCreateParams +import java.io.ByteArrayOutputStream +import java.util.concurrent.CompletableFuture +import java.util.concurrent.ExecutionException +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test + +internal class X509PublicRequestBodyLifecycleTest { + @Test + fun publicSyncAndAsyncRequestsUseFreshAttemptBodiesForRetryAndUnauthorizedReplay() { + listOf(listOf(500, 400) to 1, listOf(401, 400) to 0).forEach { (statuses, maxRetries) -> + listOf(false, true).forEach { async -> + val transport = ClosingBodyClient(statuses) + + if (async) { + val client = OpenAIClientAsyncImpl(options(transport, maxRetries)) + try { + val failure = + runCatching { client.responses().create(params()).get() } + .exceptionOrNull() + assertThat(failure).isInstanceOf(ExecutionException::class.java) + } finally { + client.close() + } + } else { + val client = OpenAIClientImpl(options(transport, maxRetries)) + try { + assertThat(runCatching { client.responses().create(params()) }.isFailure) + .isTrue() + } finally { + client.close() + } + } + + assertThat(transport.attemptBodies).hasSize(2) + assertThat(transport.attemptBodies[0]).isNotSameAs(transport.attemptBodies[1]) + assertThat(transport.payloads).hasSize(2) + assertThat(transport.payloads[0]) + .isNotEmpty() + .containsExactly(*transport.payloads[1]) + } + } + } + + private fun options(transport: HttpClient, maxRetries: Int): ClientOptions = + ClientOptions.builder() + .fixedBearerAuthentication("https://example.test/v1") + .maxRetries(maxRetries) + .buildWithFixedBearerTransport(transport, CachingAuthenticator()) + + private fun params() = + ResponseCreateParams.builder().model("gpt-4o-mini").input("Hello").build() + + private class ClosingBodyClient(statuses: List) : HttpClient { + private val statuses = ArrayDeque(statuses) + val attemptBodies = mutableListOf() + val payloads = mutableListOf() + + override fun execute(request: HttpRequest, requestOptions: RequestOptions): HttpResponse { + val body = requireNotNull(request.body) + attemptBodies += body + val output = ByteArrayOutputStream() + try { + body.writeTo(output) + payloads += output.toByteArray() + } finally { + body.close() + } + return response(statuses.removeFirst()) + } + + override fun executeAsync( + request: HttpRequest, + requestOptions: RequestOptions, + ): CompletableFuture = + CompletableFuture.completedFuture(execute(request, requestOptions)) + + override fun close() = Unit + } +} diff --git a/openai-java-core/src/test/kotlin/com/openai/core/http/AuthenticatedRequestBodyLifecycleTest.kt b/openai-java-core/src/test/kotlin/com/openai/core/http/AuthenticatedRequestBodyLifecycleTest.kt index a54059f23..cc61aca4e 100644 --- a/openai-java-core/src/test/kotlin/com/openai/core/http/AuthenticatedRequestBodyLifecycleTest.kt +++ b/openai-java-core/src/test/kotlin/com/openai/core/http/AuthenticatedRequestBodyLifecycleTest.kt @@ -1,6 +1,7 @@ package com.openai.core.http import com.openai.errors.OpenAIIoException +import java.io.ByteArrayOutputStream import java.io.OutputStream import java.time.Duration import java.util.concurrent.CompletableFuture @@ -9,6 +10,34 @@ import org.assertj.core.api.Assertions.assertThatThrownBy import org.junit.jupiter.api.Test internal class AuthenticatedRequestBodyLifecycleTest { + @Test + fun repeatableBodyRemainsOpenAcrossSyncAndAsyncRetriesAndUnauthorizedReplay() { + listOf(listOf(500, 200) to 1, listOf(401, 200) to 0).forEach { (statuses, maxRetries) -> + listOf(false, true).forEach { async -> + val body = InvalidatingRepeatableBody() + val transport = ClosingBodyClient(statuses) + val client = client(transport, CachingAuthenticator(), maxRetries) + + try { + if (async) { + client.executeAsync(request(body)).get().close() + } else { + client.execute(request(body)).close() + } + + assertThat(body.writes).isEqualTo(2) + assertThat(body.closes).isEqualTo(1) + assertThat(transport.attemptBodies).hasSize(2) + assertThat(transport.attemptBodies[0]) + .isNotSameAs(transport.attemptBodies[1]) + .isNotSameAs(body) + } finally { + client.close() + } + } + } + } + @Test fun syncAuthenticationFailureClosesRequestBodyOnce() { val failure = IllegalStateException("authentication failed") @@ -90,4 +119,54 @@ internal class AuthenticatedRequestBodyLifecycleTest { closes++ } } + + private class InvalidatingRepeatableBody : HttpRequestBody { + private var closed = false + var writes = 0 + var closes = 0 + + override fun writeTo(outputStream: OutputStream) { + check(!closed) { "request body was closed before its final attempt" } + writes++ + outputStream.write("payload".toByteArray()) + } + + override fun contentType(): String = "text/plain" + + override fun contentLength(): Long = 7 + + override fun repeatable(): Boolean = true + + override fun close() { + closes++ + closed = true + } + } + + private class ClosingBodyClient(statuses: List) : HttpClient { + private val statuses = ArrayDeque(statuses) + val attemptBodies = mutableListOf() + + override fun execute( + request: HttpRequest, + requestOptions: com.openai.core.RequestOptions, + ): HttpResponse { + val body = requireNotNull(request.body) + attemptBodies += body + try { + body.writeTo(ByteArrayOutputStream()) + } finally { + body.close() + } + return response(statuses.removeFirst()) + } + + override fun executeAsync( + request: HttpRequest, + requestOptions: com.openai.core.RequestOptions, + ): CompletableFuture = + CompletableFuture.completedFuture(execute(request, requestOptions)) + + override fun close() {} + } }