Solo's Blog

切换主题

AgentIdentity密钥模型

从域分离 Ed25519 密钥、agent/task 注册和 AgentAssertion，到 JWT 验签、Curve25519 解密与登录重试，解释 Agent Identity 的完整密钥生命周期。

基于rust-v0.150.0
CodexRustSecurityIdentityCryptography
2859
14.8m

AgentIdentity密钥模型 ​

Agent Identity 由三种不同生命周期的数据组成：Ed25519 私钥与 agent_runtime_id 是可复用的持久身份；task_id 属于单次 Codex run；服务端 Agent Identity JWT 则承载账户与私钥材料，并必须在可信 JWKS 下验证后才能成为登录身份。注册签名、请求 assertion 和 task ID 解密都使用同一份持久私钥，但 payload、编码和算法目的不同。

本文承接Secrets检测与脱敏和网络审批与规则持久化。前文解释私钥记录如何落入凭据存储，本篇只追踪密钥生成、注册、签名、JWT 验证、task ID 解密和 login bootstrap；Attestation 使用另一套协议对象，留到后文。

1. 环境路由先受限 ​

Agent Identity 不接受任意 ChatGPT base URL。ChatGptEnvironment::from_chatgpt_base_url 只把已知 production/staging URL 映射到对应 auth API；未知 URL 直接报错。这一检查防止普通自定义 endpoint 无声接管 agent registration 或 JWKS 路由。

源码位置：codex-rs/agent-identity/src/lib.rs :: ChatGptEnvironment::from_chatgpt_base_url

rust
pub fn from_chatgpt_base_url(chatgpt_base_url: &str) -> Result<Self> {
    match chatgpt_base_url.trim_end_matches('/') {
        "https://chatgpt.com"
        | "https://chatgpt.com/backend-api"
        | "https://chatgpt.com/codex"
        | "https://chatgpt.com/backend-api/codex"
        | "https://chat.openai.com"
        | "https://chat.openai.com/backend-api"
        | "https://chat.openai.com/codex"
        | "https://chat.openai.com/backend-api/codex" => Ok(Self::Production),
        "https://chatgpt-staging.com"
        | "https://chatgpt-staging.com/backend-api"
        | "https://chatgpt-staging.com/codex"
        | "https://chatgpt-staging.com/backend-api/codex" => Ok(Self::Staging),
        _ => anyhow::bail!(
            "Agent Identity only supports production and staging ChatGPT environments"
        ),
    }
}

测试用本地 mock server 时可以显式传入 auth API URL，但生产 routing 的 login 层还会验证 override 与 ChatGPT environment 的关系。

2. 密钥生成有域分离 ​

generate_agent_key_material 从 OsRng 读取 64 字节，不直接截断为 Ed25519 seed。它先向 SHA-512 写入固定 context codex-agent-identity-ed25519-v1，再写入全部随机材料，取 digest 前 32 字节构造 SigningKey。固定 context 提供用途域分离，降低同一随机材料被其他协议复用时的混淆。

源码位置：codex-rs/agent-identity/src/lib.rs :: generate_agent_key_material

rust
pub fn generate_agent_key_material() -> Result<GeneratedAgentKeyMaterial> {
    let mut seed_material = [0u8; AGENT_IDENTITY_KEY_SEED_BYTES];
    OsRng
        .try_fill_bytes(&mut seed_material)
        .context("failed to generate agent identity private key seed material")?;
    // Ed25519 stores a 32-byte seed, so derive it from all sampled seed material.
    let mut digest = Sha512::new();
    digest.update(AGENT_IDENTITY_KEY_DERIVATION_CONTEXT);
    digest.update(seed_material);
    let digest = digest.finalize();
    let mut secret_key_bytes = [0u8; 32];
    secret_key_bytes.copy_from_slice(&digest[..32]);
    let signing_key = SigningKey::from_bytes(&secret_key_bytes);
    let private_key_pkcs8 = signing_key
        .to_pkcs8_der()
        .context("failed to encode agent identity private key as PKCS#8")?;

    Ok(GeneratedAgentKeyMaterial {
        private_key_pkcs8_base64: BASE64_STANDARD.encode(private_key_pkcs8.as_bytes()),
        public_key_ssh: encode_ssh_ed25519_public_key(&signing_key.verifying_key()),
    })
}

