使用 X.509 憑證設定工作負載身分聯合

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26import os
import ssl

from openai import DefaultHttpx2Client, OpenAI
from openai.auth import x509_workload_identity

tls_context = ssl.create_default_context()
tls_context.load_cert_chain(
    certfile=os.environ["OPENAI_MTLS_CERT_CHAIN"],
    keyfile=os.environ["OPENAI_MTLS_KEY"],
)

with OpenAI(
    base_url="https://mtls.api.openai.com/v1",
    workload_identity=x509_workload_identity(
        identity_provider_id=os.environ["OPENAI_IDENTITY_PROVIDER_ID"],
        service_account_id=os.environ["OPENAI_SERVICE_ACCOUNT_ID"],
    ),
    http_client=DefaultHttpx2Client(verify=tls_context, follow_redirects=False),
) as client:
    response = client.responses.create(
        model="gpt-5.6-terra",
        input="Say hello from X.509 workload identity federation.",
    )

    print(response.output_text)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56package main

import (
	"context"
	"crypto/tls"
	"fmt"
	"log"
	"net/http"
	"os"

	"github.com/openai/openai-go/v3"
	"github.com/openai/openai-go/v3/auth"
	"github.com/openai/openai-go/v3/option"
	"github.com/openai/openai-go/v3/responses"
)

func main() {
	certificate, err := tls.LoadX509KeyPair(
		os.Getenv("OPENAI_MTLS_CERT_CHAIN"),
		os.Getenv("OPENAI_MTLS_KEY"),
	)
	if err != nil {
		log.Fatal(err)
	}

	transport, err := auth.NewX509Transport(&http.Transport{
		TLSClientConfig: &tls.Config{
			Certificates: []tls.Certificate{certificate},
			MinVersion:   tls.VersionTLS12,
		},
	})
	if err != nil {
		log.Fatal(err)
	}
	defer transport.Close()

	client := openai.NewClient(
		option.WithX509WorkloadIdentity(auth.X509WorkloadIdentity{
			IdentityProviderID: os.Getenv("OPENAI_IDENTITY_PROVIDER_ID"),
			ServiceAccountID:   os.Getenv("OPENAI_SERVICE_ACCOUNT_ID"),
			Transport:          transport,
		}),
	)

	response, err := client.Responses.New(context.Background(), responses.ResponseNewParams{
		Model: "gpt-5.6-terra",
		Input: responses.ResponseNewParamsInputUnion{
			OfString: openai.String("Say hello from X.509 workload identity federation."),
		},
	})
	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(response.OutputText())
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77import com.openai.client.OpenAIClient;
import com.openai.client.okhttp.OpenAIOkHttpClient;
import com.openai.client.okhttp.X509Transport;
import com.openai.client.okhttp.X509WorkloadIdentity;
import com.openai.models.responses.ResponseCreateParams;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.KeyStore;
import java.time.Duration;
import java.util.Arrays;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.TrustManagerFactory;
import javax.net.ssl.X509ExtendedKeyManager;
import javax.net.ssl.X509TrustManager;

char[] password = System.getenv("OPENAI_X509_KEYSTORE_PASSWORD").toCharArray();
try {
  KeyStore keyStore = KeyStore.getInstance("PKCS12");
  try (InputStream input =
      Files.newInputStream(Paths.get(System.getenv("OPENAI_X509_KEYSTORE_PATH")))) {
    keyStore.load(input, password);
  }

  KeyManagerFactory keyManagers =
      KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
  keyManagers.init(keyStore, password);
  X509ExtendedKeyManager keyManager =
      Arrays.stream(keyManagers.getKeyManagers())
          .filter(X509ExtendedKeyManager.class::isInstance)
          .map(X509ExtendedKeyManager.class::cast)
          .findFirst()
          .orElseThrow(() -> new IllegalStateException("No X.509 key manager available"));

  TrustManagerFactory trustManagers =
      TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
  trustManagers.init((KeyStore) null);
  X509TrustManager trustManager =
      Arrays.stream(trustManagers.getTrustManagers())
          .filter(X509TrustManager.class::isInstance)
          .map(X509TrustManager.class::cast)
          .findFirst()
          .orElseThrow(() -> new IllegalStateException("No X.509 trust manager available"));

  X509Transport transport =
      X509Transport.builder()
          .keyManager(keyManager)
          .certificateAlias(System.getenv("OPENAI_X509_CERTIFICATE_ALIAS"))
          .trustManager(trustManager)
          .build();
  X509WorkloadIdentity identity =
      X509WorkloadIdentity.builder()
          .identityProviderId(System.getenv("OPENAI_IDENTITY_PROVIDER_ID"))
          .serviceAccountId(System.getenv("OPENAI_SERVICE_ACCOUNT_ID"))
          .transport(transport)
          .refreshBuffer(Duration.ofMinutes(10))
          .build();

  OpenAIClient client = OpenAIOkHttpClient.builder().x509WorkloadIdentity(identity).build();
  try {
    ResponseCreateParams params =
        ResponseCreateParams.builder()
            .model("gpt-5.6-terra")
            .input("Say hello from X.509 workload identity federation.")
            .build();

    client.responses().create(params).output().stream()
        .flatMap(item -> item.message().stream())
        .flatMap(message -> message.content().stream())
        .flatMap(content -> content.outputText().stream())
        .forEach(outputText -> System.out.println(outputText.text()));
  } finally {
    client.close();
  }
} finally {
  Arrays.fill(password, '\0');
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32require "openai"
require "openssl"

certificate_chain = OpenSSL::X509::Certificate.load(
  File.binread(ENV.fetch("OPENAI_MTLS_CERT_CHAIN"))
)
certificate, *intermediates = certificate_chain
private_key = OpenSSL::PKey.read(File.binread(ENV.fetch("OPENAI_MTLS_KEY")))

http_client = OpenAI::NetHTTPClient.new do |connection|
  connection.cert = certificate
  connection.extra_chain_cert = intermediates
  connection.key = private_key
end

workload_identity = OpenAI::Auth::X509WorkloadIdentity.new(
  identity_provider_id: ENV.fetch("OPENAI_IDENTITY_PROVIDER_ID"),
  service_account_id: ENV.fetch("OPENAI_SERVICE_ACCOUNT_ID"),
  http_client: http_client
)

begin
  client = OpenAI::Client.new(api_key: nil, workload_identity: workload_identity)
  response = client.responses.create(
    model: "gpt-5.6-terra",
    input: "Say hello from X.509 workload identity federation."
  )

  puts(response.output_text)
ensure
  http_client.close
end