openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-25 18:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Codex][Codex CLI] Harden auth-failure telemetry delivery

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Colin Young
2026-04-09 10:56:10 -07:00
co-authored by Codex
parent 2b01b05bb9
commit 6b6253e48a
22 changed files with 1498 additions and 398 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+1
View File
@@ -2259,6 +2259,7 @@ dependencies = [
"codex-core",
"codex-exec-server",
"codex-features",
"codex-feedback",
"codex-protocol",
"codex-shell-command",
"codex-utils-cli",
codex-rs/app-server/src/lib.rs
-3
View File
@@ -434,7 +434,6 @@ pub async fn run_main_with_transport(
})?
}
};
if let Ok(Some(err)) = check_execpolicy_for_warnings(&config.config_layer_stack).await {
let (path, range) = exec_policy_warning_location(&err);
let message = ConfigWarningNotification {
@@ -500,7 +499,6 @@ pub async fn run_main_with_transport(
let feedback_layer = feedback.logger_layer();
let feedback_metadata_layer = feedback.metadata_layer();
let feedback_auth_event_layer = feedback.auth_event_layer();
let log_db = codex_state::StateRuntime::init(
config.sqlite_home.clone(),
config.model_provider_id.clone(),
@@ -517,7 +515,6 @@ pub async fn run_main_with_transport(
.with(stderr_fmt)
.with(feedback_layer)
.with(feedback_metadata_layer)
.with(feedback_auth_event_layer)
.with(log_db_layer)
.with(otel_logger_layer)
.with(otel_tracing_layer)
codex-rs/cloud-tasks/src/util.rs
-1
View File
@@ -60,7 +60,6 @@ pub fn extract_chatgpt_account_id(token: &str) -> Option<String> {
}
pub async fn load_auth_manager() -> Option<AuthManager> {
// TODO: pass in cli overrides once cloud tasks properly support them.
let config = Config::load_with_cli_overrides(Vec::new()).await.ok()?;
Some(AuthManager::new(
config.codex_home,
codex-rs/core/src/client.rs
+194 -22
View File
@@ -105,6 +105,7 @@ use crate::flags::CODEX_RS_SSE_FIXTURE;
use crate::model_provider_info::ModelProviderInfo;
use crate::model_provider_info::WireApi;
use crate::provider_auth::auth_manager_for_provider;
use crate::response_debug_context::ResponseDebugContext;
use crate::response_debug_context::extract_response_debug_context;
use crate::response_debug_context::extract_response_debug_context_from_api_error;
use crate::response_debug_context::telemetry_api_error_message;
@@ -364,10 +365,12 @@ impl ModelClient {
AuthRequestTelemetryContext::new(
client_setup.auth.as_ref().map(CodexAuth::auth_mode),
&client_setup.api_auth,
/*has_followup_unauthorized_retry*/ false,
PendingUnauthorizedRetry::default(),
),
RequestRouteTelemetry::for_endpoint(RESPONSES_COMPACT_ENDPOINT),
self.state.auth_env_telemetry.clone(),
self.state.provider.should_emit_sentry_auth_failures(),
);
let client =
ApiCompactClient::new(transport, client_setup.api_provider, client_setup.api_auth)
@@ -433,10 +436,12 @@ impl ModelClient {
AuthRequestTelemetryContext::new(
client_setup.auth.as_ref().map(CodexAuth::auth_mode),
&client_setup.api_auth,
/*has_followup_unauthorized_retry*/ false,
PendingUnauthorizedRetry::default(),
),
RequestRouteTelemetry::for_endpoint(MEMORIES_SUMMARIZE_ENDPOINT),
self.state.auth_env_telemetry.clone(),
self.state.provider.should_emit_sentry_auth_failures(),
);
let client =
ApiMemoriesClient::new(transport, client_setup.api_provider, client_setup.api_auth)
@@ -482,12 +487,14 @@ impl ModelClient {
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
emit_sentry_auth_failures: bool,
) -> Arc<dyn RequestTelemetry> {
let telemetry = Arc::new(ApiTelemetry::new(
session_telemetry.clone(),
auth_context,
request_route_telemetry,
auth_env_telemetry,
emit_sentry_auth_failures,
));
let request_telemetry: Arc<dyn RequestTelemetry> = telemetry;
request_telemetry
@@ -568,6 +575,7 @@ impl ModelClient {
auth_context,
request_route_telemetry,
self.state.auth_env_telemetry.clone(),
self.state.provider.should_emit_sentry_auth_failures(),
);
let websocket_connect_timeout = self.state.provider.websocket_connect_timeout();
let start = Instant::now();
@@ -632,6 +640,33 @@ impl ModelClient {
},
&self.state.auth_env_telemetry,
);
emit_sentry_auth_failure_if_unauthorized(
&FeedbackRequestTags {
endpoint: request_route_telemetry.endpoint,
auth_header_attached: auth_context.auth_header_attached,
auth_header_name: auth_context.auth_header_name,
auth_mode: auth_context.auth_mode,
auth_retry_after_unauthorized: Some(auth_context.retry_after_unauthorized),
auth_recovery_mode: auth_context.recovery_mode,
auth_recovery_phase: auth_context.recovery_phase,
auth_connection_reused: Some(false),
auth_request_id: response_debug.request_id.as_deref(),
auth_cf_ray: response_debug.cf_ray.as_deref(),
auth_error: response_debug.auth_error.as_deref(),
auth_error_code: response_debug.auth_error_code.as_deref(),
auth_recovery_followup_success: auth_context
.retry_after_unauthorized
.then_some(result.is_ok()),
auth_recovery_followup_status: auth_context
.retry_after_unauthorized
.then_some(status)
.flatten(),
},
&self.state.auth_env_telemetry,
self.state.provider.should_emit_sentry_auth_failures(),
status,
auth_context.has_followup_unauthorized_retry,
);
result
}
@@ -879,6 +914,7 @@ impl ModelClientSession {
let auth_context = AuthRequestTelemetryContext::new(
client_setup.auth.as_ref().map(CodexAuth::auth_mode),
&client_setup.api_auth,
/*has_followup_unauthorized_retry*/ false,
PendingUnauthorizedRetry::default(),
);
let connection = self
@@ -1029,11 +1065,15 @@ impl ModelClientSession {
.map(super::auth::AuthManager::unauthorized_recovery);
let mut pending_retry = PendingUnauthorizedRetry::default();
loop {
let has_followup_unauthorized_retry = auth_recovery
.as_ref()
.is_some_and(UnauthorizedRecovery::has_next);
let client_setup = self.client.current_client_setup().await?;
let transport = ReqwestTransport::new(build_reqwest_client());
let request_auth_context = AuthRequestTelemetryContext::new(
client_setup.auth.as_ref().map(CodexAuth::auth_mode),
&client_setup.api_auth,
has_followup_unauthorized_retry,
pending_retry,
);
let (request_telemetry, sse_telemetry) = Self::build_streaming_telemetry(
@@ -1041,6 +1081,10 @@ impl ModelClientSession {
request_auth_context,
RequestRouteTelemetry::for_endpoint(RESPONSES_ENDPOINT),
self.client.state.auth_env_telemetry.clone(),
self.client
.state
.provider
.should_emit_sentry_auth_failures(),
);
let compression = self.responses_request_compression(client_setup.auth.as_ref());
let options = self.build_responses_options(turn_metadata_header, compression);
@@ -1069,14 +1113,42 @@ impl ModelClientSession {
Err(ApiError::Transport(
unauthorized_transport @ TransportError::Http { status, .. },
)) if status == StatusCode::UNAUTHORIZED => {
pending_retry = PendingUnauthorizedRetry::from_recovery(
handle_unauthorized(
unauthorized_transport,
&mut auth_recovery,
session_telemetry,
)
.await?,
);
let attempted_recovery = auth_recovery.as_ref().and_then(|recovery| {
recovery
.has_next()
.then(|| (recovery.mode_name(), recovery.step_name()))
});
let debug = extract_response_debug_context(&unauthorized_transport);
match handle_unauthorized(
unauthorized_transport,
&mut auth_recovery,
session_telemetry,
)
.await
{
Ok(recovery) => {
pending_retry = PendingUnauthorizedRetry::from_recovery(recovery);
}
Err(err) => {
if request_auth_context.has_followup_unauthorized_retry {
emit_terminal_auth_failure_after_failed_recovery(
AuthRequestTelemetryContext {
recovery_mode: attempted_recovery.map(|(mode, _)| mode),
recovery_phase: attempted_recovery.map(|(_, phase)| phase),
..request_auth_context
},
RequestRouteTelemetry::for_endpoint(RESPONSES_ENDPOINT),
&self.client.state.auth_env_telemetry,
&debug,
self.client
.state
.provider
.should_emit_sentry_auth_failures(),
);
}
return Err(err);
}
}
continue;
}
Err(err) => return Err(map_api_error(err)),
@@ -1118,10 +1190,14 @@ impl ModelClientSession {
.map(super::auth::AuthManager::unauthorized_recovery);
let mut pending_retry = PendingUnauthorizedRetry::default();
loop {
let has_followup_unauthorized_retry = auth_recovery
.as_ref()
.is_some_and(UnauthorizedRecovery::has_next);
let client_setup = self.client.current_client_setup().await?;
let request_auth_context = AuthRequestTelemetryContext::new(
client_setup.auth.as_ref().map(CodexAuth::auth_mode),
&client_setup.api_auth,
has_followup_unauthorized_retry,
pending_retry,
);
let compression = self.responses_request_compression(client_setup.auth.as_ref());
@@ -1169,14 +1245,42 @@ impl ModelClientSession {
Err(ApiError::Transport(
unauthorized_transport @ TransportError::Http { status, .. },
)) if status == StatusCode::UNAUTHORIZED => {
pending_retry = PendingUnauthorizedRetry::from_recovery(
handle_unauthorized(
unauthorized_transport,
&mut auth_recovery,
session_telemetry,
)
.await?,
);
let attempted_recovery = auth_recovery.as_ref().and_then(|recovery| {
recovery
.has_next()
.then(|| (recovery.mode_name(), recovery.step_name()))
});
let debug = extract_response_debug_context(&unauthorized_transport);
match handle_unauthorized(
unauthorized_transport,
&mut auth_recovery,
session_telemetry,
)
.await
{
Ok(recovery) => {
pending_retry = PendingUnauthorizedRetry::from_recovery(recovery);
}
Err(err) => {
if request_auth_context.has_followup_unauthorized_retry {
emit_terminal_auth_failure_after_failed_recovery(
AuthRequestTelemetryContext {
recovery_mode: attempted_recovery.map(|(mode, _)| mode),
recovery_phase: attempted_recovery.map(|(_, phase)| phase),
..request_auth_context
},
RequestRouteTelemetry::for_endpoint(RESPONSES_ENDPOINT),
&self.client.state.auth_env_telemetry,
&debug,
self.client
.state
.provider
.should_emit_sentry_auth_failures(),
);
}
return Err(err);
}
}
continue;
}
Err(err) => return Err(map_api_error(err)),
@@ -1206,12 +1310,14 @@ impl ModelClientSession {
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
emit_sentry_auth_failures: bool,
) -> (Arc<dyn RequestTelemetry>, Arc<dyn SseTelemetry>) {
let telemetry = Arc::new(ApiTelemetry::new(
session_telemetry.clone(),
auth_context,
request_route_telemetry,
auth_env_telemetry,
emit_sentry_auth_failures,
));
let request_telemetry: Arc<dyn RequestTelemetry> = telemetry.clone();
let sse_telemetry: Arc<dyn SseTelemetry> = telemetry;
@@ -1224,12 +1330,14 @@ impl ModelClientSession {
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
emit_sentry_auth_failures: bool,
) -> Arc<dyn WebsocketTelemetry> {
let telemetry = Arc::new(ApiTelemetry::new(
session_telemetry.clone(),
auth_context,
request_route_telemetry,
auth_env_telemetry,
emit_sentry_auth_failures,
));
let websocket_telemetry: Arc<dyn WebsocketTelemetry> = telemetry;
websocket_telemetry
@@ -1523,6 +1631,7 @@ struct AuthRequestTelemetryContext {
auth_mode: Option<&'static str>,
auth_header_attached: bool,
auth_header_name: Option<&'static str>,
has_followup_unauthorized_retry: bool,
retry_after_unauthorized: bool,
recovery_mode: Option<&'static str>,
recovery_phase: Option<&'static str>,
@@ -1532,6 +1641,7 @@ impl AuthRequestTelemetryContext {
fn new(
auth_mode: Option<AuthMode>,
api_auth: &CoreAuthProvider,
has_followup_unauthorized_retry: bool,
retry: PendingUnauthorizedRetry,
) -> Self {
Self {
@@ -1541,6 +1651,7 @@ impl AuthRequestTelemetryContext {
}),
auth_header_attached: api_auth.auth_header_attached(),
auth_header_name: api_auth.auth_header_name(),
has_followup_unauthorized_retry,
retry_after_unauthorized: retry.retry_after_unauthorized,
recovery_mode: retry.recovery_mode,
recovery_phase: retry.recovery_phase,
@@ -1686,6 +1797,7 @@ struct ApiTelemetry {
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
emit_sentry_auth_failures: bool,
}
impl ApiTelemetry {
@@ -1694,16 +1806,68 @@ impl ApiTelemetry {
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
emit_sentry_auth_failures: bool,
) -> Self {
Self {
session_telemetry,
auth_context,
request_route_telemetry,
auth_env_telemetry,
emit_sentry_auth_failures,
}
}
}
fn emit_sentry_auth_failure_if_unauthorized(
feedback_tags: &FeedbackRequestTags<'_>,
auth_env_telemetry: &AuthEnvTelemetry,
emit_sentry_auth_failures: bool,
status: Option<u16>,
has_followup_unauthorized_retry: bool,
) {
if status == Some(http::StatusCode::UNAUTHORIZED.as_u16())
&& !has_followup_unauthorized_retry
&& emit_sentry_auth_failures
{
emit_sentry_auth_failure_event_with_auth_env(feedback_tags, auth_env_telemetry);
}
}
fn emit_terminal_auth_failure_after_failed_recovery(
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
auth_env_telemetry: &AuthEnvTelemetry,
debug: &ResponseDebugContext,
emit_sentry_auth_failures: bool,
) {
if auth_context.recovery_phase == Some("refresh_token") {
return;
}
let feedback_tags = FeedbackRequestTags {
endpoint: request_route_telemetry.endpoint,
auth_header_attached: auth_context.auth_header_attached,
auth_header_name: auth_context.auth_header_name,
auth_mode: auth_context.auth_mode,
auth_retry_after_unauthorized: Some(false),
auth_recovery_mode: auth_context.recovery_mode,
auth_recovery_phase: auth_context.recovery_phase,
auth_connection_reused: None,
auth_request_id: debug.request_id.as_deref(),
auth_cf_ray: debug.cf_ray.as_deref(),
auth_error: debug.auth_error.as_deref(),
auth_error_code: debug.auth_error_code.as_deref(),
auth_recovery_followup_success: None,
auth_recovery_followup_status: None,
};
emit_sentry_auth_failure_if_unauthorized(
&feedback_tags,
auth_env_telemetry,
emit_sentry_auth_failures,
Some(StatusCode::UNAUTHORIZED.as_u16()),
/*has_followup_unauthorized_retry*/ false,
);
}
impl RequestTelemetry for ApiTelemetry {
fn on_request(
&self,
@@ -1757,9 +1921,13 @@ impl RequestTelemetry for ApiTelemetry {
.flatten(),
};
emit_feedback_request_tags_with_auth_env(&feedback_tags, &self.auth_env_telemetry);
if status == Some(http::StatusCode::UNAUTHORIZED.as_u16()) {
emit_sentry_auth_failure_event_with_auth_env(&feedback_tags, &self.auth_env_telemetry);
}
emit_sentry_auth_failure_if_unauthorized(
&feedback_tags,
&self.auth_env_telemetry,
self.emit_sentry_auth_failures,
status,
self.auth_context.has_followup_unauthorized_retry,
);
}
}
@@ -1812,9 +1980,13 @@ impl WebsocketTelemetry for ApiTelemetry {
.flatten(),
};
emit_feedback_request_tags_with_auth_env(&feedback_tags, &self.auth_env_telemetry);
if status == Some(http::StatusCode::UNAUTHORIZED.as_u16()) {
emit_sentry_auth_failure_event_with_auth_env(&feedback_tags, &self.auth_env_telemetry);
}
emit_sentry_auth_failure_if_unauthorized(
&feedback_tags,
&self.auth_env_telemetry,
self.emit_sentry_auth_failures,
status,
self.auth_context.has_followup_unauthorized_retry && !connection_reused,
);
}
fn on_ws_event(
codex-rs/core/src/client_tests.rs
+333
View File
@@ -1,14 +1,27 @@
use super::ApiTelemetry;
use super::AuthRequestTelemetryContext;
use super::ModelClient;
use super::PendingUnauthorizedRetry;
use super::RequestRouteTelemetry;
use super::UnauthorizedRecoveryExecution;
use super::emit_terminal_auth_failure_after_failed_recovery;
use crate::auth_env_telemetry::AuthEnvTelemetry;
use crate::response_debug_context::ResponseDebugContext;
use codex_api::ApiError;
use codex_api::RequestTelemetry;
use codex_api::WebsocketTelemetry;
use codex_otel::SessionTelemetry;
use codex_protocol::ThreadId;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use http::StatusCode;
use pretty_assertions::assert_eq;
use serde_json::json;
use serial_test::serial;
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::Mutex;
fn test_model_client(session_source: SessionSource) -> ModelClient {
let provider = crate::model_provider_info::create_oss_provider_with_base_url(
@@ -72,6 +85,37 @@ fn test_session_telemetry() -> SessionTelemetry {
)
}
fn empty_auth_env_telemetry() -> AuthEnvTelemetry {
AuthEnvTelemetry {
openai_api_key_env_present: false,
codex_api_key_env_present: false,
codex_api_key_env_enabled: false,
provider_env_key_name: None,
provider_env_key_present: None,
refresh_token_url_override_present: false,
}
}
type AuthFailureReportCollector = Arc<Mutex<Vec<BTreeMap<String, String>>>>;
fn install_auth_failure_report_collector() -> (
AuthFailureReportCollector,
crate::auth::AuthFailureReporterGuard,
) {
let reported = Arc::new(Mutex::new(Vec::new()));
let guard = crate::auth::set_auth_failure_reporter({
let reported = Arc::clone(&reported);
Arc::new(move |fields| {
reported
.lock()
.expect("report collector poisoned")
.push(fields);
true
})
});
(reported, guard)
}
#[test]
fn build_subagent_headers_sets_other_subagent_label() {
let client = test_model_client(SessionSource::SubAgent(SubAgentSource::Other(
@@ -107,6 +151,7 @@ fn auth_request_telemetry_context_tracks_attached_auth_and_retry_phase() {
let auth_context = AuthRequestTelemetryContext::new(
Some(crate::auth::AuthMode::Chatgpt),
&crate::api_bridge::CoreAuthProvider::for_test(Some("access-token"), Some("workspace-123")),
/*has_followup_unauthorized_retry*/ true,
PendingUnauthorizedRetry::from_recovery(UnauthorizedRecoveryExecution {
mode: "managed",
phase: "refresh_token",
@@ -120,3 +165,291 @@ fn auth_request_telemetry_context_tracks_attached_auth_and_retry_phase() {
assert_eq!(auth_context.recovery_mode, Some("managed"));
assert_eq!(auth_context.recovery_phase, Some("refresh_token"));
}
#[test]
#[serial(auth_failure_reporter)]
fn api_telemetry_skips_auth_failure_while_followup_retry_remains() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let telemetry = ApiTelemetry::new(
test_session_telemetry(),
AuthRequestTelemetryContext {
auth_mode: Some("Chatgpt"),
auth_header_attached: true,
auth_header_name: Some("authorization"),
has_followup_unauthorized_retry: true,
retry_after_unauthorized: false,
recovery_mode: Some("managed"),
recovery_phase: Some("refresh_token"),
},
RequestRouteTelemetry::for_endpoint("/responses"),
empty_auth_env_telemetry(),
/*emit_sentry_auth_failures*/ true,
);
telemetry.on_request(
/*attempt*/ 1,
Some(StatusCode::UNAUTHORIZED),
/*error*/ None,
Default::default(),
);
assert!(reported.lock().unwrap().is_empty());
}
#[test]
#[serial(auth_failure_reporter)]
fn api_telemetry_emits_auth_failure_once_no_followup_retry_remains() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let telemetry = ApiTelemetry::new(
test_session_telemetry(),
AuthRequestTelemetryContext {
auth_mode: Some("Chatgpt"),
auth_header_attached: true,
auth_header_name: Some("authorization"),
has_followup_unauthorized_retry: false,
retry_after_unauthorized: true,
recovery_mode: Some("managed"),
recovery_phase: Some("refresh_token"),
},
RequestRouteTelemetry::for_endpoint("/responses"),
empty_auth_env_telemetry(),
/*emit_sentry_auth_failures*/ true,
);
telemetry.on_request(
/*attempt*/ 2,
Some(StatusCode::UNAUTHORIZED),
/*error*/ None,
Default::default(),
);
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(
reported[0].get("endpoint").map(String::as_str),
Some("/responses")
);
assert_eq!(
reported[0]
.get("auth_retry_after_unauthorized")
.map(String::as_str),
Some("true")
);
}
#[test]
#[serial(auth_failure_reporter)]
fn api_telemetry_skips_sentry_auth_failure_for_non_openai_provider() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let telemetry = ApiTelemetry::new(
test_session_telemetry(),
AuthRequestTelemetryContext {
auth_mode: Some("Chatgpt"),
auth_header_attached: true,
auth_header_name: Some("authorization"),
has_followup_unauthorized_retry: true,
retry_after_unauthorized: true,
recovery_mode: Some("managed"),
recovery_phase: Some("refresh_token"),
},
RequestRouteTelemetry::for_endpoint("/responses"),
empty_auth_env_telemetry(),
/*emit_sentry_auth_failures*/ false,
);
telemetry.on_request(
/*attempt*/ 1,
Some(StatusCode::UNAUTHORIZED),
/*error*/ None,
Default::default(),
);
assert!(reported.lock().unwrap().is_empty());
}
#[test]
#[serial(auth_failure_reporter)]
fn websocket_handshake_failure_emits_auth_failure_when_no_followup_retry_remains() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let telemetry = ApiTelemetry::new(
test_session_telemetry(),
AuthRequestTelemetryContext {
auth_mode: Some("Chatgpt"),
auth_header_attached: true,
auth_header_name: Some("authorization"),
has_followup_unauthorized_retry: false,
retry_after_unauthorized: true,
recovery_mode: Some("managed"),
recovery_phase: Some("refresh_token"),
},
RequestRouteTelemetry::for_endpoint("/responses"),
empty_auth_env_telemetry(),
/*emit_sentry_auth_failures*/ true,
);
telemetry.on_ws_request(
Default::default(),
Some(&ApiError::Transport(codex_api::TransportError::Http {
status: StatusCode::UNAUTHORIZED,
url: None,
headers: None,
body: None,
})),
/*connection_reused*/ false,
);
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(
reported[0].get("endpoint").map(String::as_str),
Some("/responses")
);
assert_eq!(
reported[0]
.get("auth_retry_after_unauthorized")
.map(String::as_str),
Some("true")
);
}
#[test]
#[serial(auth_failure_reporter)]
fn websocket_reused_connection_failure_emits_auth_failure_even_with_followup_retry_available() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let telemetry = ApiTelemetry::new(
test_session_telemetry(),
AuthRequestTelemetryContext {
auth_mode: Some("Chatgpt"),
auth_header_attached: true,
auth_header_name: Some("authorization"),
has_followup_unauthorized_retry: true,
retry_after_unauthorized: false,
recovery_mode: Some("managed"),
recovery_phase: Some("refresh_token"),
},
RequestRouteTelemetry::for_endpoint("/responses"),
empty_auth_env_telemetry(),
/*emit_sentry_auth_failures*/ true,
);
telemetry.on_ws_request(
Default::default(),
Some(&ApiError::Transport(codex_api::TransportError::Http {
status: StatusCode::UNAUTHORIZED,
url: None,
headers: None,
body: None,
})),
/*connection_reused*/ true,
);
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(
reported[0].get("endpoint").map(String::as_str),
Some("/responses")
);
assert_eq!(
reported[0]
.get("auth_connection_reused")
.map(String::as_str),
Some("true")
);
}
#[test]
#[serial(auth_failure_reporter)]
fn failed_recovery_before_retry_emits_terminal_auth_failure() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let auth_context = AuthRequestTelemetryContext {
auth_mode: Some("Chatgpt"),
auth_header_attached: true,
auth_header_name: Some("authorization"),
has_followup_unauthorized_retry: true,
retry_after_unauthorized: false,
recovery_mode: Some("managed"),
recovery_phase: Some("reload"),
};
let debug = ResponseDebugContext {
request_id: Some("req_failed_recovery".to_string()),
cf_ray: Some("ray_failed_recovery".to_string()),
auth_error: None,
auth_error_code: Some("token_expired".to_string()),
};
emit_terminal_auth_failure_after_failed_recovery(
auth_context,
RequestRouteTelemetry::for_endpoint("/responses"),
&empty_auth_env_telemetry(),
&debug,
/*emit_sentry_auth_failures*/ true,
);
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(
reported[0].get("endpoint").map(String::as_str),
Some("/responses")
);
assert_eq!(
reported[0].get("auth_recovery_mode").map(String::as_str),
Some("managed")
);
assert_eq!(
reported[0].get("auth_recovery_phase").map(String::as_str),
Some("reload")
);
assert_eq!(
reported[0]
.get("auth_retry_after_unauthorized")
.map(String::as_str),
Some("false")
);
assert_eq!(
reported[0].get("auth_request_id").map(String::as_str),
Some("req_failed_recovery")
);
}
#[test]
#[serial(auth_failure_reporter)]
fn refresh_token_failed_recovery_before_retry_does_not_emit_terminal_auth_failure() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let auth_context = AuthRequestTelemetryContext {
auth_mode: Some("Chatgpt"),
auth_header_attached: true,
auth_header_name: Some("authorization"),
has_followup_unauthorized_retry: true,
retry_after_unauthorized: false,
recovery_mode: Some("managed"),
recovery_phase: Some("refresh_token"),
};
let debug = ResponseDebugContext {
request_id: Some("req_refresh_failed".to_string()),
cf_ray: Some("ray_refresh_failed".to_string()),
auth_error: None,
auth_error_code: Some("token_expired".to_string()),
};
emit_terminal_auth_failure_after_failed_recovery(
auth_context,
RequestRouteTelemetry::for_endpoint("/responses"),
&empty_auth_env_telemetry(),
&debug,
/*emit_sentry_auth_failures*/ true,
);
assert!(
reported
.lock()
.expect("report collector poisoned")
.is_empty()
);
}
codex-rs/core/src/config/mod.rs
+9 -5
View File
@@ -796,6 +796,13 @@ fn load_catalog_json(path: &AbsolutePathBuf) -> std::io::Result<ModelsResponse>
Ok(catalog)
}
pub fn feedback_enabled_from_config_toml(cfg: &ConfigToml) -> bool {
cfg.feedback
.as_ref()
.and_then(|feedback| feedback.enabled)
.unwrap_or(true)
}
fn load_model_catalog(
model_catalog_json: Option<AbsolutePathBuf>,
) -> std::io::Result<Option<ModelsResponse>> {
@@ -1983,6 +1990,7 @@ impl Config {
) -> std::io::Result<Self> {
validate_model_providers(&cfg.model_providers)
.map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?;
let feedback_enabled = feedback_enabled_from_config_toml(&cfg);
// Ensure that every field of ConfigRequirements is applied to the final
// Config.
let ConfigRequirements {
@@ -2685,11 +2693,7 @@ impl Config {
.as_ref()
.and_then(|a| a.enabled)
.or(cfg.analytics.as_ref().and_then(|a| a.enabled)),
feedback_enabled: cfg
.feedback
.as_ref()
.and_then(|feedback| feedback.enabled)
.unwrap_or(true),
feedback_enabled,
tool_suggest,
tui_notifications: cfg
.tui
codex-rs/core/src/model_provider_info.rs
+37
View File
@@ -19,6 +19,7 @@ use serde::Serialize;
use std::collections::HashMap;
use std::fmt;
use std::time::Duration;
use url::Url;
const DEFAULT_STREAM_IDLE_TIMEOUT_MS: u64 = 300_000;
const DEFAULT_STREAM_MAX_RETRIES: u64 = 5;
@@ -316,6 +317,42 @@ impl ModelProviderInfo {
pub(crate) fn has_command_auth(&self) -> bool {
self.auth.is_some()
}
pub(crate) fn should_emit_sentry_auth_failures(&self) -> bool {
(self.requires_openai_auth
|| self.env_key.is_some()
|| self.auth.is_some()
|| self.experimental_bearer_token.is_some()
|| self.http_headers.as_ref().is_some_and(|headers| {
headers
.keys()
.any(|key| key.eq_ignore_ascii_case("authorization"))
})
|| self.env_http_headers.as_ref().is_some_and(|headers| {
headers
.keys()
.any(|key| key.eq_ignore_ascii_case("authorization"))
}))
&& self
.base_url
.as_deref()
.map(Self::is_openai_owned_base_url)
.unwrap_or(true)
}
fn is_openai_owned_base_url(base_url: &str) -> bool {
let Ok(parsed) = Url::parse(base_url) else {
return false;
};
let Some(host) = parsed.host_str() else {
return false;
};
host == "openai.com"
|| host.ends_with(".openai.com")
|| host == "chatgpt.com"
|| host.ends_with(".chatgpt.com")
}
}
pub const DEFAULT_LMSTUDIO_PORT: u16 = 1234;
codex-rs/core/src/model_provider_info_tests.rs
+129
View File
@@ -156,3 +156,132 @@ args = ["--format=text"]
})
);
}
#[test]
fn openai_provider_without_base_url_emits_sentry_auth_failures() {
let provider = ModelProviderInfo::create_openai_provider(/*base_url*/ None);
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn openai_provider_with_non_openai_base_url_skips_sentry_auth_failures() {
let provider =
ModelProviderInfo::create_openai_provider(Some("https://example.com/v1".to_string()));
assert!(!provider.should_emit_sentry_auth_failures());
}
#[test]
fn openai_provider_with_openai_subdomain_base_url_emits_sentry_auth_failures() {
let provider = ModelProviderInfo::create_openai_provider(Some(
"https://api-staging.openai.com/v1".to_string(),
));
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn openai_provider_with_chatgpt_subdomain_base_url_emits_sentry_auth_failures() {
let provider =
ModelProviderInfo::create_openai_provider(Some("https://ab.chatgpt.com/v1".to_string()));
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn custom_openai_provider_with_env_key_emits_sentry_auth_failures() {
let provider: ModelProviderInfo = toml::from_str(
r#"
name = "OpenAI Staging"
base_url = "https://api.openai.com/v1"
env_key = "OPENAI_API_KEY"
wire_api = "responses"
"#,
)
.expect("custom openai provider should deserialize");
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn custom_openai_provider_with_command_auth_emits_sentry_auth_failures() {
let provider: ModelProviderInfo = toml::from_str(
r#"
name = "OpenAI Staging"
base_url = "https://api.openai.com/v1"
wire_api = "responses"
[auth]
command = "print-token"
"#,
)
.expect("custom openai provider should deserialize");
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn custom_openai_provider_with_static_bearer_token_emits_sentry_auth_failures() {
let provider: ModelProviderInfo = toml::from_str(
r#"
name = "OpenAI Staging"
base_url = "https://api.openai.com/v1"
wire_api = "responses"
experimental_bearer_token = "token"
"#,
)
.expect("custom openai provider should deserialize");
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn custom_openai_provider_with_http_authorization_header_emits_sentry_auth_failures() {
let provider: ModelProviderInfo = toml::from_str(
r#"
name = "OpenAI Staging"
base_url = "https://api.openai.com/v1"
wire_api = "responses"
[http_headers]
Authorization = "Bearer token"
"#,
)
.expect("custom openai provider should deserialize");
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn custom_openai_provider_with_env_authorization_header_emits_sentry_auth_failures() {
let provider: ModelProviderInfo = toml::from_str(
r#"
name = "OpenAI Staging"
base_url = "https://api.openai.com/v1"
wire_api = "responses"
[env_http_headers]
authorization = "OPENAI_AUTH_HEADER"
"#,
)
.expect("custom openai provider should deserialize");
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn renamed_openai_provider_still_emits_sentry_auth_failures() {
let mut provider = ModelProviderInfo::create_openai_provider(/*base_url*/ None);
provider.name = "Staging".to_string();
assert!(provider.should_emit_sentry_auth_failures());
}
#[test]
fn oss_provider_skips_sentry_auth_failures() {
let provider =
create_oss_provider_with_base_url("https://api.openai.com/v1", WireApi::Responses);
assert!(!provider.should_emit_sentry_auth_failures());
}
codex-rs/core/src/models_manager/manager.rs
-4
View File
@@ -19,7 +19,6 @@ use crate::response_debug_context::extract_response_debug_context;
use crate::response_debug_context::telemetry_transport_error_message;
use crate::util::FeedbackRequestTags;
use crate::util::emit_feedback_request_tags_with_auth_env;
use crate::util::emit_sentry_auth_failure_event_with_auth_env;
use codex_api::ModelsClient;
use codex_api::RequestTelemetry;
use codex_api::ReqwestTransport;
@@ -132,9 +131,6 @@ impl RequestTelemetry for ModelsRequestTelemetry {
auth_recovery_followup_status: None,
};
emit_feedback_request_tags_with_auth_env(&feedback_tags, &self.auth_env);
if status == Some(http::StatusCode::UNAUTHORIZED.as_u16()) {
emit_sentry_auth_failure_event_with_auth_env(&feedback_tags, &self.auth_env);
}
}
}
codex-rs/core/src/models_manager/manager_tests.rs
+26 -31
View File
@@ -14,6 +14,7 @@ use http::HeaderMap;
use http::StatusCode;
use pretty_assertions::assert_eq;
use serde_json::json;
use serial_test::serial;
use std::collections::BTreeMap;
use std::num::NonZeroU64;
use std::sync::Arc;
@@ -233,6 +234,26 @@ where
}
}
type AuthFailureReportCollector = Arc<Mutex<Vec<BTreeMap<String, String>>>>;
fn install_auth_failure_report_collector() -> (
AuthFailureReportCollector,
crate::auth::AuthFailureReporterGuard,
) {
let reported = Arc::new(Mutex::new(Vec::new()));
let guard = crate::auth::set_auth_failure_reporter({
let reported = Arc::clone(&reported);
Arc::new(move |fields| {
reported
.lock()
.expect("report collector poisoned")
.push(fields);
true
})
});
(reported, guard)
}
#[tokio::test]
async fn get_model_info_tracks_fallback_usage() {
let codex_home = tempdir().expect("temp dir");
@@ -840,14 +861,9 @@ fn models_request_telemetry_emits_auth_env_feedback_tags_on_failure() {
}
#[test]
fn models_request_telemetry_emits_sentry_auth_failure_event_on_401() {
let tags = Arc::new(Mutex::new(BTreeMap::new()));
let _guard = tracing_subscriber::registry()
.with(TagCollectorLayer {
target: crate::util::SENTRY_AUTH_FAILURES_TARGET,
tags: tags.clone(),
})
.set_default();
#[serial(auth_failure_reporter)]
fn models_request_telemetry_does_not_emit_sentry_auth_failure_event_on_401() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let telemetry = ModelsRequestTelemetry {
auth_mode: Some(TelemetryAuthMode::Chatgpt.to_string()),
@@ -877,7 +893,7 @@ fn models_request_telemetry_emits_sentry_auth_failure_event_on_401() {
.unwrap(),
);
telemetry.on_request(
1,
/*attempt*/ 1,
Some(StatusCode::UNAUTHORIZED),
Some(&TransportError::Http {
status: StatusCode::UNAUTHORIZED,
@@ -888,28 +904,7 @@ fn models_request_telemetry_emits_sentry_auth_failure_event_on_401() {
Duration::from_millis(17),
);
let tags = tags.lock().unwrap().clone();
assert_eq!(
tags.get("report_kind").map(String::as_str),
Some("auth_failure_auto")
);
assert_eq!(tags.get("endpoint").map(String::as_str), Some("/models"));
assert_eq!(
tags.get("auth_header_attached").map(String::as_str),
Some("true")
);
assert_eq!(
tags.get("auth_header_name").map(String::as_str),
Some("authorization")
);
assert_eq!(
tags.get("auth_request_id").map(String::as_str),
Some("req-models-401")
);
assert_eq!(
tags.get("auth_cf_ray").map(String::as_str),
Some("ray-models-401")
);
assert!(reported.lock().unwrap().is_empty());
}
#[test]
codex-rs/core/src/util.rs
+94 -34
View File
@@ -1,3 +1,4 @@
use std::collections::BTreeMap;
use std::path::Path;
use std::path::PathBuf;
use std::time::Duration;
@@ -11,8 +12,6 @@ use crate::parse_command::shlex_join;
const INITIAL_DELAY_MS: u64 = 200;
const BACKOFF_FACTOR: f64 = 2.0;
pub(crate) const SENTRY_AUTH_FAILURES_TARGET: &str = "sentry_auth_failures";
/// Emit structured feedback metadata as key/value pairs.
///
/// This logs a tracing event with `target: "feedback_tags"`. If
@@ -206,38 +205,99 @@ pub(crate) fn emit_sentry_auth_failure_event_with_auth_env(
tags: &FeedbackRequestTags<'_>,
auth_env: &AuthEnvTelemetry,
) {
tracing::event!(
target: SENTRY_AUTH_FAILURES_TARGET,
tracing::Level::ERROR,
report_kind = "auth_failure_auto",
endpoint = tags.endpoint,
auth_header_attached = tags.auth_header_attached,
auth_header_name = tags.auth_header_name.unwrap_or(""),
auth_mode = tags.auth_mode.unwrap_or(""),
auth_retry_after_unauthorized = tags
.auth_retry_after_unauthorized
.unwrap_or(false),
auth_recovery_mode = tags.auth_recovery_mode.unwrap_or(""),
auth_recovery_phase = tags.auth_recovery_phase.unwrap_or(""),
auth_connection_reused = tags.auth_connection_reused.unwrap_or(false),
auth_request_id = tags.auth_request_id.unwrap_or(""),
auth_cf_ray = tags.auth_cf_ray.unwrap_or(""),
auth_error = tags.auth_error.unwrap_or(""),
auth_error_code = tags.auth_error_code.unwrap_or(""),
auth_recovery_followup_success = tags
.auth_recovery_followup_success
.unwrap_or(false),
auth_recovery_followup_status = tags
.auth_recovery_followup_status
.unwrap_or_default(),
auth_env_openai_api_key_present = auth_env.openai_api_key_env_present,
auth_env_codex_api_key_present = auth_env.codex_api_key_env_present,
auth_env_codex_api_key_enabled = auth_env.codex_api_key_env_enabled,
auth_env_provider_key_name = auth_env.provider_env_key_name.as_deref().unwrap_or(""),
auth_env_provider_key_present = auth_env.provider_env_key_present.unwrap_or(false),
auth_env_refresh_token_url_override_present = auth_env.refresh_token_url_override_present,
cli_version = env!("CARGO_PKG_VERSION"),
);
let mut fields = BTreeMap::from([
("report_kind".to_string(), "auth_failure_auto".to_string()),
("endpoint".to_string(), tags.endpoint.to_string()),
(
"auth_header_attached".to_string(),
tags.auth_header_attached.to_string(),
),
(
"auth_retry_after_unauthorized".to_string(),
tags.auth_retry_after_unauthorized
.unwrap_or(false)
.to_string(),
),
(
"auth_env_openai_api_key_present".to_string(),
auth_env.openai_api_key_env_present.to_string(),
),
(
"auth_env_codex_api_key_present".to_string(),
auth_env.codex_api_key_env_present.to_string(),
),
(
"auth_env_codex_api_key_enabled".to_string(),
auth_env.codex_api_key_env_enabled.to_string(),
),
(
"auth_env_refresh_token_url_override_present".to_string(),
auth_env.refresh_token_url_override_present.to_string(),
),
(
"cli_version".to_string(),
env!("CARGO_PKG_VERSION").to_string(),
),
]);
if let Some(auth_header_name) = tags.auth_header_name {
fields.insert("auth_header_name".to_string(), auth_header_name.to_string());
}
if let Some(auth_mode) = tags.auth_mode {
fields.insert("auth_mode".to_string(), auth_mode.to_string());
}
if let Some(auth_recovery_mode) = tags.auth_recovery_mode {
fields.insert(
"auth_recovery_mode".to_string(),
auth_recovery_mode.to_string(),
);
}
if let Some(auth_recovery_phase) = tags.auth_recovery_phase {
fields.insert(
"auth_recovery_phase".to_string(),
auth_recovery_phase.to_string(),
);
}
if let Some(auth_connection_reused) = tags.auth_connection_reused {
fields.insert(
"auth_connection_reused".to_string(),
auth_connection_reused.to_string(),
);
}
if let Some(auth_request_id) = tags.auth_request_id {
fields.insert("auth_request_id".to_string(), auth_request_id.to_string());
}
if let Some(auth_cf_ray) = tags.auth_cf_ray {
fields.insert("auth_cf_ray".to_string(), auth_cf_ray.to_string());
}
if let Some(auth_error_code) = tags.auth_error_code {
fields.insert("auth_error_code".to_string(), auth_error_code.to_string());
}
if let Some(auth_recovery_followup_success) = tags.auth_recovery_followup_success {
fields.insert(
"auth_recovery_followup_success".to_string(),
auth_recovery_followup_success.to_string(),
);
}
if let Some(auth_recovery_followup_status) = tags.auth_recovery_followup_status {
fields.insert(
"auth_recovery_followup_status".to_string(),
auth_recovery_followup_status.to_string(),
);
}
if let Some(auth_env_provider_key_name) = auth_env.provider_env_key_name.as_deref() {
fields.insert(
"auth_env_provider_key_name".to_string(),
auth_env_provider_key_name.to_string(),
);
}
if let Some(auth_env_provider_key_present) = auth_env.provider_env_key_present {
fields.insert(
"auth_env_provider_key_present".to_string(),
auth_env_provider_key_present.to_string(),
);
}
let _ = crate::auth::report_auth_failure(fields);
}
pub fn backoff(attempt: u64) -> Duration {
codex-rs/core/src/util_tests.rs
+157 -24
View File
@@ -1,5 +1,6 @@
use super::*;
use crate::auth_env_telemetry::AuthEnvTelemetry;
use serial_test::serial;
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::Mutex;
@@ -64,6 +65,26 @@ where
}
}
type AuthFailureReportCollector = Arc<Mutex<Vec<BTreeMap<String, String>>>>;
fn install_auth_failure_report_collector() -> (
AuthFailureReportCollector,
crate::auth::AuthFailureReporterGuard,
) {
let reported = Arc::new(Mutex::new(Vec::new()));
let guard = crate::auth::set_auth_failure_reporter({
let reported = Arc::clone(&reported);
Arc::new(move |fields| {
reported
.lock()
.expect("report collector poisoned")
.push(fields);
true
})
});
(reported, guard)
}
#[test]
fn emit_feedback_request_tags_records_sentry_feedback_fields() {
let tags = Arc::new(Mutex::new(BTreeMap::new()));
@@ -313,16 +334,9 @@ fn emit_feedback_request_tags_preserves_latest_auth_fields_after_unauthorized()
}
#[test]
fn emit_sentry_auth_failure_event_records_non_feedback_fields() {
let tags = Arc::new(Mutex::new(BTreeMap::new()));
let event_count = Arc::new(Mutex::new(0));
let _guard = tracing_subscriber::registry()
.with(TagCollectorLayer {
target: SENTRY_AUTH_FAILURES_TARGET,
tags: tags.clone(),
event_count: event_count.clone(),
})
.set_default();
#[serial(auth_failure_reporter)]
fn emit_sentry_auth_failure_event_keeps_post_refresh_retry_reports_when_override_present() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let auth_env = AuthEnvTelemetry {
openai_api_key_env_present: true,
@@ -341,7 +355,7 @@ fn emit_sentry_auth_failure_event_records_non_feedback_fields() {
auth_mode: Some("chatgpt"),
auth_retry_after_unauthorized: Some(false),
auth_recovery_mode: Some("managed"),
auth_recovery_phase: Some("reload"),
auth_recovery_phase: Some("refresh_token"),
auth_connection_reused: Some(false),
auth_request_id: Some("req-123"),
auth_cf_ray: Some("ray-123"),
@@ -353,26 +367,145 @@ fn emit_sentry_auth_failure_event_records_non_feedback_fields() {
&auth_env,
);
let tags = tags.lock().unwrap().clone();
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(
tags.get("report_kind").map(String::as_str),
Some("auth_failure_auto")
);
assert_eq!(tags.get("endpoint").map(String::as_str), Some("/responses"));
assert_eq!(
tags.get("auth_header_attached").map(String::as_str),
Some("true")
reported[0].get("endpoint").map(String::as_str),
Some("/responses")
);
assert_eq!(
tags.get("auth_header_name").map(String::as_str),
Some("authorization")
reported[0].get("auth_recovery_phase").map(String::as_str),
Some("refresh_token")
);
assert_eq!(
tags.get("auth_request_id").map(String::as_str),
reported[0].get("auth_request_id").map(String::as_str),
Some("req-123")
);
assert_eq!(tags.get("auth_cf_ray").map(String::as_str), Some("ray-123"));
assert_eq!(*event_count.lock().unwrap(), 1);
}
#[test]
#[serial(auth_failure_reporter)]
fn emit_sentry_auth_failure_event_keeps_non_refresh_reports_when_override_present() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
let auth_env = AuthEnvTelemetry {
openai_api_key_env_present: true,
codex_api_key_env_present: false,
codex_api_key_env_enabled: true,
provider_env_key_name: Some("configured".to_string()),
provider_env_key_present: Some(true),
refresh_token_url_override_present: true,
};
emit_sentry_auth_failure_event_with_auth_env(
&FeedbackRequestTags {
endpoint: "/responses/compact",
auth_header_attached: true,
auth_header_name: Some("authorization"),
auth_mode: Some("openai_api_key"),
auth_retry_after_unauthorized: Some(false),
auth_recovery_mode: None,
auth_recovery_phase: None,
auth_connection_reused: None,
auth_request_id: Some("req-non-refresh"),
auth_cf_ray: Some("ray-non-refresh"),
auth_error: Some("missing_authorization_header"),
auth_error_code: Some("invalid_api_key"),
auth_recovery_followup_success: None,
auth_recovery_followup_status: None,
},
&auth_env,
);
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(
reported[0].get("endpoint").map(String::as_str),
Some("/responses/compact")
);
assert_eq!(
reported[0].get("auth_request_id").map(String::as_str),
Some("req-non-refresh")
);
}
#[test]
#[serial(auth_failure_reporter)]
fn emit_sentry_auth_failure_event_omits_missing_followup_metadata() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
emit_sentry_auth_failure_event_with_auth_env(
&FeedbackRequestTags {
endpoint: "/responses/compact",
auth_header_attached: true,
auth_header_name: Some("authorization"),
auth_mode: Some("chatgpt"),
auth_retry_after_unauthorized: Some(false),
auth_recovery_mode: None,
auth_recovery_phase: None,
auth_connection_reused: None,
auth_request_id: Some("req-no-followup"),
auth_cf_ray: Some("ray-no-followup"),
auth_error: Some("plain body"),
auth_error_code: Some("token_expired"),
auth_recovery_followup_success: None,
auth_recovery_followup_status: None,
},
&AuthEnvTelemetry {
openai_api_key_env_present: false,
codex_api_key_env_present: false,
codex_api_key_env_enabled: true,
provider_env_key_name: None,
provider_env_key_present: None,
refresh_token_url_override_present: false,
},
);
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(reported[0].get("auth_recovery_followup_success"), None);
assert_eq!(reported[0].get("auth_recovery_followup_status"), None);
}
#[test]
#[serial(auth_failure_reporter)]
fn emit_sentry_auth_failure_event_uses_direct_reporter_when_available() {
let (reported, _reporter_guard) = install_auth_failure_report_collector();
emit_sentry_auth_failure_event_with_auth_env(
&FeedbackRequestTags {
endpoint: "/responses",
auth_header_attached: true,
auth_header_name: Some("authorization"),
auth_mode: Some("chatgpt"),
auth_retry_after_unauthorized: Some(true),
auth_recovery_mode: Some("managed"),
auth_recovery_phase: Some("refresh_token"),
auth_connection_reused: Some(false),
auth_request_id: Some("req-direct"),
auth_cf_ray: Some("ray-direct"),
auth_error: Some("plain body"),
auth_error_code: Some("token_expired"),
auth_recovery_followup_success: Some(false),
auth_recovery_followup_status: Some(401),
},
&AuthEnvTelemetry {
openai_api_key_env_present: false,
codex_api_key_env_present: false,
codex_api_key_env_enabled: true,
provider_env_key_name: Some("OPENAI_API_KEY".to_string()),
provider_env_key_present: Some(false),
refresh_token_url_override_present: false,
},
);
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(
reported[0].get("auth_request_id").map(String::as_str),
Some("req-direct")
);
assert_eq!(reported[0].get("auth_error").map(String::as_str), None);
}
#[test]
codex-rs/exec/src/lib.rs
+1 -18
View File
@@ -66,7 +66,6 @@ use codex_core::config_loader::format_config_error_with_source;
use codex_core::format_exec_policy_error_with_source;
use codex_core::path_utils;
use codex_feedback::CodexFeedback;
use codex_feedback::upload_auth_failure_event_tags;
use codex_git_utils::get_git_repo_root;
use codex_otel::set_parent_from_context;
use codex_otel::traceparent_context_from_env;
@@ -91,7 +90,6 @@ use std::io::IsTerminal;
use std::io::Read;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use supports_color::Stream;
use tokio::sync::mpsc;
use tracing::Instrument;
@@ -258,14 +256,6 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
}
};
let feedback = CodexFeedback::new();
let _auth_failure_reporter_guard =
codex_core::auth::set_auth_failure_reporter(Arc::new(|tags| {
if let Err(err) = upload_auth_failure_event_tags(tags) {
tracing::warn!(error = %err, "failed to upload auth failure event");
}
}));
#[allow(clippy::print_stderr)]
let config_toml = match load_config_as_toml_with_cli_overrides(
&codex_home,
@@ -420,15 +410,8 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
let otel_tracing_layer = otel.as_ref().and_then(|o| o.tracing_layer());
let feedback_layer = feedback.logger_layer();
let feedback_metadata_layer = feedback.metadata_layer();
let feedback_auth_event_layer = feedback.auth_event_layer();
let _ = tracing_subscriber::registry()
.with(fmt_layer)
.with(feedback_layer)
.with(feedback_metadata_layer)
.with(feedback_auth_event_layer)
.with(otel_tracing_layer)
.with(otel_logger_layer)
.try_init();
@@ -453,7 +436,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
cli_overrides: run_cli_overrides,
loader_overrides: run_loader_overrides,
cloud_requirements: run_cloud_requirements,
feedback,
feedback: CodexFeedback::new(),
config_warnings,
session_source: SessionSource::Exec,
enable_codex_api_key_env: true,
codex-rs/feedback/src/lib.rs
+320 -55
View File
@@ -7,6 +7,7 @@ use std::io::{self};
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
@@ -31,9 +32,22 @@ const SENTRY_DSN: &str =
const SENTRY_DSN_OVERRIDE_ENV_VAR: &str = "CODEX_SENTRY_DSN_OVERRIDE";
const UPLOAD_TIMEOUT_SECS: u64 = 10;
const FEEDBACK_TAGS_TARGET: &str = "feedback_tags";
const SENTRY_AUTH_FAILURES_TARGET: &str = "sentry_auth_failures";
const AUTH_FAILURE_REPORT_KIND: &str = "auth_failure_auto";
const MAX_FEEDBACK_TAGS: usize = 64;
const AUTH_FAILURE_UPLOAD_QUEUE_CAPACITY: usize = 32;
const FALLBACK_AUTH_FAILURE_UPLOAD_QUEUE_CAPACITY: usize = 32;
#[derive(Clone)]
struct AuthFailureUploadTask {
tags: BTreeMap<String, String>,
dsn_override: Option<String>,
}
static AUTH_FAILURE_UPLOAD_QUEUE: OnceLock<std::sync::mpsc::SyncSender<AuthFailureUploadTask>> =
OnceLock::new();
static FALLBACK_AUTH_FAILURE_UPLOAD_QUEUE: OnceLock<
std::sync::mpsc::SyncSender<AuthFailureUploadTask>,
> = OnceLock::new();
#[derive(Clone)]
pub struct CodexFeedback {
@@ -96,18 +110,6 @@ impl CodexFeedback {
.with_filter(Targets::new().with_target(FEEDBACK_TAGS_TARGET, Level::TRACE))
}
/// Returns a [`tracing_subscriber`] layer that uploads lightweight auth-failure Sentry events.
///
/// Events with `target: "sentry_auth_failures"` are converted into searchable Sentry issues
/// without relying on a manual `/feedback` upload.
pub fn auth_event_layer<S>(&self) -> impl Layer<S> + Send + Sync + 'static
where
S: tracing::Subscriber + for<'a> LookupSpan<'a>,
{
AuthFailureSentryLayer
.with_filter(Targets::new().with_target(SENTRY_AUTH_FAILURES_TARGET, Level::TRACE))
}
pub fn snapshot(&self, session_id: Option<ThreadId>) -> FeedbackSnapshot {
let bytes = {
let guard = self.inner.ring.lock().expect("mutex poisoned");
@@ -227,6 +229,15 @@ pub struct FeedbackSnapshot {
pub thread_id: String,
}
struct FeedbackUploadParams<'a> {
classification: &'a str,
reason: Option<&'a str>,
include_logs: bool,
extra_attachment_paths: &'a [PathBuf],
session_source: Option<SessionSource>,
logs_override: Option<Vec<u8>>,
}
impl FeedbackSnapshot {
pub(crate) fn as_bytes(&self) -> &[u8] {
&self.bytes
@@ -266,6 +277,24 @@ impl FeedbackSnapshot {
extra_attachment_paths: &[PathBuf],
session_source: Option<SessionSource>,
logs_override: Option<Vec<u8>>,
) -> Result<()> {
self.upload_feedback_with_dsn_override(
FeedbackUploadParams {
classification,
reason,
include_logs,
extra_attachment_paths,
session_source,
logs_override,
},
std::env::var(SENTRY_DSN_OVERRIDE_ENV_VAR).ok().as_deref(),
)
}
fn upload_feedback_with_dsn_override(
&self,
params: FeedbackUploadParams<'_>,
dsn_override: Option<&str>,
) -> Result<()> {
use sentry::protocol::Envelope;
use sentry::protocol::EnvelopeItem;
@@ -273,18 +302,21 @@ impl FeedbackSnapshot {
use sentry::protocol::Level;
use std::collections::BTreeMap;
let client = build_sentry_client()?;
let client = build_sentry_client_with_dsn_override(dsn_override)?;
let cli_version = env!("CARGO_PKG_VERSION");
let mut tags = BTreeMap::from([
(String::from("thread_id"), self.thread_id.to_string()),
(String::from("classification"), classification.to_string()),
(
String::from("classification"),
params.classification.to_string(),
),
(String::from("cli_version"), cli_version.to_string()),
]);
if let Some(source) = session_source.as_ref() {
if let Some(source) = params.session_source.as_ref() {
tags.insert(String::from("session_source"), source.to_string());
}
if let Some(r) = reason {
if let Some(r) = params.reason {
tags.insert(String::from("reason"), r.to_string());
}
@@ -304,7 +336,7 @@ impl FeedbackSnapshot {
}
}
let level = match classification {
let level = match params.classification {
"bug" | "bad_result" | "safety_check" => Level::Error,
_ => Level::Info,
};
@@ -312,7 +344,7 @@ impl FeedbackSnapshot {
let mut envelope = Envelope::new();
let title = format!(
"[{}]: Codex session {}",
display_classification(classification),
display_classification(params.classification),
self.thread_id
);
@@ -322,7 +354,7 @@ impl FeedbackSnapshot {
tags,
..Default::default()
};
if let Some(r) = reason {
if let Some(r) = params.reason {
use sentry::protocol::Exception;
use sentry::protocol::Values;
@@ -334,9 +366,11 @@ impl FeedbackSnapshot {
}
envelope.add_item(EnvelopeItem::Event(event));
for attachment in
self.feedback_attachments(include_logs, extra_attachment_paths, logs_override)
{
for attachment in self.feedback_attachments(
params.include_logs,
params.extra_attachment_paths,
params.logs_override,
) {
envelope.add_item(EnvelopeItem::Attachment(attachment));
}
@@ -401,12 +435,6 @@ impl FeedbackSnapshot {
}
}
fn build_sentry_client() -> Result<sentry::Client> {
build_sentry_client_with_dsn_override(
std::env::var(SENTRY_DSN_OVERRIDE_ENV_VAR).ok().as_deref(),
)
}
fn build_sentry_client_with_dsn_override(dsn_override: Option<&str>) -> Result<sentry::Client> {
use std::str::FromStr;
use std::sync::Arc;
@@ -466,30 +494,6 @@ where
}
}
#[derive(Clone, Copy)]
struct AuthFailureSentryLayer;
impl<S> Layer<S> for AuthFailureSentryLayer
where
S: tracing::Subscriber + for<'a> LookupSpan<'a>,
{
fn on_event(&self, event: &Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) {
if event.metadata().target() != SENTRY_AUTH_FAILURES_TARGET {
return;
}
let mut visitor = FeedbackTagsVisitor::default();
event.record(&mut visitor);
if visitor.tags.is_empty() {
return;
}
if let Err(err) = upload_auth_failure_event_tags(visitor.tags) {
tracing::warn!(error = %err, "failed to upload auth failure event");
}
}
}
#[derive(Default)]
struct FeedbackTagsVisitor {
tags: BTreeMap<String, String>,
@@ -544,7 +548,6 @@ fn auth_failure_grouping_key(tags: &BTreeMap<String, String>) -> Vec<String> {
let error_code = tags
.get("auth_error_code")
.filter(|value| !value.is_empty())
.or_else(|| tags.get("auth_error").filter(|value| !value.is_empty()))
.cloned()
.unwrap_or_else(|| "unknown".to_string());
let auth_header_attached = tags
@@ -590,6 +593,87 @@ pub fn upload_auth_failure_event_tags(tags: BTreeMap<String, String>) -> Result<
)
}
pub fn enqueue_auth_failure_event_tags(tags: BTreeMap<String, String>) -> bool {
let task = AuthFailureUploadTask {
tags: finalize_auth_failure_tags(tags),
dsn_override: std::env::var(SENTRY_DSN_OVERRIDE_ENV_VAR).ok(),
};
enqueue_auth_failure_upload(auth_failure_upload_queue_sender(), "primary", task.clone())
|| enqueue_auth_failure_upload(
fallback_auth_failure_upload_queue_sender(),
"fallback",
task,
)
}
#[cfg(test)]
fn enqueue_auth_failure_event_with_dsn_override(
tags: BTreeMap<String, String>,
dsn_override: Option<String>,
) -> bool {
enqueue_auth_failure_upload(
auth_failure_upload_queue_sender(),
"primary",
AuthFailureUploadTask {
tags: finalize_auth_failure_tags(tags),
dsn_override,
},
)
}
fn enqueue_auth_failure_upload(
sender: &std::sync::mpsc::SyncSender<AuthFailureUploadTask>,
lane: &'static str,
task: AuthFailureUploadTask,
) -> bool {
match sender.try_send(task) {
Ok(()) => true,
Err(err) => {
tracing::warn!(error = %err, lane, "failed to enqueue auth failure event");
false
}
}
}
fn auth_failure_upload_queue_sender() -> &'static std::sync::mpsc::SyncSender<AuthFailureUploadTask>
{
auth_failure_upload_queue_sender_for(
&AUTH_FAILURE_UPLOAD_QUEUE,
AUTH_FAILURE_UPLOAD_QUEUE_CAPACITY,
"primary",
)
}
fn fallback_auth_failure_upload_queue_sender()
-> &'static std::sync::mpsc::SyncSender<AuthFailureUploadTask> {
auth_failure_upload_queue_sender_for(
&FALLBACK_AUTH_FAILURE_UPLOAD_QUEUE,
FALLBACK_AUTH_FAILURE_UPLOAD_QUEUE_CAPACITY,
"fallback",
)
}
fn auth_failure_upload_queue_sender_for(
slot: &'static OnceLock<std::sync::mpsc::SyncSender<AuthFailureUploadTask>>,
capacity: usize,
lane: &'static str,
) -> &'static std::sync::mpsc::SyncSender<AuthFailureUploadTask> {
slot.get_or_init(|| {
let (tx, rx) = std::sync::mpsc::sync_channel::<AuthFailureUploadTask>(capacity);
std::thread::spawn(move || {
while let Ok(task) = rx.recv() {
if let Err(err) = upload_auth_failure_event_with_dsn_override(
task.tags,
task.dsn_override.as_deref(),
) {
tracing::warn!(error = %err, lane, "failed to upload auth failure event");
}
}
});
tx
})
}
fn upload_auth_failure_event_with_dsn_override(
tags: BTreeMap<String, String>,
dsn_override: Option<&str>,
@@ -601,10 +685,20 @@ fn upload_auth_failure_event_with_dsn_override(
let mut envelope = Envelope::new();
envelope.add_item(EnvelopeItem::Event(build_auth_failure_event(tags)));
client.send_envelope(envelope);
client.flush(Some(Duration::from_secs(UPLOAD_TIMEOUT_SECS)));
ensure_auth_failure_event_flushed(
client.flush(Some(Duration::from_secs(UPLOAD_TIMEOUT_SECS))),
)?;
Ok(())
}
fn ensure_auth_failure_event_flushed(flushed: bool) -> Result<()> {
if flushed {
Ok(())
} else {
Err(anyhow!("timed out flushing auth failure event"))
}
}
#[cfg(test)]
mod tests {
use std::ffi::OsStr;
@@ -687,6 +781,29 @@ mod tests {
);
}
#[test]
fn auth_failure_grouping_key_does_not_use_raw_error_text() {
let grouping_key = auth_failure_grouping_key(&BTreeMap::from([
(String::from("endpoint"), String::from("/responses")),
(
String::from("auth_error"),
String::from("request-specific plaintext body"),
),
(String::from("auth_header_attached"), String::from("true")),
]));
assert_eq!(
grouping_key,
vec![
"codex".to_string(),
AUTH_FAILURE_REPORT_KIND.to_string(),
"/responses".to_string(),
"unknown".to_string(),
"true".to_string(),
]
);
}
#[test]
fn build_auth_failure_event_sets_stable_message_and_tags() {
let event = build_auth_failure_event(BTreeMap::from([
@@ -781,6 +898,154 @@ mod tests {
assert!(request_text.contains("\"auth_error_code\":\"refresh_token_reused\""));
}
#[test]
fn enqueue_auth_failure_event_tags_posts_envelope_to_overridden_dsn() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind local listener");
listener
.set_nonblocking(false)
.expect("listener should stay blocking");
let addr = listener.local_addr().expect("local addr");
let (tx, rx) = mpsc::channel();
let server = thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept envelope request");
let mut buffer = Vec::new();
let mut headers_end = None;
while headers_end.is_none() {
let mut chunk = [0_u8; 4096];
let read = stream.read(&mut chunk).expect("read request headers");
buffer.extend_from_slice(&chunk[..read]);
headers_end = buffer.windows(4).position(|window| window == b"\r\n\r\n");
}
let headers_end = headers_end.expect("headers terminator should exist") + 4;
let headers = String::from_utf8_lossy(&buffer[..headers_end]);
let content_length = headers
.lines()
.find_map(|line| {
let (name, value) = line.split_once(':')?;
if name.eq_ignore_ascii_case("content-length") {
Some(value.trim().parse::<usize>().expect("content-length"))
} else {
None
}
})
.expect("content-length header");
while buffer.len() < headers_end + content_length {
let mut chunk = [0_u8; 4096];
let read = stream.read(&mut chunk).expect("read request body");
buffer.extend_from_slice(&chunk[..read]);
}
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
.expect("write response");
tx.send(buffer).expect("capture request");
});
enqueue_auth_failure_event_with_dsn_override(
BTreeMap::from([
(String::from("endpoint"), String::from("/oauth/token")),
(
String::from("auth_error_code"),
String::from("refresh_token_reused"),
),
(String::from("auth_header_attached"), String::from("false")),
]),
Some(format!("http://public@127.0.0.1:{}/1", addr.port())),
);
let request = rx
.recv_timeout(StdDuration::from_secs(5))
.expect("receive envelope request");
server.join().expect("server thread should exit");
let request_text = String::from_utf8_lossy(&request);
assert!(request_text.contains("POST /api/1/envelope/"));
assert!(request_text.contains("\"report_kind\":\"auth_failure_auto\""));
assert!(request_text.contains("\"endpoint\":\"/oauth/token\""));
assert!(request_text.contains("\"auth_header_attached\":\"false\""));
}
#[test]
fn upload_feedback_posts_envelope_to_overridden_dsn() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind local listener");
listener
.set_nonblocking(false)
.expect("listener should stay blocking");
let addr = listener.local_addr().expect("local addr");
let (tx, rx) = mpsc::channel();
let server = thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept envelope request");
let mut buffer = Vec::new();
let mut headers_end = None;
while headers_end.is_none() {
let mut chunk = [0_u8; 4096];
let read = stream.read(&mut chunk).expect("read request headers");
buffer.extend_from_slice(&chunk[..read]);
headers_end = buffer.windows(4).position(|window| window == b"\r\n\r\n");
}
let headers_end = headers_end.expect("headers terminator should exist") + 4;
let headers = String::from_utf8_lossy(&buffer[..headers_end]);
let content_length = headers
.lines()
.find_map(|line| {
let (name, value) = line.split_once(':')?;
if name.eq_ignore_ascii_case("content-length") {
Some(value.trim().parse::<usize>().expect("content-length"))
} else {
None
}
})
.expect("content-length header");
while buffer.len() < headers_end + content_length {
let mut chunk = [0_u8; 4096];
let read = stream.read(&mut chunk).expect("read request body");
buffer.extend_from_slice(&chunk[..read]);
}
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
.expect("write response");
tx.send(buffer).expect("capture request");
});
let snapshot = CodexFeedback::new().snapshot(/*session_id*/ None);
let dsn = format!("http://public@127.0.0.1:{}/1", addr.port());
snapshot
.upload_feedback_with_dsn_override(
FeedbackUploadParams {
classification: "idea",
reason: Some("manual"),
include_logs: false,
extra_attachment_paths: &[],
session_source: None,
logs_override: None,
},
Some(&dsn),
)
.expect("upload feedback");
let request = rx
.recv_timeout(StdDuration::from_secs(5))
.expect("receive envelope request");
server.join().expect("server thread should exit");
let request_text = String::from_utf8_lossy(&request);
assert!(request_text.contains("POST /api/1/envelope/"));
assert!(request_text.contains("\"classification\":\"idea\""));
assert!(request_text.contains("\"reason\":\"manual\""));
}
#[test]
fn ensure_auth_failure_event_flushed_rejects_timeouts() {
assert!(ensure_auth_failure_event_flushed(/*flushed*/ true).is_ok());
assert_eq!(
ensure_auth_failure_event_flushed(/*flushed*/ false)
.expect_err("flush timeout should fail")
.to_string(),
"timed out flushing auth failure event"
);
}
#[test]
fn feedback_attachments_gate_connectivity_diagnostics() {
let extra_filename = format!("codex-feedback-extra-{}.jsonl", ThreadId::new());
codex-rs/login/src/auth/auth_tests.rs
+50 -119
View File
@@ -10,20 +10,14 @@ use base64::Engine;
use codex_protocol::config_types::ForcedLoginMethod;
use codex_protocol::config_types::ModelProviderAuthInfo;
use pretty_assertions::assert_eq;
use reqwest::header::HeaderMap;
use reqwest::header::HeaderValue;
use serde::Serialize;
use serde_json::json;
use serial_test::serial;
use std::collections::BTreeMap;
use std::sync::Arc;
use tempfile::TempDir;
use std::sync::Mutex;
use tempfile::TempDir;
use tempfile::tempdir;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
#[tokio::test]
async fn refresh_without_id_token() {
@@ -800,125 +794,62 @@ fn missing_plan_type_maps_to_unknown() {
}
#[test]
fn refresh_token_reused_builds_sentry_auth_failure_fields() {
let mut headers = HeaderMap::new();
headers.insert("x-request-id", HeaderValue::from_static("req_refresh_123"));
headers.insert("cf-ray", HeaderValue::from_static("abc123-SJC"));
assert_eq!(
refresh_failure_sentry_auth_fields(
&headers,
r#"{"error":{"code":"refresh_token_reused","message":"refresh token already used"}}"#
),
BTreeMap::from([
("report_kind".to_string(), "auth_failure_auto".to_string()),
("endpoint".to_string(), "/oauth/token".to_string()),
("auth_header_attached".to_string(), "true".to_string()),
("auth_header_name".to_string(), "authorization".to_string()),
("auth_mode".to_string(), "Chatgpt".to_string()),
("auth_request_id".to_string(), "req_refresh_123".to_string()),
("auth_cf_ray".to_string(), "abc123-SJC".to_string()),
(
"auth_error".to_string(),
"refresh token already used".to_string()
),
(
"auth_error_code".to_string(),
"refresh_token_reused".to_string()
),
(
"cli_version".to_string(),
env!("CARGO_PKG_VERSION").to_string()
),
])
);
}
#[tokio::test]
#[serial(codex_api_key)]
async fn refresh_token_reused_reports_fields_through_auth_failure_reporter() {
let codex_home = tempdir().unwrap();
write_auth_file(
AuthFileParams {
openai_api_key: None,
chatgpt_plan_type: Some("pro".to_string()),
chatgpt_account_id: Some("org_mine".to_string()),
},
codex_home.path(),
)
.expect("failed to write auth file");
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/oauth/token"))
.respond_with(
ResponseTemplate::new(401)
.insert_header("x-request-id", "req_refresh_123")
.insert_header("cf-ray", "abc123-SJC")
.set_body_json(json!({
"error": {
"code": "refresh_token_reused",
"message": "refresh token already used"
}
})),
)
.mount(&server)
.await;
let _refresh_url = EnvVarGuard::set(
REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR,
&format!("{}/oauth/token", server.uri()),
);
let reported = Arc::new(Mutex::new(Vec::new()));
let _reporter_guard = set_auth_failure_reporter({
let reported = Arc::clone(&reported);
#[serial(auth_failure_reporter)]
fn nested_auth_failure_reporter_guard_restores_previous_reporter() {
let outer_reported = Arc::new(Mutex::new(Vec::new()));
let inner_reported = Arc::new(Mutex::new(Vec::new()));
let outer_guard = set_auth_failure_reporter({
let outer_reported = Arc::clone(&outer_reported);
Arc::new(move |fields| {
reported
outer_reported
.lock()
.expect("report collector poisoned")
.expect("outer collector poisoned")
.push(fields);
true
})
});
let manager = AuthManager::shared(
codex_home.path().to_path_buf(),
false,
AuthCredentialsStoreMode::File,
);
{
let _inner_guard = set_auth_failure_reporter({
let inner_reported = Arc::clone(&inner_reported);
Arc::new(move |fields| {
inner_reported
.lock()
.expect("inner collector poisoned")
.push(fields);
true
})
});
assert!(report_auth_failure(BTreeMap::from([(
"endpoint".to_string(),
"/responses".to_string(),
)])));
}
let err = manager
.refresh_token_from_authority()
.await
.expect_err("refresh should fail permanently");
assert_eq!(
err.failed_reason(),
Some(RefreshTokenFailedReason::Exhausted)
);
assert!(report_auth_failure(BTreeMap::from([(
"endpoint".to_string(),
"/models".to_string(),
)])));
drop(outer_guard);
let reported = reported.lock().expect("report collector poisoned");
assert_eq!(reported.len(), 1);
assert_eq!(
reported[0],
BTreeMap::from([
("report_kind".to_string(), "auth_failure_auto".to_string()),
("endpoint".to_string(), "/oauth/token".to_string()),
("auth_header_attached".to_string(), "true".to_string()),
("auth_header_name".to_string(), "authorization".to_string()),
("auth_mode".to_string(), "Chatgpt".to_string()),
("auth_request_id".to_string(), "req_refresh_123".to_string()),
("auth_cf_ray".to_string(), "abc123-SJC".to_string()),
(
"auth_error".to_string(),
"refresh token already used".to_string()
),
(
"auth_error_code".to_string(),
"refresh_token_reused".to_string()
),
(
"cli_version".to_string(),
env!("CARGO_PKG_VERSION").to_string()
),
])
inner_reported
.lock()
.expect("inner collector poisoned")
.clone(),
vec![BTreeMap::from([(
"endpoint".to_string(),
"/responses".to_string(),
)])]
);
assert_eq!(
outer_reported
.lock()
.expect("outer collector poisoned")
.clone(),
vec![BTreeMap::from([(
"endpoint".to_string(),
"/models".to_string(),
)])]
);
}
codex-rs/login/src/auth/manager.rs
+52 -71
View File
@@ -2,11 +2,10 @@ use async_trait::async_trait;
use chrono::Utc;
use once_cell::sync::Lazy;
use reqwest::StatusCode;
use reqwest::header::HeaderMap;
use serde::Deserialize;
use serde::Serialize;
#[cfg(test)]
use serial_test::serial;
use std::cell::Cell;
use std::collections::BTreeMap;
use std::env;
use std::fmt::Debug;
@@ -86,10 +85,15 @@ const REFRESH_TOKEN_UNKNOWN_MESSAGE: &str =
const REFRESH_TOKEN_ACCOUNT_MISMATCH_MESSAGE: &str = "Your access token could not be refreshed because you have since logged out or signed in to another account. Please sign in again.";
const REFRESH_TOKEN_URL: &str = "https://auth.openai.com/oauth/token";
pub const REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR: &str = "CODEX_REFRESH_TOKEN_URL_OVERRIDE";
const SENTRY_AUTH_FAILURES_TARGET: &str = "sentry_auth_failures";
static AUTH_FAILURE_REPORTER: Lazy<
RwLock<Option<Arc<dyn Fn(BTreeMap<String, String>) + Send + Sync + 'static>>>,
> = Lazy::new(|| RwLock::new(None));
type AuthFailureReporter = Arc<dyn Fn(BTreeMap<String, String>) -> bool + Send + Sync + 'static>;
static AUTH_FAILURE_REPORTER: Lazy<RwLock<Option<AuthFailureReporter>>> =
Lazy::new(|| RwLock::new(None));
#[cfg(test)]
static AUTH_FAILURE_REPORTER_TEST_LOCK: Lazy<Mutex<()>> = Lazy::new(|| Mutex::new(()));
#[cfg(test)]
thread_local! {
static AUTH_FAILURE_REPORTER_TEST_DEPTH: Cell<u32> = const { Cell::new(0) };
}
#[derive(Debug, Error)]
pub enum RefreshTokenError {
@@ -99,23 +103,58 @@ pub enum RefreshTokenError {
Transient(#[from] std::io::Error),
}
pub struct AuthFailureReporterGuard;
pub struct AuthFailureReporterGuard {
previous: Option<AuthFailureReporter>,
#[cfg(test)]
_test_lock: Option<std::sync::MutexGuard<'static, ()>>,
}
impl Drop for AuthFailureReporterGuard {
fn drop(&mut self) {
if let Ok(mut reporter) = AUTH_FAILURE_REPORTER.write() {
*reporter = None;
*reporter = self.previous.take();
}
#[cfg(test)]
AUTH_FAILURE_REPORTER_TEST_DEPTH.with(|depth| {
depth.set(depth.get().saturating_sub(1));
});
}
}
pub fn set_auth_failure_reporter(
reporter: Arc<dyn Fn(BTreeMap<String, String>) + Send + Sync + 'static>,
) -> AuthFailureReporterGuard {
pub fn set_auth_failure_reporter(reporter: AuthFailureReporter) -> AuthFailureReporterGuard {
#[cfg(test)]
let test_lock = AUTH_FAILURE_REPORTER_TEST_DEPTH.with(|depth| {
if depth.get() == 0 {
depth.set(1);
Some(
AUTH_FAILURE_REPORTER_TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
)
} else {
depth.set(depth.get() + 1);
None
}
});
let mut previous = None;
if let Ok(mut current) = AUTH_FAILURE_REPORTER.write() {
*current = Some(reporter);
previous = current.replace(reporter);
}
AuthFailureReporterGuard {
previous,
#[cfg(test)]
_test_lock: test_lock,
}
}
pub fn report_auth_failure(fields: BTreeMap<String, String>) -> bool {
if let Ok(reporter) = AUTH_FAILURE_REPORTER.read()
&& let Some(reporter) = reporter.as_ref()
{
reporter(fields)
} else {
false
}
AuthFailureReporterGuard
}
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -707,11 +746,9 @@ async fn request_chatgpt_token_refresh(
.map_err(|err| RefreshTokenError::Transient(std::io::Error::other(err)))?;
Ok(refresh_response)
} else {
let headers = response.headers().clone();
let body = response.text().await.unwrap_or_default();
tracing::error!("Failed to refresh token: {status}: {body}");
if status == StatusCode::UNAUTHORIZED {
emit_sentry_auth_failure_event_for_refresh_failure(&headers, &body);
let failed = classify_refresh_token_failure(&body);
Err(RefreshTokenError::Permanent(failed))
} else {
@@ -723,62 +760,6 @@ async fn request_chatgpt_token_refresh(
}
}
fn emit_sentry_auth_failure_event_for_refresh_failure(headers: &HeaderMap, body: &str) {
let fields = refresh_failure_sentry_auth_fields(headers, body);
if let Ok(reporter) = AUTH_FAILURE_REPORTER.read()
&& let Some(reporter) = reporter.as_ref()
{
reporter(fields);
return;
}
tracing::event!(
target: SENTRY_AUTH_FAILURES_TARGET,
tracing::Level::ERROR,
report_kind = fields.get("report_kind").map(String::as_str).unwrap_or(""),
endpoint = fields.get("endpoint").map(String::as_str).unwrap_or(""),
auth_header_attached = fields
.get("auth_header_attached")
.is_some_and(|value| value == "true"),
auth_header_name = fields.get("auth_header_name").map(String::as_str).unwrap_or(""),
auth_mode = fields.get("auth_mode").map(String::as_str).unwrap_or(""),
auth_request_id = fields.get("auth_request_id").map(String::as_str).unwrap_or(""),
auth_cf_ray = fields.get("auth_cf_ray").map(String::as_str).unwrap_or(""),
auth_error = fields.get("auth_error").map(String::as_str).unwrap_or(""),
auth_error_code = fields.get("auth_error_code").map(String::as_str).unwrap_or(""),
cli_version = fields.get("cli_version").map(String::as_str).unwrap_or(""),
);
}
fn refresh_failure_sentry_auth_fields(headers: &HeaderMap, body: &str) -> BTreeMap<String, String> {
let auth_request_id = headers
.get("x-request-id")
.and_then(|value| value.to_str().ok())
.unwrap_or("");
let auth_cf_ray = headers
.get("cf-ray")
.and_then(|value| value.to_str().ok())
.unwrap_or("");
let auth_error_code = extract_refresh_token_error_code(body).unwrap_or_default();
let auth_error = try_parse_error_message(body);
BTreeMap::from([
("report_kind".to_string(), "auth_failure_auto".to_string()),
("endpoint".to_string(), "/oauth/token".to_string()),
("auth_header_attached".to_string(), "true".to_string()),
("auth_header_name".to_string(), "authorization".to_string()),
("auth_mode".to_string(), "Chatgpt".to_string()),
("auth_request_id".to_string(), auth_request_id.to_string()),
("auth_cf_ray".to_string(), auth_cf_ray.to_string()),
("auth_error".to_string(), auth_error),
("auth_error_code".to_string(), auth_error_code),
(
"cli_version".to_string(),
env!("CARGO_PKG_VERSION").to_string(),
),
])
}
fn classify_refresh_token_failure(body: &str) -> RefreshTokenFailedError {
let code = extract_refresh_token_error_code(body);
codex-rs/login/src/lib.rs
+1
View File
@@ -38,6 +38,7 @@ pub use auth::load_auth_dot_json;
pub use auth::login_with_api_key;
pub use auth::logout;
pub use auth::read_openai_api_key_from_env;
pub use auth::report_auth_failure;
pub use auth::save_auth;
pub use auth::set_auth_failure_reporter;
pub use codex_app_server_protocol::AuthMode;
codex-rs/mcp-server/Cargo.toml
+1
View File
@@ -21,6 +21,7 @@ codex-arg0 = { workspace = true }
codex-core = { workspace = true }
codex-exec-server = { workspace = true }
codex-features = { workspace = true }
codex-feedback = { workspace = true }
codex-protocol = { workspace = true }
codex-shell-command = { workspace = true }
codex-utils-cli = { workspace = true }
codex-rs/mcp-server/src/lib.rs
+5
View File
@@ -8,6 +8,7 @@ use std::sync::Arc;
use codex_arg0::Arg0DispatchPaths;
use codex_core::config::Config;
use codex_exec_server::EnvironmentManager;
use codex_feedback::enqueue_auth_failure_event_tags;
use codex_utils_cli::CliConfigOverrides;
use rmcp::model::ClientNotification;
@@ -71,6 +72,10 @@ pub async fn run_main(
.map_err(|e| {
std::io::Error::new(ErrorKind::InvalidData, format!("error loading config: {e}"))
})?;
let feedback_enabled = config.feedback_enabled;
let _auth_failure_reporter_guard = feedback_enabled.then(|| {
codex_core::auth::set_auth_failure_reporter(Arc::new(enqueue_auth_failure_event_tags))
});
let otel = codex_core::otel_init::build_provider(
&config,
codex-rs/tui/src/app.rs
+86 -7
View File
@@ -978,6 +978,7 @@ pub(crate) struct App {
/// transcript cells.
pub(crate) backtrack_render_pending: bool,
pub(crate) feedback: codex_feedback::CodexFeedback,
auth_failure_reporter_guard: Option<codex_core::auth::AuthFailureReporterGuard>,
feedback_audience: FeedbackAudience,
remote_app_server_url: Option<String>,
remote_app_server_auth_token: Option<String>,
@@ -1052,6 +1053,25 @@ fn active_turn_missing_steer_error(error: &TypedRequestError) -> bool {
}
impl App {
fn set_config(&mut self, config: Config) {
self.config = config;
self.chat_widget.sync_runtime_config(&self.config);
self.reconcile_auth_failure_reporting();
}
fn reconcile_auth_failure_reporting(&mut self) {
if self.config.feedback_enabled {
if self.auth_failure_reporter_guard.is_none() {
self.auth_failure_reporter_guard =
Some(codex_core::auth::set_auth_failure_reporter(Arc::new(
codex_feedback::enqueue_auth_failure_event_tags,
)));
}
} else {
self.auth_failure_reporter_guard = None;
}
}
pub fn chatwidget_init_for_forked_or_resumed_thread(
&self,
tui: &mut tui::Tui,
@@ -1096,8 +1116,7 @@ impl App {
.rebuild_config_for_cwd(self.chat_widget.config_ref().cwd.to_path_buf())
.await?;
self.apply_runtime_policy_overrides(&mut config);
self.config = config;
self.chat_widget.sync_plugin_mentions_config(&self.config);
self.set_config(config);
Ok(())
}
@@ -1349,7 +1368,7 @@ impl App {
return;
}
self.config = next_config;
self.set_config(next_config);
for (feature, effective_enabled) in feature_updates_to_apply {
self.chat_widget
.set_feature_enabled(feature, effective_enabled);
@@ -3222,7 +3241,7 @@ impl App {
tracing::warn!("failed to unsubscribe tracked thread {thread_id}: {err}");
}
}
self.config = config.clone();
self.set_config(config.clone());
match app_server.start_thread(&config).await {
Ok(started) => {
if let Err(err) = self
@@ -3710,6 +3729,7 @@ impl App {
backtrack: BacktrackState::default(),
backtrack_render_pending: false,
feedback: feedback.clone(),
auth_failure_reporter_guard: None,
feedback_audience,
remote_app_server_url,
remote_app_server_auth_token,
@@ -3727,6 +3747,7 @@ impl App {
pending_primary_events: VecDeque::new(),
pending_app_server_requests: PendingAppServerRequests::default(),
};
app.reconcile_auth_failure_reporting();
if let Some(started) = initial_started_thread {
app.enqueue_primary_thread_session(started.session, started.turns)
.await?;
@@ -4034,7 +4055,7 @@ impl App {
{
Ok(resumed) => {
self.shutdown_current_thread(app_server).await;
self.config = resume_config;
self.set_config(resume_config);
tui.set_notification_method(self.config.tui_notification_method);
self.file_search
.update_search_dir(self.config.cwd.to_path_buf());
@@ -4960,7 +4981,7 @@ impl App {
) {
return Ok(AppRunControl::Continue);
}
self.config = config;
self.set_config(config);
self.runtime_approval_policy_override =
Some(self.config.permissions.approval_policy.value());
self.chat_widget
@@ -4984,7 +5005,7 @@ impl App {
) {
return Ok(AppRunControl::Continue);
}
self.config = config;
self.set_config(config);
if let Err(err) = self.chat_widget.set_sandbox_policy(policy_for_chat) {
tracing::warn!(%err, "failed to set sandbox policy on chat config");
self.chat_widget
@@ -8926,6 +8947,7 @@ guardian_approval = true
backtrack: BacktrackState::default(),
backtrack_render_pending: false,
feedback: codex_feedback::CodexFeedback::new(),
auth_failure_reporter_guard: None,
feedback_audience: FeedbackAudience::External,
remote_app_server_url: None,
remote_app_server_auth_token: None,
@@ -8980,6 +9002,7 @@ guardian_approval = true
backtrack: BacktrackState::default(),
backtrack_render_pending: false,
feedback: codex_feedback::CodexFeedback::new(),
auth_failure_reporter_guard: None,
feedback_audience: FeedbackAudience::External,
remote_app_server_url: None,
remote_app_server_auth_token: None,
@@ -9919,6 +9942,62 @@ guardian_approval = true
Ok(())
}
#[tokio::test]
async fn refresh_in_memory_config_from_disk_reconciles_auth_failure_reporting() -> Result<()> {
let mut app = make_test_app().await;
let codex_home = tempdir()?;
app.config.codex_home = codex_home.path().to_path_buf();
app.config.feedback_enabled = false;
app.reconcile_auth_failure_reporting();
assert!(app.auth_failure_reporter_guard.is_none());
ConfigEditsBuilder::new(&app.config.codex_home)
.with_edits([ConfigEdit::SetPath {
segments: vec!["feedback".to_string(), "enabled".to_string()],
value: true.into(),
}])
.apply()
.await
.expect("persist feedback opt-in");
app.refresh_in_memory_config_from_disk().await?;
assert!(app.config.feedback_enabled);
assert!(app.auth_failure_reporter_guard.is_some());
assert!(app.chat_widget.config_ref().feedback_enabled);
ConfigEditsBuilder::new(&app.config.codex_home)
.with_edits([ConfigEdit::SetPath {
segments: vec!["feedback".to_string(), "enabled".to_string()],
value: false.into(),
}])
.apply()
.await
.expect("persist feedback opt-out");
app.refresh_in_memory_config_from_disk().await?;
assert!(!app.config.feedback_enabled);
assert!(app.auth_failure_reporter_guard.is_none());
assert!(!app.chat_widget.config_ref().feedback_enabled);
Ok(())
}
#[tokio::test]
async fn set_config_syncs_chat_widget_feedback_enabled() -> Result<()> {
let mut app = make_test_app().await;
let mut next_config = app.config.clone();
next_config.feedback_enabled = !app.chat_widget.config_ref().feedback_enabled;
app.set_config(next_config.clone());
assert_eq!(app.config.feedback_enabled, next_config.feedback_enabled);
assert_eq!(
app.chat_widget.config_ref().feedback_enabled,
next_config.feedback_enabled
);
Ok(())
}
#[tokio::test]
async fn rebuild_config_for_resume_or_fallback_uses_current_config_on_same_cwd_error()
-> Result<()> {
codex-rs/tui/src/chatwidget.rs
+2 -1
View File
@@ -10582,9 +10582,10 @@ impl ChatWidget {
self.bottom_pane.set_plugin_mentions(Some(plugins));
}
pub(crate) fn sync_plugin_mentions_config(&mut self, config: &Config) {
pub(crate) fn sync_runtime_config(&mut self, config: &Config) {
self.config.features = config.features.clone();
self.config.config_layer_stack = config.config_layer_stack.clone();
self.config.feedback_enabled = config.feedback_enabled;
}
pub(crate) fn open_review_popup(&mut self) {
codex-rs/tui/src/lib.rs
-3
View File
@@ -760,7 +760,6 @@ pub async fn run_main(
cloud_requirements.clone(),
)
.await;
#[allow(clippy::print_stderr)]
match check_execpolicy_for_warnings(&config.config_layer_stack).await {
Ok(None) => {}
@@ -842,7 +841,6 @@ pub async fn run_main(
let feedback = codex_feedback::CodexFeedback::new();
let feedback_layer = feedback.logger_layer();
let feedback_metadata_layer = feedback.metadata_layer();
let feedback_auth_event_layer = feedback.auth_event_layer();
if cli.oss && model_provider_override.is_some() {
// We're in the oss section, so provider_id should be Some
@@ -896,7 +894,6 @@ pub async fn run_main(
.with(file_layer)
.with(feedback_layer)
.with(feedback_metadata_layer)
.with(feedback_auth_event_layer)
.with(log_db_layer)
.with(otel_logger_layer)
.with(otel_tracing_layer)