函数没有显式 zeroize seed_material、digest 或 secret_key_bytes；返回的 PKCS#8 base64 本身也是普通 String。私钥保护依赖上层凭据存储和进程边界，不能只看密码学算法。

3. 私钥与task分离 ​

AgentIdentityKey 只借用 durable agent_runtime_id 和 PKCS#8 private key，不包含 task ID。task ID 属于一次 run，由 task registration 产生；同一 durable identity 可以为不同运行注册不同 task。

源码位置：codex-rs/agent-identity/src/lib.rs :: AgentIdentityKey

rust
/// Borrowed durable signing material for a registered agent identity.
///
/// This intentionally does not include a task id. Task ids are scoped to a
/// single Codex run, while the agent runtime id and private key are the
/// reusable identity material used to register and sign that run task.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct AgentIdentityKey<'a> {
    pub agent_runtime_id: &'a str,
    pub private_key_pkcs8_base64: &'a str,
}

login 层的 AgentIdentityAuthRecord 同时保存 durable fields 与可选 task_id；构造 AgentIdentityAuth 时，缺失 task ID 会触发新注册，初始化完成后 run_task_id() 才保证可用。

源码位置：codex-rs/login/src/auth/agent_identity.rs :: AgentIdentityAuth::from_record

rust
pub async fn from_record(
    mut record: AgentIdentityAuthRecord,
    agent_identity_authapi_base_url: &str,
    auth_route_config: &AuthRouteConfig,
) -> std::io::Result<Self> {
    public_key_ssh_from_private_key_pkcs8_base64(&record.agent_private_key)
        .map_err(std::io::Error::other)?;
    if record_needs_task_registration(&record) {
        record.task_id = Some(
            register_task_for_record_with_retries(
                &record,
                agent_identity_authapi_base_url,
                auth_route_config,
            )
            .await?,
        );
    }
    Ok(Self {
        record: Arc::new(record),
    })
}

4. Agent注册 ​

首次 managed ChatGPT registration 生成 key material，发送 ABOM、公钥、capabilities 和可选 FedRAMP header。私钥不会出现在 RegisterAgentRequest；服务端返回 agent_runtime_id，login 层再把 runtime ID 与 private key 组成 record。

源码位置：codex-rs/agent-identity/src/lib.rs :: register_agent_identity

rust
let request = RegisterAgentRequest {
    abom,
    agent_public_key: key_material.public_key_ssh.clone(),
    capabilities,
    ttl: None,
};

let mut request_builder = client
    .post(&url)
    .bearer_auth(access_token)
    .json(&request)
    .timeout(AGENT_REGISTRATION_TIMEOUT);
if is_fedramp_account {
    request_builder = request_builder.header("X-OpenAI-Fedramp", "true");
}

源码位置：codex-rs/agent-identity/src/lib.rs :: build_abom

rust
pub fn build_abom(session_source: SessionSource) -> AgentBillOfMaterials {
    AgentBillOfMaterials {
        agent_version: env!("CARGO_PKG_VERSION").to_string(),
        agent_harness_id: match &session_source {
            SessionSource::VSCode => "codex-app".to_string(),
            SessionSource::Cli
            | SessionSource::Exec
            | SessionSource::Mcp
            | SessionSource::Custom(_)
            | SessionSource::Internal(_)
            | SessionSource::SubAgent(_)
            | SessionSource::Unknown => "codex-cli".to_string(),
        },
        running_location: format!("{}-{}", session_source, std::env::consts::OS),
    }
}

ABOM 描述版本、harness 和运行位置，不是远端 attestation 证明。

5. Task注册签名 ​

task registration payload 是 agent_runtime_id:timestamp。timestamp 使用 UTC RFC3339 秒精度，request body 只包含 timestamp 和 signature。服务端可根据已注册公钥验证请求来自 durable identity。

