i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 01:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: add oauth refresh diagnostics

Browse Source
This commit is contained in:
fawney19
2026-04-28 00:18:51 +08:00
parent 83e1f99ccf
commit 6426b5d80e
2 changed files with 185 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
apps/aether-gateway/src/state/oauth.rs
View File

@@ -16,6 +16,7 @@ use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKe
use base64::{engine::general_purpose::STANDARD, Engine as _};
use flate2::read::{DeflateDecoder, GzDecoder};
use serde_json::{json, Map, Value};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::io::Read;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -51,6 +52,34 @@ fn tagged_reason(reason: Option<&str>, prefix: &str) -> Option<String> {
})
}
fn oauth_auth_config_refresh_token_fingerprint(auth_config: Option<&str>) -> Option<String> {
let parsed = auth_config
.map(str::trim)
.filter(|value| !value.is_empty())
.and_then(|value| serde_json::from_str::<Value>(value).ok())?;
oauth_metadata_refresh_token_fingerprint(Some(&parsed))
}
fn oauth_metadata_refresh_token_fingerprint(metadata: Option<&Value>) -> Option<String> {
metadata
.and_then(Value::as_object)
.and_then(|object| object.get("refresh_token"))
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(secret_fingerprint)
}
fn secret_fingerprint(value: &str) -> String {
let digest = Sha256::digest(value.as_bytes());
let mut fingerprint = String::with_capacity(16);
for byte in digest.iter().take(8) {
use std::fmt::Write as _;
let _ = write!(&mut fingerprint, "{byte:02x}");
}
fingerprint
}
fn oauth_invalid_reason_is_account_block(reason: Option<&str>) -> bool {
reason
.map(str::trim)
@@ -820,6 +849,23 @@ impl AppState {
let mut current_transport = transport.clone();
current_transport.key.decrypted_api_key = "__placeholder__".to_string();
let executor = GatewayLocalOAuthHttpExecutor { state: self };
let transport_refresh_token_fingerprint = oauth_auth_config_refresh_token_fingerprint(
current_transport.key.decrypted_auth_config.as_deref(),
)
.unwrap_or_else(|| "-".to_string());
tracing::info!(
key_id = %current_transport.key.id,
provider_id = %current_transport.provider.id,
provider_type = %current_transport.provider.provider_type,
transport_refresh_token_fingerprint = %transport_refresh_token_fingerprint,
has_transport_auth_config = current_transport
.key
.decrypted_auth_config
.as_deref()
.map(str::trim)
.is_some_and(|value| !value.is_empty()),
"gateway manual oauth refresh starting"
);
for _ in 0..2 {
let resolution = self
@@ -1008,13 +1054,34 @@ impl AppState {
let current_status_snapshot = latest_key.status_snapshot.take();
latest_key.status_snapshot =
sync_provider_key_oauth_status_snapshot(current_status_snapshot, &latest_key);
if self
let updated = self
.update_provider_catalog_key(&latest_key)
.await?
.is_some()
{
.is_some();
if updated {
self.clear_provider_transport_snapshot_cache();
}
let metadata_refresh_token_fingerprint =
oauth_metadata_refresh_token_fingerprint(entry.metadata.as_ref())
.unwrap_or_else(|| "-".to_string());
tracing::info!(
key_id = %key_id,
provider_id = %transport.provider.id,
provider_type = %transport.provider.provider_type,
updated,
metadata_has_refresh_token = entry
.metadata
.as_ref()
.and_then(|value| value.as_object())
.and_then(|object| object.get("refresh_token"))
.and_then(|value| value.as_str())
.map(str::trim)
.is_some_and(|value| !value.is_empty()),
metadata_refresh_token_fingerprint = %metadata_refresh_token_fingerprint,
expires_at_unix_secs = ?entry.expires_at_unix_secs,
cleared_provider_transport_snapshot_cache = updated,
"gateway local oauth refresh entry persisted"
);
Ok(())
}
@@ -1082,6 +1149,15 @@ impl AppState {
self.clear_provider_transport_snapshot_cache();
let _ = self.invalidate_local_oauth_refresh_entry(key_id).await;
}
tracing::info!(
key_id = %key_id,
provider_id = %transport.provider.id,
provider_type = %transport.provider.provider_type,
status_code,
updated,
cleared_provider_transport_snapshot_cache = updated,
"gateway local oauth refresh failure state persisted"
);
Ok(updated)
}

111
crates/aether-provider-transport/src/generic_oauth.rs
View File

@@ -3,6 +3,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
use async_trait::async_trait;
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use url::form_urlencoded;
use super::oauth_refresh::{
@@ -256,15 +257,47 @@ impl LocalOAuthRefreshAdapter for GenericOAuthRefreshAdapter {
else {
return Ok(None);
};
let mut metadata = self
.base_auth_config(transport, entry)
let (base_auth_config_source, base_auth_config) = if let Some(value) =
entry.and_then(|cached| Self::auth_config_from_entry(transport, cached))
{
("cached_entry", Some(value))
} else {
let value = Self::auth_config_from_transport(transport);
let source = if value.is_some() {
"transport_auth_config"
} else {
"none"
};
(source, value)
};
let mut metadata = base_auth_config
.and_then(|value| value.as_object().cloned())
.unwrap_or_default();
let Some(refresh_token) = metadata.get("refresh_token").and_then(non_empty_string) else {
tracing::warn!(
key_id = %transport.key.id,
provider_id = %transport.provider.id,
provider_type = template.provider_type,
auth_config_source = base_auth_config_source,
"gateway generic oauth refresh skipped because auth_config has no refresh_token"
);
return Ok(None);
};
let token_url = self.token_url_for_template(template);
let request_refresh_token_fingerprint = secret_fingerprint(refresh_token.as_str());
tracing::info!(
key_id = %transport.key.id,
provider_id = %transport.provider.id,
endpoint_id = %transport.endpoint.id,
provider_type = template.provider_type,
auth_config_source = base_auth_config_source,
request_refresh_token_fingerprint = %request_refresh_token_fingerprint,
request_refresh_token_len = refresh_token.len(),
token_url = %token_url,
uses_json_payload = template.uses_json_payload,
"gateway generic oauth refresh request prepared"
);
let scope = (!template.scopes.is_empty()).then(|| template.scopes.join(" "));
let response = if template.uses_json_payload {
let mut body = serde_json::Map::from_iter([
@@ -339,10 +372,21 @@ impl LocalOAuthRefreshAdapter for GenericOAuthRefreshAdapter {
let status = reqwest::StatusCode::from_u16(response.status_code).unwrap_or_default();
let body = response.body_text;
if !status.is_success() {
let body_excerpt = truncate_body(&body);
tracing::warn!(
key_id = %transport.key.id,
provider_id = %transport.provider.id,
endpoint_id = %transport.endpoint.id,
provider_type = template.provider_type,
status_code = status.as_u16(),
request_refresh_token_fingerprint = %request_refresh_token_fingerprint,
body_excerpt = %body_excerpt,
"gateway generic oauth refresh returned error status"
);
return Err(LocalOAuthRefreshError::HttpStatus {
provider_type: template.provider_type,
status_code: status.as_u16(),
body_excerpt: truncate_body(&body),
body_excerpt,
});
}
@@ -364,8 +408,19 @@ impl LocalOAuthRefreshAdapter for GenericOAuthRefreshAdapter {
Value::String(template.provider_type.to_string()),
);
metadata.insert("updated_at".to_string(), json!(current_unix_secs()));
if let Some(refresh_token) = payload.get("refresh_token").and_then(non_empty_string) {
metadata.insert("refresh_token".to_string(), Value::String(refresh_token));
let response_refresh_token = payload.get("refresh_token").and_then(non_empty_string);
let response_refresh_token_fingerprint = response_refresh_token
.as_deref()
.map(secret_fingerprint)
.unwrap_or_else(|| "-".to_string());
let response_refresh_token_rotated = response_refresh_token
.as_deref()
.map(|value| value != refresh_token.as_str());
if let Some(refresh_token) = response_refresh_token.as_ref() {
metadata.insert(
"refresh_token".to_string(),
Value::String(refresh_token.clone()),
);
}
if let Some(token_type) = payload.get("token_type").and_then(non_empty_string) {
metadata.insert("token_type".to_string(), Value::String(token_type));
@@ -381,6 +436,42 @@ impl LocalOAuthRefreshAdapter for GenericOAuthRefreshAdapter {
metadata.remove("expires_at");
}
}
let stored_refresh_token_fingerprint = metadata
.get("refresh_token")
.and_then(non_empty_string)
.map(|value| secret_fingerprint(value.as_str()))
.unwrap_or_else(|| "-".to_string());
let stored_refresh_token_source = if response_refresh_token.is_some() {
"response"
} else {
"existing"
};
tracing::info!(
key_id = %transport.key.id,
provider_id = %transport.provider.id,
endpoint_id = %transport.endpoint.id,
provider_type = template.provider_type,
status_code = status.as_u16(),
request_refresh_token_fingerprint = %request_refresh_token_fingerprint,
response_has_refresh_token = response_refresh_token.is_some(),
response_refresh_token_fingerprint = %response_refresh_token_fingerprint,
response_refresh_token_rotated = ?response_refresh_token_rotated,
stored_refresh_token_source = stored_refresh_token_source,
stored_refresh_token_fingerprint = %stored_refresh_token_fingerprint,
expires_at_unix_secs = ?expires_at_unix_secs,
"gateway generic oauth refresh succeeded"
);
if response_refresh_token.is_none() && template.provider_type == "codex" {
tracing::warn!(
key_id = %transport.key.id,
provider_id = %transport.provider.id,
endpoint_id = %transport.endpoint.id,
provider_type = template.provider_type,
request_refresh_token_fingerprint = %request_refresh_token_fingerprint,
stored_refresh_token_fingerprint = %stored_refresh_token_fingerprint,
"gateway codex oauth refresh succeeded without replacement refresh_token"
);
}
Ok(Some(self.build_cached_entry(
template,
@@ -443,6 +534,16 @@ fn non_empty_string(value: &Value) -> Option<String> {
.map(ToOwned::to_owned)
}
fn secret_fingerprint(value: &str) -> String {
let digest = Sha256::digest(value.as_bytes());
let mut fingerprint = String::with_capacity(16);
for byte in digest.iter().take(8) {
use std::fmt::Write as _;
let _ = write!(&mut fingerprint, "{byte:02x}");
}
fingerprint
}
fn current_unix_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)