mirror of
https://github.com/fawney19/Aether.git
synced 2026-10-07 18:07:47 +08:00
fix(ci): align lint-safe security paths
This commit is contained in:
@@ -914,7 +914,7 @@ impl GatewayDataState {
|
||||
|
||||
Ok(Some(LdapAuthProvisioningResult {
|
||||
user: outcome.user,
|
||||
owned_wallet_id: initialized.created.then(|| initialized.wallet.id),
|
||||
owned_wallet_id: initialized.created.then_some(initialized.wallet.id),
|
||||
}))
|
||||
}
|
||||
|
||||
|
||||
@@ -433,7 +433,7 @@ fn smtp_read_response<T: std::io::BufRead>(reader: &mut T) -> Result<(u16, Strin
|
||||
if message
|
||||
.len()
|
||||
.checked_add(additional)
|
||||
.map_or(true, |length| length > SMTP_MAX_RESPONSE_BYTES)
|
||||
.is_none_or(|length| length > SMTP_MAX_RESPONSE_BYTES)
|
||||
{
|
||||
return Err(GatewayError::Internal(
|
||||
"smtp response exceeds the allowed size".to_string(),
|
||||
@@ -476,7 +476,7 @@ fn read_smtp_response_line<T: std::io::BufRead>(
|
||||
if line
|
||||
.len()
|
||||
.checked_add(take)
|
||||
.map_or(true, |length| length > SMTP_MAX_RESPONSE_LINE_BYTES)
|
||||
.is_none_or(|length| length > SMTP_MAX_RESPONSE_LINE_BYTES)
|
||||
{
|
||||
return Err(std::io::Error::new(
|
||||
std::io::ErrorKind::InvalidData,
|
||||
|
||||
@@ -39,7 +39,6 @@ use aether_scheduler_core::SchedulerRequestCandidateStatusUpdate;
|
||||
use aether_usage_runtime::{
|
||||
build_lifecycle_usage_seed, build_stream_terminal_usage_payload_seed,
|
||||
build_terminal_usage_context_seed, stream_report_represents_failure,
|
||||
DEFAULT_USAGE_RESPONSE_BODY_CAPTURE_LIMIT_BYTES,
|
||||
};
|
||||
use base64::Engine as _;
|
||||
use serde_json::Value;
|
||||
@@ -433,8 +432,7 @@ impl AttemptBodyCapture {
|
||||
if bytes.is_empty() || self.truncated {
|
||||
return;
|
||||
}
|
||||
let max_bytes = DEFAULT_USAGE_RESPONSE_BODY_CAPTURE_LIMIT_BYTES
|
||||
.min(crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES);
|
||||
let max_bytes = crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES;
|
||||
if self.buffer.len() >= max_bytes {
|
||||
self.truncated = true;
|
||||
return;
|
||||
@@ -448,10 +446,7 @@ impl AttemptBodyCapture {
|
||||
}
|
||||
|
||||
pub(crate) fn encode(&self) -> (Option<String>, Option<UsageBodyCaptureState>) {
|
||||
self.encode_with_limit(
|
||||
DEFAULT_USAGE_RESPONSE_BODY_CAPTURE_LIMIT_BYTES
|
||||
.min(crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES),
|
||||
)
|
||||
self.encode_with_limit(crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES)
|
||||
}
|
||||
|
||||
fn encode_with_limit(
|
||||
@@ -1435,8 +1430,7 @@ mod stage_tests {
|
||||
#[test]
|
||||
fn body_capture_encodes_inline_and_empty_states() {
|
||||
assert_eq!(
|
||||
super::DEFAULT_USAGE_RESPONSE_BODY_CAPTURE_LIMIT_BYTES
|
||||
.min(crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES),
|
||||
crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES,
|
||||
crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES
|
||||
);
|
||||
|
||||
|
||||
@@ -585,7 +585,7 @@ fn harden_execution_runtime_socket(
|
||||
// runners. The validated canonical parent and inode identity still
|
||||
// close the replacement window without relying on that differing
|
||||
// device number.
|
||||
|| (cfg!(target_os = "linux") && metadata.ino() != stat.st_ino as u64)
|
||||
|| (cfg!(target_os = "linux") && metadata.ino() != stat.st_ino)
|
||||
{
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::PermissionDenied,
|
||||
|
||||
@@ -25,7 +25,6 @@ use aether_usage_runtime::{
|
||||
build_lifecycle_usage_seed, build_stream_terminal_usage_payload_seed,
|
||||
build_sync_terminal_usage_payload_seed, build_terminal_usage_context_seed, LifecycleUsageSeed,
|
||||
SyncTerminalUsagePayloadSeed, TerminalUsageContextSeed, UsageRequestRecordLevel,
|
||||
DEFAULT_USAGE_RESPONSE_BODY_CAPTURE_LIMIT_BYTES,
|
||||
};
|
||||
use async_stream::stream;
|
||||
use axum::body::{Body, Bytes};
|
||||
@@ -399,8 +398,7 @@ fn direct_passthrough_mode() -> DirectPassthroughMode {
|
||||
fn stream_body_buffer_limit_for_record_level(record_level: UsageRequestRecordLevel) -> usize {
|
||||
match record_level {
|
||||
UsageRequestRecordLevel::Basic => BASIC_STREAM_BODY_ANALYSIS_LIMIT_BYTES,
|
||||
UsageRequestRecordLevel::Full => DEFAULT_USAGE_RESPONSE_BODY_CAPTURE_LIMIT_BYTES
|
||||
.min(crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES),
|
||||
UsageRequestRecordLevel::Full => crate::execution_runtime::MAX_STREAM_BODY_CAPTURE_BYTES,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -686,8 +686,9 @@ pub(crate) async fn read_admin_external_models_cache(
|
||||
match fetch_admin_external_models_from_source(state, request_id, proxy_node_id.as_deref()).await
|
||||
{
|
||||
Ok(payload) => {
|
||||
if let Err(_) =
|
||||
store_admin_external_models_cache(state, proxy_node_id.as_deref(), &payload).await
|
||||
if store_admin_external_models_cache(state, proxy_node_id.as_deref(), &payload)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
warn!("failed to store fetched external models cache");
|
||||
}
|
||||
|
||||
@@ -314,7 +314,7 @@ async fn release_codex_agent_identity_leases(
|
||||
leases: Vec<RuntimeLockLease>,
|
||||
) {
|
||||
for lease in leases {
|
||||
if let Err(_) = state.runtime_state().lock_release(&lease).await {
|
||||
if state.runtime_state().lock_release(&lease).await.is_err() {
|
||||
tracing::warn!(
|
||||
lock_key = %lease.key,
|
||||
"gateway Agent Identity enrollment lock release failed"
|
||||
|
||||
@@ -552,7 +552,13 @@ pub(super) async fn seed_provider_oauth_pool_score(
|
||||
now_unix_secs,
|
||||
pool_config.score_rules,
|
||||
);
|
||||
if let Err(_) = state.app().data.upsert_pool_member_score(upsert).await {
|
||||
if state
|
||||
.app()
|
||||
.data
|
||||
.upsert_pool_member_score(upsert)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
tracing::debug!(
|
||||
provider_id = %provider_id,
|
||||
key_id = %key.id,
|
||||
|
||||
@@ -7632,10 +7632,11 @@ impl<'a> AdminAppState<'a> {
|
||||
// Legacy uploads historically normalize an omitted rate limit to zero. Rollback
|
||||
// checkpoints instead preserve the nullable database value exactly.
|
||||
let rate_limit = imported_rate_limit.unwrap_or(0);
|
||||
let rate_limit_value = mode
|
||||
.is_rollback_checkpoint()
|
||||
.then_some(imported_rate_limit)
|
||||
.unwrap_or(Some(rate_limit));
|
||||
let rate_limit_value = if mode.is_rollback_checkpoint() {
|
||||
imported_rate_limit
|
||||
} else {
|
||||
Some(rate_limit)
|
||||
};
|
||||
let concurrent_limit = invalid_value!(imported_optional_i32(
|
||||
key.get("concurrent_limit"),
|
||||
"concurrent_limit"
|
||||
@@ -8364,7 +8365,7 @@ impl<'a> AdminAppState<'a> {
|
||||
)));
|
||||
}
|
||||
if let Some(wallet_id) = created_wallet_id {
|
||||
if let Some(journal) = mutation_journal.as_deref_mut() {
|
||||
if let Some(journal) = mutation_journal {
|
||||
journal
|
||||
.user_wallet_snapshots
|
||||
.insert((user_id.to_string(), wallet_id), synced);
|
||||
@@ -8433,7 +8434,7 @@ impl<'a> AdminAppState<'a> {
|
||||
)));
|
||||
}
|
||||
if let Some(wallet_id) = created_wallet_id {
|
||||
if let Some(journal) = mutation_journal.as_deref_mut() {
|
||||
if let Some(journal) = mutation_journal {
|
||||
journal
|
||||
.api_key_wallet_snapshots
|
||||
.insert((api_key_id.to_string(), wallet_id), synced);
|
||||
|
||||
@@ -349,6 +349,9 @@ fn load_and_sanitize_update_history(path: &Path) -> (Vec<UpdateHistoryEntry>, bo
|
||||
}
|
||||
|
||||
fn sanitize_update_history_entries(entries: &mut Vec<UpdateHistoryEntry>) -> bool {
|
||||
// Deliberately use a non-short-circuiting fold: every historical entry
|
||||
// must be sanitized even after one entry changes.
|
||||
#[allow(clippy::unnecessary_fold)]
|
||||
let mut changed = entries.iter_mut().fold(false, |changed, entry| {
|
||||
sanitize_update_history_entry(entry) || changed
|
||||
});
|
||||
@@ -455,7 +458,7 @@ fn read_update_metadata_file(path: &Path, max_bytes: usize) -> Result<Option<Vec
|
||||
if bytes.len() > max_bytes {
|
||||
return Err("更新元数据超过大小限制".to_string());
|
||||
}
|
||||
return Ok(Some(bytes));
|
||||
Ok(Some(bytes))
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
@@ -536,7 +539,7 @@ fn write_update_metadata_atomic(path: &Path, bytes: &[u8]) -> Result<(), String>
|
||||
if result.is_err() {
|
||||
let _ = unix_update_unlink_at(&parent, &temp_name);
|
||||
}
|
||||
return result;
|
||||
result
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
@@ -711,7 +714,7 @@ fn remove_update_metadata_file(path: &Path) -> Result<(), String> {
|
||||
parent
|
||||
.sync_all()
|
||||
.map_err(|err| format!("同步更新元数据目录失败: {err}"))?;
|
||||
return Ok(());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
@@ -1867,7 +1870,7 @@ fn switch_current_symlink_at(base_dir: &Path, version: &str) -> Result<(), Strin
|
||||
parent
|
||||
.sync_all()
|
||||
.map_err(|err| format!("同步版本入口目录失败: {err}"))?;
|
||||
return Ok(());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
|
||||
@@ -607,17 +607,15 @@ fn payment_order_payload(
|
||||
// A gateway response is a live checkout capability, not durable order
|
||||
// history. Once the order is paid, terminal, or expired, suppress URLs,
|
||||
// form parameters, and provider metadata from the public payload.
|
||||
let gateway_response = record
|
||||
.status
|
||||
.eq_ignore_ascii_case("pending")
|
||||
.then(|| {
|
||||
record
|
||||
.expires_at_unix_secs
|
||||
.is_some_and(|expires_at| expires_at > Utc::now().timestamp().max(0) as u64)
|
||||
})
|
||||
.unwrap_or(false)
|
||||
.then(|| record.gateway_response.clone())
|
||||
.flatten();
|
||||
let gateway_response = if record.status.eq_ignore_ascii_case("pending")
|
||||
&& record
|
||||
.expires_at_unix_secs
|
||||
.is_some_and(|expires_at| expires_at > Utc::now().timestamp().max(0) as u64)
|
||||
{
|
||||
record.gateway_response.clone()
|
||||
} else {
|
||||
None
|
||||
};
|
||||
json!({
|
||||
"id": record.id,
|
||||
"order_no": record.order_no,
|
||||
|
||||
@@ -383,8 +383,7 @@ fn forwarded_header_last(headers: &http::HeaderMap, name: &str) -> Option<String
|
||||
.filter_map(|value| value.to_str().ok())
|
||||
.flat_map(|value| value.split(','))
|
||||
.map(str::trim)
|
||||
.filter(|value| !value.is_empty())
|
||||
.last()
|
||||
.rfind(|value| !value.is_empty())
|
||||
.map(ToOwned::to_owned)
|
||||
}
|
||||
|
||||
|
||||
@@ -1054,11 +1054,14 @@ fn redirect_to(target: &str, params: Option<RedirectParams>) -> Response<Body> {
|
||||
fn build_redirect_location(target: &str, params: Option<RedirectParams>) -> String {
|
||||
let relative_target =
|
||||
url::Url::parse(target).is_err() && target.starts_with('/') && !target.starts_with("//");
|
||||
let Ok(mut url) = url::Url::parse(target).or_else(|_| {
|
||||
relative_target
|
||||
.then(|| url::Url::parse("http://aether.invalid").and_then(|base| base.join(target)))
|
||||
.unwrap_or_else(|| Err(url::ParseError::RelativeUrlWithoutBase))
|
||||
}) else {
|
||||
let parsed_target = url::Url::parse(target).or_else(|_| {
|
||||
if relative_target {
|
||||
url::Url::parse("http://aether.invalid").and_then(|base| base.join(target))
|
||||
} else {
|
||||
Err(url::ParseError::RelativeUrlWithoutBase)
|
||||
}
|
||||
});
|
||||
let Ok(mut url) = parsed_target else {
|
||||
return target.to_string();
|
||||
};
|
||||
match params {
|
||||
|
||||
@@ -147,7 +147,7 @@ fn validate_gateway_data_encryption_key(value: Option<&str>) -> Result<(), &'sta
|
||||
let Some(value) = value.map(str::trim).filter(|value| !value.is_empty()) else {
|
||||
return Ok(());
|
||||
};
|
||||
if value.as_bytes().len() < MIN_GATEWAY_DATA_ENCRYPTION_KEY_BYTES {
|
||||
if value.len() < MIN_GATEWAY_DATA_ENCRYPTION_KEY_BYTES {
|
||||
return Err("gateway data encryption key must contain at least 32 bytes");
|
||||
}
|
||||
if INSECURE_GATEWAY_DATA_ENCRYPTION_KEYS.contains(&value) {
|
||||
@@ -2984,7 +2984,7 @@ fn read_data_import_input_with_limit(path: &Path, limit: usize) -> io::Result<St
|
||||
// A file can grow after metadata() returns. Reading one extra byte catches
|
||||
// that race without allowing the input buffer to exceed the configured
|
||||
// parser budget.
|
||||
let read_limit = limit.checked_add(1).unwrap_or(usize::MAX);
|
||||
let read_limit = limit.saturating_add(1);
|
||||
// Do not reserve the whole metadata length: sparse or concurrently grown
|
||||
// files can advertise a huge size while containing little data, and a
|
||||
// single capacity reservation would otherwise become a local DoS vector.
|
||||
|
||||
@@ -258,7 +258,7 @@ pub(crate) async fn resolve_identity_oauth_login_user(
|
||||
.initialize_auth_user_wallet_with_outcome(&user.id, initial_gift, false)
|
||||
.await
|
||||
{
|
||||
Ok(Some(outcome)) => outcome.created.then(|| outcome.wallet.id),
|
||||
Ok(Some(outcome)) => outcome.created.then_some(outcome.wallet.id),
|
||||
Ok(None) => {
|
||||
let _ = state
|
||||
.rollback_provisional_auth_user_with_wallet(&user.id, None)
|
||||
|
||||
@@ -1538,7 +1538,7 @@ impl AppState {
|
||||
body_excerpt,
|
||||
..
|
||||
}) if matches!(status_code, 400 | 401 | 403) => {
|
||||
if let Err(_) = self
|
||||
if self
|
||||
.persist_local_oauth_refresh_failure_state(
|
||||
¤t_transport,
|
||||
status_code,
|
||||
@@ -1546,6 +1546,7 @@ impl AppState {
|
||||
false,
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
key_id = %current_transport.key.id,
|
||||
@@ -1596,13 +1597,14 @@ impl AppState {
|
||||
.await;
|
||||
return Ok(None);
|
||||
}
|
||||
if let Err(_) = self
|
||||
if self
|
||||
.persist_local_oauth_refresh_entry(
|
||||
¤t_transport,
|
||||
&refreshed_entry,
|
||||
expected_credential_fence.as_ref(),
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
key_id = %current_transport.key.id,
|
||||
@@ -1792,13 +1794,14 @@ impl AppState {
|
||||
.await;
|
||||
return Ok(None);
|
||||
}
|
||||
if let Err(_) = self
|
||||
if self
|
||||
.persist_local_oauth_refresh_entry(
|
||||
¤t_transport,
|
||||
&refreshed_entry,
|
||||
expected_credential_fence.as_ref(),
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
key_id = %current_transport.key.id,
|
||||
@@ -1854,7 +1857,7 @@ impl AppState {
|
||||
let Some(lease) = lease else {
|
||||
return;
|
||||
};
|
||||
if let Err(_) = self.runtime_state.lock_release(&lease).await {
|
||||
if self.runtime_state.lock_release(&lease).await.is_err() {
|
||||
tracing::warn!(
|
||||
key_id = %lease.key,
|
||||
"gateway local oauth refresh distributed lease release failed"
|
||||
|
||||
@@ -79,9 +79,11 @@ fn config_f64(value: Option<&serde_json::Value>, default: f64) -> f64 {
|
||||
|
||||
fn config_percent(value: Option<&serde_json::Value>) -> f64 {
|
||||
let value = config_f64(value, 0.0);
|
||||
(value.is_finite() && value > 0.0 && value <= 100.0)
|
||||
.then_some(value)
|
||||
.unwrap_or(0.0)
|
||||
if value.is_finite() && value > 0.0 && value <= 100.0 {
|
||||
value
|
||||
} else {
|
||||
0.0
|
||||
}
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
|
||||
@@ -315,7 +315,7 @@ fn canonicalize_internal_report_json(value: &Value) -> Value {
|
||||
),
|
||||
Value::Object(object) => {
|
||||
let mut entries = object.iter().collect::<Vec<_>>();
|
||||
entries.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));
|
||||
entries.sort_unstable_by_key(|(left, _)| *left);
|
||||
Value::Object(Map::from_iter(entries.into_iter().map(|(key, value)| {
|
||||
(key.clone(), canonicalize_internal_report_json(value))
|
||||
})))
|
||||
|
||||
Reference in New Issue
Block a user