源码位置：codex-rs/agent-identity/src/lib.rs :: sign_task_registration_payload

rust
pub fn sign_task_registration_payload(
    key: AgentIdentityKey<'_>,
    timestamp: &str,
) -> Result<String> {
    let signing_key = signing_key_from_private_key_pkcs8_base64(key.private_key_pkcs8_base64)?;
    let payload = format!("{}:{timestamp}", key.agent_runtime_id);
    Ok(BASE64_STANDARD.encode(signing_key.sign(payload.as_bytes()).to_bytes()))
}

源码位置：codex-rs/agent-identity/src/lib.rs :: register_agent_task

rust
let timestamp = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
let request = RegisterTaskRequest {
    signature: sign_task_registration_payload(key, &timestamp)?,
    timestamp,
};
let url = agent_task_registration_url(agent_identity_authapi_base_url, key.agent_runtime_id);

let response = client
    .post(url)
    .timeout(AGENT_TASK_REGISTRATION_TIMEOUT)
    .json(&request)
    .send()
    .await
    .context("failed to register agent task")?;

返回体兼容 snake_case/camelCase，并可以直接返回 task ID，也可以返回 encrypted task ID。

6. AgentAssertion ​

访问 task API 时，客户端生成 AgentAssertion <token>。token 是 base64url-no-pad 编码的 JSON envelope，内部 signature 绑定 agent_runtime_id:task_id:timestamp。它不是三段式 JWT，也不使用 JWT header/kid。

源码位置：codex-rs/agent-identity/src/lib.rs :: authorization_header_for_agent_task

rust
pub fn authorization_header_for_agent_task(
    key: AgentIdentityKey<'_>,
    task_id: &str,
) -> Result<String> {
    let timestamp = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
    let envelope = AgentAssertionEnvelope {
        agent_runtime_id: key.agent_runtime_id.to_string(),
        task_id: task_id.to_string(),
        timestamp: timestamp.clone(),
        signature: sign_agent_assertion_payload(key, task_id, &timestamp)?,
    };
    let serialized_assertion = serialize_agent_assertion(&envelope)?;
    Ok(format!("AgentAssertion {serialized_assertion}"))
}

源码位置：codex-rs/agent-identity/src/lib.rs :: sign_agent_assertion_payload、serialize_agent_assertion

rust
fn sign_agent_assertion_payload(
    key: AgentIdentityKey<'_>,
    task_id: &str,
    timestamp: &str,
) -> Result<String> {
    let signing_key = signing_key_from_private_key_pkcs8_base64(key.private_key_pkcs8_base64)?;
    let payload = format!("{}:{task_id}:{timestamp}", key.agent_runtime_id);
    Ok(BASE64_STANDARD.encode(signing_key.sign(payload.as_bytes()).to_bytes()))
}

fn serialize_agent_assertion(envelope: &AgentAssertionEnvelope) -> Result<String> {
    let payload = serde_json::to_vec(&BTreeMap::from([
        ("agent_runtime_id", envelope.agent_runtime_id.as_str()),
        ("signature", envelope.signature.as_str()),
        ("task_id", envelope.task_id.as_str()),
        ("timestamp", envelope.timestamp.as_str()),
    ]))
    .context("failed to serialize agent assertion envelope")?;
    Ok(URL_SAFE_NO_PAD.encode(payload))
}

7. JWT有两种模式 ​

decode_agent_identity_jwt 的 jwks 参数决定安全语义。传入 JWKS 时，它读取 header kid、查找 trusted JWK、强制 RS256，并验证固定 audience、issuer 及所需 claims。传入 None 时只拆分 JWT、base64url decode payload 并反序列化，不验证 signature、issuer、audience 或 expiration。

源码位置：codex-rs/agent-identity/src/lib.rs :: decode_agent_identity_jwt

