From 75dfa8f281fa6343e716ae7c1ab1bfb53207d72f Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Thu, 27 Aug 2026 00:50:33 +0000 Subject: [PATCH 1/4] feat(auth): add X.509 token exchange --- .../com/openai/client/okhttp/OkHttpClient.kt | 5 +- .../openai/client/okhttp/X509TokenExchange.kt | 297 +++++++++++ .../openai/client/okhttp/OkHttpClientTest.kt | 74 +++ .../client/okhttp/X509TokenExchangeTest.kt | 494 ++++++++++++++++++ .../okhttp/X509WireTestInfrastructure.kt | 24 + 5 files changed, 893 insertions(+), 1 deletion(-) create mode 100644 openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt create mode 100644 openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/OkHttpClient.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/OkHttpClient.kt index 5a034f7bd..13ee78016 100644 --- a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/OkHttpClient.kt +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/OkHttpClient.kt @@ -64,7 +64,10 @@ internal constructor(@JvmSynthetic internal val okHttpClient: okhttp3.OkHttpClie call.enqueue( object : Callback { override fun onResponse(call: Call, response: Response) { - future.complete(response.toHttpResponse()) + val httpResponse = response.toHttpResponse() + if (!future.complete(httpResponse)) { + httpResponse.close() + } } override fun onFailure(call: Call, e: IOException) { diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt new file mode 100644 index 000000000..3b7bf4380 --- /dev/null +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt @@ -0,0 +1,297 @@ +package com.openai.client.okhttp + +import com.fasterxml.jackson.core.JsonProcessingException +import com.fasterxml.jackson.core.StreamReadFeature +import com.fasterxml.jackson.databind.DeserializationFeature +import com.fasterxml.jackson.databind.JsonNode +import com.openai.core.http.Headers +import com.openai.core.http.HttpClient +import com.openai.core.http.HttpMethod +import com.openai.core.http.HttpRequest +import com.openai.core.http.HttpRequestBody +import com.openai.core.http.HttpResponse +import com.openai.core.jsonMapper +import com.openai.errors.OpenAIInvalidDataException +import com.openai.errors.OpenAIIoException +import com.openai.errors.UnexpectedStatusCodeException +import java.io.IOException +import java.io.OutputStream +import java.time.Duration +import java.util.concurrent.CompletableFuture +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.SynchronousQueue +import java.util.concurrent.ThreadFactory +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference + +/** One validated access token. Its string representation never includes the credential. */ +internal class X509AccessToken(val value: String, val expiresIn: Duration) { + override fun toString(): String = "X509AccessToken{value=, expiresIn=$expiresIn}" +} + +/** Executes the fixed X.509 workload-identity token exchange without caching or retries. */ +internal class X509TokenExchange( + private val identityProviderId: String, + private val serviceAccountId: String, + private val httpClient: HttpClient, +) : AutoCloseable { + private val jsonMapper = jsonMapper() + private val responseReader = + jsonMapper + .reader() + .with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS) + .with(StreamReadFeature.STRICT_DUPLICATE_DETECTION) + private val responseExecutor = + ThreadPoolExecutor( + 0, + MAX_RESPONSE_THREADS, + RESPONSE_THREAD_IDLE_SECONDS, + TimeUnit.SECONDS, + SynchronousQueue(), + ResponseThreadFactory, + ) + private val operations = ConcurrentHashMap.newKeySet() + private val closed = AtomicBoolean() + + init { + require(identityProviderId.isNotBlank()) { "identityProviderId must not be blank" } + require(serviceAccountId.isNotBlank()) { "serviceAccountId must not be blank" } + } + + fun execute(): X509AccessToken { + checkOpen() + return httpClient.execute(request()).use(::parse) + } + + fun executeAsync(): CompletableFuture { + checkOpen() + val operation = AsyncOperation() + operations.add(operation) + operation.result.whenComplete { _, _ -> operations.remove(operation) } + + val responseFuture = + try { + httpClient.executeAsync(request()) + } catch (error: Throwable) { + operations.remove(operation) + throw error + } + operation.responseFuture.set(responseFuture) + responseFuture.whenComplete(operation::accept) + if (closed.get() || operation.result.isCancelled) { + operation.result.cancel(true) + responseFuture.cancel(true) + } + return operation.result + } + + override fun close() { + if (closed.compareAndSet(false, true)) { + operations.toTypedArray().forEach { operation -> operation.result.cancel(true) } + responseExecutor.shutdownNow() + } + } + + private fun checkOpen() { + check(!closed.get()) { "X.509 token exchange is closed" } + } + + private inner class AsyncOperation { + val result = CompletableFuture() + val responseFuture = AtomicReference?>() + private val activeResponse = AtomicReference() + + init { + result.whenComplete { _, _ -> + if (result.isCancelled) { + responseFuture.get()?.cancel(true) + activeResponse.getAndSet(null)?.close() + } + } + } + + fun accept(response: HttpResponse?, error: Throwable?) { + if (error != null) { + if (!result.isDone) result.completeExceptionally(error) + return + } + if (response == null) { + result.completeExceptionally( + IllegalStateException("X.509 token exchange completed without a response") + ) + return + } + + val lease = ResponseLease(response) + activeResponse.set(lease) + if (result.isDone) { + close(lease) + return + } + try { + responseExecutor.execute { process(lease) } + } catch (_: java.util.concurrent.RejectedExecutionException) { + close(lease) + if (!result.isDone) { + result.completeExceptionally( + OpenAIIoException("X.509 token exchange response processing unavailable") + ) + } + } + } + + private fun process(lease: ResponseLease) { + try { + val token = lease.use { if (result.isDone) null else parse(lease.response) } + if (token != null && !result.isDone) result.complete(token) + } catch (error: Throwable) { + if (!result.isDone) result.completeExceptionally(error) + } finally { + activeResponse.compareAndSet(lease, null) + } + } + + private fun close(lease: ResponseLease) { + activeResponse.compareAndSet(lease, null) + lease.close() + } + } + + private class ResponseLease(val response: HttpResponse) : AutoCloseable { + private val closed = AtomicBoolean() + + override fun close() { + if (closed.compareAndSet(false, true)) response.close() + } + } + + private fun request(): HttpRequest { + val bytes = + jsonMapper.writeValueAsBytes( + linkedMapOf( + "grant_type" to TOKEN_EXCHANGE_GRANT_TYPE, + "subject_token_type" to X509_TOKEN_TYPE, + "identity_provider_id" to identityProviderId, + "service_account_id" to serviceAccountId, + ) + ) + val body = + object : HttpRequestBody { + override fun writeTo(outputStream: OutputStream) = outputStream.write(bytes) + + override fun contentType(): String = "application/json" + + override fun contentLength(): Long = bytes.size.toLong() + + override fun repeatable(): Boolean = true + + override fun close() {} + } + return HttpRequest.builder() + .method(HttpMethod.POST) + .baseUrl(TOKEN_EXCHANGE_URL) + .body(body) + .build() + } + + private fun parse(response: HttpResponse): X509AccessToken { + val statusCode = response.statusCode() + if (statusCode != 200) { + throw UnexpectedStatusCodeException.builder() + .statusCode(statusCode) + .headers(safeDiagnosticHeaders(response.headers())) + .build() + } + + val node: JsonNode = + try { + responseReader.readTree(response.body()) + } catch (error: JsonProcessingException) { + transportFailure(error)?.let { throw readFailure(it) } + throw invalidResponse() + } catch (error: IOException) { + throw readFailure(error) + } ?: throw invalidResponse() + + val accessToken = + node.text("access_token")?.takeIf(BEARER_TOKEN_PATTERN::matches) + ?: throw invalidResponse("access_token") + if (!node.text("token_type").equals("Bearer", ignoreCase = true)) { + throw invalidResponse("token_type") + } + if (node.text("issued_token_type") != ACCESS_TOKEN_TYPE) { + throw invalidResponse("issued_token_type") + } + val expiresIn = + node + .get("expires_in") + ?.takeIf { it.isIntegralNumber && it.canConvertToLong() } + ?.longValue() + ?.takeIf { it in 1..MAX_TOKEN_LIFETIME_SECONDS } + ?: throw invalidResponse("expires_in") + return X509AccessToken(accessToken, Duration.ofSeconds(expiresIn)) + } + + private fun JsonNode.text(name: String): String? = + get(name)?.takeIf(JsonNode::isTextual)?.asText()?.takeIf(String::isNotBlank) + + private fun invalidResponse(field: String? = null): OpenAIInvalidDataException = + OpenAIInvalidDataException( + if (field == null) "Invalid X.509 token exchange response" + else "Invalid X.509 token exchange response field: $field" + ) + + private fun readFailure(cause: IOException): OpenAIIoException = + OpenAIIoException("Failed to read X.509 token exchange response", cause) + + private fun transportFailure(error: JsonProcessingException): IOException? { + var cause = error.cause + while (cause != null && cause !== error) { + if (cause is IOException && cause !is JsonProcessingException) return cause + cause = cause.cause + } + return null + } + + private fun safeDiagnosticHeaders(headers: Headers): Headers = + Headers.builder() + .apply { SAFE_DIAGNOSTIC_HEADERS.forEach { name -> put(name, headers.values(name)) } } + .build() + + private companion object { + const val TOKEN_EXCHANGE_URL = "https://mtls.auth.openai.com/oauth/token" + const val TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange" + const val X509_TOKEN_TYPE = "urn:openai:params:oauth:token-type:x509" + const val ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" + const val MAX_TOKEN_LIFETIME_SECONDS = 3600L + const val MAX_RESPONSE_THREADS = 4 + const val RESPONSE_THREAD_IDLE_SECONDS = 30L + val BEARER_TOKEN_PATTERN = Regex("[A-Za-z0-9._~+/-]+=*") + val SAFE_DIAGNOSTIC_HEADERS = + setOf( + "Content-Length", + "Content-Type", + "Date", + "OpenAI-Request-ID", + "Request-ID", + "Retry-After", + "Retry-After-Ms", + "Traceparent", + "Tracestate", + "X-Request-ID", + "X-Should-Retry", + ) + } +} + +private object ResponseThreadFactory : ThreadFactory { + private val threadNumber = AtomicInteger() + + override fun newThread(runnable: Runnable): Thread = + Thread(runnable, "openai-x509-response-${threadNumber.incrementAndGet()}").apply { + isDaemon = true + } +} diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/OkHttpClientTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/OkHttpClientTest.kt index b202c83c6..7f87a5137 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/OkHttpClientTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/OkHttpClientTest.kt @@ -5,6 +5,17 @@ import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo import com.github.tomakehurst.wiremock.junit5.WireMockTest import com.openai.core.http.HttpMethod import com.openai.core.http.HttpRequest +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import okhttp3.Call +import okhttp3.EventListener +import okhttp3.Interceptor +import okhttp3.MediaType +import okhttp3.Protocol +import okhttp3.Response +import okhttp3.ResponseBody +import okio.Buffer +import okio.BufferedSource import org.assertj.core.api.Assertions.assertThat import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test @@ -41,4 +52,67 @@ internal class OkHttpClientTest { // Should have cancelled the underlying call assertThat(call.isCanceled()).isTrue() } + + @Test + fun executeAsync_whenResponseLosesCancellationRace_closesDroppedResponse() { + val handoffStarted = CountDownLatch(1) + val releaseHandoff = CountDownLatch(1) + val responseBody = TrackingResponseBody() + val interceptor = Interceptor { chain -> + Response.Builder() + .request(chain.request()) + .protocol(Protocol.HTTP_1_1) + .code(200) + .message("OK") + .body(responseBody) + .build() + } + val handoffListener = + object : EventListener() { + override fun callEnd(call: Call) { + handoffStarted.countDown() + check(releaseHandoff.await(5, TimeUnit.SECONDS)) + } + } + val client = + OkHttpClient( + okhttp3.OkHttpClient.Builder() + .addInterceptor(interceptor) + .eventListener(handoffListener) + .build() + ) + + client.use { + val responseFuture = + client.executeAsync( + HttpRequest.builder() + .method(HttpMethod.POST) + .baseUrl("https://example.test") + .build() + ) + assertThat(handoffStarted.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(responseFuture.cancel(true)).isTrue() + + releaseHandoff.countDown() + + assertThat(responseBody.closed.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(responseFuture.isCancelled).isTrue() + } + } +} + +private class TrackingResponseBody : ResponseBody() { + val closed = CountDownLatch(1) + private val source = Buffer() + + override fun contentType(): MediaType? = null + + override fun contentLength(): Long = 0 + + override fun source(): BufferedSource = source + + override fun close() { + super.close() + closed.countDown() + } } diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt new file mode 100644 index 000000000..5b4f8ddb7 --- /dev/null +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt @@ -0,0 +1,494 @@ +package com.openai.client.okhttp + +import com.fasterxml.jackson.databind.ObjectMapper +import com.fasterxml.jackson.databind.node.ObjectNode +import com.openai.core.RequestOptions +import com.openai.core.Timeout +import com.openai.core.http.Headers +import com.openai.core.http.HttpClient +import com.openai.core.http.HttpRequest +import com.openai.core.http.HttpResponse +import com.openai.errors.OpenAIInvalidDataException +import com.openai.errors.OpenAIIoException +import com.openai.errors.UnexpectedStatusCodeException +import java.io.ByteArrayInputStream +import java.io.IOException +import java.io.InputStream +import java.net.SocketTimeoutException +import java.security.cert.X509Certificate +import java.time.Duration +import java.util.concurrent.CompletableFuture +import java.util.concurrent.CountDownLatch +import java.util.concurrent.ExecutionException +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import okhttp3.mockwebserver.MockResponse +import okhttp3.tls.HandshakeCertificates +import org.assertj.core.api.Assertions.assertThat +import org.assertj.core.api.Assertions.assertThatThrownBy +import org.junit.jupiter.api.Test + +internal class X509TokenExchangeTest { + + @Test + fun exchangesSynchronouslyOverPinnedMtlsUsingTheExactProtocol() { + exchangesOverPinnedMtlsUsingTheExactProtocol(async = false) + } + + @Test + fun exchangesAsynchronouslyOverPinnedMtlsUsingTheExactProtocol() { + exchangesOverPinnedMtlsUsingTheExactProtocol(async = true) + } + + private fun exchangesOverPinnedMtlsUsingTheExactProtocol(async: Boolean) { + val clientIdentity = X509TestIdentity.create("exchange client") + X509TestPeer(AUTH_HOST, clientIdentity.root.certificate).use { authPeer -> + authPeer.enqueue( + MockResponse().setHeader("Content-Type", "application/json").setBody(TOKEN_RESPONSE) + ) + val transport = transport(clientIdentity, listOf(authPeer.serverRootCertificate)) + + val token = + transport.bindForTest(Timeout.default(), authPeer.proxy, authPeer.proxy).use { bound + -> + X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, bound.exchangeClient).use { + exchange -> + if (async) exchange.executeAsync().get(5, TimeUnit.SECONDS) + else exchange.execute() + } + } + + assertThat(token.value).isEqualTo(ACCESS_TOKEN) + assertThat(token.expiresIn).isEqualTo(Duration.ofHours(1)) + assertThat(token.toString()).doesNotContain(ACCESS_TOKEN) + + val connect = authPeer.takeRequest() + val request = authPeer.takeRequest() + assertThat(connect.requestLine).isEqualTo("CONNECT $AUTH_HOST:443 HTTP/1.1") + assertThat(request.method).isEqualTo("POST") + assertThat(request.path).isEqualTo("/oauth/token") + assertThat(request.getHeader("Authorization")).isNull() + assertThat(request.getHeader("Cookie")).isNull() + assertThat(request.getHeader("Content-Type")).startsWith("application/json") + assertThat(ObjectMapper().readTree(request.body.readUtf8())) + .isEqualTo(ObjectMapper().readTree(TOKEN_REQUEST)) + val presentedChain = requireNotNull(request.handshake).peerCertificates + assertThat(presentedChain.first()).isEqualTo(clientIdentity.leaf.certificate) + assertThat(presentedChain).contains(clientIdentity.root.certificate) + assertThat(authPeer.requestedServerNames).containsExactly(AUTH_HOST) + } + } + + @Test + fun rejectsMalformedOrSemanticallyInvalidSuccessResponsesWithoutLeakingCredentials() { + val invalidBodies = + mapOf( + "malformed" to "not-json", + "trailing garbage" to "${validResponse()} trailing", + "second root" to "${validResponse()} {}", + "duplicate access_token" to + validResponseWithDuplicate( + "access_token", + "\"secret-duplicate-token-must-not-leak\"", + ), + "duplicate expires_in" to validResponseWithDuplicate("expires_in", "60"), + "missing access_token" to validResponseWithout("access_token"), + "missing token_type" to validResponseWithout("token_type"), + "null token_type" to validResponseWithNull("token_type"), + "missing issued_token_type" to validResponseWithout("issued_token_type"), + "null issued_token_type" to validResponseWithNull("issued_token_type"), + "invalid access_token" to validResponse(accessToken = "secret token with spaces"), + "invalid token_type" to + validResponse(accessToken = "secret-token-must-not-leak", tokenType = "MAC"), + "invalid issued_token_type" to validResponse(issuedTokenType = "refresh_token"), + "missing expires_in" to validResponseWithout("expires_in"), + "zero expires_in" to validResponse(expiresIn = "0"), + "fractional expires_in" to validResponse(expiresIn = "1.5"), + "expires_in above maximum" to validResponse(expiresIn = "3601"), + "overflow expires_in" to validResponse(expiresIn = "9223372036854775808"), + ) + + invalidBodies.forEach { (description, body) -> + listOf(false, true).forEach { async -> + val response = TestResponse(200, body) + + assertThat(exchangeFailure(response, async)) + .describedAs("$description (async=$async)") + .isInstanceOf(OpenAIInvalidDataException::class.java) + .hasMessageNotContaining("secret") + .hasMessageNotContaining(body) + assertThat(response.closed).describedAs(description).isTrue() + } + } + } + + @Test + fun acceptsLargeForwardCompatibleResponsesWithoutAnArbitraryLimit() { + val node = ObjectMapper().readTree(validResponse()) + (node as ObjectNode).put("forward_compatible_field", "x".repeat(2 * 1024 * 1024)) + val response = TestResponse(200, node.toString()) + + val token = + X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, SingleResponseClient(response)).use { + it.execute() + } + + assertThat(token.value).isEqualTo(ACCESS_TOKEN) + assertThat(response.closed).isTrue() + } + + @Test + fun failureStatusDoesNotReadBodyAndRetainsOnlySafeDiagnosticHeaders() { + listOf(false, true).forEach { async -> + val response = + TestResponse( + 503, + """{"error_description":"secret diagnostic"}""", + Headers.builder() + .put("Set-Cookie", "session=secret-cookie") + .put("Authorization", "Bearer secret-token") + .put("X-Api-Key", "secret-api-key") + .put("X-Request-ID", "req_safe") + .put("Retry-After", "1") + .build(), + ) + + val failure = exchangeFailure(response, async) + + assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) + val statusError = failure as UnexpectedStatusCodeException + assertThat(statusError.statusCode()).isEqualTo(503) + assertThat(statusError.headers().values("Set-Cookie")).isEmpty() + assertThat(statusError.headers().values("Authorization")).isEmpty() + assertThat(statusError.headers().values("X-Api-Key")).isEmpty() + assertThat(statusError.headers().values("X-Request-ID")).containsExactly("req_safe") + assertThat(statusError.headers().values("Retry-After")).containsExactly("1") + assertThat(statusError.toString()) + .doesNotContain("secret-cookie", "secret-token", "secret-api-key") + assertThat(response.bodyRead).isFalse() + assertThat(response.closed).isTrue() + } + } + + @Test + fun rejectsNon200SuccessWithoutReadingTheBody() { + listOf(false, true).forEach { async -> + val response = TestResponse(201, validResponse()) + + val failure = exchangeFailure(response, async) + + assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) + assertThat((failure as UnexpectedStatusCodeException).statusCode()).isEqualTo(201) + assertThat(response.bodyRead).isFalse() + assertThat(response.closed).isTrue() + } + } + + @Test + fun preservesIssuerBodyTimeoutAndIoFailuresAsSanitizedRetryableIo() { + listOf( + SocketTimeoutException("issuer body stalled"), + IOException("issuer body disconnected"), + ) + .forEach { cause -> + listOf(false, true).forEach { async -> + val response = FailingBodyResponse(cause) + + val failure = exchangeFailure(response, async) + + assertThat(failure).isInstanceOf(OpenAIIoException::class.java) + assertThat(failure).hasMessage("Failed to read X.509 token exchange response") + assertThat(failure.cause).isSameAs(cause) + assertThat(response.closed).isTrue() + } + } + } + + @Test + fun completedAsyncResponseParsesOnOwnedIoThreadAfterReturningCancellationHandle() { + val response = BlockingResponse() + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, SingleResponseClient(response)) + + val result = exchange.executeAsync() + assertThat(response.bodyStarted.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(response.bodyThreadName).startsWith("openai-x509-response-") + assertThat(result.cancel(true)).isTrue() + + assertThat(response.closed.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(response.closeCount).hasValue(1) + exchange.close() + } + + @Test + fun cancelingBeforeResponseDeliveryCancelsTheUnderlyingCall() { + val responseFuture = CompletableFuture() + val client = DeferredResponseClient(responseFuture) + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) + + val result = exchange.executeAsync() + assertThat(result.cancel(true)).isTrue() + + assertThat(responseFuture.isCancelled).isTrue() + assertThat(client.closed).isFalse() + exchange.close() + assertThat(client.closed).isFalse() + } + + @Test + fun cancelingClosesALateResponseWhenUnderlyingCancellationLosesTheRace() { + val responseFuture = NonCancellableFuture() + val exchange = + X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, DeferredResponseClient(responseFuture)) + val result = exchange.executeAsync() + val response = TestResponse(200, validResponse()) + + assertThat(result.cancel(true)).isTrue() + assertThat(responseFuture.complete(response)).isTrue() + + assertThat(result.isCancelled).isTrue() + assertThat(response.bodyRead).isFalse() + assertThat(response.closeCount).hasValue(1) + exchange.close() + } + + @Test + fun responseExecutorIsBoundedAndClosesRejectedResponses() { + val responses = List(5) { BlockingResponse() } + val client = SequenceResponseClient(responses) + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) + val results = mutableListOf>() + + repeat(4) { index -> + results += exchange.executeAsync() + assertThat(responses[index].bodyStarted.await(5, TimeUnit.SECONDS)).isTrue() + } + results += exchange.executeAsync() + + assertThatThrownBy { results.last().get(5, TimeUnit.SECONDS) } + .isInstanceOf(ExecutionException::class.java) + .hasCauseInstanceOf(OpenAIIoException::class.java) + assertThat(responses.last().closed.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(responses.last().bodyStarted.count).isEqualTo(1) + + exchange.close() + assertThat(results.take(4)).allMatch(CompletableFuture<*>::isCancelled) + assertThat(responses.take(4)).allMatch { it.closed.await(5, TimeUnit.SECONDS) } + assertThat(client.closed).isFalse() + } + + private fun exchangeFailure(response: HttpResponse, async: Boolean): Throwable { + val result = + runCatching { + X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, SingleResponseClient(response)) + .use { exchange -> + if (async) exchange.executeAsync().get(5, TimeUnit.SECONDS) + else exchange.execute() + } + } + .exceptionOrNull() ?: error("Expected X.509 exchange to fail") + return if (result is ExecutionException) result.cause ?: result else result + } + + private fun transport( + clientIdentity: X509TestIdentity, + trustedServerRoots: Iterable, + ): X509Transport { + val trustManager = + HandshakeCertificates.Builder() + .apply { + trustedServerRoots.forEach { certificate -> addTrustedCertificate(certificate) } + } + .build() + .trustManager + return X509Transport.builder() + .keyManager(x509TestKeyManager(mapOf(CERTIFICATE_ALIAS to clientIdentity))) + .certificateAlias(CERTIFICATE_ALIAS) + .trustManager(trustManager) + .build() + } + + private fun validResponseWithout(field: String): String = + (ObjectMapper().readTree(validResponse()) as ObjectNode).apply { remove(field) }.toString() + + private fun validResponseWithNull(field: String): String = + (ObjectMapper().readTree(validResponse()) as ObjectNode).apply { putNull(field) }.toString() + + private fun validResponseWithDuplicate(field: String, firstValue: String): String = + validResponse().replaceFirst("\"$field\":", "\"$field\": $firstValue,\n \"$field\":") + + private fun validResponse( + accessToken: String = ACCESS_TOKEN, + tokenType: String = "Bearer", + issuedTokenType: String = ACCESS_TOKEN_TYPE, + expiresIn: String = "3600", + ): String = + """ + { + "access_token": "$accessToken", + "issued_token_type": "$issuedTokenType", + "token_type": "$tokenType", + "expires_in": $expiresIn + } + """ + .trimIndent() + + private companion object { + const val AUTH_HOST = "mtls.auth.openai.com" + const val CERTIFICATE_ALIAS = "x509" + const val IDP_ID = "idp_test" + const val SERVICE_ACCOUNT_ID = "svc_acct_test" + const val ACCESS_TOKEN = "test-x509-access-token" + const val ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" + val TOKEN_REQUEST = + """ + { + "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", + "subject_token_type": "urn:openai:params:oauth:token-type:x509", + "identity_provider_id": "$IDP_ID", + "service_account_id": "$SERVICE_ACCOUNT_ID" + } + """ + .trimIndent() + val TOKEN_RESPONSE = + """ + { + "access_token": "$ACCESS_TOKEN", + "issued_token_type": "$ACCESS_TOKEN_TYPE", + "token_type": "Bearer", + "expires_in": 3600, + "forward_compatible_field": true + } + """ + .trimIndent() + } +} + +private open class SingleResponseClient(private val response: HttpResponse) : HttpClient { + var closed = false + private set + + override fun execute(request: HttpRequest, requestOptions: RequestOptions): HttpResponse = + response + + override fun executeAsync( + request: HttpRequest, + requestOptions: RequestOptions, + ): CompletableFuture = CompletableFuture.completedFuture(response) + + override fun close() { + closed = true + } +} + +private class DeferredResponseClient(private val responseFuture: CompletableFuture) : + HttpClient { + var closed = false + private set + + override fun execute(request: HttpRequest, requestOptions: RequestOptions): HttpResponse = + error("Unexpected synchronous call") + + override fun executeAsync( + request: HttpRequest, + requestOptions: RequestOptions, + ): CompletableFuture = responseFuture + + override fun close() { + closed = true + } +} + +private class NonCancellableFuture : CompletableFuture() { + override fun cancel(mayInterruptIfRunning: Boolean): Boolean = false +} + +private class SequenceResponseClient(responses: List) : HttpClient { + private val responses = ArrayDeque(responses) + var closed = false + private set + + override fun execute(request: HttpRequest, requestOptions: RequestOptions): HttpResponse = + error("Unexpected synchronous call") + + override fun executeAsync( + request: HttpRequest, + requestOptions: RequestOptions, + ): CompletableFuture = CompletableFuture.completedFuture(responses.removeFirst()) + + override fun close() { + closed = true + } +} + +private class BlockingResponse : HttpResponse { + val bodyStarted = CountDownLatch(1) + val closed = CountDownLatch(1) + val closeCount = AtomicInteger() + + @Volatile var bodyThreadName: String? = null + + override fun statusCode(): Int = 200 + + override fun headers(): Headers = Headers.builder().build() + + override fun body(): InputStream = + object : InputStream() { + override fun read(): Int { + bodyThreadName = Thread.currentThread().name + bodyStarted.countDown() + closed.await() + throw IOException("response closed") + } + } + + override fun close() { + closeCount.incrementAndGet() + closed.countDown() + } +} + +private class TestResponse( + private val statusCode: Int, + body: String, + private val responseHeaders: Headers = Headers.builder().build(), +) : HttpResponse { + private val bytes = body.toByteArray() + val closeCount = AtomicInteger() + var bodyRead = false + private set + + var closed = false + private set + + override fun statusCode(): Int = statusCode + + override fun headers(): Headers = responseHeaders + + override fun body(): ByteArrayInputStream { + bodyRead = true + return ByteArrayInputStream(bytes) + } + + override fun close() { + closeCount.incrementAndGet() + closed = true + } +} + +private class FailingBodyResponse(private val failure: IOException) : HttpResponse { + var closed = false + private set + + override fun statusCode(): Int = 200 + + override fun headers(): Headers = Headers.builder().build() + + override fun body(): InputStream = + object : InputStream() { + override fun read(): Int = throw failure + } + + override fun close() { + closed = true + } +} 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 14749d382..fa6932d00 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 @@ -2,14 +2,17 @@ package com.openai.client.okhttp import java.net.Proxy import java.net.Socket +import java.security.KeyStore import java.security.SecureRandom import java.security.cert.X509Certificate import java.util.concurrent.CopyOnWriteArrayList import javax.net.ssl.ExtendedSSLSession +import javax.net.ssl.KeyManagerFactory import javax.net.ssl.SNIHostName import javax.net.ssl.SSLContext import javax.net.ssl.SSLEngine import javax.net.ssl.SSLSocket +import javax.net.ssl.X509ExtendedKeyManager import javax.net.ssl.X509ExtendedTrustManager import javax.net.ssl.X509TrustManager import okhttp3.OkHttpClient @@ -110,6 +113,27 @@ internal inline fun OkHttpClient.useTestClient(block: (OkHttpClient) -> T): cache?.close() } +internal fun x509TestKeyManager(identities: Map): X509ExtendedKeyManager { + val password = "test password".toCharArray() + val keyStore = + KeyStore.getInstance("PKCS12").apply { + load(null, null) + identities.forEach { (alias, identity) -> + setKeyEntry( + alias, + identity.leaf.keyPair.private, + password, + arrayOf(identity.leaf.certificate, identity.root.certificate), + ) + } + } + val keyManagerFactory = + KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()).apply { + init(keyStore, password) + } + return keyManagerFactory.keyManagers.filterIsInstance().single() +} + private class RecordingClientTrustManager(private val delegate: X509TrustManager) : X509ExtendedTrustManager() { val requestedServerNames = CopyOnWriteArrayList() From f29339ab38c331fa132d571b9b9777019dd2357b Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Thu, 27 Aug 2026 02:10:32 +0000 Subject: [PATCH 2/4] fix(auth): address X.509 exchange review feedback --- .../openai/client/okhttp/X509TokenExchange.kt | 228 +++++++++++++++--- .../client/okhttp/X509TokenExchangeTest.kt | 120 +++++++-- 2 files changed, 293 insertions(+), 55 deletions(-) diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt index 3b7bf4380..7fe8bd25a 100644 --- a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt @@ -1,9 +1,9 @@ package com.openai.client.okhttp +import com.fasterxml.jackson.core.JsonParser import com.fasterxml.jackson.core.JsonProcessingException +import com.fasterxml.jackson.core.JsonToken import com.fasterxml.jackson.core.StreamReadFeature -import com.fasterxml.jackson.databind.DeserializationFeature -import com.fasterxml.jackson.databind.JsonNode import com.openai.core.http.Headers import com.openai.core.http.HttpClient import com.openai.core.http.HttpMethod @@ -14,8 +14,10 @@ import com.openai.core.jsonMapper import com.openai.errors.OpenAIInvalidDataException import com.openai.errors.OpenAIIoException import com.openai.errors.UnexpectedStatusCodeException +import com.openai.models.ErrorObject import java.io.IOException import java.io.OutputStream +import java.io.Writer import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.concurrent.ConcurrentHashMap @@ -40,10 +42,7 @@ internal class X509TokenExchange( ) : AutoCloseable { private val jsonMapper = jsonMapper() private val responseReader = - jsonMapper - .reader() - .with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS) - .with(StreamReadFeature.STRICT_DUPLICATE_DETECTION) + jsonMapper.reader().with(StreamReadFeature.STRICT_DUPLICATE_DETECTION) private val responseExecutor = ThreadPoolExecutor( 0, @@ -200,43 +199,161 @@ internal class X509TokenExchange( private fun parse(response: HttpResponse): X509AccessToken { val statusCode = response.statusCode() if (statusCode != 200) { - throw UnexpectedStatusCodeException.builder() - .statusCode(statusCode) - .headers(safeDiagnosticHeaders(response.headers())) - .build() + val builder = + UnexpectedStatusCodeException.builder() + .statusCode(statusCode) + .headers(safeDiagnosticHeaders(response.headers())) + readOAuthError(response)?.let(builder::error) + throw builder.build() } - val node: JsonNode = - try { - responseReader.readTree(response.body()) - } catch (error: JsonProcessingException) { - transportFailure(error)?.let { throw readFailure(it) } - throw invalidResponse() - } catch (error: IOException) { - throw readFailure(error) - } ?: throw invalidResponse() - - val accessToken = - node.text("access_token")?.takeIf(BEARER_TOKEN_PATTERN::matches) + return try { + responseReader.createParser(response.body()).use(::parseSuccessResponse) + } catch (error: JsonProcessingException) { + transportFailure(error)?.let { throw readFailure(it) } + throw invalidResponse() + } catch (error: IOException) { + throw readFailure(error) + } + } + + private fun parseSuccessResponse(parser: JsonParser): X509AccessToken { + if (parser.nextToken() != JsonToken.START_OBJECT) throw invalidResponse() + + var accessToken: String? = null + var tokenType: BoundedText? = null + var issuedTokenType: BoundedText? = null + var expiresIn: Long? = null + while (parser.nextToken() != JsonToken.END_OBJECT) { + if (parser.currentToken() != JsonToken.FIELD_NAME) throw invalidResponse() + val field = parser.currentName() + val valueToken = parser.nextToken() ?: throw invalidResponse() + when (field) { + "access_token" -> + accessToken = + parser + .takeIf { valueToken == JsonToken.VALUE_STRING } + ?.text + ?.takeIf(String::isNotBlank) + "token_type" -> + tokenType = + parser + .takeIf { valueToken == JsonToken.VALUE_STRING } + ?.boundedText(MAX_TOKEN_TYPE_CHARS) + "issued_token_type" -> + issuedTokenType = + parser + .takeIf { valueToken == JsonToken.VALUE_STRING } + ?.boundedText(MAX_ISSUED_TOKEN_TYPE_CHARS) + "expires_in" -> + expiresIn = + parser + .takeIf { valueToken == JsonToken.VALUE_NUMBER_INT } + ?.longValue + ?.takeIf { it > 0 } + else -> parser.skipChildren() + } + } + if (parser.nextToken() != null) throw invalidResponse() + + val validatedAccessToken = + accessToken?.takeIf(BEARER_TOKEN_PATTERN::matches) ?: throw invalidResponse("access_token") - if (!node.text("token_type").equals("Bearer", ignoreCase = true)) { + if ( + tokenType?.takeUnless(BoundedText::truncated)?.value?.let { + it.equals("Bearer", ignoreCase = true) + } != true + ) { throw invalidResponse("token_type") } - if (node.text("issued_token_type") != ACCESS_TOKEN_TYPE) { + if (issuedTokenType?.takeUnless(BoundedText::truncated)?.value != ACCESS_TOKEN_TYPE) { throw invalidResponse("issued_token_type") } - val expiresIn = - node - .get("expires_in") - ?.takeIf { it.isIntegralNumber && it.canConvertToLong() } - ?.longValue() - ?.takeIf { it in 1..MAX_TOKEN_LIFETIME_SECONDS } - ?: throw invalidResponse("expires_in") - return X509AccessToken(accessToken, Duration.ofSeconds(expiresIn)) + return X509AccessToken( + validatedAccessToken, + Duration.ofSeconds(expiresIn ?: throw invalidResponse("expires_in")), + ) + } + + private fun readOAuthError(response: HttpResponse): ErrorObject? = + try { + responseReader.createParser(response.body()).use(::parseOAuthError) + } catch (error: JsonProcessingException) { + transportFailure(error)?.let { throw readFailure(it) } + null + } catch (error: IOException) { + throw readFailure(error) + } catch (_: RuntimeException) { + null + } + + private fun parseOAuthError(parser: JsonParser): ErrorObject? { + if (parser.nextToken() != JsonToken.START_OBJECT) return null + + var errorCode: BoundedText? = null + var errorDescription: BoundedText? = null + while (parser.nextToken() != JsonToken.END_OBJECT) { + if (parser.currentToken() != JsonToken.FIELD_NAME) return null + val field = parser.currentName() + val valueToken = parser.nextToken() ?: return null + when (field) { + "error" -> + errorCode = + parser + .takeIf { valueToken == JsonToken.VALUE_STRING } + ?.boundedText(MAX_OAUTH_ERROR_CODE_CHARS) + "error_description" -> + errorDescription = + parser + .takeIf { valueToken == JsonToken.VALUE_STRING } + ?.boundedText(MAX_OAUTH_ERROR_DESCRIPTION_CHARS) + else -> parser.skipChildren() + } + } + if (parser.nextToken() != null) return null + + val safeCode = + errorCode + ?.takeUnless(BoundedText::truncated) + ?.value + ?.takeIf(OAUTH_ERROR_CODE_PATTERN::matches) + ?.takeUnless(SENSITIVE_DIAGNOSTIC_NAME_PATTERN::containsMatchIn) + val safeDescription = errorDescription?.value?.let(::sanitizeOAuthErrorDescription) + val message = safeDescription ?: safeCode ?: return null + return ErrorObject.builder() + .code(safeCode) + .message(message) + .param(null) + .type("oauth_error") + .build() } - private fun JsonNode.text(name: String): String? = - get(name)?.takeIf(JsonNode::isTextual)?.asText()?.takeIf(String::isNotBlank) + private fun JsonParser.boundedText(maxChars: Int): BoundedText { + val writer = BoundedTextWriter(maxChars) + getText(writer) + return writer.result() + } + + private fun sanitizeOAuthErrorDescription(value: String): String? { + var sanitized = + value + .map { character -> if (character.code in 0x20..0x7E) character else ' ' } + .joinToString("") + HEADER_CREDENTIAL_PATTERN.find(sanitized)?.let { match -> + sanitized = "${sanitized.substring(0, match.range.first).trimEnd()} " + } + sanitized = + AUTH_SCHEME_CREDENTIAL_PATTERN.replace(sanitized) { match -> + "${match.groupValues[1]} " + } + sanitized = + NAMED_CREDENTIAL_PATTERN.replace(sanitized) { match -> + "${match.groupValues[1]}=" + } + sanitized = JWT_CREDENTIAL_PATTERN.replace(sanitized, "") + sanitized = LONG_CREDENTIAL_CANDIDATE_PATTERN.replace(sanitized, "") + return sanitized.trim().takeIf(String::isNotEmpty) + } private fun invalidResponse(field: String? = null): OpenAIInvalidDataException = OpenAIInvalidDataException( @@ -266,10 +383,31 @@ internal class X509TokenExchange( const val TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange" const val X509_TOKEN_TYPE = "urn:openai:params:oauth:token-type:x509" const val ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" - const val MAX_TOKEN_LIFETIME_SECONDS = 3600L const val MAX_RESPONSE_THREADS = 4 const val RESPONSE_THREAD_IDLE_SECONDS = 30L + const val MAX_TOKEN_TYPE_CHARS = 32 + const val MAX_ISSUED_TOKEN_TYPE_CHARS = 128 + const val MAX_OAUTH_ERROR_CODE_CHARS = 128 + const val MAX_OAUTH_ERROR_DESCRIPTION_CHARS = 1024 val BEARER_TOKEN_PATTERN = Regex("[A-Za-z0-9._~+/-]+=*") + val OAUTH_ERROR_CODE_PATTERN = Regex("[A-Za-z0-9._~-]+") + val SENSITIVE_DIAGNOSTIC_NAME_PATTERN = + Regex( + "(?i)(?:authorization|cookie|session|api[-_]?key|access[-_]?token|" + + "refresh[-_]?token|subject[-_]?token|client[-_]?secret|password)" + ) + val HEADER_CREDENTIAL_PATTERN = + Regex("(?i)\\b(?:authorization|cookie|set-cookie)\\b\\s*[:=]") + val NAMED_CREDENTIAL_PATTERN = + Regex( + "(?i)\\b(authorization|cookie|set-cookie|session|api[-_ ]?key|" + + "access[-_ ]?token|refresh[-_ ]?token|subject[-_ ]?token|" + + "client[-_ ]?secret|password)\\b\\s*[:=]\\s*[^\\s,;]+" + ) + val AUTH_SCHEME_CREDENTIAL_PATTERN = Regex("(?i)\\b(Bearer|Basic)\\s+[^\\s,;]+") + val JWT_CREDENTIAL_PATTERN = + Regex("\\b[A-Za-z0-9_-]{8,}\\.[A-Za-z0-9_-]{8,}\\.[A-Za-z0-9_-]{8,}\\b") + val LONG_CREDENTIAL_CANDIDATE_PATTERN = Regex("[A-Za-z0-9._~+/=-]{24,}") val SAFE_DIAGNOSTIC_HEADERS = setOf( "Content-Length", @@ -287,6 +425,26 @@ internal class X509TokenExchange( } } +private data class BoundedText(val value: String, val truncated: Boolean) + +private class BoundedTextWriter(private val maxChars: Int) : Writer() { + private val value = StringBuilder(maxChars) + private var truncated = false + + override fun write(characters: CharArray, offset: Int, length: Int) { + val retained = minOf(length, maxChars - value.length) + if (retained > 0) value.append(characters, offset, retained) + if (retained < length) truncated = true + } + + override fun flush() {} + + override fun close() {} + + fun result(): BoundedText = + BoundedText(value.toString() + if (truncated) "..." else "", truncated) +} + private object ResponseThreadFactory : ThreadFactory { private val threadNumber = AtomicInteger() diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt index 5b4f8ddb7..1c7a2b23c 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt @@ -104,7 +104,6 @@ internal class X509TokenExchangeTest { "missing expires_in" to validResponseWithout("expires_in"), "zero expires_in" to validResponse(expiresIn = "0"), "fractional expires_in" to validResponse(expiresIn = "1.5"), - "expires_in above maximum" to validResponse(expiresIn = "3601"), "overflow expires_in" to validResponse(expiresIn = "9223372036854775808"), ) @@ -124,9 +123,22 @@ internal class X509TokenExchangeTest { @Test fun acceptsLargeForwardCompatibleResponsesWithoutAnArbitraryLimit() { - val node = ObjectMapper().readTree(validResponse()) - (node as ObjectNode).put("forward_compatible_field", "x".repeat(2 * 1024 * 1024)) - val response = TestResponse(200, node.toString()) + val response = + TestResponse( + 200, + """ + { + "forward_compatible_field": { + "nested": [{"value": "${"x".repeat(2 * 1024 * 1024)}"}] + }, + "access_token": "$ACCESS_TOKEN", + "issued_token_type": "$ACCESS_TOKEN_TYPE", + "token_type": "Bearer", + "expires_in": 3600 + } + """ + .trimIndent(), + ) val token = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, SingleResponseClient(response)).use { @@ -138,12 +150,35 @@ internal class X509TokenExchangeTest { } @Test - fun failureStatusDoesNotReadBodyAndRetainsOnlySafeDiagnosticHeaders() { + fun acceptsPositiveTokenLifetimesAboveOneHour() { + listOf(false, true).forEach { async -> + val response = TestResponse(200, validResponse(expiresIn = "86400")) + + val token = + X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, SingleResponseClient(response)).use { + exchange -> + if (async) exchange.executeAsync().get(5, TimeUnit.SECONDS) + else exchange.execute() + } + + assertThat(token.expiresIn).isEqualTo(Duration.ofDays(1)) + assertThat(response.closed).isTrue() + } + } + + @Test + fun failureStatusPreservesRedactedOAuthDiagnosticsAndSafeHeaders() { listOf(false, true).forEach { async -> val response = TestResponse( 503, - """{"error_description":"secret diagnostic"}""", + """ + { + "error": "invalid_grant", + "error_description": "Certificate is not authorized; Authorization: Bearer secret-body-token; Cookie: session=secret-body-cookie; access_token=secret-body-api-key" + } + """ + .trimIndent(), Headers.builder() .put("Set-Cookie", "session=secret-cookie") .put("Authorization", "Bearer secret-token") @@ -163,27 +198,68 @@ internal class X509TokenExchangeTest { assertThat(statusError.headers().values("X-Api-Key")).isEmpty() assertThat(statusError.headers().values("X-Request-ID")).containsExactly("req_safe") assertThat(statusError.headers().values("Retry-After")).containsExactly("1") + assertThat(statusError.code()).contains("invalid_grant") + assertThat(statusError.message).contains("Certificate is not authorized") assertThat(statusError.toString()) - .doesNotContain("secret-cookie", "secret-token", "secret-api-key") - assertThat(response.bodyRead).isFalse() + .doesNotContain( + "secret-cookie", + "secret-token", + "secret-api-key", + "secret-body-token", + "secret-body-cookie", + "secret-body-api-key", + ) + assertThat(statusError.body().toString()) + .doesNotContain("secret-body-token", "secret-body-cookie", "secret-body-api-key") + assertThat(response.bodyRead).isTrue() assertThat(response.closed).isTrue() } } @Test - fun rejectsNon200SuccessWithoutReadingTheBody() { + fun rejectsNon200SuccessWithoutRetainingCredentialFields() { listOf(false, true).forEach { async -> - val response = TestResponse(201, validResponse()) + val response = + TestResponse(201, validResponse(accessToken = "secret-token-must-not-leak")) val failure = exchangeFailure(response, async) assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) assertThat((failure as UnexpectedStatusCodeException).statusCode()).isEqualTo(201) - assertThat(response.bodyRead).isFalse() + assertThat(failure.toString()).doesNotContain("secret-token-must-not-leak") + assertThat(response.bodyRead).isTrue() assertThat(response.closed).isTrue() } } + @Test + fun redactsCompleteHeaderLikeValuesFromOAuthDiagnostics() { + mapOf( + "Authorization: Digest username=alice, response=secret-digest-value" to + "secret-digest-value", + "Cookie: first=secret-first-cookie; second=secret-second-cookie" to + "secret-second-cookie", + ) + .forEach { (diagnostic, secret) -> + listOf(false, true).forEach { async -> + val response = + TestResponse( + 400, + """{"error":"invalid_grant","error_description":"Safe prefix; $diagnostic"}""", + ) + + val failure = exchangeFailure(response, async) + + assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) + assertThat(failure.message).contains("Safe prefix", "") + assertThat(failure.toString()).doesNotContain(secret) + assertThat((failure as UnexpectedStatusCodeException).body().toString()) + .doesNotContain(secret) + assertThat(response.closed).isTrue() + } + } + } + @Test fun preservesIssuerBodyTimeoutAndIoFailuresAsSanitizedRetryableIo() { listOf( @@ -191,15 +267,18 @@ internal class X509TokenExchangeTest { IOException("issuer body disconnected"), ) .forEach { cause -> - listOf(false, true).forEach { async -> - val response = FailingBodyResponse(cause) + listOf(200, 503).forEach { statusCode -> + listOf(false, true).forEach { async -> + val response = FailingBodyResponse(cause, statusCode) - val failure = exchangeFailure(response, async) + val failure = exchangeFailure(response, async) - assertThat(failure).isInstanceOf(OpenAIIoException::class.java) - assertThat(failure).hasMessage("Failed to read X.509 token exchange response") - assertThat(failure.cause).isSameAs(cause) - assertThat(response.closed).isTrue() + assertThat(failure).isInstanceOf(OpenAIIoException::class.java) + assertThat(failure) + .hasMessage("Failed to read X.509 token exchange response") + assertThat(failure.cause).isSameAs(cause) + assertThat(response.closed).isTrue() + } } } } @@ -475,11 +554,12 @@ private class TestResponse( } } -private class FailingBodyResponse(private val failure: IOException) : HttpResponse { +private class FailingBodyResponse(private val failure: IOException, private val statusCode: Int) : + HttpResponse { var closed = false private set - override fun statusCode(): Int = 200 + override fun statusCode(): Int = statusCode override fun headers(): Headers = Headers.builder().build() From 9da1d2f25e8aabedfed6d0cae6828133a57789c6 Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Thu, 27 Aug 2026 03:51:12 +0000 Subject: [PATCH 3/4] fix(auth): harden X.509 response processing --- .../openai/client/okhttp/X509TokenExchange.kt | 326 ++++++++++++++---- .../client/okhttp/X509TokenExchangeTest.kt | 208 +++++++++-- 2 files changed, 439 insertions(+), 95 deletions(-) diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt index 7fe8bd25a..487cb626f 100644 --- a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt @@ -16,12 +16,16 @@ import com.openai.errors.OpenAIIoException import com.openai.errors.UnexpectedStatusCodeException import com.openai.models.ErrorObject import java.io.IOException +import java.io.InputStreamReader import java.io.OutputStream -import java.io.Writer +import java.io.Reader +import java.nio.charset.CharacterCodingException +import java.nio.charset.CodingErrorAction +import java.nio.charset.StandardCharsets import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.SynchronousQueue +import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.ThreadFactory import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit @@ -45,13 +49,14 @@ internal class X509TokenExchange( jsonMapper.reader().with(StreamReadFeature.STRICT_DUPLICATE_DETECTION) private val responseExecutor = ThreadPoolExecutor( - 0, - MAX_RESPONSE_THREADS, - RESPONSE_THREAD_IDLE_SECONDS, - TimeUnit.SECONDS, - SynchronousQueue(), - ResponseThreadFactory, - ) + MAX_RESPONSE_THREADS, + MAX_RESPONSE_THREADS, + RESPONSE_THREAD_IDLE_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(), + ResponseThreadFactory, + ) + .apply { allowCoreThreadTimeOut(true) } private val operations = ConcurrentHashMap.newKeySet() private val closed = AtomicBoolean() @@ -208,10 +213,15 @@ internal class X509TokenExchange( } return try { - responseReader.createParser(response.body()).use(::parseSuccessResponse) + responseParser(response, SUCCESS_STRING_LIMITS).use(::parseSuccessResponse) } catch (error: JsonProcessingException) { + if (isInvalidResponseFailure(error)) throw invalidResponse() transportFailure(error)?.let { throw readFailure(it) } throw invalidResponse() + } catch (_: X509ResponseConstraintException) { + throw invalidResponse() + } catch (_: CharacterCodingException) { + throw invalidResponse() } catch (error: IOException) { throw readFailure(error) } @@ -221,8 +231,8 @@ internal class X509TokenExchange( if (parser.nextToken() != JsonToken.START_OBJECT) throw invalidResponse() var accessToken: String? = null - var tokenType: BoundedText? = null - var issuedTokenType: BoundedText? = null + var tokenType: String? = null + var issuedTokenType: String? = null var expiresIn: Long? = null while (parser.nextToken() != JsonToken.END_OBJECT) { if (parser.currentToken() != JsonToken.FIELD_NAME) throw invalidResponse() @@ -236,15 +246,9 @@ internal class X509TokenExchange( ?.text ?.takeIf(String::isNotBlank) "token_type" -> - tokenType = - parser - .takeIf { valueToken == JsonToken.VALUE_STRING } - ?.boundedText(MAX_TOKEN_TYPE_CHARS) + tokenType = parser.takeIf { valueToken == JsonToken.VALUE_STRING }?.text "issued_token_type" -> - issuedTokenType = - parser - .takeIf { valueToken == JsonToken.VALUE_STRING } - ?.boundedText(MAX_ISSUED_TOKEN_TYPE_CHARS) + issuedTokenType = parser.takeIf { valueToken == JsonToken.VALUE_STRING }?.text "expires_in" -> expiresIn = parser @@ -259,14 +263,10 @@ internal class X509TokenExchange( val validatedAccessToken = accessToken?.takeIf(BEARER_TOKEN_PATTERN::matches) ?: throw invalidResponse("access_token") - if ( - tokenType?.takeUnless(BoundedText::truncated)?.value?.let { - it.equals("Bearer", ignoreCase = true) - } != true - ) { + if (tokenType?.equals("Bearer", ignoreCase = true) != true) { throw invalidResponse("token_type") } - if (issuedTokenType?.takeUnless(BoundedText::truncated)?.value != ACCESS_TOKEN_TYPE) { + if (issuedTokenType != ACCESS_TOKEN_TYPE) { throw invalidResponse("issued_token_type") } return X509AccessToken( @@ -277,9 +277,15 @@ internal class X509TokenExchange( private fun readOAuthError(response: HttpResponse): ErrorObject? = try { - responseReader.createParser(response.body()).use(::parseOAuthError) + responseParser(response, OAUTH_ERROR_STRING_LIMITS).use(::parseOAuthError) } catch (error: JsonProcessingException) { - transportFailure(error)?.let { throw readFailure(it) } + if (!isInvalidResponseFailure(error)) { + transportFailure(error)?.let { throw readFailure(it) } + } + null + } catch (_: X509ResponseConstraintException) { + null + } catch (_: CharacterCodingException) { null } catch (error: IOException) { throw readFailure(error) @@ -290,23 +296,16 @@ internal class X509TokenExchange( private fun parseOAuthError(parser: JsonParser): ErrorObject? { if (parser.nextToken() != JsonToken.START_OBJECT) return null - var errorCode: BoundedText? = null - var errorDescription: BoundedText? = null + var errorCode: String? = null + var errorDescription: String? = null while (parser.nextToken() != JsonToken.END_OBJECT) { if (parser.currentToken() != JsonToken.FIELD_NAME) return null val field = parser.currentName() val valueToken = parser.nextToken() ?: return null when (field) { - "error" -> - errorCode = - parser - .takeIf { valueToken == JsonToken.VALUE_STRING } - ?.boundedText(MAX_OAUTH_ERROR_CODE_CHARS) + "error" -> errorCode = parser.takeIf { valueToken == JsonToken.VALUE_STRING }?.text "error_description" -> - errorDescription = - parser - .takeIf { valueToken == JsonToken.VALUE_STRING } - ?.boundedText(MAX_OAUTH_ERROR_DESCRIPTION_CHARS) + errorDescription = parser.takeIf { valueToken == JsonToken.VALUE_STRING }?.text else -> parser.skipChildren() } } @@ -314,11 +313,11 @@ internal class X509TokenExchange( val safeCode = errorCode - ?.takeUnless(BoundedText::truncated) - ?.value ?.takeIf(OAUTH_ERROR_CODE_PATTERN::matches) ?.takeUnless(SENSITIVE_DIAGNOSTIC_NAME_PATTERN::containsMatchIn) - val safeDescription = errorDescription?.value?.let(::sanitizeOAuthErrorDescription) + ?.takeUnless(JWT_CREDENTIAL_PATTERN::containsMatchIn) + ?.takeUnless(LONG_CREDENTIAL_CANDIDATE_PATTERN::containsMatchIn) + val safeDescription = errorDescription?.let(::sanitizeOAuthErrorDescription) val message = safeDescription ?: safeCode ?: return null return ErrorObject.builder() .code(safeCode) @@ -328,10 +327,13 @@ internal class X509TokenExchange( .build() } - private fun JsonParser.boundedText(maxChars: Int): BoundedText { - val writer = BoundedTextWriter(maxChars) - getText(writer) - return writer.result() + private fun responseParser(response: HttpResponse, stringLimits: Map): JsonParser { + val decoder = + StandardCharsets.UTF_8.newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT) + val reader = BoundedFieldReader(InputStreamReader(response.body(), decoder), stringLimits) + return responseReader.createParser(reader) } private fun sanitizeOAuthErrorDescription(value: String): String? { @@ -339,17 +341,13 @@ internal class X509TokenExchange( value .map { character -> if (character.code in 0x20..0x7E) character else ' ' } .joinToString("") - HEADER_CREDENTIAL_PATTERN.find(sanitized)?.let { match -> + NAMED_CREDENTIAL_ASSIGNMENT_PATTERN.find(sanitized)?.let { match -> sanitized = "${sanitized.substring(0, match.range.first).trimEnd()} " } - sanitized = - AUTH_SCHEME_CREDENTIAL_PATTERN.replace(sanitized) { match -> - "${match.groupValues[1]} " - } - sanitized = - NAMED_CREDENTIAL_PATTERN.replace(sanitized) { match -> - "${match.groupValues[1]}=" - } + AUTH_SCHEME_CREDENTIAL_PATTERN.find(sanitized)?.let { match -> + sanitized = + sanitized.substring(0, match.range.first) + "${match.groupValues[1]} " + } sanitized = JWT_CREDENTIAL_PATTERN.replace(sanitized, "") sanitized = LONG_CREDENTIAL_CANDIDATE_PATTERN.replace(sanitized, "") return sanitized.trim().takeIf(String::isNotEmpty) @@ -373,9 +371,24 @@ internal class X509TokenExchange( return null } + private fun isInvalidResponseFailure(error: Throwable): Boolean { + var cause: Throwable? = error + while (cause != null) { + if (cause is X509ResponseConstraintException || cause is CharacterCodingException) { + return true + } + cause = cause.cause + } + return false + } + private fun safeDiagnosticHeaders(headers: Headers): Headers = Headers.builder() - .apply { SAFE_DIAGNOSTIC_HEADERS.forEach { name -> put(name, headers.values(name)) } } + .apply { + SAFE_DIAGNOSTIC_HEADERS.forEach { name -> + put(name, headers.values(name).mapNotNull(::sanitizeOAuthErrorDescription)) + } + } .build() private companion object { @@ -389,6 +402,16 @@ internal class X509TokenExchange( const val MAX_ISSUED_TOKEN_TYPE_CHARS = 128 const val MAX_OAUTH_ERROR_CODE_CHARS = 128 const val MAX_OAUTH_ERROR_DESCRIPTION_CHARS = 1024 + val SUCCESS_STRING_LIMITS = + mapOf( + "token_type" to MAX_TOKEN_TYPE_CHARS, + "issued_token_type" to MAX_ISSUED_TOKEN_TYPE_CHARS, + ) + val OAUTH_ERROR_STRING_LIMITS = + mapOf( + "error" to MAX_OAUTH_ERROR_CODE_CHARS, + "error_description" to MAX_OAUTH_ERROR_DESCRIPTION_CHARS, + ) val BEARER_TOKEN_PATTERN = Regex("[A-Za-z0-9._~+/-]+=*") val OAUTH_ERROR_CODE_PATTERN = Regex("[A-Za-z0-9._~-]+") val SENSITIVE_DIAGNOSTIC_NAME_PATTERN = @@ -396,15 +419,13 @@ internal class X509TokenExchange( "(?i)(?:authorization|cookie|session|api[-_]?key|access[-_]?token|" + "refresh[-_]?token|subject[-_]?token|client[-_]?secret|password)" ) - val HEADER_CREDENTIAL_PATTERN = - Regex("(?i)\\b(?:authorization|cookie|set-cookie)\\b\\s*[:=]") - val NAMED_CREDENTIAL_PATTERN = + val NAMED_CREDENTIAL_ASSIGNMENT_PATTERN = Regex( - "(?i)\\b(authorization|cookie|set-cookie|session|api[-_ ]?key|" + + "(?i)\\b(?:authorization|cookie|set-cookie|session|api[-_ ]?key|" + "access[-_ ]?token|refresh[-_ ]?token|subject[-_ ]?token|" + - "client[-_ ]?secret|password)\\b\\s*[:=]\\s*[^\\s,;]+" + "client[-_ ]?secret|password)\\b\\s*[:=]" ) - val AUTH_SCHEME_CREDENTIAL_PATTERN = Regex("(?i)\\b(Bearer|Basic)\\s+[^\\s,;]+") + val AUTH_SCHEME_CREDENTIAL_PATTERN = Regex("(?i)\\b(Bearer|Basic)\\s+") val JWT_CREDENTIAL_PATTERN = Regex("\\b[A-Za-z0-9_-]{8,}\\.[A-Za-z0-9_-]{8,}\\.[A-Za-z0-9_-]{8,}\\b") val LONG_CREDENTIAL_CANDIDATE_PATTERN = Regex("[A-Za-z0-9._~+/=-]{24,}") @@ -425,24 +446,181 @@ internal class X509TokenExchange( } } -private data class BoundedText(val value: String, val truncated: Boolean) +private class X509ResponseConstraintException : IOException("Invalid X.509 issuer response") + +/** + * Stops recognized top-level strings before Jackson materializes an oversized value. Jackson still + * owns JSON grammar, duplicate detection, unknown-value skipping, and trailing-token validation. + */ +private class BoundedFieldReader( + private val delegate: Reader, + private val stringLimits: Map, +) : Reader() { + private enum class StringKind { + FIELD_NAME, + FIELD_VALUE, + OTHER, + } + + private val maxFieldNameChars = stringLimits.keys.maxOf(String::length) + private var depth = 0 + private var previousSignificant: Char? = null + private var stringKind: StringKind? = null + private var fieldName = StringBuilder(maxFieldNameChars) + private var fieldNameTooLong = false + private var pendingFieldName: String? = null + private var valueLimit: Int? = null + private var valueChars = 0 + private var escaped = false + private var unicodeEscapeDigits = 0 + private var unicodeEscapeValue = 0 + private var firstCharacter = true + + override fun read(characters: CharArray, offset: Int, length: Int): Int { + if (length == 0) return 0 + + while (true) { + val read = delegate.read(characters, offset, length) + if (read <= 0) return read + + var monitoredOffset = offset + var monitoredLength = read + if (firstCharacter) { + firstCharacter = false + if (characters[monitoredOffset] == UTF8_BOM) { + monitoredOffset++ + monitoredLength-- + if (monitoredLength == 0) continue + characters.copyInto( + characters, + destinationOffset = offset, + startIndex = monitoredOffset, + endIndex = monitoredOffset + monitoredLength, + ) + monitoredOffset = offset + } + } + + for (index in monitoredOffset until monitoredOffset + monitoredLength) { + inspect(characters[index]) + } + return monitoredLength + } + } + + override fun close() = delegate.close() + + private fun inspect(character: Char) { + if (stringKind != null) { + inspectString(character) + return + } + + if (character == '"') { + startString() + return + } + if (!character.isWhitespace()) { + when (character) { + '{', + '[' -> depth++ + '}', + ']' -> depth-- + } + previousSignificant = character + } + } -private class BoundedTextWriter(private val maxChars: Int) : Writer() { - private val value = StringBuilder(maxChars) - private var truncated = false + private fun startString() { + stringKind = + when { + depth == 1 && (previousSignificant == '{' || previousSignificant == ',') -> { + fieldName = StringBuilder(maxFieldNameChars) + fieldNameTooLong = false + StringKind.FIELD_NAME + } + depth == 1 && previousSignificant == ':' -> { + valueLimit = stringLimits[pendingFieldName] + valueChars = 0 + StringKind.FIELD_VALUE + } + else -> StringKind.OTHER + } + escaped = false + unicodeEscapeDigits = 0 + unicodeEscapeValue = 0 + } - override fun write(characters: CharArray, offset: Int, length: Int) { - val retained = minOf(length, maxChars - value.length) - if (retained > 0) value.append(characters, offset, retained) - if (retained < length) truncated = true + private fun inspectString(character: Char) { + if (unicodeEscapeDigits > 0) { + val digit = Character.digit(character, 16) + if (digit < 0) { + fieldNameTooLong = true + } else { + unicodeEscapeValue = (unicodeEscapeValue shl 4) or digit + } + unicodeEscapeDigits-- + if (unicodeEscapeDigits == 0) acceptDecoded(unicodeEscapeValue.toChar()) + return + } + if (escaped) { + escaped = false + if (character == 'u') { + unicodeEscapeDigits = 4 + unicodeEscapeValue = 0 + } else { + acceptDecoded( + when (character) { + '"', + '\\', + '/' -> character + 'b' -> '\b' + 'f' -> '\u000C' + 'n' -> '\n' + 'r' -> '\r' + 't' -> '\t' + else -> character + } + ) + } + return + } + when (character) { + '\\' -> escaped = true + '"' -> finishString() + else -> acceptDecoded(character) + } } - override fun flush() {} + private fun acceptDecoded(character: Char) { + when (stringKind) { + StringKind.FIELD_NAME -> { + if (fieldName.length < maxFieldNameChars) fieldName.append(character) + else fieldNameTooLong = true + } + StringKind.FIELD_VALUE -> { + valueChars++ + if (valueLimit?.let { valueChars > it } == true) { + throw X509ResponseConstraintException() + } + } + StringKind.OTHER, + null -> {} + } + } - override fun close() {} + private fun finishString() { + if (stringKind == StringKind.FIELD_NAME) { + pendingFieldName = fieldName.takeUnless { fieldNameTooLong }?.toString() + } + stringKind = null + valueLimit = null + previousSignificant = '"' + } - fun result(): BoundedText = - BoundedText(value.toString() + if (truncated) "..." else "", truncated) + private companion object { + const val UTF8_BOM = '\uFEFF' + } } private object ResponseThreadFactory : ThreadFactory { diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt index 1c7a2b23c..65fc8b431 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt @@ -25,7 +25,6 @@ import java.util.concurrent.atomic.AtomicInteger import okhttp3.mockwebserver.MockResponse import okhttp3.tls.HandshakeCertificates import org.assertj.core.api.Assertions.assertThat -import org.assertj.core.api.Assertions.assertThatThrownBy import org.junit.jupiter.api.Test internal class X509TokenExchangeTest { @@ -149,6 +148,48 @@ internal class X509TokenExchangeTest { assertThat(response.closed).isTrue() } + @Test + fun acceptsAccessTokensAboveMetadataDiagnosticLimits() { + val largeAccessToken = "a".repeat(2 * 1024) + + listOf(false, true).forEach { async -> + val response = TestResponse(200, validResponse(accessToken = largeAccessToken)) + + val token = + X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, SingleResponseClient(response)).use { + exchange -> + if (async) exchange.executeAsync().get(5, TimeUnit.SECONDS) + else exchange.execute() + } + + assertThat(token.value).isEqualTo(largeAccessToken) + assertThat(response.closed).isTrue() + } + } + + @Test + fun rejectsOversizedRecognizedSuccessStringsBeforeReadingTheCompleteValue() { + mapOf( + "token_type" to validResponse(tokenType = OVERSIZED_RECOGNIZED_VALUE), + "issued_token_type" to validResponse(issuedTokenType = OVERSIZED_RECOGNIZED_VALUE), + "token\\u005ftype" to + validResponse(tokenType = OVERSIZED_RECOGNIZED_VALUE) + .replaceFirst("token_type", "token\\u005ftype"), + ) + .forEach { (field, body) -> + listOf(false, true).forEach { async -> + val response = TestResponse(200, body) + + assertThat(exchangeFailure(response, async)) + .describedAs("$field (async=$async)") + .isInstanceOf(OpenAIInvalidDataException::class.java) + .hasMessage("Invalid X.509 token exchange response") + assertThat(response.bytesRead.get()).isLessThan(body.toByteArray().size) + assertThat(response.closed).isTrue() + } + } + } + @Test fun acceptsPositiveTokenLifetimesAboveOneHour() { listOf(false, true).forEach { async -> @@ -185,6 +226,8 @@ internal class X509TokenExchangeTest { .put("X-Api-Key", "secret-api-key") .put("X-Request-ID", "req_safe") .put("Retry-After", "1") + .put("Traceparent", "aaaaaaaa.bbbbbbbb.cccccccc") + .put("Tracestate", "vendor=sk-test-not-a-real-credential-00000000") .build(), ) @@ -198,6 +241,8 @@ internal class X509TokenExchangeTest { assertThat(statusError.headers().values("X-Api-Key")).isEmpty() assertThat(statusError.headers().values("X-Request-ID")).containsExactly("req_safe") assertThat(statusError.headers().values("Retry-After")).containsExactly("1") + assertThat(statusError.headers().values("Traceparent")).containsExactly("") + assertThat(statusError.headers().values("Tracestate")).containsExactly("") assertThat(statusError.code()).contains("invalid_grant") assertThat(statusError.message).contains("Certificate is not authorized") assertThat(statusError.toString()) @@ -208,6 +253,13 @@ internal class X509TokenExchangeTest { "secret-body-token", "secret-body-cookie", "secret-body-api-key", + "aaaaaaaa.bbbbbbbb.cccccccc", + "sk-test-not-a-real-credential-00000000", + ) + assertThat(statusError.headers().toString()) + .doesNotContain( + "aaaaaaaa.bbbbbbbb.cccccccc", + "sk-test-not-a-real-credential-00000000", ) assertThat(statusError.body().toString()) .doesNotContain("secret-body-token", "secret-body-cookie", "secret-body-api-key") @@ -239,13 +291,23 @@ internal class X509TokenExchangeTest { "secret-digest-value", "Cookie: first=secret-first-cookie; second=secret-second-cookie" to "secret-second-cookie", + "client_secret=\"part-one,part-two\"" to "part-two", + "password=\"part-one;part-two\"" to "part-two", + "api_key=\"part-one part-two\"" to "part-two", + "Bearer \"part-one,part-two\"" to "part-two", ) .forEach { (diagnostic, secret) -> listOf(false, true).forEach { async -> val response = TestResponse( 400, - """{"error":"invalid_grant","error_description":"Safe prefix; $diagnostic"}""", + ObjectMapper() + .writeValueAsString( + mapOf( + "error" to "invalid_grant", + "error_description" to "Safe prefix; $diagnostic", + ) + ), ) val failure = exchangeFailure(response, async) @@ -260,6 +322,59 @@ internal class X509TokenExchangeTest { } } + @Test + fun rejectsCredentialLikeOAuthErrorCodesFromDiagnostics() { + listOf("sk-test-not-a-real-credential-00000000", "aaaaaaaa.bbbbbbbb.cccccccc").forEach { + credential -> + listOf(false, true).forEach { async -> + val response = + TestResponse( + 400, + """{"error":"$credential","error_description":"Issuer rejected certificate"}""", + ) + + val failure = exchangeFailure(response, async) + + assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) + val statusError = failure as UnexpectedStatusCodeException + assertThat(statusError.code()).isEmpty() + assertThat(statusError.message).contains("Issuer rejected certificate") + assertThat(statusError.message).doesNotContain(credential) + assertThat(statusError.body().toString()).doesNotContain(credential) + assertThat(statusError.toString()).doesNotContain(credential) + assertThat(response.closed).isTrue() + } + } + } + + @Test + fun ignoresOversizedOAuthDiagnosticsBeforeReadingTheCompleteValue() { + mapOf( + "error" to """{"error":"$OVERSIZED_RECOGNIZED_VALUE","error_description":"safe"}""", + "error_description" to + """{"error":"invalid_grant","error_description":"$OVERSIZED_RECOGNIZED_VALUE"}""", + "error\\u005fdescription" to + "{\"error\":\"invalid_grant\",\"error\\u005fdescription\":" + + "\"$OVERSIZED_RECOGNIZED_VALUE\"}", + ) + .forEach { (field, body) -> + listOf(false, true).forEach { async -> + val response = TestResponse(503, body) + + val failure = exchangeFailure(response, async) + + assertThat(failure) + .describedAs("$field (async=$async)") + .isInstanceOf(UnexpectedStatusCodeException::class.java) + val statusError = failure as UnexpectedStatusCodeException + assertThat(statusError.statusCode()).isEqualTo(503) + assertThat(statusError.code()).isEmpty() + assertThat(response.bytesRead.get()).isLessThan(body.toByteArray().size) + assertThat(response.closed).isTrue() + } + } + } + @Test fun preservesIssuerBodyTimeoutAndIoFailuresAsSanitizedRetryableIo() { listOf( @@ -331,27 +446,32 @@ internal class X509TokenExchangeTest { } @Test - fun responseExecutorIsBoundedAndClosesRejectedResponses() { - val responses = List(5) { BlockingResponse() } + fun responseExecutorQueuesValidResponsesBeyondItsOwnedThreadCount() { + val release = CountDownLatch(1) + val activeResponses = List(4) { GatedResponse(validResponse(), release) } + val queuedResponse = TestResponse(200, validResponse()) + val responses = activeResponses + queuedResponse val client = SequenceResponseClient(responses) val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) - val results = mutableListOf>() + try { + val results = responses.map { exchange.executeAsync() } - repeat(4) { index -> - results += exchange.executeAsync() - assertThat(responses[index].bodyStarted.await(5, TimeUnit.SECONDS)).isTrue() + activeResponses.forEach { response -> + assertThat(response.bodyStarted.await(5, TimeUnit.SECONDS)).isTrue() + } + assertThat(queuedResponse.bodyRead).isFalse() + assertThat(results.last().isDone).isFalse() + + release.countDown() + + assertThat(results.map { it.get(5, TimeUnit.SECONDS).value }) + .containsExactlyElementsOf(List(5) { ACCESS_TOKEN }) + assertThat(activeResponses).allMatch { it.closed } + assertThat(queuedResponse.closed).isTrue() + } finally { + release.countDown() + exchange.close() } - results += exchange.executeAsync() - - assertThatThrownBy { results.last().get(5, TimeUnit.SECONDS) } - .isInstanceOf(ExecutionException::class.java) - .hasCauseInstanceOf(OpenAIIoException::class.java) - assertThat(responses.last().closed.await(5, TimeUnit.SECONDS)).isTrue() - assertThat(responses.last().bodyStarted.count).isEqualTo(1) - - exchange.close() - assertThat(results.take(4)).allMatch(CompletableFuture<*>::isCancelled) - assertThat(responses.take(4)).allMatch { it.closed.await(5, TimeUnit.SECONDS) } assertThat(client.closed).isFalse() } @@ -418,6 +538,7 @@ internal class X509TokenExchangeTest { const val SERVICE_ACCOUNT_ID = "svc_acct_test" const val ACCESS_TOKEN = "test-x509-access-token" const val ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" + val OVERSIZED_RECOGNIZED_VALUE = "x".repeat(256 * 1024) val TOKEN_REQUEST = """ { @@ -526,12 +647,51 @@ private class BlockingResponse : HttpResponse { } } +private class GatedResponse(body: String, private val release: CountDownLatch) : HttpResponse { + private val bytes = body.toByteArray() + val bodyStarted = CountDownLatch(1) + var closed = false + private set + + override fun statusCode(): Int = 200 + + override fun headers(): Headers = Headers.builder().build() + + override fun body(): InputStream = + object : ByteArrayInputStream(bytes) { + override fun read(): Int { + awaitRelease() + return super.read() + } + + override fun read(buffer: ByteArray, offset: Int, length: Int): Int { + awaitRelease() + return super.read(buffer, offset, length) + } + + private fun awaitRelease() { + bodyStarted.countDown() + try { + release.await() + } catch (error: InterruptedException) { + Thread.currentThread().interrupt() + throw IOException("Interrupted while waiting to read test response", error) + } + } + } + + override fun close() { + closed = true + } +} + private class TestResponse( private val statusCode: Int, body: String, private val responseHeaders: Headers = Headers.builder().build(), ) : HttpResponse { private val bytes = body.toByteArray() + val bytesRead = AtomicInteger() val closeCount = AtomicInteger() var bodyRead = false private set @@ -543,9 +703,15 @@ private class TestResponse( override fun headers(): Headers = responseHeaders - override fun body(): ByteArrayInputStream { + override fun body(): InputStream { bodyRead = true - return ByteArrayInputStream(bytes) + return object : ByteArrayInputStream(bytes) { + override fun read(): Int = + super.read().also { if (it >= 0) bytesRead.incrementAndGet() } + + override fun read(buffer: ByteArray, offset: Int, length: Int): Int = + super.read(buffer, offset, length).also { if (it > 0) bytesRead.addAndGet(it) } + } } override fun close() { From e319e8a56a207694c56b822b30b1c56dc0e70a8c Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Thu, 27 Aug 2026 05:24:25 +0000 Subject: [PATCH 4/4] fix(auth): close X.509 feedback gaps --- .../openai/client/okhttp/X509TokenExchange.kt | 315 ++++++++----- .../client/okhttp/X509TokenExchangeTest.kt | 417 ++++++++++++++++-- 2 files changed, 577 insertions(+), 155 deletions(-) diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt index 487cb626f..73818bdcf 100644 --- a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/X509TokenExchange.kt @@ -23,9 +23,10 @@ import java.nio.charset.CharacterCodingException import java.nio.charset.CodingErrorAction import java.nio.charset.StandardCharsets import java.time.Duration +import java.util.concurrent.ArrayBlockingQueue import java.util.concurrent.CompletableFuture import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.ExecutionException import java.util.concurrent.ThreadFactory import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit @@ -53,10 +54,11 @@ internal class X509TokenExchange( MAX_RESPONSE_THREADS, RESPONSE_THREAD_IDLE_SECONDS, TimeUnit.SECONDS, - LinkedBlockingQueue(), + ArrayBlockingQueue(MAX_QUEUED_EXCHANGES), ResponseThreadFactory, ) .apply { allowCoreThreadTimeOut(true) } + private val lifecycleLock = Any() private val operations = ConcurrentHashMap.newKeySet() private val closed = AtomicBoolean() @@ -71,59 +73,87 @@ internal class X509TokenExchange( } fun executeAsync(): CompletableFuture { - checkOpen() - val operation = AsyncOperation() - operations.add(operation) - operation.result.whenComplete { _, _ -> operations.remove(operation) } - - val responseFuture = - try { - httpClient.executeAsync(request()) - } catch (error: Throwable) { - operations.remove(operation) - throw error + val operation = + synchronized(lifecycleLock) { + checkOpen() + AsyncOperation().also { operation -> + operations.add(operation) + try { + responseExecutor.execute(operation) + } catch (_: java.util.concurrent.RejectedExecutionException) { + operation.result.completeExceptionally( + OpenAIIoException("X.509 token exchange processing unavailable") + ) + } + } } - operation.responseFuture.set(responseFuture) - responseFuture.whenComplete(operation::accept) - if (closed.get() || operation.result.isCancelled) { - operation.result.cancel(true) - responseFuture.cancel(true) - } return operation.result } override fun close() { - if (closed.compareAndSet(false, true)) { - operations.toTypedArray().forEach { operation -> operation.result.cancel(true) } - responseExecutor.shutdownNow() - } + val cancellations = + synchronized(lifecycleLock) { + if (closed.compareAndSet(false, true)) { + operations + .toTypedArray() + .filter { operation -> operation.prepareCancellation(true) } + .also { responseExecutor.shutdown() } + } else { + emptyList() + } + } + cancellations.forEach { operation -> operation.publishCancellation(true) } } private fun checkOpen() { check(!closed.get()) { "X.509 token exchange is closed" } } - private inner class AsyncOperation { - val result = CompletableFuture() + private inner class AsyncOperation : Runnable { + private val terminal = AtomicBoolean() + private val cancellationRequested = AtomicBoolean() + private val enrollmentLock = Any() val responseFuture = AtomicReference?>() + private val responseLeaseFuture = CompletableFuture() private val activeResponse = AtomicReference() + val result = OperationResult(this) - init { - result.whenComplete { _, _ -> - if (result.isCancelled) { - responseFuture.get()?.cancel(true) - activeResponse.getAndSet(null)?.close() - } + override fun run() { + try { + if (!initiateRequest()) return + val lease = + try { + responseLeaseFuture.get() + } catch (error: ExecutionException) { + throw error.cause ?: error + } + val token = lease.use { if (terminal.get()) null else parse(lease.response) } + if (token != null) result.complete(token) + } catch (error: Throwable) { + if (!closed.get() && !terminal.get()) result.completeExceptionally(error) + } finally { + activeResponse.getAndSet(null)?.close() } } - fun accept(response: HttpResponse?, error: Throwable?) { + private fun initiateRequest(): Boolean = + synchronized(enrollmentLock) { + if (terminal.get() || closed.get()) return@synchronized false + + val future = httpClient.executeAsync(request()) + future.whenComplete(::acceptResponse) + responseFuture.set(future) + if (terminal.get() || closed.get()) future.cancel(true) + true + } + + private fun acceptResponse(response: HttpResponse?, error: Throwable?) { if (error != null) { - if (!result.isDone) result.completeExceptionally(error) + responseLeaseFuture.completeExceptionally(error) return } if (response == null) { - result.completeExceptionally( + responseLeaseFuture.completeExceptionally( IllegalStateException("X.509 token exchange completed without a response") ) return @@ -131,39 +161,74 @@ internal class X509TokenExchange( val lease = ResponseLease(response) activeResponse.set(lease) - if (result.isDone) { - close(lease) - return - } - try { - responseExecutor.execute { process(lease) } - } catch (_: java.util.concurrent.RejectedExecutionException) { - close(lease) - if (!result.isDone) { - result.completeExceptionally( - OpenAIIoException("X.509 token exchange response processing unavailable") - ) + if (terminal.get() && activeResponse.compareAndSet(lease, null)) lease.close() + responseLeaseFuture.complete(lease) + } + + private fun cancelResources(mayInterruptIfRunning: Boolean) { + val completedFuture = + synchronized(enrollmentLock) { + val future = responseFuture.get() + val canceled = future?.cancel(mayInterruptIfRunning) == true + activeResponse.getAndSet(null)?.close() + future?.takeIf { !canceled && it.isDone } } + if (completedFuture != null) { + responseLeaseFuture.handle { _, _ -> Unit }.join() + activeResponse.getAndSet(null)?.close() } } - private fun process(lease: ResponseLease) { - try { - val token = lease.use { if (result.isDone) null else parse(lease.response) } - if (token != null && !result.isDone) result.complete(token) - } catch (error: Throwable) { - if (!result.isDone) result.completeExceptionally(error) - } finally { - activeResponse.compareAndSet(lease, null) - } + fun cancel(mayInterruptIfRunning: Boolean): Boolean { + if (!prepareCancellation(mayInterruptIfRunning)) return false + return publishCancellation(mayInterruptIfRunning) + } + + fun prepareCancellation(mayInterruptIfRunning: Boolean): Boolean { + val firstCompletion = terminal.compareAndSet(false, true) + if (firstCompletion) cancellationRequested.set(true) + if (cancellationRequested.get()) cancelResources(mayInterruptIfRunning) + if (!firstCompletion) return false + + responseExecutor.remove(this) + operations.remove(this) + return true + } + + fun publishCancellation(mayInterruptIfRunning: Boolean): Boolean = + result.publishCancellation(mayInterruptIfRunning) + + fun complete(value: X509AccessToken): Boolean { + if (!terminal.compareAndSet(false, true)) return false + operations.remove(this) + return result.publishValue(value) } - private fun close(lease: ResponseLease) { - activeResponse.compareAndSet(lease, null) - lease.close() + fun completeExceptionally(error: Throwable): Boolean { + if (!terminal.compareAndSet(false, true)) return false + operations.remove(this) + return result.publishException(error) } } + private inner class OperationResult(private val operation: AsyncOperation) : + CompletableFuture() { + override fun cancel(mayInterruptIfRunning: Boolean): Boolean = + operation.cancel(mayInterruptIfRunning) + + override fun complete(value: X509AccessToken): Boolean = operation.complete(value) + + override fun completeExceptionally(error: Throwable): Boolean = + operation.completeExceptionally(error) + + fun publishCancellation(mayInterruptIfRunning: Boolean): Boolean = + super.cancel(mayInterruptIfRunning) + + fun publishValue(value: X509AccessToken): Boolean = super.complete(value) + + fun publishException(error: Throwable): Boolean = super.completeExceptionally(error) + } + private class ResponseLease(val response: HttpResponse) : AutoCloseable { private val closed = AtomicBoolean() @@ -239,22 +304,28 @@ internal class X509TokenExchange( val field = parser.currentName() val valueToken = parser.nextToken() ?: throw invalidResponse() when (field) { - "access_token" -> - accessToken = - parser - .takeIf { valueToken == JsonToken.VALUE_STRING } - ?.text - ?.takeIf(String::isNotBlank) - "token_type" -> - tokenType = parser.takeIf { valueToken == JsonToken.VALUE_STRING }?.text - "issued_token_type" -> - issuedTokenType = parser.takeIf { valueToken == JsonToken.VALUE_STRING }?.text - "expires_in" -> - expiresIn = - parser - .takeIf { valueToken == JsonToken.VALUE_NUMBER_INT } - ?.longValue - ?.takeIf { it > 0 } + "access_token" -> { + if (valueToken != JsonToken.VALUE_STRING) { + throw invalidResponse("access_token") + } + accessToken = parser.text.takeIf(String::isNotBlank) + } + "token_type" -> { + if (valueToken != JsonToken.VALUE_STRING) throw invalidResponse("token_type") + tokenType = parser.text + } + "issued_token_type" -> { + if (valueToken != JsonToken.VALUE_STRING) { + throw invalidResponse("issued_token_type") + } + issuedTokenType = parser.text + } + "expires_in" -> { + if (valueToken != JsonToken.VALUE_NUMBER_INT) { + throw invalidResponse("expires_in") + } + expiresIn = parser.longValue.takeIf { it > 0 } + } else -> parser.skipChildren() } } @@ -297,28 +368,23 @@ internal class X509TokenExchange( if (parser.nextToken() != JsonToken.START_OBJECT) return null var errorCode: String? = null - var errorDescription: String? = null while (parser.nextToken() != JsonToken.END_OBJECT) { if (parser.currentToken() != JsonToken.FIELD_NAME) return null val field = parser.currentName() val valueToken = parser.nextToken() ?: return null when (field) { - "error" -> errorCode = parser.takeIf { valueToken == JsonToken.VALUE_STRING }?.text - "error_description" -> - errorDescription = parser.takeIf { valueToken == JsonToken.VALUE_STRING }?.text + "error" -> { + if (valueToken != JsonToken.VALUE_STRING) return null + errorCode = parser.text + } + "error_description" -> if (valueToken != JsonToken.VALUE_STRING) return null else -> parser.skipChildren() } } if (parser.nextToken() != null) return null - val safeCode = - errorCode - ?.takeIf(OAUTH_ERROR_CODE_PATTERN::matches) - ?.takeUnless(SENSITIVE_DIAGNOSTIC_NAME_PATTERN::containsMatchIn) - ?.takeUnless(JWT_CREDENTIAL_PATTERN::containsMatchIn) - ?.takeUnless(LONG_CREDENTIAL_CANDIDATE_PATTERN::containsMatchIn) - val safeDescription = errorDescription?.let(::sanitizeOAuthErrorDescription) - val message = safeDescription ?: safeCode ?: return null + val safeCode = errorCode?.takeIf(SAFE_OAUTH_ERROR_CODES::contains) + val message = safeCode ?: return null return ErrorObject.builder() .code(safeCode) .message(message) @@ -336,23 +402,6 @@ internal class X509TokenExchange( return responseReader.createParser(reader) } - private fun sanitizeOAuthErrorDescription(value: String): String? { - var sanitized = - value - .map { character -> if (character.code in 0x20..0x7E) character else ' ' } - .joinToString("") - NAMED_CREDENTIAL_ASSIGNMENT_PATTERN.find(sanitized)?.let { match -> - sanitized = "${sanitized.substring(0, match.range.first).trimEnd()} " - } - AUTH_SCHEME_CREDENTIAL_PATTERN.find(sanitized)?.let { match -> - sanitized = - sanitized.substring(0, match.range.first) + "${match.groupValues[1]} " - } - sanitized = JWT_CREDENTIAL_PATTERN.replace(sanitized, "") - sanitized = LONG_CREDENTIAL_CANDIDATE_PATTERN.replace(sanitized, "") - return sanitized.trim().takeIf(String::isNotEmpty) - } - private fun invalidResponse(field: String? = null): OpenAIInvalidDataException = OpenAIInvalidDataException( if (field == null) "Invalid X.509 token exchange response" @@ -386,18 +435,42 @@ internal class X509TokenExchange( Headers.builder() .apply { SAFE_DIAGNOSTIC_HEADERS.forEach { name -> - put(name, headers.values(name).mapNotNull(::sanitizeOAuthErrorDescription)) + put( + name, + headers.values(name).mapNotNull { value -> + safeDiagnosticHeaderValue(name, value) + }, + ) } } .build() + private fun safeDiagnosticHeaderValue(name: String, value: String): String? { + if (name in OPAQUE_DIAGNOSTIC_HEADERS) { + return if (value.isEmpty()) null else "" + } + if (value.length > MAX_DIAGNOSTIC_HEADER_CHARS) return null + + return when (name) { + "Content-Length", + "Retry-After", + "Retry-After-Ms" -> value.takeIf { it.isNotEmpty() && it.all { it in '0'..'9' } } + "Content-Type" -> + value.trim().lowercase().takeIf(SAFE_DIAGNOSTIC_CONTENT_TYPES::contains) + "X-Should-Retry" -> value.lowercase().takeIf { it == "true" || it == "false" } + else -> null + } + } + private companion object { const val TOKEN_EXCHANGE_URL = "https://mtls.auth.openai.com/oauth/token" const val TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange" const val X509_TOKEN_TYPE = "urn:openai:params:oauth:token-type:x509" const val ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" const val MAX_RESPONSE_THREADS = 4 + const val MAX_QUEUED_EXCHANGES = MAX_RESPONSE_THREADS const val RESPONSE_THREAD_IDLE_SECONDS = 30L + const val MAX_DIAGNOSTIC_HEADER_CHARS = 64 const val MAX_TOKEN_TYPE_CHARS = 32 const val MAX_ISSUED_TOKEN_TYPE_CHARS = 128 const val MAX_OAUTH_ERROR_CODE_CHARS = 128 @@ -413,27 +486,20 @@ internal class X509TokenExchange( "error_description" to MAX_OAUTH_ERROR_DESCRIPTION_CHARS, ) val BEARER_TOKEN_PATTERN = Regex("[A-Za-z0-9._~+/-]+=*") - val OAUTH_ERROR_CODE_PATTERN = Regex("[A-Za-z0-9._~-]+") - val SENSITIVE_DIAGNOSTIC_NAME_PATTERN = - Regex( - "(?i)(?:authorization|cookie|session|api[-_]?key|access[-_]?token|" + - "refresh[-_]?token|subject[-_]?token|client[-_]?secret|password)" - ) - val NAMED_CREDENTIAL_ASSIGNMENT_PATTERN = - Regex( - "(?i)\\b(?:authorization|cookie|set-cookie|session|api[-_ ]?key|" + - "access[-_ ]?token|refresh[-_ ]?token|subject[-_ ]?token|" + - "client[-_ ]?secret|password)\\b\\s*[:=]" + val SAFE_OAUTH_ERROR_CODES = + setOf( + "invalid_client", + "invalid_grant", + "invalid_request", + "invalid_scope", + "invalid_target", + "unauthorized_client", + "unsupported_grant_type", ) - val AUTH_SCHEME_CREDENTIAL_PATTERN = Regex("(?i)\\b(Bearer|Basic)\\s+") - val JWT_CREDENTIAL_PATTERN = - Regex("\\b[A-Za-z0-9_-]{8,}\\.[A-Za-z0-9_-]{8,}\\.[A-Za-z0-9_-]{8,}\\b") - val LONG_CREDENTIAL_CANDIDATE_PATTERN = Regex("[A-Za-z0-9._~+/=-]{24,}") val SAFE_DIAGNOSTIC_HEADERS = setOf( "Content-Length", "Content-Type", - "Date", "OpenAI-Request-ID", "Request-ID", "Retry-After", @@ -443,6 +509,15 @@ internal class X509TokenExchange( "X-Request-ID", "X-Should-Retry", ) + val OPAQUE_DIAGNOSTIC_HEADERS = + setOf("OpenAI-Request-ID", "Request-ID", "Traceparent", "Tracestate", "X-Request-ID") + val SAFE_DIAGNOSTIC_CONTENT_TYPES = + setOf( + "application/json", + "application/json; charset=utf-8", + "application/problem+json", + "application/problem+json; charset=utf-8", + ) } } diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt index 65fc8b431..1c150d508 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/X509TokenExchangeTest.kt @@ -18,10 +18,12 @@ import java.net.SocketTimeoutException import java.security.cert.X509Certificate import java.time.Duration import java.util.concurrent.CompletableFuture +import java.util.concurrent.ConcurrentLinkedQueue import java.util.concurrent.CountDownLatch import java.util.concurrent.ExecutionException import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger +import java.util.function.BiConsumer import okhttp3.mockwebserver.MockResponse import okhttp3.tls.HandshakeCertificates import org.assertj.core.api.Assertions.assertThat @@ -190,6 +192,24 @@ internal class X509TokenExchangeTest { } } + @Test + fun rejectsNestedRecognizedSuccessFieldsBeforeReadingLargeNestedStrings() { + listOf("access_token", "token_type", "issued_token_type", "expires_in").forEach { field -> + val body = + validResponseWithRawValue(field, """{"$field":"$OVERSIZED_RECOGNIZED_VALUE"}""") + listOf(false, true).forEach { async -> + val response = TestResponse(200, body) + + assertThat(exchangeFailure(response, async)) + .describedAs("$field (async=$async)") + .isInstanceOf(OpenAIInvalidDataException::class.java) + .hasMessage("Invalid X.509 token exchange response field: $field") + assertThat(response.bytesRead.get()).isLessThan(body.toByteArray().size) + assertThat(response.closed).isTrue() + } + } + } + @Test fun acceptsPositiveTokenLifetimesAboveOneHour() { listOf(false, true).forEach { async -> @@ -208,7 +228,7 @@ internal class X509TokenExchangeTest { } @Test - fun failureStatusPreservesRedactedOAuthDiagnosticsAndSafeHeaders() { + fun failureStatusPreservesSafeOAuthCodeAndHeadersWithoutFreeFormDescription() { listOf(false, true).forEach { async -> val response = TestResponse( @@ -239,14 +259,15 @@ internal class X509TokenExchangeTest { assertThat(statusError.headers().values("Set-Cookie")).isEmpty() assertThat(statusError.headers().values("Authorization")).isEmpty() assertThat(statusError.headers().values("X-Api-Key")).isEmpty() - assertThat(statusError.headers().values("X-Request-ID")).containsExactly("req_safe") + assertThat(statusError.headers().values("X-Request-ID")).containsExactly("") assertThat(statusError.headers().values("Retry-After")).containsExactly("1") assertThat(statusError.headers().values("Traceparent")).containsExactly("") assertThat(statusError.headers().values("Tracestate")).containsExactly("") assertThat(statusError.code()).contains("invalid_grant") - assertThat(statusError.message).contains("Certificate is not authorized") + assertThat(statusError.message).contains("invalid_grant") assertThat(statusError.toString()) .doesNotContain( + "Certificate is not authorized", "secret-cookie", "secret-token", "secret-api-key", @@ -268,6 +289,47 @@ internal class X509TokenExchangeTest { } } + @Test + fun doesNotExposeShortOpaqueSecretsFromDiagnosticHeaders() { + listOf(false, true).forEach { async -> + val response = + TestResponse( + 400, + """{"error":"invalid_request"}""", + Headers.builder() + .put("X-Request-ID", "short-request-secret") + .put("Tracestate", "short-trace-secret") + .put("Retry-After", "short-retry-secret") + .put("Content-Type", "short-content-secret") + .build(), + ) + + val failure = exchangeFailure(response, async) + + assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) + val statusError = failure as UnexpectedStatusCodeException + assertThat(statusError.headers().values("X-Request-ID")).containsExactly("") + assertThat(statusError.headers().values("Tracestate")).containsExactly("") + assertThat(statusError.headers().values("Retry-After")).isEmpty() + assertThat(statusError.headers().values("Content-Type")).isEmpty() + assertThat(statusError.headers().toString()) + .doesNotContain( + "short-request-secret", + "short-trace-secret", + "short-retry-secret", + "short-content-secret", + ) + assertThat(failure.toString()) + .doesNotContain( + "short-request-secret", + "short-trace-secret", + "short-retry-secret", + "short-content-secret", + ) + assertThat(response.closed).isTrue() + } + } + @Test fun rejectsNon200SuccessWithoutRetainingCredentialFields() { listOf(false, true).forEach { async -> @@ -285,8 +347,9 @@ internal class X509TokenExchangeTest { } @Test - fun redactsCompleteHeaderLikeValuesFromOAuthDiagnostics() { + fun doesNotExposeFreeFormOAuthErrorDescriptions() { mapOf( + "short-unlabeled-secret" to "short-unlabeled-secret", "Authorization: Digest username=alice, response=secret-digest-value" to "secret-digest-value", "Cookie: first=secret-first-cookie; second=secret-second-cookie" to @@ -313,10 +376,11 @@ internal class X509TokenExchangeTest { val failure = exchangeFailure(response, async) assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) - assertThat(failure.message).contains("Safe prefix", "") + val statusError = failure as UnexpectedStatusCodeException + assertThat(statusError.code()).contains("invalid_grant") + assertThat(failure.message).doesNotContain("Safe prefix", diagnostic, secret) assertThat(failure.toString()).doesNotContain(secret) - assertThat((failure as UnexpectedStatusCodeException).body().toString()) - .doesNotContain(secret) + assertThat(statusError.body().toString()).doesNotContain(diagnostic, secret) assertThat(response.closed).isTrue() } } @@ -324,27 +388,31 @@ internal class X509TokenExchangeTest { @Test fun rejectsCredentialLikeOAuthErrorCodesFromDiagnostics() { - listOf("sk-test-not-a-real-credential-00000000", "aaaaaaaa.bbbbbbbb.cccccccc").forEach { - credential -> - listOf(false, true).forEach { async -> - val response = - TestResponse( - 400, - """{"error":"$credential","error_description":"Issuer rejected certificate"}""", - ) - - val failure = exchangeFailure(response, async) - - assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) - val statusError = failure as UnexpectedStatusCodeException - assertThat(statusError.code()).isEmpty() - assertThat(statusError.message).contains("Issuer rejected certificate") - assertThat(statusError.message).doesNotContain(credential) - assertThat(statusError.body().toString()).doesNotContain(credential) - assertThat(statusError.toString()).doesNotContain(credential) - assertThat(response.closed).isTrue() + listOf( + "short-unlabeled-secret", + "sk-test-not-a-real-credential-00000000", + "aaaaaaaa.bbbbbbbb.cccccccc", + ) + .forEach { credential -> + listOf(false, true).forEach { async -> + val response = + TestResponse( + 400, + """{"error":"$credential","error_description":"Issuer rejected certificate"}""", + ) + + val failure = exchangeFailure(response, async) + + assertThat(failure).isInstanceOf(UnexpectedStatusCodeException::class.java) + val statusError = failure as UnexpectedStatusCodeException + assertThat(statusError.code()).isEmpty() + assertThat(statusError.message).doesNotContain("Issuer rejected certificate") + assertThat(statusError.message).doesNotContain(credential) + assertThat(statusError.body().toString()).doesNotContain(credential) + assertThat(statusError.toString()).doesNotContain(credential) + assertThat(response.closed).isTrue() + } } - } } @Test @@ -375,6 +443,32 @@ internal class X509TokenExchangeTest { } } + @Test + fun ignoresNestedOAuthDiagnosticsBeforeReadingLargeNestedStrings() { + mapOf( + "error" to + """{"error":{"error":"$OVERSIZED_RECOGNIZED_VALUE"},"error_description":"safe"}""", + "error_description" to + """{"error":"invalid_grant","error_description":{"error_description":"$OVERSIZED_RECOGNIZED_VALUE"}}""", + ) + .forEach { (field, body) -> + listOf(false, true).forEach { async -> + val response = TestResponse(503, body) + + val failure = exchangeFailure(response, async) + + assertThat(failure) + .describedAs("$field (async=$async)") + .isInstanceOf(UnexpectedStatusCodeException::class.java) + val statusError = failure as UnexpectedStatusCodeException + assertThat(statusError.statusCode()).isEqualTo(503) + assertThat(statusError.code()).isEmpty() + assertThat(response.bytesRead.get()).isLessThan(body.toByteArray().size) + assertThat(response.closed).isTrue() + } + } + } + @Test fun preservesIssuerBodyTimeoutAndIoFailuresAsSanitizedRetryableIo() { listOf( @@ -420,6 +514,7 @@ internal class X509TokenExchangeTest { val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) val result = exchange.executeAsync() + assertThat(client.started.await(5, TimeUnit.SECONDS)).isTrue() assertThat(result.cancel(true)).isTrue() assertThat(responseFuture.isCancelled).isTrue() @@ -428,25 +523,185 @@ internal class X509TokenExchangeTest { assertThat(client.closed).isFalse() } + @Test + fun cancellationWaitsForInFlightRequestEnrollmentBeforePublishing() { + val responseFuture = CompletableFuture() + val client = BlockingStartClient(responseFuture) + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) + val result = exchange.executeAsync() + val cancellationPublished = CountDownLatch(1) + val cancellationFinished = CountDownLatch(1) + result.whenComplete { _, _ -> cancellationPublished.countDown() } + + assertThat(client.started.await(5, TimeUnit.SECONDS)).isTrue() + Thread { + result.cancel(true) + cancellationFinished.countDown() + } + .start() + + try { + assertThat(cancellationFinished.await(200, TimeUnit.MILLISECONDS)).isFalse() + assertThat(cancellationPublished.count).isEqualTo(1) + + client.release.countDown() + + assertThat(cancellationFinished.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(cancellationPublished.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(result.isCancelled).isTrue() + assertThat(responseFuture.isCancelled).isTrue() + } finally { + client.release.countDown() + exchange.close() + } + } + + @Test + fun cancellationWaitsForCompletedResponseHandoffBeforePublishing() { + val response = BlockingResponse() + val responseFuture = BlockingHandoffFuture(response) + val client = DeferredResponseClient(responseFuture) + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) + val result = exchange.executeAsync() + val cancellationPublished = CountDownLatch(1) + val cancellationFinished = CountDownLatch(1) + result.whenComplete { _, _ -> cancellationPublished.countDown() } + + assertThat(responseFuture.handoffStarted.await(5, TimeUnit.SECONDS)).isTrue() + Thread { + result.cancel(true) + cancellationFinished.countDown() + } + .start() + + try { + assertThat(cancellationFinished.await(200, TimeUnit.MILLISECONDS)).isFalse() + assertThat(cancellationPublished.count).isEqualTo(1) + + responseFuture.releaseHandoff.countDown() + + assertThat(cancellationFinished.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(cancellationPublished.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(response.closed.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(response.bodyStarted.count).isEqualTo(1) + assertThat(response.closeCount).hasValue(1) + } finally { + responseFuture.releaseHandoff.countDown() + exchange.close() + } + } + @Test fun cancelingClosesALateResponseWhenUnderlyingCancellationLosesTheRace() { val responseFuture = NonCancellableFuture() - val exchange = - X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, DeferredResponseClient(responseFuture)) + val client = DeferredResponseClient(responseFuture) + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) val result = exchange.executeAsync() - val response = TestResponse(200, validResponse()) + val response = BlockingResponse() + assertThat(client.started.await(5, TimeUnit.SECONDS)).isTrue() assertThat(result.cancel(true)).isTrue() assertThat(responseFuture.complete(response)).isTrue() assertThat(result.isCancelled).isTrue() - assertThat(response.bodyRead).isFalse() + assertThat(response.bodyStarted.count).isEqualTo(1) + assertThat(response.closed.await(5, TimeUnit.SECONDS)).isTrue() assertThat(response.closeCount).hasValue(1) exchange.close() } @Test - fun responseExecutorQueuesValidResponsesBeyondItsOwnedThreadCount() { + fun cancellationClosesActiveResponseBeforeRunningCallerCallbacks() { + val response = BlockingResponse() + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, SingleResponseClient(response)) + val result = exchange.executeAsync() + val callbackStarted = CountDownLatch(1) + val releaseCallback = CountDownLatch(1) + val cancellationFinished = CountDownLatch(1) + + assertThat(response.bodyStarted.await(5, TimeUnit.SECONDS)).isTrue() + result.whenComplete { _, _ -> + callbackStarted.countDown() + releaseCallback.await() + } + Thread { + result.cancel(true) + cancellationFinished.countDown() + } + .start() + + try { + assertThat(callbackStarted.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(response.closed.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(response.closeCount).hasValue(1) + } finally { + releaseCallback.countDown() + assertThat(cancellationFinished.await(5, TimeUnit.SECONDS)).isTrue() + exchange.close() + } + } + + @Test + fun closeClosesALateResponseWhenUnderlyingCancellationLosesTheRace() { + val responseFuture = NonCancellableFuture() + val client = DeferredResponseClient(responseFuture) + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) + val result = exchange.executeAsync() + val response = BlockingResponse() + + assertThat(client.started.await(5, TimeUnit.SECONDS)).isTrue() + exchange.close() + assertThat(responseFuture.complete(response)).isTrue() + + assertThat(result.isCancelled).isTrue() + assertThat(response.bodyStarted.count).isEqualTo(1) + assertThat(response.closed.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(response.closeCount).hasValue(1) + } + + @Test + fun closeCleansAllAdmittedOperationsBeforeRunningCallerCallbacks() { + val responses = List(4) { BlockingResponse() } + val client = SequenceResponseClient(responses) + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) + val activeResults = responses.map { exchange.executeAsync() } + val queuedResult = exchange.executeAsync() + val results = activeResults + queuedResult + val callbackStarted = CountDownLatch(1) + val releaseCallbacks = CountDownLatch(1) + val closeFinished = CountDownLatch(1) + + responses.forEach { response -> + assertThat(response.bodyStarted.await(5, TimeUnit.SECONDS)).isTrue() + } + assertThat(client.executeCount).hasValue(4) + results.forEach { result -> + result.whenComplete { _, _ -> + callbackStarted.countDown() + releaseCallbacks.await() + } + } + Thread { + exchange.close() + closeFinished.countDown() + } + .start() + + try { + assertThat(callbackStarted.await(5, TimeUnit.SECONDS)).isTrue() + responses.forEach { response -> + assertThat(response.closed.await(5, TimeUnit.SECONDS)).isTrue() + assertThat(response.closeCount).hasValue(1) + } + assertThat(client.executeCount).hasValue(4) + } finally { + releaseCallbacks.countDown() + assertThat(closeFinished.await(5, TimeUnit.SECONDS)).isTrue() + } + } + + @Test + fun asyncExecutorQueuesWholeExchangesBeforeIssuingAdditionalRequests() { val release = CountDownLatch(1) val activeResponses = List(4) { GatedResponse(validResponse(), release) } val queuedResponse = TestResponse(200, validResponse()) @@ -459,6 +714,7 @@ internal class X509TokenExchangeTest { activeResponses.forEach { response -> assertThat(response.bodyStarted.await(5, TimeUnit.SECONDS)).isTrue() } + assertThat(client.executeCount).hasValue(4) assertThat(queuedResponse.bodyRead).isFalse() assertThat(results.last().isDone).isFalse() @@ -475,6 +731,51 @@ internal class X509TokenExchangeTest { assertThat(client.closed).isFalse() } + @Test + fun asyncExecutorBoundsAndRemovesCanceledQueuedExchanges() { + val release = CountDownLatch(1) + val activeResponses = List(4) { GatedResponse(validResponse(), release) } + val admittedResponses = List(4) { TestResponse(200, validResponse()) } + val client = SequenceResponseClient(activeResponses + admittedResponses) + val exchange = X509TokenExchange(IDP_ID, SERVICE_ACCOUNT_ID, client) + try { + val activeResults = List(4) { exchange.executeAsync() } + val queuedResults = List(4) { exchange.executeAsync() } + + activeResponses.forEach { response -> + assertThat(response.bodyStarted.await(5, TimeUnit.SECONDS)).isTrue() + } + assertThat(client.executeCount).hasValue(4) + + val overflow = exchange.executeAsync() + val overflowFailure = + runCatching { overflow.get(5, TimeUnit.SECONDS) }.exceptionOrNull() + ?: error("Expected bounded X.509 executor rejection") + assertThat(overflowFailure).isInstanceOf(ExecutionException::class.java) + assertThat(overflowFailure.cause).isInstanceOf(OpenAIIoException::class.java) + assertThat(overflowFailure.cause) + .hasMessage("X.509 token exchange processing unavailable") + + assertThat(queuedResults.first().cancel(true)).isTrue() + val replacement = exchange.executeAsync() + assertThat(replacement.isDone).isFalse() + assertThat(client.executeCount).hasValue(4) + + release.countDown() + + val successful = activeResults + queuedResults.drop(1) + replacement + assertThat(successful.map { it.get(5, TimeUnit.SECONDS).value }) + .containsExactlyElementsOf(List(8) { ACCESS_TOKEN }) + assertThat(queuedResults.first().isCancelled).isTrue() + assertThat(client.executeCount).hasValue(8) + assertThat(activeResponses).allMatch { it.closed } + assertThat(admittedResponses).allMatch { it.closed } + } finally { + release.countDown() + exchange.close() + } + } + private fun exchangeFailure(response: HttpResponse, async: Boolean): Throwable { val result = runCatching { @@ -515,6 +816,10 @@ internal class X509TokenExchangeTest { private fun validResponseWithDuplicate(field: String, firstValue: String): String = validResponse().replaceFirst("\"$field\":", "\"$field\": $firstValue,\n \"$field\":") + private fun validResponseWithRawValue(field: String, value: String): String = + validResponse() + .replaceFirst(Regex("\"$field\"\\s*:\\s*(?:\"[^\"]*\"|[0-9]+)"), "\"$field\": $value") + private fun validResponse( accessToken: String = ACCESS_TOKEN, tokenType: String = "Bearer", @@ -582,6 +887,7 @@ private open class SingleResponseClient(private val response: HttpResponse) : Ht private class DeferredResponseClient(private val responseFuture: CompletableFuture) : HttpClient { + val started = CountDownLatch(1) var closed = false private set @@ -591,19 +897,55 @@ private class DeferredResponseClient(private val responseFuture: CompletableFutu override fun executeAsync( request: HttpRequest, requestOptions: RequestOptions, - ): CompletableFuture = responseFuture + ): CompletableFuture = responseFuture.also { started.countDown() } override fun close() { closed = true } } +private class BlockingStartClient(private val responseFuture: CompletableFuture) : + HttpClient { + val started = CountDownLatch(1) + val release = CountDownLatch(1) + + override fun execute(request: HttpRequest, requestOptions: RequestOptions): HttpResponse = + error("Unexpected synchronous call") + + override fun executeAsync( + request: HttpRequest, + requestOptions: RequestOptions, + ): CompletableFuture { + started.countDown() + release.await() + return responseFuture + } + + override fun close() {} +} + private class NonCancellableFuture : CompletableFuture() { override fun cancel(mayInterruptIfRunning: Boolean): Boolean = false } +private class BlockingHandoffFuture(value: T) : CompletableFuture() { + val handoffStarted = CountDownLatch(1) + val releaseHandoff = CountDownLatch(1) + + init { + complete(value) + } + + override fun whenComplete(action: BiConsumer): CompletableFuture { + handoffStarted.countDown() + releaseHandoff.await() + return super.whenComplete(action) + } +} + private class SequenceResponseClient(responses: List) : HttpClient { - private val responses = ArrayDeque(responses) + private val responses = ConcurrentLinkedQueue(responses) + val executeCount = AtomicInteger() var closed = false private set @@ -613,7 +955,12 @@ private class SequenceResponseClient(responses: List) : HttpClient override fun executeAsync( request: HttpRequest, requestOptions: RequestOptions, - ): CompletableFuture = CompletableFuture.completedFuture(responses.removeFirst()) + ): CompletableFuture { + executeCount.incrementAndGet() + return CompletableFuture.completedFuture( + responses.poll() ?: error("No X.509 test response remains") + ) + } override fun close() { closed = true