feat(auth): add X.509 workload token exchange

.github/workflows/x509-live-smoke.yml

@@ -0,0 +1,69 @@
+name: X.509 live smoke
+
+on:
+  workflow_dispatch:
+    inputs:
+      run_x509:
+        description: Confirm the protected production X.509 smoke test
+        required: true
+        default: false
+        type: boolean
+
+permissions: {}
+
+concurrency:
+  group: x509-live-smoke
+  cancel-in-progress: false
+
+jobs:
+  smoke:
+    if: >-
+      github.repository == 'openai/openai-java' &&
+      github.ref == 'refs/heads/main' &&
+      inputs.run_x509
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    # Configure this environment with an independent SDK-team required reviewer and no bypass.
+    environment: x509-live-smoke
+    permissions:
+      contents: read
+
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v6
+        with:
+          persist-credentials: false
+
+      - name: Set up Java
+        uses: actions/setup-java@b6effb05e454b25005698d916606bdc6ffcbf961 # v5
+        with:
+          distribution: temurin
+          java-version: 21
+
+      - name: Run certificate-only production smoke
+        shell: bash
+        env:
+          OPENAI_X509_KEYSTORE_P12_BASE64: ${{ secrets.OPENAI_X509_KEYSTORE_P12_BASE64 }}
+          OPENAI_X509_KEYSTORE_PASSWORD: ${{ secrets.OPENAI_X509_KEYSTORE_PASSWORD }}
+          OPENAI_X509_CERTIFICATE_ALIAS: ${{ secrets.OPENAI_X509_CERTIFICATE_ALIAS }}
+          OPENAI_X509_IDENTITY_PROVIDER_ID: ${{ secrets.OPENAI_X509_IDENTITY_PROVIDER_ID }}
+          OPENAI_X509_SERVICE_ACCOUNT_ID: ${{ secrets.OPENAI_X509_SERVICE_ACCOUNT_ID }}
+        run: |
+          set -euo pipefail
+          umask 077
+          for name in \
+            OPENAI_X509_KEYSTORE_P12_BASE64 \
+            OPENAI_X509_KEYSTORE_PASSWORD \
+            OPENAI_X509_CERTIFICATE_ALIAS \
+            OPENAI_X509_IDENTITY_PROVIDER_ID \
+            OPENAI_X509_SERVICE_ACCOUNT_ID; do
+            if [[ -z "${!name:-}" ]]; then
+              echo "Missing required X.509 smoke-test configuration: $name" >&2
+              exit 1
+            fi
+          done
+          x509_tmp_dir="$(mktemp -d "$RUNNER_TEMP/openai-x509-live.XXXXXX")"
+          trap 'rm -rf -- "$x509_tmp_dir"' EXIT
+          export OPENAI_X509_KEYSTORE_PATH="$x509_tmp_dir/identity.p12"
+          printf '%s' "$OPENAI_X509_KEYSTORE_P12_BASE64" | base64 --decode > "$OPENAI_X509_KEYSTORE_PATH"
+          chmod 600 "$OPENAI_X509_KEYSTORE_PATH"
+          ./gradlew :openai-java-example:run -Pexample=X509WorkloadIdentity --no-daemon

README.md