rust
pub fn decode_agent_identity_jwt(
    jwt: &str,
    jwks: Option<&JwkSet>,
) -> Result<AgentIdentityJwtClaims> {
    let Some(jwks) = jwks else {
        return decode_agent_identity_jwt_payload(jwt);
    };

    let header = decode_header(jwt).context("failed to decode agent identity JWT header")?;
    let kid = header
        .kid
        .context("agent identity JWT header does not include a kid")?;
    let jwk = jwks
        .find(&kid)
        .with_context(|| format!("agent identity JWT kid {kid} is not trusted"))?;
    let decoding_key = DecodingKey::from_jwk(jwk).context("failed to build JWT decoding key")?;
    let mut validation = Validation::new(Algorithm::RS256);
    validation.set_audience(&[AGENT_IDENTITY_JWT_AUDIENCE]);
    validation.set_issuer(&[AGENT_IDENTITY_JWT_ISSUER]);
    validation.required_spec_claims.insert("iss".to_string());
    validation.required_spec_claims.insert("aud".to_string());
    decode::<AgentIdentityJwtClaims>(jwt, &decoding_key, &validation)
        .map(|data| data.claims)
        .context("failed to verify agent identity JWT")
}

“可以解析”不等于“可信”。无 JWKS 模式只适合在后续一定会验签的流程中做初步 shape/account 检查。

8. Login层强制验签 ​

verified_record_from_jwt 先调用无 JWKS parser 验证结构可转 record，然后根据允许的 endpoint 获取 JWKS，再调用带 JWKS 的验证模式，最终只返回 verified claims。AgentIdentityAuth::from_jwt 随后注册 run task。

源码位置：codex-rs/login/src/auth/agent_identity.rs :: verified_record_from_jwt

rust
pub(super) async fn verified_record_from_jwt(
    jwt: &str,
    chatgpt_base_url: &str,
    auth_route_config: &AuthRouteConfig,
) -> std::io::Result<AgentIdentityAuthRecord> {
    AgentIdentityAuthRecord::from_agent_identity_jwt(jwt)?;
    let jwks_base_url =
        match agent_identity_endpoint_override(CODEX_AGENT_IDENTITY_JWKS_BASE_URL_ENV_VAR) {
            Some(base_url) => {
                if !agent_identity_jwks_base_url_matches(chatgpt_base_url, &base_url) {
                    ChatGptEnvironment::from_chatgpt_base_url(chatgpt_base_url)
                        .map_err(std::io::Error::other)?;
                }
                base_url
            }
            None => chatgpt_base_url.to_string(),
        };
    let jwks_url = agent_identity_jwks_url(&jwks_base_url);
    let client = create_default_auth_client(&jwks_url, auth_route_config)?;
    let jwks = fetch_agent_identity_jwks(&client, &jwks_base_url)
        .await
        .map_err(std::io::Error::other)?;
    let claims = decode_agent_identity_jwt(jwt, Some(&jwks)).map_err(std::io::Error::other)?;
    Ok(claims.into())
}

其他仅用于读取账户字段的代码若调用 AgentIdentityAuthRecord::from_agent_identity_jwt，必须把结果视为未验证 metadata，不能用于 authorization。

9. Task ID解密 ​

task registration response 优先接受明文 task ID；只有缺失时才读取 encrypted task ID。解密 key 从 Ed25519 SigningKey seed 做 SHA-512，再按 X25519 clamping 规则得到 Curve25519 secret，最后调用 sealed-box unseal。

源码位置：codex-rs/agent-identity/src/lib.rs :: task_id_from_register_task_response、decrypt_task_id_response

rust
if let Some(task_id) = response.task_id.or(response.task_id_camel) {
    return Ok(task_id);
}
let encrypted_task_id = response
    .encrypted_task_id
    .or(response.encrypted_task_id_camel)
    .context("agent task registration response omitted task id")?;
decrypt_task_id_response(key, &encrypted_task_id)

源码位置：codex-rs/agent-identity/src/lib.rs :: curve25519_secret_key_from_signing_key

