将 Oracle Identity Cloud Service (IDCS) 访问 Token 换取短期有效的 OpenAI 访问 Token,即可将 Oracle Cloud Infrastructure (OCI) 用作工作负载身份提供方。OCI 实例主体对发送到同一租户内身份域的 Token 交换请求进行签名。OpenAI 验证返回的 Token,并授权 OCI 工作负载以映射的 OpenAI 服务账户身份执行操作。
如果您使用 Codex,请按照本页说明获取并检查 Oracle Token。然后配置 Codex 工作负载身份,将该 Token 写入文件,并让 Codex 使用此文件。本页的服务账户映射和 SDK 示例适用于 OpenAI API。
此配置不需要 OpenAI API 密钥、自定义 Oracle OAuth 资源应用,也不需要向自定义应用授予动态组权限。
在具有实例主体的 OCI Compute 实例上运行您的工作负载。对于 Oracle Kubernetes Engine (OKE),请确认签署请求所使用的身份:标准实例主体签名器通常标识的是工作节点,而非单个 Kubernetes Pod。
签名器从 OCI 实例元数据服务获取凭据。请验证工作负载能否访问链路本地元数据端点:
123curl --fail --silent \
--header "Authorization: Bearer Oracle" \
http://169.254.169.254/opc/v2/instance/id
工作负载还必须能够向其租户内的身份域发出出站 HTTPS 请求。元数据端点本身不需要 NAT 网关或互联网连接。
使用 OCI Python SDK 中的 InstancePrincipalsSecurityTokenSigner,对发送到您身份域的 OAuth Token 交换请求进行签名:
POST https://<identity-domain>/oauth2/v1/token
Content-Type: application/x-www-form-urlencoded;charset=utf-8
grant_type=urn:ietf:params:oauth:grant-type:token-exchange
scope=urn:opc:idm:__myscopes__
requested_token_type=urn:ietf:params:oauth:token-type:access_token
urn:opc:idm:__myscopes__ 作用域使用实例主体已有的授权。将返回的 IDCS 访问 Token 用作 OpenAI 工作负载身份联合的主体 Token。不要将 Oracle Token 的受众替换为 https://api.openai.com/v1;请使用实际 Oracle Token 中的受众来配置 OpenAI 身份提供方。
将 TOKEN 设置为实际 OCI 工作负载生成的访问 Token,然后使用现有的本地 JWT 解码器检查其声明:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18const parts = process.env.TOKEN?.split(".") ?? [];
if (parts.length !== 3) {
throw new Error("Expected a compact JWT with three segments");
}
if (!/^[A-Za-z0-9_-]+$/.test(parts[1]) || parts[1].length % 4 === 1) {
throw new Error("JWT payload is not valid Base64URL");
}
const bytes = Buffer.from(parts[1], "base64url");
if (bytes.toString("base64url") !== parts[1]) {
throw new Error("JWT payload is not valid Base64URL");
}
const decoded = new TextDecoder("utf-8", { fatal: true }).decode(bytes);
const claims = JSON.parse(decoded);
if (claims === null || Array.isArray(claims) || typeof claims !== "object") {
throw new Error("JWT payload is not a JSON object");
}
console.log(decoded);
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 base64
import json
import os
import re
def reject_non_json_constant(value):
raise ValueError(f"JWT payload contains non-JSON constant: {value}")
parts = os.environ.get("TOKEN", "").split(".")
if len(parts) != 3:
raise ValueError("Expected a compact JWT with three segments")
payload = parts[1]
if re.fullmatch(r"[A-Za-z0-9_-]+", payload) is None or len(payload) % 4 == 1:
raise ValueError("JWT payload is not valid Base64URL")
padded_payload = payload + "=" * (-len(payload) % 4)
decoded = base64.b64decode(padded_payload, altchars=b"-_", validate=True)
if base64.urlsafe_b64encode(decoded).rstrip(b"=").decode("ascii") != payload:
raise ValueError("JWT payload is not valid Base64URL")
decoded_text = decoded.decode("utf-8")
claims = json.loads(decoded_text, parse_constant=reject_non_json_constant)
if not isinstance(claims, dict):
raise ValueError("JWT payload is not a JSON object")
print(decoded_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
56
57
58
59
60
61
62
63
64
65
66
67
68
69package main
import (
"bytes"
"encoding/base64"
"encoding/json"
"fmt"
"os"
"strings"
"unicode/utf8"
)
func decodeSegment(segment string) (json.RawMessage, error) {
if !isBase64URLSegment(segment) {
return nil, fmt.Errorf("JWT segment is not valid Base64URL")
}
decoded, err := base64.RawURLEncoding.DecodeString(segment)
if err != nil {
return nil, err
}
if base64.RawURLEncoding.EncodeToString(decoded) != segment {
return nil, fmt.Errorf("JWT segment is not valid Base64URL")
}
if !utf8.Valid(decoded) {
return nil, fmt.Errorf("JWT segment is not valid UTF-8")
}
var value json.RawMessage
if err := json.Unmarshal(decoded, &value); err != nil {
return nil, err
}
if trimmed := bytes.TrimSpace(value); len(trimmed) == 0 || trimmed[0] != '{' {
return nil, fmt.Errorf("JWT segment is not a JSON object")
}
return value, nil
}
func isBase64URLSegment(segment string) bool {
if segment == "" || len(segment)%4 == 1 {
return false
}
for _, character := range segment {
if !('A' <= character && character <= 'Z') &&
!('a' <= character && character <= 'z') &&
!('0' <= character && character <= '9') &&
character != '-' &&
character != '_' {
return false
}
}
return true
}
func main() {
parts := strings.Split(os.Getenv("TOKEN"), ".")
if len(parts) != 3 {
panic("Expected a compact JWT with three segments")
}
payload, err := decodeSegment(parts[1])
if err != nil {
panic(err)
}
formatted, err := json.MarshalIndent(payload, "", " ")
if err != nil {
panic(err)
}
fmt.Println(string(formatted))
}
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
77// Add Jackson (com.fasterxml.jackson.core:jackson-databind) to your project.
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.CodingErrorAction;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
public final class DecodeJwtPayloadExample {
private static final ObjectMapper JSON =
new ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
private DecodeJwtPayloadExample() {}
static String decodeUtf8(byte[] bytes) throws IOException {
try {
return StandardCharsets.UTF_8
.newDecoder()
.onMalformedInput(CodingErrorAction.REPORT)
.onUnmappableCharacter(CodingErrorAction.REPORT)
.decode(ByteBuffer.wrap(bytes))
.toString();
} catch (CharacterCodingException exception) {
throw new IOException("JWT segment is not valid UTF-8", exception);
}
}
static String decodeSegment(String segment) throws IOException {
if (!isBase64UrlSegment(segment)) {
throw new IllegalArgumentException("JWT segment is not valid Base64URL");
}
byte[] bytes = Base64.getUrlDecoder().decode(segment);
if (!Base64.getUrlEncoder().withoutPadding().encodeToString(bytes).equals(segment)) {
throw new IllegalArgumentException("JWT segment is not valid Base64URL");
}
String decoded = decodeUtf8(bytes);
JsonNode value = JSON.readTree(decoded);
if (value == null || value.isMissingNode() || !value.isObject()) {
throw new IOException("JWT segment is not a JSON object");
}
return decoded;
}
static boolean isBase64UrlSegment(String segment) {
if (segment.isEmpty() || segment.length() % 4 == 1) {
return false;
}
return segment
.chars()
.allMatch(
character ->
character >= 'A' && character <= 'Z'
|| character >= 'a' && character <= 'z'
|| character >= '0' && character <= '9'
|| character == '-'
|| character == '_');
}
static String[] requireCompactJwt(String token) {
if (token == null) {
throw new IllegalArgumentException("Expected a compact JWT with three segments");
}
String[] parts = token.split("\\.", -1);
if (parts.length != 3) {
throw new IllegalArgumentException("Expected a compact JWT with three segments");
}
return parts;
}
public static void main(String[] args) throws IOException {
String[] parts = requireCompactJwt(System.getenv("TOKEN"));
System.out.println(decodeSegment(parts[1]));
}
}
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
59using System.Text;
using System.Text.Json;
static string DecodeSegment(string segment)
{
if (
segment.Length % 4 == 1 ||
segment.Any(
character =>
!(
character is >= 'A' and <= 'Z' ||
character is >= 'a' and <= 'z' ||
character is >= '0' and <= '9' ||
character is '-' or '_'
)
)
)
{
throw new FormatException("JWT segment is not valid Base64URL");
}
byte[] decoded = Convert.FromBase64String(
segment.Replace('-', '+').Replace('_', '/') +
new string('=', (4 - segment.Length % 4) % 4)
);
string canonicalSegment = Convert
.ToBase64String(decoded)
.TrimEnd('=')
.Replace('+', '-')
.Replace('/', '_');
if (canonicalSegment != segment)
{
throw new FormatException("JWT segment is not valid Base64URL");
}
string decodedJson = new UTF8Encoding(false, true).GetString(decoded);
using JsonDocument document = JsonDocument.Parse(decodedJson);
if (document.RootElement.ValueKind is not JsonValueKind.Object)
{
throw new FormatException("JWT segment is not a JSON object");
}
return decodedJson;
}
string? token = Environment.GetEnvironmentVariable("TOKEN");
if (token is null)
{
throw new InvalidOperationException(
"Expected a compact JWT with three segments"
);
}
string[] parts = token.Split('.');
if (parts.Length != 3)
{
throw new InvalidOperationException(
"Expected a compact JWT with three segments"
);
}
Console.WriteLine(DecodeSegment(parts[1]));
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
26require "base64"
require "json"
parts = ENV.fetch("TOKEN", "").split(".", -1)
raise "Expected a compact JWT with three segments" unless parts.length == 3
unless parts[1].match?(/\A[A-Za-z0-9_-]+\z/) && parts[1].length % 4 != 1
raise "JWT payload is not valid Base64URL"
end
begin
payload = Base64.urlsafe_decode64(parts[1].ljust((parts[1].length + 3) & ~3, "="))
rescue ArgumentError
raise "JWT payload is not valid Base64URL"
end
unless Base64.urlsafe_encode64(payload, padding: false) == parts[1]
raise "JWT payload is not valid Base64URL"
end
payload.force_encoding(Encoding::UTF_8)
raise "JWT payload is not valid UTF-8" unless payload.valid_encoding?
claims = JSON.parse(payload)
raise "JWT payload is not a JSON object" unless claims.is_a?(Hash)
puts(payload)
解码器会检查 Token,但不会验证其签名。请将原始 Token 视为敏感信息,不要将其写入日志,也不要将生产环境的 Token 粘贴到第三方 JWT 解码器中。
解码后的 Oracle 访问 Token 可能包含以下声明:
123456789101112131415{
"iss": "https://identity.oraclecloud.com/",
"aud": [
"https://idcs-example.us-phoenix-1.identity.oraclecloud.com",
"https://idcs-example.identity.oraclecloud.com"
],
"sub_type": "instance",
"ipst_instance": "ocid1.instance.oc1.phx.<instance-id>",
"ipst_compartment": "ocid1.compartment.oc1..<compartment-id>",
"domain_id": "ocid1.domain.oc1..<domain-id>",
"ca_ocid": "ocid1.tenancy.oc1..<tenancy-id>",
"tenant": "idcs-example",
"exp": 1782369434,
"iat": 1782365834
}
请以您自己的身份域签发的 Token 为准。配置时,准确填写 iss 值以及该 Token 的其中一个 aud 值。在授权工作负载时,优先使用不可变的 ipst_instance、ipst_compartment、domain_id 和 ca_ocid 声明。
为您的 Oracle 身份域创建工作负载身份提供方,然后为可以使用目标 OpenAI 服务账户的 OCI 实例或区间添加映射。
-
创建工作负载身份提供方。 将 名称 设置为唯一值,例如 oracle-cloud-prod。使用 描述来标识受信任的工作负载,例如 Production OCI instance principal。
-
设置签发者和受众。 将 OIDC 签发者 URL 设置为 Token 的 iss 声明,例如 https://identity.oraclecloud.com/。将 受众 设置为同一 Token 中的其中一个 aud 值。
-
如果支持,请配置租户专属的 OIDC 发现。 如果 高级下显示 使用自定义 URL 进行 OIDC 发现 ,请将其启用。将 自定义 OIDC 发现 URL 设置为您的租户专属身份域,例如 https://idcs-example.identity.oraclecloud.com。OpenAI 会获取 https://idcs-example.identity.oraclecloud.com/.well-known/openid-configuration,然后使用发现文档中的 jwks_uri 获取租户的签名公钥。如果未显示自定义发现选项,请改为启用 使用上传的 JWKS 验证 Token ,并上传从 https://<identity-domain>/admin/v1/SigningCert/jwk 获取的公钥 JWKS。
-
仅在需要派生属性时添加属性转换。 您可以在服务账户映射断言中直接使用原始 Oracle 声明,例如 ipst_instance、ipst_compartment、domain_id 和 ca_ocid。若要显式派生一个实例属性,请输入 instance 并使用表达式 assertion.ipst_instance,以创建 openai.instance。
Oracle 的 OpenID Connect 发现参考资料说明了自定义发现为何重要:发现文档可以声明全局签发者 https://identity.oraclecloud.com/,同时在租户专属身份域上发布 Token 端点和 jwks_uri。请在 OIDC 签发者 URL 中保留全局签发者,并将租户域用于 自定义 OIDC 发现 URL。
如果您的身份域在 Token 签发者地址发布发现元数据,
请保持自定义发现处于禁用状态,并使用标准 OIDC 发现。如果 OpenAI
无法访问租户的发现文档或签名密钥端点,请禁用
自定义发现,启用 使用上传的 JWKS 验证 Token,并
上传租户的公钥 JWKS,获取地址为
https://<identity-domain>/admin/v1/SigningCert/jwk。自定义发现与
上传的 JWKS 不能同时启用。请在
Oracle 轮换签名证书时更新上传的密钥。
-
创建服务账户映射。 将 名称 设置为唯一值,例如 oracle-instance-prod,并添加用于标识受信任 OCI 工作负载的描述。
-
匹配范围最小的稳定 OCI 身份。 若要向单个实例授予访问权限,请将 键 设置为 ipst_instance,并将 值 准确设置为已验证 Token 中的实例 OCID。若要向同一区间内的实例授予访问权限,请将 键 设置为 ipst_compartment,并将 值 准确设置为该区间的 OCID。
-
根据需要添加身份域和租户边界限制。 为 domain_id 或 ca_ocid 添加更多映射行,将工作负载限制在特定的 Oracle 身份域或租户内。如果 Token 包含 sub_type 声明,且您希望要求使用实例主体,请添加该声明并将值设置为 instance。所有映射行都必须匹配。
-
选择 OpenAI 目标。 将 项目 设置为服务账户所属的项目,然后选择受信任的 OCI 工作负载可以使用的 服务账户 。
-
根据需要缩小 API 权限范围。 仅选择工作负载所需的 权限 。映射权限可以限制所选服务账户的权限,但不能授予该服务账户尚未拥有的权限。
使用标准实例主体签名器的 OKE 工作负载会继承
工作节点的身份。实例级映射授权的是该节点,而不只是
某个 Pod。如果您需要隔离共享同一工作节点的 Pod,
请使用受支持且范围更精细的 OCI 工作负载身份。
安装 OpenAI、OCI 和 Requests Python 包:
pip install openai oci requests
对于 Ruby,请安装 OpenAI 和 OCI gem:
gem install openai oci
将 OCI_IDENTITY_DOMAIN_URL 设置为与工作负载位于同一租户内的身份域的基础 URL。将 OPENAI_IDENTITY_PROVIDER_ID 和 OPENAI_SERVICE_ACCOUNT_ID 设置为您在 OpenAI 身份提供方和服务账户映射中获得的相应 ID。
以下示例使用 OCI 实例主体对 Oracle Token 交换请求进行签名,将 IDCS 访问 Token 返回给 OpenAI SDK,并让 SDK 在需要时将其换取短期有效的 OpenAI 访问 Token:
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
53import os
import oci
import requests
from openai import OpenAI
from openai.auth import SubjectTokenProvider
def oracle_instance_principal_token_provider(
identity_domain_url: str,
) -> SubjectTokenProvider:
def get_token() -> str:
signer = oci.auth.signers.InstancePrincipalsSecurityTokenSigner()
response = requests.post(
f"{identity_domain_url.rstrip('/')}/oauth2/v1/token",
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"scope": "urn:opc:idm:__myscopes__",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
},
headers={
"Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
},
auth=signer,
timeout=30,
)
response.raise_for_status()
token = response.json().get("access_token")
if not isinstance(token, str) or not token:
raise RuntimeError("Oracle IDCS did not return an access token.")
return token
return {"token_type": "jwt", "get_token": get_token}
client = OpenAI(
workload_identity={
"identity_provider_id": os.environ["OPENAI_IDENTITY_PROVIDER_ID"],
"service_account_id": os.environ["OPENAI_SERVICE_ACCOUNT_ID"],
"provider": oracle_instance_principal_token_provider(
os.environ["OCI_IDENTITY_DOMAIN_URL"]
),
},
)
response = client.responses.create(
model="gpt-5.6-terra",
input="Say hello from Oracle Cloud Infrastructure 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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105require "json"
require "net/http"
require "oci"
require "openai"
require "uri"
class OracleInstancePrincipalTokenProvider
include OpenAI::Auth::SubjectTokenProvider
def initialize(identity_domain_url:)
@identity_domain_url = identity_domain_url.sub(%r{/+\z}, "")
end
def token_type
OpenAI::Auth::TokenType::JWT
end
def get_token
uri = URI("#{@identity_domain_url}/oauth2/v1/token")
unless uri.is_a?(URI::HTTPS)
raise OpenAI::Errors::SubjectTokenProviderError.new(
message: "Oracle identity domain URL must use HTTPS",
provider: "oracle-instance-principal"
)
end
body = URI.encode_www_form(
grant_type: "urn:ietf:params:oauth:grant-type:token-exchange",
scope: "urn:opc:idm:__myscopes__",
requested_token_type: "urn:ietf:params:oauth:token-type:access_token"
)
headers = {
"content-type": "application/x-www-form-urlencoded;charset=utf-8"
}
signer = OCI::Auth::Signers::InstancePrincipalsSecurityTokenSigner.new
signer.sign(:post, uri.to_s, headers, body)
request = Net::HTTP::Post.new(uri)
headers.each { |name, value| request[name.to_s] = value }
request.body = body
response = Net::HTTP.start(
uri.hostname,
uri.port,
use_ssl: true,
open_timeout: 10,
read_timeout: 30
) do |http|
http.request(request)
end
unless response.is_a?(Net::HTTPSuccess)
raise OpenAI::Errors::SubjectTokenProviderError.new(
message: "Oracle identity token request failed with status #{response.code}",
provider: "oracle-instance-principal"
)
end
token = JSON.parse(response.body).fetch("access_token")
unless token.is_a?(String) && !token.empty?
raise OpenAI::Errors::SubjectTokenProviderError.new(
message: "Oracle identity domain did not return an access token",
provider: "oracle-instance-principal"
)
end
token
rescue JSON::ParserError
raise OpenAI::Errors::SubjectTokenProviderError.new(
message: "Oracle identity token response was not valid JSON",
provider: "oracle-instance-principal"
), cause: nil
rescue KeyError
raise OpenAI::Errors::SubjectTokenProviderError.new(
message: "Oracle identity domain did not return an access token",
provider: "oracle-instance-principal"
), cause: nil
rescue SystemCallError, Timeout::Error => error
raise OpenAI::Errors::SubjectTokenProviderError.new(
message: "Failed to request Oracle identity token: #{error.message}",
provider: "oracle-instance-principal",
cause: error
)
end
end
provider = OracleInstancePrincipalTokenProvider.new(
identity_domain_url: ENV.fetch("OCI_IDENTITY_DOMAIN_URL")
)
workload_identity = OpenAI::Auth::WorkloadIdentity.new(
identity_provider_id: ENV.fetch("OPENAI_IDENTITY_PROVIDER_ID"),
service_account_id: ENV.fetch("OPENAI_SERVICE_ACCOUNT_ID"),
provider: provider
)
client = OpenAI::Client.new(workload_identity: workload_identity)
response = client.responses.create(
model: "gpt-5.6-terra",
input: "Say hello from Oracle Cloud Infrastructure workload identity federation."
)
puts(response.output_text)
当 OpenAI SDK 需要续期工作负载身份凭据时,主体 Token 提供方会请求新的 Oracle Token。切勿打印或持久化存储 Oracle 主体 Token 或换取的 OpenAI 访问 Token。
- 如果只有一个工作负载应具有访问权限,请使用
ipst_instance 映射单个实例。
- 仅当区间内所有符合条件的实例都应共享此映射时,才使用
ipst_compartment。
- 添加
domain_id 或 ca_ocid,以强制实施身份域和租户边界限制。
- 为每个应用和环境使用单独的 OpenAI 服务账户。
- 在依赖 Pod 级隔离之前,请验证 OKE Token 是否代表工作节点。
- 使用已签发 Oracle Token 中的受众,不要假定它是 OpenAI 专属的受众。
- 如果您的身份域无法使用 OIDC 发现,请在 Oracle 轮换签名密钥时轮换上传的公钥。