mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-21 11:56:38 +00:00
fix(runtime): remove high-impact unwrap paths (#3755)
* fix(runtime): remove high-impact unwrap paths * fix(runtime): propagate managed SSE metadata errors * fix(runtime): add typed OPA config errors * fix(runtime): harden config and credential helpers * fix(runtime): remove SSE hmac unwraps * fix(runtime): complete SSE helper error propagation * fix(trusted-proxies): avoid legacy global init panics * test(credentials): allow deprecated rpc token check * fix(storage): harden object lock retention parsing * chore(checks): refresh layer dependency baseline * chore(checks): refresh layer dependency baseline * Update layer-dependency-baseline.txt Signed-off-by: houseme <housemecn@gmail.com> * test(context): avoid clone on copy boot time --------- Signed-off-by: houseme <housemecn@gmail.com> Co-authored-by: Zhengchao An <anzhengchao@gmail.com> Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
@@ -243,7 +243,9 @@ fn derive_rpc_secret(access_key: &str, secret_key: &str) -> Option<String> {
|
||||
return None;
|
||||
}
|
||||
|
||||
let mut mac = <HmacSha256 as KeyInit>::new_from_slice(secret_key.as_bytes()).expect("HMAC can take key of any size");
|
||||
let Ok(mut mac) = <HmacSha256 as KeyInit>::new_from_slice(secret_key.as_bytes()) else {
|
||||
return None;
|
||||
};
|
||||
mac.update(RPC_SECRET_DERIVATION_CONTEXT);
|
||||
mac.update(&[0]);
|
||||
mac.update(access_key.as_bytes());
|
||||
@@ -284,7 +286,7 @@ pub fn try_get_rpc_token() -> std::io::Result<String> {
|
||||
|
||||
#[deprecated(note = "use try_get_rpc_token to handle missing RPC secrets explicitly")]
|
||||
pub fn get_rpc_token() -> String {
|
||||
try_get_rpc_token().expect(RPC_SECRET_REQUIRED_MESSAGE)
|
||||
try_get_rpc_token().unwrap_or_default()
|
||||
}
|
||||
|
||||
/// A wrapper struct for masking sensitive strings in Debug implementations.
|
||||
@@ -596,6 +598,15 @@ mod tests {
|
||||
assert_string_return(get_rpc_token);
|
||||
}
|
||||
|
||||
#[allow(deprecated)]
|
||||
#[test]
|
||||
fn test_get_rpc_token_matches_fallible_api_contract() {
|
||||
match try_get_rpc_token() {
|
||||
Ok(secret) => assert_eq!(get_rpc_token(), secret),
|
||||
Err(_) => assert_eq!(get_rpc_token(), ""),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_rpc_secret_accepts_non_default_secret() {
|
||||
assert_eq!(
|
||||
|
||||
@@ -36,7 +36,25 @@ pub struct AuthZPlugin {
|
||||
args: Args,
|
||||
}
|
||||
|
||||
fn check() -> Result<(), String> {
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum OpaConfigError {
|
||||
#[error("Missing required env var: {0}")]
|
||||
MissingRequiredEnv(&'static str),
|
||||
#[error("Invalid env vars: {0:?}")]
|
||||
InvalidEnvVars(HashMap<String, String>),
|
||||
#[error("Error getting env var {name}: {source:?}")]
|
||||
EnvRead {
|
||||
name: &'static str,
|
||||
#[source]
|
||||
source: env::VarError,
|
||||
},
|
||||
#[error("OPA returned an error: {0}")]
|
||||
InvalidStatus(reqwest::StatusCode),
|
||||
#[error("Error connecting to OPA: {0}")]
|
||||
Connection(reqwest::Error),
|
||||
}
|
||||
|
||||
fn check() -> Result<(), OpaConfigError> {
|
||||
let env_list = env::vars();
|
||||
let mut candidate = HashMap::new();
|
||||
let prefix = format!("{ENV_PREFIX}{POLICY_PLUGIN_SUB_SYS}").to_uppercase();
|
||||
@@ -48,17 +66,17 @@ fn check() -> Result<(), String> {
|
||||
|
||||
//check required env vars
|
||||
if candidate.remove(ENV_POLICY_PLUGIN_OPA_URL).is_none() {
|
||||
return Err(format!("Missing required env var: {ENV_POLICY_PLUGIN_OPA_URL}"));
|
||||
return Err(OpaConfigError::MissingRequiredEnv(ENV_POLICY_PLUGIN_OPA_URL));
|
||||
}
|
||||
|
||||
// check optional env vars
|
||||
candidate.remove(ENV_POLICY_PLUGIN_AUTH_TOKEN);
|
||||
if !candidate.is_empty() {
|
||||
return Err(format!("Invalid env vars: {candidate:?}"));
|
||||
return Err(OpaConfigError::InvalidEnvVars(candidate));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
async fn validate(config: &Args) -> Result<(), String> {
|
||||
async fn validate(config: &Args) -> Result<(), OpaConfigError> {
|
||||
let client = reqwest::Client::new();
|
||||
|
||||
match client.post(&config.url).send().await {
|
||||
@@ -68,31 +86,34 @@ async fn validate(config: &Args) -> Result<(), String> {
|
||||
info!("OPA is ready to accept requests.");
|
||||
}
|
||||
_ => {
|
||||
return Err(format!("OPA returned an error: {}", resp.status()));
|
||||
return Err(OpaConfigError::InvalidStatus(resp.status()));
|
||||
}
|
||||
};
|
||||
}
|
||||
Err(err) => {
|
||||
return Err(format!("Error connecting to OPA: {err}"));
|
||||
return Err(OpaConfigError::Connection(err));
|
||||
}
|
||||
};
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn lookup_config() -> Result<Args, String> {
|
||||
pub async fn lookup_config() -> Result<Args, OpaConfigError> {
|
||||
let args = Args::default();
|
||||
|
||||
let get_cfg =
|
||||
|cfg: &str| -> Result<String, String> { env::var(cfg).map_err(|e| format!("Error getting env var {cfg}: {e:?}")) };
|
||||
let get_cfg = |cfg: &'static str| -> Result<String, OpaConfigError> {
|
||||
env::var(cfg).map_err(|source| OpaConfigError::EnvRead { name: cfg, source })
|
||||
};
|
||||
|
||||
let url = get_cfg(ENV_POLICY_PLUGIN_OPA_URL);
|
||||
if url.is_err() {
|
||||
info!("OPA is not enabled.");
|
||||
return Ok(args);
|
||||
}
|
||||
let url = match get_cfg(ENV_POLICY_PLUGIN_OPA_URL) {
|
||||
Ok(url) => url,
|
||||
Err(_) => {
|
||||
info!("OPA is not enabled.");
|
||||
return Ok(args);
|
||||
}
|
||||
};
|
||||
check()?;
|
||||
let args = Args {
|
||||
url: url.ok().unwrap(),
|
||||
url,
|
||||
auth_token: get_cfg(ENV_POLICY_PLUGIN_AUTH_TOKEN).unwrap_or_default(),
|
||||
};
|
||||
validate(&args).await?;
|
||||
@@ -111,7 +132,10 @@ impl AuthZPlugin {
|
||||
.http2_keep_alive_interval(Some(Duration::from_secs(30)))
|
||||
.http2_keep_alive_timeout(Duration::from_secs(15))
|
||||
.build()
|
||||
.unwrap();
|
||||
.unwrap_or_else(|err| {
|
||||
error!("failed to build OPA HTTP client, falling back to default reqwest client: {err}");
|
||||
reqwest::Client::new()
|
||||
});
|
||||
|
||||
Self { client, args: config }
|
||||
}
|
||||
@@ -235,7 +259,7 @@ mod tests {
|
||||
temp_env::with_var("RUSTFS_POLICY_PLUGIN_AUTH_TOKEN", Some("test-token"), || {
|
||||
let result = check();
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Missing required env var"));
|
||||
assert!(matches!(result.unwrap_err(), OpaConfigError::MissingRequiredEnv(_)));
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -250,7 +274,7 @@ mod tests {
|
||||
|| {
|
||||
let result = check();
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Invalid env vars"));
|
||||
assert!(matches!(result.unwrap_err(), OpaConfigError::InvalidEnvVars(_)));
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -270,6 +294,21 @@ mod tests {
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_lookup_config_uses_env_url_without_unwrap_path() {
|
||||
temp_env::with_vars(
|
||||
[
|
||||
("RUSTFS_POLICY_PLUGIN_URL", Some("http://localhost:8181")),
|
||||
("RUSTFS_POLICY_PLUGIN_AUTH_TOKEN", Some("token")),
|
||||
],
|
||||
|| {
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
let result = rt.block_on(async { lookup_config().await });
|
||||
assert!(result.is_err(), "lookup should fail validation without panicking when OPA is unreachable");
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_args_enable() {
|
||||
// Test Args enable method
|
||||
|
||||
@@ -31,9 +31,30 @@ static METRICS: OnceLock<Option<ProxyMetrics>> = OnceLock::new();
|
||||
/// Global instance of the trusted proxy layer.
|
||||
static PROXY_LAYER: OnceLock<LegacyTrustedProxyLayer> = OnceLock::new();
|
||||
|
||||
/// Disabled fallback layer used when legacy trusted proxies are not enabled.
|
||||
static DISABLED_PROXY_LAYER: OnceLock<LegacyTrustedProxyLayer> = OnceLock::new();
|
||||
|
||||
/// Global flag indicating if the trusted proxy middleware is enabled.
|
||||
static ENABLED: OnceLock<bool> = OnceLock::new();
|
||||
|
||||
fn load_config() -> &'static Arc<AppConfig> {
|
||||
CONFIG.get_or_init(|| Arc::new(ConfigLoader::from_env_or_default()))
|
||||
}
|
||||
|
||||
fn load_metrics(config: &AppConfig, enabled: bool) -> &'static Option<ProxyMetrics> {
|
||||
METRICS.get_or_init(|| {
|
||||
if config.monitoring.metrics_enabled {
|
||||
Some(default_proxy_metrics(enabled))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn disabled_layer() -> &'static LegacyTrustedProxyLayer {
|
||||
DISABLED_PROXY_LAYER.get_or_init(LegacyTrustedProxyLayer::disabled)
|
||||
}
|
||||
|
||||
/// Initializes the global trusted proxy system.
|
||||
///
|
||||
/// This function should be called once at the start of the application.
|
||||
@@ -54,24 +75,11 @@ pub fn init() {
|
||||
return;
|
||||
}
|
||||
|
||||
let config = CONFIG.get_or_init(|| Arc::new(ConfigLoader::from_env_or_default())).clone();
|
||||
let config = load_config().clone();
|
||||
let metrics = load_metrics(&config, enabled).clone();
|
||||
|
||||
METRICS.get_or_init(|| {
|
||||
if config.monitoring.metrics_enabled {
|
||||
Some(default_proxy_metrics(enabled))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
|
||||
PROXY_LAYER.get_or_init(|| {
|
||||
LegacyTrustedProxyLayer::with_cache_config(
|
||||
config.proxy.clone(),
|
||||
config.cache.clone(),
|
||||
METRICS.get().and_then(|m| m.clone()),
|
||||
enabled,
|
||||
)
|
||||
});
|
||||
PROXY_LAYER
|
||||
.get_or_init(|| LegacyTrustedProxyLayer::with_cache_config(config.proxy.clone(), config.cache.clone(), metrics, enabled));
|
||||
|
||||
tracing::info!(
|
||||
event = "trusted_proxies.lifecycle",
|
||||
@@ -90,25 +98,27 @@ pub fn init() {
|
||||
/// Returns a reference to the global trusted proxy layer.
|
||||
///
|
||||
/// This layer can be used to wrap Axum services or other Tower-compatible services.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if `init()` has not been called.
|
||||
pub fn layer() -> &'static LegacyTrustedProxyLayer {
|
||||
PROXY_LAYER
|
||||
.get()
|
||||
.expect("Trusted proxy system not initialized. Call init() first.")
|
||||
if let Some(layer) = PROXY_LAYER.get() {
|
||||
return layer;
|
||||
}
|
||||
|
||||
init();
|
||||
|
||||
if let Some(layer) = PROXY_LAYER.get() {
|
||||
return layer;
|
||||
}
|
||||
|
||||
disabled_layer()
|
||||
}
|
||||
|
||||
/// Returns a reference to the global configuration.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if `init()` has not been called.
|
||||
pub fn config() -> &'static AppConfig {
|
||||
CONFIG
|
||||
.get()
|
||||
.expect("Trusted proxy system not initialized. Call init() first.")
|
||||
if CONFIG.get().is_none() {
|
||||
init();
|
||||
}
|
||||
|
||||
load_config().as_ref()
|
||||
}
|
||||
|
||||
/// Returns a reference to the global metrics collector, if enabled.
|
||||
@@ -120,3 +130,23 @@ pub fn metrics() -> Option<&'static ProxyMetrics> {
|
||||
pub fn is_enabled() -> bool {
|
||||
*ENABLED.get_or_init(|| rustfs_utils::get_env_bool(ENV_TRUSTED_PROXY_ENABLED, DEFAULT_TRUSTED_PROXY_ENABLED))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn legacy_layer_is_available_without_explicit_init() {
|
||||
let layer = layer();
|
||||
assert_eq!(layer.is_enabled(), is_enabled());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_config_is_available_without_explicit_init() {
|
||||
let expected = ConfigLoader::from_env_or_default();
|
||||
let config = config();
|
||||
|
||||
assert_eq!(config.server_addr, expected.server_addr);
|
||||
assert_eq!(config.monitoring.metrics_enabled, expected.monitoring.metrics_enabled);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user