rust
fn curve25519_secret_key_from_signing_key(signing_key: &SigningKey) -> Curve25519SecretKey {
    let digest = Sha512::digest(signing_key.to_bytes());
    let mut secret_key = [0u8; 32];
    secret_key.copy_from_slice(&digest[..32]);
    secret_key[0] &= 248;
    secret_key[31] &= 127;
    secret_key[31] |= 64;
    Curve25519SecretKey::from(secret_key)
}

这不是把 Ed25519 私钥字节直接重解释为 X25519 key，而是按转换规则重新派生。

10. 有限重试 ​

registration 只把 HTTP 429、5xx，以及 timeout/connect/request transport errors 判为 retryable。login 层最多重试固定次数；403、签名/格式错误等 hard failure 不进入重试。managed ChatGPT bootstrap 的 retry exhaustion 可以被分类为 bootstrap unavailable，JWT 路径的 task registration 则保持严格失败。

源码位置：codex-rs/agent-identity/src/lib.rs :: is_retryable_registration_error

rust
fn is_retryable_registration_status(status: StatusCode) -> bool {
    status == StatusCode::TOO_MANY_REQUESTS || status.is_server_error()
}

源码位置：codex-rs/login/src/auth/agent_identity.rs :: retry_registration

rust
pub(super) async fn retry_registration<T, F, Fut>(mut operation: F) -> std::io::Result<T>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = std::io::Result<T>>,
{
    let mut attempt = 1;
    loop {
        match operation().await {
            Ok(value) => return Ok(value),
            Err(err)
                if attempt < MAX_AGENT_IDENTITY_BOOTSTRAP_ATTEMPTS
                    && is_retryable_io_registration_error(&err) =>
            {
                tracing::warn!(
                    attempt,
                    max_attempts = MAX_AGENT_IDENTITY_BOOTSTRAP_ATTEMPTS,
                    error = %err,
                    "agent identity registration attempt failed; retrying"
                );
                attempt += 1;
            }
            Err(err) => return Err(err),
        }
    }
}

11. 测试与边界 ​

核心测试覆盖 assertion envelope 和签名验证、JWT unverified parse 与 JWKS verified mode、issuer/audience、untrusted kid、environment routing、registration URLs、retry classification、task response 和 Curve25519 conversion。login tests 再用 mock HTTP server 验证 task registration 与 transient retry。

源码位置：codex-rs/agent-identity/src/lib.rs :: authorization_header_for_agent_task_serializes_signed_agent_assertion

rust
signing_key
    .verifying_key()
    .verify(
        format!(
            "{}:{}:{}",
            envelope.agent_runtime_id, envelope.task_id, envelope.timestamp
        )
        .as_bytes(),
        &signature,
    )
    .expect("signature should verify");

源码位置：codex-rs/agent-identity/src/lib.rs :: decode_agent_identity_jwt_rejects_untrusted_kid

rust
decode_agent_identity_jwt(&jwt, Some(&jwks)).expect_err("JWT should not verify");

这些断言证明本地编码、payload 绑定、JWKS 验签和 retry 分类；不能证明远端服务接受注册、JWKS 永远新鲜、私钥存储安全、服务器 timestamp freshness policy 或 assertion replay 防护。decode_agent_identity_jwt(..., None) 的测试只证明 payload parser，不是验签通过。

在源码仓库中可运行：

text
cd codex-rs
cargo test -p codex-agent-identity --lib -- --test-threads=1
cargo test -p codex-login --lib agent_identity -- --test-threads=1

补充 Agent 与 Task 两级密钥关系，以及注册、断言和验签的时序。

12. 阅读闭环 ​

建议按 environment routing → 64-byte/domain-separated key generation → PKCS#8/SSH encoding → agent registration → task registration → AgentAssertion → JWT parser/verified mode → login bootstrap → Curve25519 task decryption → retry classification 阅读。读完后应能解释：durable identity 与 run task 为什么分离；AgentAssertion 为什么不是 JWT；无 JWKS decode 为什么不能用于授权；以及同一 Ed25519 key 如何分别承担签名和 task response 解密。

下一篇进入 Keyring 凭据存储与回退策略。