@@ -203,6 +203,45 @@ OpenAIClient client = OpenAIOkHttpClient.builder()
     .build();
 ```
 
+#### X.509 workload identity federation (preview)
+
+X.509 workload identity exchanges a client certificate for a short-lived bearer token and then
+presents the same certificate to the OpenAI mTLS API. It does not use an API key:
+
+```java
+X509WorkloadIdentity identity = X509WorkloadIdentity.builder()
+    .identityProviderId("your-identity-provider-id")
+    .serviceAccountId("your-service-account-id")
+    .build();
+
+X509Transport transport = X509Transport.builder()
+    .keyManager(keyManager)
+    .certificateAlias("workload-certificate")
+    .trustManager(trustManager)
+    .build();
+
+OpenAIClient client = OpenAIOkHttpClient.x509Builder(identity, transport).build();
+```
+
+`keyManager` must contain the private key and certificate chain for the selected alias;
+`trustManager` verifies OpenAI's servers. See
+[`X509WorkloadIdentityExample`](openai-java-example/src/main/java/com/openai/example/X509WorkloadIdentityExample.java)
+for a complete PKCS#12 example.
+
+This mode deliberately fixes both network destinations: token exchange goes directly to
+`https://mtls.auth.openai.com/oauth/token`, and API requests go directly to
+`https://mtls.api.openai.com/v1`. It rejects API keys, admin keys, `fromEnv()`, custom base URLs,
+organization/project headers, proxies, redirects, and custom transports. A client owns one
+generation-scoped token cache; concurrent requests share exchanges, and transient exchange and API
+failures share one retry budget and total deadline. Separately, one `401` can invalidate the exact
+rejected token and replay a repeatable request once.
+
+The bearer token is not cryptographically certificate-bound unless the service includes and
+enforces a confirmation (`cnf`) claim. Treat the token as a credential: never log it, and do not
+forward it outside the fixed mTLS client. To rotate a certificate, build a new key manager,
+transport, and client, atomically direct new work to that client, then drain and close the old
+client. Mutating a key store behind a live client is unsupported.
+
 #### Kubernetes service account token provider
 
 ```java

buildSrc/src/main/kotlin/com/openai/gradle/CoreCompilationShards.kt

@@ -109,15 +109,19 @@ object CoreCompilationShards {
 
     private val clientBaseSources =
         setOf(
+            "com/openai/core/CancellationPropagatingFuture.kt",
             "com/openai/core/ClientOptions.kt",
+            "com/openai/core/ClientOptionsView.kt",
             "com/openai/core/PrepareRequest.kt",
             "com/openai/core/Properties.kt",
             "com/openai/core/RequestOptions.kt",
             "com/openai/core/http/AuthenticatingHttpClient.kt",
             "com/openai/core/http/HttpClient.kt",
             "com/openai/core/http/LoggingHttpClient.kt",
+            "com/openai/core/http/PipelineRequestBody.kt",
             "com/openai/core/http/PhantomReachableClosingHttpClient.kt",
             "com/openai/core/http/RetryingHttpClient.kt",
+            "com/openai/core/http/RetryingHttpClientOrchestrator.kt",
             "com/openai/core/http/WorkloadIdentityHttpClient.kt",
         )
 

openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/OkHttpClient.kt

@@ -17,8 +17,10 @@ import java.net.Proxy
 import java.time.Duration
 import java.util.concurrent.CancellationException
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 import javax.net.ssl.HostnameVerifier
 import javax.net.ssl.SSLSocketFactory
 import javax.net.ssl.X509TrustManager
@@ -27,6 +29,7 @@ import okhttp3.Call
 import okhttp3.Callback
 import okhttp3.ConnectionPool
 import okhttp3.Dispatcher
+import okhttp3.EventListener
 import okhttp3.HttpUrl
 import okhttp3.HttpUrl.Companion.toHttpUrl
 import okhttp3.MediaType
@@ -40,17 +43,30 @@ import okio.buffer
 import okio.sink
 
 class OkHttpClient
-internal constructor(@JvmSynthetic internal val okHttpClient: okhttp3.OkHttpClient) : HttpClient {
+private constructor(
+    @get:JvmSynthetic internal val okHttpClient: okhttp3.OkHttpClient,
+    private val callTracker: CallTracker,
+) : HttpClient {
 
     override fun execute(request: HttpRequest, requestOptions: RequestOptions): HttpResponse {
         val call = newCall(request, requestOptions)
-
+        var failure: Throwable? = null
         return try {
             call.execute().toHttpResponse()
         } catch (e: IOException) {
-            throw OpenAIIoException("Request failed", e)
+            val ioFailure = OpenAIIoException("Request failed", e)
+            failure = ioFailure
+            throw ioFailure
+        } catch (t: Throwable) {
+            failure = t
+            throw t
         } finally {
-            request.body?.close()
+            val primaryFailure = failure
+            if (primaryFailure == null) {
+                request.body?.close()
+            } else {
+                request.body.closeSuppressing(primaryFailure)
+            }
         }
     }
 
@@ -64,7 +80,7 @@ internal constructor(@JvmSynthetic internal val okHttpClient: okhttp3.OkHttpClie
         call.enqueue(
             object : Callback {
                 override fun onResponse(call: Call, response: Response) {
-                    future.complete(response.toHttpResponse())
+                    completeOrCloseResponse(future, response.toHttpResponse())
                 }
 
                 override fun onFailure(call: Call, e: IOException) {
@@ -73,7 +89,7 @@ internal constructor(@JvmSynthetic internal val okHttpClient: okhttp3.OkHttpClie
             }
         )
 
-        future.whenComplete { _, e ->
+        future.whenComplete { response, e ->
             if (e is CancellationException) {
                 call.cancel()
             }
@@ -84,24 +100,32 @@ internal constructor(@JvmSynthetic internal val okHttpClient: okhttp3.OkHttpClie
     }
 
     override fun close() {
-        okHttpClient.dispatcher.executorService.shutdown()
-        okHttpClient.connectionPool.evictAll()
-        okHttpClient.cache?.close()
+        if (callTracker.close()) {
+            okHttpClient.dispatcher.executorService.shutdown()
+            okHttpClient.connectionPool.evictAll()
+            okHttpClient.cache?.close()
+        }
     }
 
     private fun newCall(request: HttpRequest, requestOptions: RequestOptions): Call {
-        val clientBuilder = okHttpClient.newBuilder()
-
-        requestOptions.timeout?.let {
-            clientBuilder
-                .connectTimeout(it.connect())
-                .readTimeout(it.read())
-                .writeTimeout(it.write())
-                .callTimeout(it.request())
-        }
+        return try {
+            callTracker.ensureOpen()
+            val clientBuilder = okHttpClient.newBuilder()
+
+            requestOptions.timeout?.let {
+                clientBuilder
+                    .connectTimeout(it.connect())
+                    .readTimeout(it.read())
+                    .writeTimeout(it.write())
+                    .callTimeout(it.request())
+            }
 
-        val client = clientBuilder.build()
-        return client.newCall(request.toRequest(client))
+            val client = clientBuilder.build()
+            client.newCall(request.toRequest(client))
+        } catch (failure: Throwable) {
+            request.body.closeSuppressing(failure)
+            throw failure
+        }
     }
 
     companion object {
@@ -174,11 +198,13 @@ internal constructor(@JvmSynthetic internal val okHttpClient: okhttp3.OkHttpClie
             this.hostnameVerifier = hostnameVerifier
         }
 
-        fun build(): OkHttpClient =
-            OkHttpClient(
+        fun build(): OkHttpClient {
+            val callTracker = CallTracker()
+            return OkHttpClient(
                 okhttp3.OkHttpClient.Builder()
                     // `RetryingHttpClient` handles retries if the user enabled them.
                     .retryOnConnectionFailure(false)
+                    .eventListenerFactory(callTracker.eventListenerFactory)
                     .followRedirects(followRedirects)
                     .followSslRedirects(followRedirects)
                     .connectTimeout(timeout.connect())
@@ -235,8 +261,57 @@ internal constructor(@JvmSynthetic internal val okHttpClient: okhttp3.OkHttpClie
                         // We usually make all our requests to the same host so it makes sense to
                         // raise the per-host limit to the overall limit.
                         dispatcher.maxRequestsPerHost = dispatcher.maxRequests
-                    }
+                    },
+                callTracker,
             )
+        }
+    }
+}
+
+private fun HttpRequestBody?.closeSuppressing(failure: Throwable) {
+    try {
+        this?.close()
+    } catch (closeFailure: Throwable) {
+        if (closeFailure !== failure) {
+            failure.addSuppressed(closeFailure)
+        }
+    }
+}
+
+private class CallTracker {
+    private val closed = AtomicBoolean()
+    private val activeCalls = ConcurrentHashMap.newKeySet<Call>()
+
+    val eventListenerFactory =
+        EventListener.Factory {
+            object : EventListener() {
+                override fun callStart(call: Call) {
+                    activeCalls.add(call)
+                    if (closed.get()) {
+                        call.cancel()
+                    }
+                }
+
+                override fun callEnd(call: Call) {
+                    activeCalls.remove(call)
+                }
+
+                override fun callFailed(call: Call, ioe: IOException) {
+                    activeCalls.remove(call)
+                }
+            }
+        }
+
+    fun ensureOpen() {
+        check(!closed.get()) { "HTTP client is closed" }
+    }
+
+    fun close(): Boolean {
+        if (!closed.compareAndSet(false, true)) {
+            return false
+        }
+        activeCalls.forEach(Call::cancel)
+        return true
     }
 }
 
@@ -357,6 +432,15 @@ private fun Response.toHttpResponse(): HttpResponse {
     }
 }
 
+internal fun completeOrCloseResponse(
+    future: CompletableFuture<HttpResponse>,
+    response: HttpResponse,
+) {
+    if (!future.complete(response)) {
+        response.close()
+    }
+}
+
 private fun okhttp3.Headers.toHeaders(): Headers {
     val headersBuilder = Headers.builder()
     forEach { (name, value) -> headersBuilder.put(name, value) }