gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-09 14:49:25 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
d9e0a25174e7ed7e76548908bd01e2a1db000998
rustfs/rustfs/src/admin/handlers/config_admin.rs
T
Zhengchao An d9e0a25174 fix(oidc): support separate discovery issuer (#5149)
2026-07-23 02:44:32 +00:00

2641 lines
100 KiB
Rust
Raw Blame History

// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::admin::auth::validate_admin_request;
use crate::admin::handlers::supervise_admin_mutation;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::{
current_app_context, current_object_store_handle_for_context, current_server_config_for_context, publish_server_config,
};
use crate::admin::service::config::{
CONFIG_WORKER_RELOAD_FAILURE_STATE, EVENT_CONFIG_WORKER_RELOAD_FAILED, FULL_CONFIG_WORKER_SUBSYSTEMS, LOG_COMPONENT_ADMIN,
LOG_SUBSYSTEM_CONFIG, PreparedRuntimeConfig, apply_dynamic_config_for_subsystem, is_dynamic_config_subsystem,
preflight_dynamic_config_reload, prepare_server_config, signal_config_snapshot_reload_checked,
signal_dynamic_config_reload_checked,
};
use crate::admin::storage_api::config::storageclass::{INLINE_BLOCK_ENV, OPTIMIZE_ENV, RRS_ENV, STANDARD_ENV};
use crate::admin::storage_api::config::{
RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config, read_admin_config, read_admin_config_without_migrate,
read_admin_config_without_migrate_no_lock, save_admin_config, save_admin_server_config_no_lock,
with_admin_server_config_write_lock,
};
use crate::admin::storage_api::contract::list::ListOperations as _;
use crate::admin::utils::{encode_compatible_admin_payload, is_compat_admin_request, read_compatible_admin_body};
use crate::auth::{check_key_valid, get_session_token};
use crate::error::ApiError;
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use http::{HeaderMap, HeaderValue, Uri};
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_config::audit::{
AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS, ENV_AUDIT_MQTT_BROKER, ENV_AUDIT_MQTT_ENABLE, ENV_AUDIT_MQTT_KEEP_ALIVE_INTERVAL,
ENV_AUDIT_MQTT_PASSWORD, ENV_AUDIT_MQTT_QOS, ENV_AUDIT_MQTT_QUEUE_DIR, ENV_AUDIT_MQTT_QUEUE_LIMIT,
ENV_AUDIT_MQTT_RECONNECT_INTERVAL, ENV_AUDIT_MQTT_TOPIC, ENV_AUDIT_MQTT_USERNAME, ENV_AUDIT_WEBHOOK_AUTH_TOKEN,
ENV_AUDIT_WEBHOOK_CLIENT_CERT, ENV_AUDIT_WEBHOOK_CLIENT_KEY, ENV_AUDIT_WEBHOOK_ENABLE, ENV_AUDIT_WEBHOOK_ENDPOINT,
ENV_AUDIT_WEBHOOK_QUEUE_DIR, ENV_AUDIT_WEBHOOK_QUEUE_LIMIT,
};
use rustfs_config::notify::{
ENV_NOTIFY_MQTT_BROKER, ENV_NOTIFY_MQTT_ENABLE, ENV_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL, ENV_NOTIFY_MQTT_PASSWORD,
ENV_NOTIFY_MQTT_QOS, ENV_NOTIFY_MQTT_QUEUE_DIR, ENV_NOTIFY_MQTT_QUEUE_LIMIT, ENV_NOTIFY_MQTT_RECONNECT_INTERVAL,
ENV_NOTIFY_MQTT_TOPIC, ENV_NOTIFY_MQTT_USERNAME, ENV_NOTIFY_WEBHOOK_AUTH_TOKEN, ENV_NOTIFY_WEBHOOK_CLIENT_CERT,
ENV_NOTIFY_WEBHOOK_CLIENT_KEY, ENV_NOTIFY_WEBHOOK_ENABLE, ENV_NOTIFY_WEBHOOK_ENDPOINT, ENV_NOTIFY_WEBHOOK_QUEUE_DIR,
ENV_NOTIFY_WEBHOOK_QUEUE_LIMIT, NOTIFY_MQTT_SUB_SYS, NOTIFY_SUB_SYSTEMS, NOTIFY_WEBHOOK_SUB_SYS,
};
use rustfs_config::oidc::{
ENV_IDENTITY_OPENID_CLAIM_NAME, ENV_IDENTITY_OPENID_CLAIM_PREFIX, ENV_IDENTITY_OPENID_CLIENT_ID,
ENV_IDENTITY_OPENID_CLIENT_SECRET, ENV_IDENTITY_OPENID_CONFIG_URL, ENV_IDENTITY_OPENID_DISPLAY_NAME,
ENV_IDENTITY_OPENID_EMAIL_CLAIM, ENV_IDENTITY_OPENID_ENABLE, ENV_IDENTITY_OPENID_GROUPS_CLAIM, ENV_IDENTITY_OPENID_ISSUER,
ENV_IDENTITY_OPENID_REDIRECT_URI, ENV_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC, ENV_IDENTITY_OPENID_ROLE_POLICY,
ENV_IDENTITY_OPENID_SCOPES, ENV_IDENTITY_OPENID_USERNAME_CLAIM, IDENTITY_OPENID_SUB_SYS, OIDC_CLAIM_NAME, OIDC_CLAIM_PREFIX,
OIDC_CLIENT_ID, OIDC_CLIENT_SECRET, OIDC_CONFIG_URL, OIDC_DISPLAY_NAME, OIDC_EMAIL_CLAIM, OIDC_GROUPS_CLAIM, OIDC_ISSUER,
OIDC_REDIRECT_URI, OIDC_REDIRECT_URI_DYNAMIC, OIDC_ROLE_POLICY, OIDC_SCOPES, OIDC_USERNAME_CLAIM,
};
use rustfs_config::server_config::{Config as ServerConfig, DEFAULT_KVS, KV, KVS};
use rustfs_config::{
COMMENT_KEY, DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, ENV_SCANNER_ALERT_EXCESS_FOLDERS,
ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, ENV_SCANNER_ALERT_EXCESS_VERSIONS, ENV_SCANNER_BITROT_CYCLE_SECS,
ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES,
ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, ENV_SCANNER_IDLE_MODE,
ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED,
ENV_SCANNER_START_DELAY_SECS, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, HEAL_BITROT_CYCLE, HEAL_SUB_SYS,
MAX_ADMIN_REQUEST_BODY_SIZE, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR,
MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, SCANNER_ALERT_EXCESS_FOLDERS,
SCANNER_ALERT_EXCESS_VERSION_SIZE, SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT,
SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY,
SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED,
SCANNER_START_DELAY, SCANNER_SUB_SYS, SCANNER_YIELD_EVERY_N_OBJECTS, WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE,
WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR,
WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL,
};
use rustfs_credentials::Credentials;
use rustfs_policy::policy::action::{Action, AdminAction};
use s3s::header::CONTENT_TYPE;
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
use serde::Serialize;
use std::collections::{BTreeSet, HashMap};
use std::env;
use std::future::Future;
use time::OffsetDateTime;
use tracing::warn;
use uuid::Uuid;
const REDACTED_VALUE: &str = "*redacted*";
const OCTET_STREAM_CONTENT_TYPE: &str = "application/octet-stream";
const JSON_CONTENT_TYPE: &str = "application/json";
const TEXT_CONTENT_TYPE: &str = "text/plain; charset=utf-8";
const CONFIG_HISTORY_PREFIX: &str = "config/history";
const CONFIG_HISTORY_SUFFIX: &str = ".kv";
const CONFIG_APPLIED_HEADER: &str = "x-rustfs-config-applied";
const CONFIG_APPLIED_COMPAT_HEADER: &str = "x-minio-config-applied";
const CONFIG_APPLIED_TRUE: &str = "true";
const DEFAULT_COMMENT_DESCRIPTION: &str = "optionally add a comment to this setting";
#[derive(Debug, Clone, PartialEq, Eq)]
struct ConfigEntry {
key: String,
value: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ConfigDirective {
sub_system: String,
target: String,
entries: Vec<ConfigEntry>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ConfigSelector {
sub_system: String,
target: Option<String>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct HelpSubSystemMetadata {
key: &'static str,
description: &'static str,
multiple_targets: bool,
keys: &'static [HelpKeyMetadata],
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct HelpKeyMetadata {
key: &'static str,
type_name: &'static str,
description: &'static str,
optional: bool,
}
#[derive(Debug, Serialize, PartialEq, Eq)]
struct ConfigHelpResponse {
#[serde(rename = "subSys")]
sub_sys: String,
description: String,
#[serde(rename = "multipleTargets")]
multiple_targets: bool,
#[serde(rename = "keysHelp")]
keys_help: Vec<ConfigHelpEntry>,
}
#[derive(Debug, Serialize, PartialEq, Eq)]
struct ConfigHelpEntry {
key: String,
#[serde(rename = "type")]
type_name: String,
description: String,
optional: bool,
#[serde(rename = "multipleTargets")]
multiple_targets: bool,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
struct ConfigHistoryEntry {
#[serde(rename = "RestoreID")]
restore_id: String,
#[serde(rename = "CreateTime", with = "time::serde::rfc3339")]
create_time: OffsetDateTime,
#[serde(rename = "Data", skip_serializing_if = "Option::is_none")]
data: Option<String>,
}
const STORAGE_CLASS_HELP_KEYS: &[HelpKeyMetadata] = &[
HelpKeyMetadata {
key: "standard",
type_name: "string",
description: "set the parity count for default standard storage class",
optional: true,
},
HelpKeyMetadata {
key: "rrs",
type_name: "string",
description: "set the parity count for reduced redundancy storage class",
optional: true,
},
HelpKeyMetadata {
key: "optimize",
type_name: "string",
description: "optimize parity calculation for standard storage class, set 'capacity' for capacity optimized",
optional: true,
},
HelpKeyMetadata {
key: "inline_block",
type_name: "string",
description: "set the shard size threshold considered for inline blocks",
optional: true,
},
];
const SCANNER_HELP_KEYS: &[HelpKeyMetadata] = &[
HelpKeyMetadata {
key: SCANNER_SPEED,
type_name: "fastest|fast|default|slow|slowest",
description: "set scanner throttling preset",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_DELAY,
type_name: "float",
description: "override scanner sleep multiplier derived from speed",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_MAX_WAIT,
type_name: "seconds",
description: "override scanner maximum sleep duration derived from speed",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_CYCLE,
type_name: "seconds",
description: "override scanner cycle interval in seconds",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_START_DELAY,
type_name: "seconds",
description: "set scanner startup delay in seconds; used as legacy cycle interval when cycle is unset",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_CYCLE_MAX_DURATION,
type_name: "seconds",
description: "cap one scanner cycle runtime in seconds, 0 disables the cap",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_CYCLE_MAX_OBJECTS,
type_name: "number",
description: "cap objects processed by one scanner cycle, 0 disables the cap",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_CYCLE_MAX_DIRECTORIES,
type_name: "number",
description: "cap directories entered by one scanner cycle, 0 disables the cap",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_BITROT_CYCLE,
type_name: "seconds|off",
description: "set periodic deep bitrot scan cycle, 0 scans deeply every cycle, off disables periodic deep scans",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_IDLE_MODE,
type_name: "on|off",
description: "enable scanner throttling sleeps between operations",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_CACHE_SAVE_TIMEOUT,
type_name: "seconds",
description: "set scanner data-usage cache save timeout in seconds",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_MAX_CONCURRENT_SET_SCANS,
type_name: "number",
description: "cap concurrent scanner set tasks, 0 uses topology defaults",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_MAX_CONCURRENT_DISK_SCANS,
type_name: "number",
description: "cap concurrent disk bucket walks per set, 0 uses disk-count defaults",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_YIELD_EVERY_N_OBJECTS,
type_name: "number",
description: "yield to the async runtime after this many scanned objects, 0 disables extra object-count yields",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_ALERT_EXCESS_VERSIONS,
type_name: "number",
description: "object version count threshold for scanner alerts",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_ALERT_EXCESS_VERSION_SIZE,
type_name: "bytes",
description: "retained object version size threshold for scanner alerts",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_ALERT_EXCESS_FOLDERS,
type_name: "number",
description: "direct subfolder count threshold for scanner alerts",
optional: true,
},
];
const HEAL_HELP_KEYS: &[HelpKeyMetadata] = &[HelpKeyMetadata {
key: HEAL_BITROT_CYCLE,
type_name: "seconds|off",
description: "set scanner-driven periodic deep bitrot scan cycle, 0 scans deeply every cycle, off disables periodic deep scans",
optional: true,
}];
const OIDC_HELP_KEYS: &[HelpKeyMetadata] = &[
HelpKeyMetadata {
key: OIDC_CONFIG_URL,
type_name: "url",
description: "openid discovery document URL e.g. \"https://accounts.google.com/.well-known/openid-configuration\"",
optional: false,
},
HelpKeyMetadata {
key: OIDC_ISSUER,
type_name: "url",
description: "expected OpenID issuer URL when it differs from config_url",
optional: true,
},
HelpKeyMetadata {
key: OIDC_CLIENT_ID,
type_name: "string",
description: "unique public identifier for the client application",
optional: false,
},
HelpKeyMetadata {
key: OIDC_CLIENT_SECRET,
type_name: "string",
description: "secret for the client application identifier",
optional: true,
},
HelpKeyMetadata {
key: OIDC_SCOPES,
type_name: "csv",
description: "comma-separated list of OpenID scopes for the server",
optional: true,
},
HelpKeyMetadata {
key: OIDC_REDIRECT_URI,
type_name: "url",
description: "static redirect URI used when dynamic redirects are disabled",
optional: true,
},
HelpKeyMetadata {
key: OIDC_REDIRECT_URI_DYNAMIC,
type_name: "on|off",
description: "enable Host header based dynamic redirect URI",
optional: true,
},
HelpKeyMetadata {
key: OIDC_CLAIM_NAME,
type_name: "string",
description: "JWT canned policy claim name",
optional: true,
},
HelpKeyMetadata {
key: OIDC_CLAIM_PREFIX,
type_name: "string",
description: "prefix added to claims before policy mapping",
optional: true,
},
HelpKeyMetadata {
key: OIDC_ROLE_POLICY,
type_name: "string",
description: "IAM access policies mapped to this client application and identity provider",
optional: true,
},
HelpKeyMetadata {
key: OIDC_DISPLAY_NAME,
type_name: "string",
description: "friendly display name for this provider",
optional: true,
},
HelpKeyMetadata {
key: OIDC_GROUPS_CLAIM,
type_name: "string",
description: "claim name containing group memberships",
optional: true,
},
HelpKeyMetadata {
key: OIDC_EMAIL_CLAIM,
type_name: "string",
description: "claim name containing the user email",
optional: true,
},
HelpKeyMetadata {
key: OIDC_USERNAME_CLAIM,
type_name: "string",
description: "claim name containing the username",
optional: true,
},
];
const WEBHOOK_HELP_KEYS: &[HelpKeyMetadata] = &[
HelpKeyMetadata {
key: WEBHOOK_ENDPOINT,
type_name: "url",
description: "webhook server endpoint e.g. \"http://localhost:8080/rustfs/events\"",
optional: false,
},
HelpKeyMetadata {
key: WEBHOOK_AUTH_TOKEN,
type_name: "string",
description: "opaque string or JWT authorization token",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_QUEUE_DIR,
type_name: "path",
description: "staging dir for undelivered messages e.g. '/home/events'",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_QUEUE_LIMIT,
type_name: "number",
description: "maximum limit for undelivered messages",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_CLIENT_CERT,
type_name: "string",
description: "client cert for webhook mTLS authentication",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_CLIENT_KEY,
type_name: "string",
description: "client cert key for webhook mTLS authentication",
optional: true,
},
];
const AUDIT_WEBHOOK_HELP_KEYS: &[HelpKeyMetadata] = &[
HelpKeyMetadata {
key: WEBHOOK_ENDPOINT,
type_name: "url",
description: "HTTP(s) endpoint e.g. \"http://localhost:8080/rustfs/logs/audit\"",
optional: false,
},
HelpKeyMetadata {
key: WEBHOOK_AUTH_TOKEN,
type_name: "string",
description: "opaque string or JWT authorization token",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_CLIENT_CERT,
type_name: "string",
description: "mTLS certificate for webhook authentication",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_CLIENT_KEY,
type_name: "string",
description: "mTLS certificate key for webhook authentication",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_BATCH_SIZE,
type_name: "number",
description: "number of events per HTTP send to the webhook target",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_QUEUE_LIMIT,
type_name: "number",
description: "channel queue size for webhook targets",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_QUEUE_DIR,
type_name: "path",
description: "staging dir for undelivered audit messages e.g. '/home/audit-events'",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_MAX_RETRY,
type_name: "number",
description: "maximum retry count before audit events are dropped",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_RETRY_INTERVAL,
type_name: "duration",
description: "sleep between retries e.g. '10s'",
optional: true,
},
HelpKeyMetadata {
key: WEBHOOK_HTTP_TIMEOUT,
type_name: "duration",
description: "maximum duration for each HTTP request",
optional: true,
},
];
const MQTT_HELP_KEYS: &[HelpKeyMetadata] = &[
HelpKeyMetadata {
key: MQTT_BROKER,
type_name: "uri",
description: "MQTT server endpoint e.g. `tcp://localhost:1883`",
optional: false,
},
HelpKeyMetadata {
key: MQTT_TOPIC,
type_name: "string",
description: "name of the MQTT topic to publish",
optional: false,
},
HelpKeyMetadata {
key: MQTT_USERNAME,
type_name: "string",
description: "MQTT username",
optional: true,
},
HelpKeyMetadata {
key: MQTT_PASSWORD,
type_name: "string",
description: "MQTT password",
optional: true,
},
HelpKeyMetadata {
key: MQTT_QOS,
type_name: "number",
description: "quality of service priority for MQTT delivery",
optional: true,
},
HelpKeyMetadata {
key: MQTT_KEEP_ALIVE_INTERVAL,
type_name: "duration",
description: "keep-alive interval for MQTT connections in s,m,h,d",
optional: true,
},
HelpKeyMetadata {
key: MQTT_RECONNECT_INTERVAL,
type_name: "duration",
description: "reconnect interval for MQTT connections in s,m,h,d",
optional: true,
},
HelpKeyMetadata {
key: MQTT_QUEUE_DIR,
type_name: "path",
description: "staging dir for undelivered messages e.g. '/home/events'",
optional: true,
},
HelpKeyMetadata {
key: MQTT_QUEUE_LIMIT,
type_name: "number",
description: "maximum limit for undelivered messages",
optional: true,
},
];
const HELP_SUBSYSTEMS: &[HelpSubSystemMetadata] = &[
HelpSubSystemMetadata {
key: STORAGE_CLASS_SUB_SYS,
description: "define object level redundancy",
multiple_targets: false,
keys: STORAGE_CLASS_HELP_KEYS,
},
HelpSubSystemMetadata {
key: SCANNER_SUB_SYS,
description: "configure background data scanner scheduling, throttling, bitrot, and observability thresholds",
multiple_targets: false,
keys: SCANNER_HELP_KEYS,
},
HelpSubSystemMetadata {
key: HEAL_SUB_SYS,
description: "configure heal and scanner-driven bitrot behavior",
multiple_targets: false,
keys: HEAL_HELP_KEYS,
},
HelpSubSystemMetadata {
key: IDENTITY_OPENID_SUB_SYS,
description: "enable OpenID SSO support",
multiple_targets: true,
keys: OIDC_HELP_KEYS,
},
HelpSubSystemMetadata {
key: AUDIT_WEBHOOK_SUB_SYS,
description: "send audit logs to webhook endpoints",
multiple_targets: true,
keys: AUDIT_WEBHOOK_HELP_KEYS,
},
HelpSubSystemMetadata {
key: AUDIT_MQTT_SUB_SYS,
description: "send audit logs to MQTT endpoints",
multiple_targets: true,
keys: MQTT_HELP_KEYS,
},
HelpSubSystemMetadata {
key: NOTIFY_WEBHOOK_SUB_SYS,
description: "publish bucket notifications to webhook endpoints",
multiple_targets: true,
keys: WEBHOOK_HELP_KEYS,
},
HelpSubSystemMetadata {
key: NOTIFY_MQTT_SUB_SYS,
description: "publish bucket notifications to MQTT endpoints",
multiple_targets: true,
keys: MQTT_HELP_KEYS,
},
];
pub fn register_config_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
r.insert(
Method::GET,
format!("{ADMIN_PREFIX}/v3/get-config-kv").as_str(),
AdminOperation(&GetConfigKVHandler {}),
)?;
r.insert(
Method::PUT,
format!("{ADMIN_PREFIX}/v3/set-config-kv").as_str(),
AdminOperation(&SetConfigKVHandler {}),
)?;
r.insert(
Method::DELETE,
format!("{ADMIN_PREFIX}/v3/del-config-kv").as_str(),
AdminOperation(&DelConfigKVHandler {}),
)?;
r.insert(
Method::GET,
format!("{ADMIN_PREFIX}/v3/help-config-kv").as_str(),
AdminOperation(&HelpConfigKVHandler {}),
)?;
r.insert(
Method::GET,
format!("{ADMIN_PREFIX}/v3/list-config-history-kv").as_str(),
AdminOperation(&ListConfigHistoryKVHandler {}),
)?;
r.insert(
Method::DELETE,
format!("{ADMIN_PREFIX}/v3/clear-config-history-kv").as_str(),
AdminOperation(&ClearConfigHistoryKVHandler {}),
)?;
r.insert(
Method::PUT,
format!("{ADMIN_PREFIX}/v3/restore-config-history-kv").as_str(),
AdminOperation(&RestoreConfigHistoryKVHandler {}),
)?;
r.insert(
Method::GET,
format!("{ADMIN_PREFIX}/v3/config").as_str(),
AdminOperation(&GetConfigHandler {}),
)?;
r.insert(
Method::PUT,
format!("{ADMIN_PREFIX}/v3/config").as_str(),
AdminOperation(&SetConfigHandler {}),
)?;
Ok(())
}
fn extract_query_params(uri: &Uri) -> HashMap<String, String> {
let mut params = HashMap::new();
if let Some(query) = uri.query() {
for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
params.insert(key.into_owned(), value.into_owned());
}
}
params
}
async fn validate_config_admin_request(req: &S3Request<Body>) -> S3Result<Credentials> {
let Some(input_cred) = req.credentials.as_ref() else {
return Err(s3_error!(InvalidRequest, "missing credentials"));
};
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let remote_addr = req
.extensions
.get::<Option<RemoteAddr>>()
.and_then(|opt| opt.map(|addr| addr.0));
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::ConfigUpdateAdminAction)],
remote_addr,
)
.await?;
Ok(cred)
}
fn header_value(content_type: &str) -> S3Result<HeaderValue> {
HeaderValue::from_str(content_type)
.map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid content type: {err}")))
}
fn response_with_content_type(status: StatusCode, body: Vec<u8>, content_type: &str) -> S3Result<S3Response<(StatusCode, Body)>> {
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, header_value(content_type)?);
Ok(S3Response::with_headers((status, Body::from(body)), headers))
}
fn encode_config_payload(path: &str, secret_key: &str, data: Vec<u8>, plain_content_type: &str) -> S3Result<(Vec<u8>, String)> {
if is_compat_admin_request(path) {
let (encoded, _) = encode_compatible_admin_payload(path, secret_key, data)?;
Ok((encoded, OCTET_STREAM_CONTENT_TYPE.to_string()))
} else {
Ok((data, plain_content_type.to_string()))
}
}
fn success_response(config_applied: bool) -> S3Result<S3Response<(StatusCode, Body)>> {
if !config_applied {
return Ok(S3Response::new((StatusCode::OK, Body::default())));
}
let mut headers = HeaderMap::new();
headers.insert(CONFIG_APPLIED_HEADER, header_value(CONFIG_APPLIED_TRUE)?);
headers.insert(CONFIG_APPLIED_COMPAT_HEADER, header_value(CONFIG_APPLIED_TRUE)?);
Ok(S3Response::with_headers((StatusCode::OK, Body::default()), headers))
}
fn object_store() -> S3Result<std::sync::Arc<crate::admin::storage_api::runtime::ECStore>> {
let context = current_app_context();
current_object_store_handle_for_context(context.as_deref())
.ok_or_else(|| s3_error!(InternalError, "server storage not initialized"))
}
async fn load_server_config_from_store() -> S3Result<ServerConfig> {
let store = object_store()?;
read_admin_config_without_migrate(store)
.await
.map_err(ApiError::from)
.map_err(Into::into)
}
async fn load_server_config_from_store_locked() -> S3Result<ServerConfig> {
let store = object_store()?;
read_admin_config_without_migrate_no_lock(store)
.await
.map_err(ApiError::from)
.map_err(Into::into)
}
async fn load_active_server_config() -> S3Result<ServerConfig> {
if let Ok(config) = load_server_config_from_store().await {
return Ok(config);
}
let context = current_app_context();
current_server_config_for_context(context.as_deref())
.ok_or_else(|| s3_error!(InternalError, "server config is not initialized"))
}
async fn save_server_config_to_store_locked(config: &ServerConfig) -> S3Result<()> {
let store = object_store()?;
save_admin_server_config_no_lock(store, config)
.await
.map_err(ApiError::from)
.map_err(Into::into)
}
fn history_object_name(restore_id: &str) -> String {
format!("{CONFIG_HISTORY_PREFIX}/{restore_id}{CONFIG_HISTORY_SUFFIX}")
}
fn config_update_sub_system(directives: &[ConfigDirective]) -> S3Result<Option<&str>> {
let sub_systems = directives
.iter()
.map(|directive| directive.sub_system.as_str())
.collect::<BTreeSet<_>>();
if sub_systems.len() > 1 {
return Err(s3_error!(InvalidRequest, "config update must target a single subsystem"));
}
Ok(sub_systems.iter().next().copied())
}
fn validate_config_directives(directives: &[ConfigDirective]) -> S3Result<()> {
if DEFAULT_KVS.get().is_none() {
crate::admin::storage_api::config::init_admin_config_defaults();
}
let Some(defaults) = DEFAULT_KVS.get() else {
return Err(s3_error!(InternalError, "config defaults are not initialized"));
};
for directive in directives {
let Some(default_kvs) = defaults.get(&directive.sub_system) else {
return Err(s3_error!(InvalidRequest, "unsupported config subsystem '{}'", directive.sub_system));
};
let valid_keys = default_kvs.keys().into_iter().collect::<BTreeSet<_>>();
for entry in &directive.entries {
if !valid_keys.contains(&entry.key) {
return Err(s3_error!(
InvalidRequest,
"unsupported config key '{}' for subsystem '{}'",
entry.key,
directive.sub_system
));
}
}
}
Ok(())
}
fn history_restore_id_from_name(name: &str) -> Option<String> {
name.strip_prefix(&format!("{CONFIG_HISTORY_PREFIX}/"))?
.strip_suffix(CONFIG_HISTORY_SUFFIX)
.map(ToString::to_string)
}
fn trim_history_entries(mut entries: Vec<ConfigHistoryEntry>, count: Option<usize>) -> Vec<ConfigHistoryEntry> {
entries.sort_by_key(|lhs| lhs.create_time);
if let Some(count) = count
&& entries.len() > count
{
entries.drain(0..entries.len() - count);
}
entries
}
async fn save_server_config_history(data: &[u8]) -> S3Result<String> {
let restore_id = Uuid::new_v4().to_string();
let store = object_store()?;
save_admin_config(store, &history_object_name(&restore_id), data.to_vec())
.await
.map_err(ApiError::from)
.map_err(S3Error::from)?;
Ok(restore_id)
}
async fn read_server_config_history(restore_id: &str) -> S3Result<Vec<u8>> {
let store = object_store()?;
read_admin_config(store, &history_object_name(restore_id))
.await
.map_err(ApiError::from)
.map_err(Into::into)
}
async fn delete_server_config_history(restore_id: &str) -> S3Result<()> {
let store = object_store()?;
delete_admin_config(store, &history_object_name(restore_id))
.await
.map_err(ApiError::from)
.map_err(Into::into)
}
async fn list_server_config_history(with_data: bool, count: Option<usize>) -> S3Result<Vec<ConfigHistoryEntry>> {
let store = object_store()?;
let mut continuation_token = None;
let mut entries = Vec::new();
loop {
let page = store
.clone()
.list_objects_v2(
RUSTFS_META_BUCKET,
CONFIG_HISTORY_PREFIX,
continuation_token.clone(),
None,
1000,
false,
None,
false,
)
.await
.map_err(ApiError::from)
.map_err(S3Error::from)?;
for object in page.objects {
if object.is_dir {
continue;
}
let Some(restore_id) = history_restore_id_from_name(&object.name) else {
continue;
};
let data = if with_data {
Some(
String::from_utf8(
read_admin_config(store.clone(), &object.name)
.await
.map_err(ApiError::from)
.map_err(S3Error::from)?,
)
.map_err(ApiError::other)
.map_err(S3Error::from)?,
)
} else {
None
};
entries.push(ConfigHistoryEntry {
restore_id,
create_time: object.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH),
data,
});
}
if !page.is_truncated {
break;
}
continuation_token = page.next_continuation_token;
if continuation_token.is_none() {
break;
}
}
Ok(trim_history_entries(entries, count))
}
fn normalize_target(target: &str) -> String {
if target.trim().is_empty() || target == "default" || target == DEFAULT_DELIMITER {
DEFAULT_DELIMITER.to_string()
} else {
target.trim().to_string()
}
}
fn format_scope(sub_system: &str, target: &str) -> String {
if target == DEFAULT_DELIMITER {
sub_system.to_string()
} else {
format!("{sub_system}:{target}")
}
}
fn escape_config_value(value: &str) -> String {
value.replace('\\', "\\\\").replace('"', "\\\"")
}
fn format_kv_pair(key: &str, value: &str) -> String {
format!(r#"{key}="{}""#, escape_config_value(value))
}
fn tokenize_config_line(line: &str) -> S3Result<Vec<String>> {
let mut tokens = Vec::new();
let mut current = String::new();
let mut quote: Option<char> = None;
let mut escaped = false;
for ch in line.chars() {
if escaped {
current.push(ch);
escaped = false;
continue;
}
if ch == '\\' {
escaped = true;
continue;
}
if let Some(active_quote) = quote {
if ch == active_quote {
quote = None;
} else {
current.push(ch);
}
continue;
}
match ch {
'"' | '\'' => quote = Some(ch),
c if c.is_whitespace() => {
if !current.is_empty() {
tokens.push(std::mem::take(&mut current));
}
}
_ => current.push(ch),
}
}
if escaped {
current.push('\\');
}
if quote.is_some() {
return Err(s3_error!(InvalidRequest, "unterminated quoted config value"));
}
if !current.is_empty() {
tokens.push(current);
}
Ok(tokens)
}
fn parse_directive_scope(scope: &str) -> S3Result<(String, String)> {
let (sub_system, target) = match scope.split_once(':') {
Some((sub_system, target)) => (sub_system.trim(), normalize_target(target)),
None => (scope.trim(), DEFAULT_DELIMITER.to_string()),
};
if sub_system.is_empty() {
return Err(s3_error!(InvalidRequest, "missing config subsystem"));
}
Ok((sub_system.to_string(), target))
}
fn parse_config_directives(input: &str, allow_bare_keys: bool) -> S3Result<Vec<ConfigDirective>> {
let mut directives = Vec::new();
for raw_line in input.lines() {
let line = raw_line.trim();
if line.is_empty() || line.starts_with('#') {
continue;
}
let tokens = tokenize_config_line(line)?;
if tokens.is_empty() {
continue;
}
let (sub_system, target) = parse_directive_scope(&tokens[0])?;
let mut entries = Vec::new();
for token in tokens.iter().skip(1) {
if let Some((key, value)) = token.split_once('=') {
let key = key.trim();
if key.is_empty() {
return Err(s3_error!(InvalidRequest, "config key cannot be empty"));
}
entries.push(ConfigEntry {
key: key.to_string(),
value: Some(value.to_string()),
});
} else if allow_bare_keys {
entries.push(ConfigEntry {
key: token.trim().to_string(),
value: None,
});
} else {
return Err(s3_error!(InvalidRequest, "config assignment must use key=value syntax"));
}
}
directives.push(ConfigDirective {
sub_system,
target,
entries,
});
}
Ok(directives)
}
fn parse_config_selector(input: &str) -> S3Result<ConfigSelector> {
let raw = input.trim();
if raw.is_empty() {
return Err(s3_error!(InvalidRequest, "missing config key selector"));
}
let (sub_system, target) = match raw.split_once(':') {
Some((sub_system, target)) => (sub_system.trim(), Some(normalize_target(target))),
None => (raw, None),
};
if sub_system.is_empty() {
return Err(s3_error!(InvalidRequest, "missing config subsystem"));
}
Ok(ConfigSelector {
sub_system: sub_system.to_string(),
target,
})
}
fn set_kvs_value(kvs: &mut KVS, key: &str, value: String) {
if let Some(existing) = kvs.0.iter_mut().find(|entry| entry.key == key) {
existing.value = value;
return;
}
kvs.0.push(KV {
key: key.to_string(),
value,
hidden_if_empty: false,
});
}
fn is_sensitive_key_name(key: &str) -> bool {
let normalized = key.trim().to_ascii_lowercase();
normalized.contains("secret")
|| normalized.contains("password")
|| normalized == "token"
|| normalized.ends_with("_token")
|| normalized == "client_key"
|| normalized.ends_with("_client_key")
|| normalized == "tls_client_key"
|| normalized.ends_with("_tls_client_key")
|| normalized == "private_key"
|| normalized.ends_with("_private_key")
}
fn apply_set_directives(config: &mut ServerConfig, directives: &[ConfigDirective]) -> S3Result<()> {
for directive in directives {
let targets = config.0.entry(directive.sub_system.clone()).or_default();
let kvs = targets.entry(directive.target.clone()).or_default();
for entry in &directive.entries {
let value = entry.value.clone().ok_or_else(|| {
s3_error!(
InvalidRequest,
"config key '{}' in subsystem '{}' is missing a value",
entry.key,
directive.sub_system
)
})?;
set_kvs_value(kvs, &entry.key, value);
}
}
config.set_defaults();
Ok(())
}
fn apply_delete_directives(config: &mut ServerConfig, directives: &[ConfigDirective]) {
for directive in directives {
let mut remove_subsystem = false;
if let Some(targets) = config.0.get_mut(&directive.sub_system) {
if directive.entries.is_empty() {
targets.remove(&directive.target);
} else if let Some(kvs) = targets.get_mut(&directive.target) {
let keys = directive
.entries
.iter()
.map(|entry| entry.key.as_str())
.collect::<BTreeSet<_>>();
kvs.0.retain(|entry| !keys.contains(entry.key.as_str()));
if kvs.0.is_empty() {
targets.remove(&directive.target);
}
}
remove_subsystem = targets.is_empty();
}
if remove_subsystem {
config.0.remove(&directive.sub_system);
}
}
config.set_defaults();
}
fn render_entry_value(entry: &KV, redact_secrets: bool) -> String {
if redact_secrets && (entry.hidden_if_empty || is_sensitive_key_name(&entry.key)) && !entry.value.trim().is_empty() {
REDACTED_VALUE.to_string()
} else {
entry.value.clone()
}
}
fn should_render_config_entry(entry: &KV) -> bool {
if entry.hidden_if_empty && entry.value.trim().is_empty() {
return false;
}
if entry.key == ENABLE_KEY
&& entry
.value
.parse::<rustfs_config::EnableState>()
.map(|state| state.is_enabled())
.unwrap_or(false)
{
return false;
}
true
}
fn sorted_kv_entries(kvs: &KVS) -> Vec<&KV> {
let mut entries = kvs.0.iter().filter(|entry| entry.key != COMMENT_KEY).collect::<Vec<_>>();
entries.sort_by(|lhs, rhs| lhs.key.cmp(&rhs.key));
entries
}
fn render_scope_line(sub_system: &str, target: &str, kvs: &KVS, redact_secrets: bool) -> Option<String> {
let entries = sorted_kv_entries(kvs)
.into_iter()
.filter(|entry| should_render_config_entry(entry))
.collect::<Vec<_>>();
if entries.is_empty() {
return None;
}
let pairs = entries
.into_iter()
.map(|entry| format_kv_pair(&entry.key, &render_entry_value(entry, redact_secrets)))
.collect::<Vec<_>>();
Some(format!("{} {}", format_scope(sub_system, target), pairs.join(" ")))
}
fn env_var_name_for_target(sub_system: &str, target: &str, key: &str) -> String {
let base = env_help_key(sub_system, key);
if lookup_help_subsystem(sub_system).is_some_and(|metadata| metadata.multiple_targets) && target != DEFAULT_DELIMITER {
format!("{base}_{}", target.to_ascii_uppercase())
} else {
base
}
}
fn config_target_keys(kvs: &KVS) -> Vec<&str> {
let mut keys = sorted_kv_entries(kvs)
.into_iter()
.map(|entry| entry.key.as_str())
.collect::<Vec<_>>();
if !keys.contains(&COMMENT_KEY) {
keys.push(COMMENT_KEY);
}
keys
}
fn read_non_empty_env_vars() -> Vec<(String, String)> {
env::vars().filter(|(_, value)| !value.is_empty()).collect()
}
fn discover_env_targets(sub_system: &str, kvs: &KVS, env_vars: &[(String, String)]) -> Vec<String> {
if !lookup_help_subsystem(sub_system).is_some_and(|metadata| metadata.multiple_targets) {
return Vec::new();
}
let mut targets = BTreeSet::new();
for key in config_target_keys(kvs) {
let prefix = format!("{}_", env_var_name_for_target(sub_system, DEFAULT_DELIMITER, key));
for (name, _) in env_vars {
if let Some(target) = name.strip_prefix(&prefix)
&& !target.is_empty()
{
targets.insert(target.to_ascii_lowercase());
}
}
}
targets.into_iter().collect()
}
fn render_env_override_lines(
sub_system: &str,
target: &str,
kvs: &KVS,
redact_secrets: bool,
env_vars: &[(String, String)],
) -> Vec<String> {
let mut lines = Vec::new();
for key in config_target_keys(kvs) {
let env_name = env_var_name_for_target(sub_system, target, key);
let Some((_, value)) = env_vars.iter().find(|(name, _)| name == &env_name) else {
continue;
};
if redact_secrets && is_sensitive_key_name(key) {
lines.push(format!("# {env_name}={REDACTED_VALUE}"));
continue;
}
lines.push(format!("# {env_name}={value}"));
}
lines
}
fn render_selected_config(config: &ServerConfig, selector: &ConfigSelector, redact_secrets: bool) -> S3Result<Vec<u8>> {
let Some(targets) = config.0.get(&selector.sub_system) else {
return Err(s3_error!(InvalidRequest, "config subsystem '{}' not found", selector.sub_system));
};
let env_vars = read_non_empty_env_vars();
let mut lines = Vec::new();
let mut sorted_targets = targets.iter().collect::<Vec<_>>();
sorted_targets.sort_by_key(|(lhs, _)| *lhs);
if let Some(target) = selector.target.as_ref() {
let default_kvs = targets.get(DEFAULT_DELIMITER).cloned().unwrap_or_else(KVS::new);
let discovered_targets = discover_env_targets(&selector.sub_system, &default_kvs, &env_vars);
let kvs = if let Some(kvs) = targets.get(target) {
kvs
} else if discovered_targets.iter().any(|name| name == target) {
&default_kvs
} else {
return Err(s3_error!(
InvalidRequest,
"config target '{}' not found for subsystem '{}'",
target,
selector.sub_system
));
};
let scope_start = lines.len();
lines.extend(render_env_override_lines(&selector.sub_system, target, kvs, redact_secrets, &env_vars));
if let Some(line) = render_scope_line(&selector.sub_system, target, kvs, redact_secrets) {
lines.push(line);
} else if lines.len() > scope_start {
lines.push(format_scope(&selector.sub_system, target));
}
} else {
let default_kvs = targets.get(DEFAULT_DELIMITER).cloned().unwrap_or_else(KVS::new);
let mut target_names = sorted_targets
.iter()
.map(|(target, _)| (*target).clone())
.collect::<BTreeSet<_>>();
for target in discover_env_targets(&selector.sub_system, &default_kvs, &env_vars) {
target_names.insert(target);
}
for target in target_names {
let kvs = targets.get(&target).unwrap_or(&default_kvs);
let scope_start = lines.len();
lines.extend(render_env_override_lines(&selector.sub_system, &target, kvs, redact_secrets, &env_vars));
if let Some(line) = render_scope_line(&selector.sub_system, &target, kvs, redact_secrets) {
lines.push(line);
} else if lines.len() > scope_start {
lines.push(format_scope(&selector.sub_system, &target));
}
}
}
Ok(lines.join("\n").into_bytes())
}
fn render_full_config(config: &ServerConfig) -> Vec<u8> {
let mut subsystems = config.0.iter().collect::<Vec<_>>();
subsystems.sort_by_key(|(lhs, _)| *lhs);
let mut lines = Vec::new();
for (sub_system, targets) in subsystems {
let mut sorted_targets = targets.iter().collect::<Vec<_>>();
sorted_targets.sort_by_key(|(lhs, _)| *lhs);
for (target, kvs) in sorted_targets {
if let Some(line) = render_scope_line(sub_system, target, kvs, false) {
lines.push(line);
}
}
}
lines.join("\n").into_bytes()
}
fn lookup_help_subsystem(sub_system: &str) -> Option<&'static HelpSubSystemMetadata> {
HELP_SUBSYSTEMS.iter().find(|metadata| metadata.key == sub_system)
}
fn normalize_help_subsystem(sub_system: &str) -> &str {
sub_system
.split_once(':')
.map(|(base, _)| base.trim())
.unwrap_or_else(|| sub_system.trim())
}
fn env_help_key(sub_system: &str, key: &str) -> String {
match (sub_system, key) {
(STORAGE_CLASS_SUB_SYS, "standard") => STANDARD_ENV.to_string(),
(STORAGE_CLASS_SUB_SYS, "rrs") => RRS_ENV.to_string(),
(STORAGE_CLASS_SUB_SYS, "optimize") => OPTIMIZE_ENV.to_string(),
(STORAGE_CLASS_SUB_SYS, "inline_block") => INLINE_BLOCK_ENV.to_string(),
(SCANNER_SUB_SYS, SCANNER_SPEED) => ENV_SCANNER_SPEED.to_string(),
(SCANNER_SUB_SYS, SCANNER_DELAY) => ENV_SCANNER_DELAY.to_string(),
(SCANNER_SUB_SYS, SCANNER_MAX_WAIT) => ENV_SCANNER_MAX_WAIT_SECS.to_string(),
(SCANNER_SUB_SYS, SCANNER_CYCLE) => ENV_SCANNER_CYCLE.to_string(),
(SCANNER_SUB_SYS, SCANNER_START_DELAY) => ENV_SCANNER_START_DELAY_SECS.to_string(),
(SCANNER_SUB_SYS, SCANNER_CYCLE_MAX_DURATION) => ENV_SCANNER_CYCLE_MAX_DURATION_SECS.to_string(),
(SCANNER_SUB_SYS, SCANNER_CYCLE_MAX_OBJECTS) => ENV_SCANNER_CYCLE_MAX_OBJECTS.to_string(),
(SCANNER_SUB_SYS, SCANNER_CYCLE_MAX_DIRECTORIES) => ENV_SCANNER_CYCLE_MAX_DIRECTORIES.to_string(),
(SCANNER_SUB_SYS, SCANNER_BITROT_CYCLE) => ENV_SCANNER_BITROT_CYCLE_SECS.to_string(),
(SCANNER_SUB_SYS, SCANNER_IDLE_MODE) => ENV_SCANNER_IDLE_MODE.to_string(),
(SCANNER_SUB_SYS, SCANNER_CACHE_SAVE_TIMEOUT) => ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS.to_string(),
(SCANNER_SUB_SYS, SCANNER_MAX_CONCURRENT_SET_SCANS) => ENV_SCANNER_MAX_CONCURRENT_SET_SCANS.to_string(),
(SCANNER_SUB_SYS, SCANNER_MAX_CONCURRENT_DISK_SCANS) => ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS.to_string(),
(SCANNER_SUB_SYS, SCANNER_YIELD_EVERY_N_OBJECTS) => ENV_SCANNER_YIELD_EVERY_N_OBJECTS.to_string(),
(SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_VERSIONS) => ENV_SCANNER_ALERT_EXCESS_VERSIONS.to_string(),
(SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_VERSION_SIZE) => ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE.to_string(),
(SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_FOLDERS) => ENV_SCANNER_ALERT_EXCESS_FOLDERS.to_string(),
(HEAL_SUB_SYS, HEAL_BITROT_CYCLE) => ENV_SCANNER_BITROT_CYCLE_SECS.to_string(),
(IDENTITY_OPENID_SUB_SYS, ENABLE_KEY) => ENV_IDENTITY_OPENID_ENABLE.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_CONFIG_URL) => ENV_IDENTITY_OPENID_CONFIG_URL.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_ISSUER) => ENV_IDENTITY_OPENID_ISSUER.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_CLIENT_ID) => ENV_IDENTITY_OPENID_CLIENT_ID.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_CLIENT_SECRET) => ENV_IDENTITY_OPENID_CLIENT_SECRET.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_SCOPES) => ENV_IDENTITY_OPENID_SCOPES.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_REDIRECT_URI) => ENV_IDENTITY_OPENID_REDIRECT_URI.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_REDIRECT_URI_DYNAMIC) => ENV_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_CLAIM_NAME) => ENV_IDENTITY_OPENID_CLAIM_NAME.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_CLAIM_PREFIX) => ENV_IDENTITY_OPENID_CLAIM_PREFIX.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_ROLE_POLICY) => ENV_IDENTITY_OPENID_ROLE_POLICY.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_DISPLAY_NAME) => ENV_IDENTITY_OPENID_DISPLAY_NAME.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_GROUPS_CLAIM) => ENV_IDENTITY_OPENID_GROUPS_CLAIM.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_EMAIL_CLAIM) => ENV_IDENTITY_OPENID_EMAIL_CLAIM.to_string(),
(IDENTITY_OPENID_SUB_SYS, OIDC_USERNAME_CLAIM) => ENV_IDENTITY_OPENID_USERNAME_CLAIM.to_string(),
(NOTIFY_WEBHOOK_SUB_SYS, ENABLE_KEY) => ENV_NOTIFY_WEBHOOK_ENABLE.to_string(),
(NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_ENDPOINT) => ENV_NOTIFY_WEBHOOK_ENDPOINT.to_string(),
(NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_AUTH_TOKEN) => ENV_NOTIFY_WEBHOOK_AUTH_TOKEN.to_string(),
(NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_QUEUE_LIMIT) => ENV_NOTIFY_WEBHOOK_QUEUE_LIMIT.to_string(),
(NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_QUEUE_DIR) => ENV_NOTIFY_WEBHOOK_QUEUE_DIR.to_string(),
(NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_CLIENT_CERT) => ENV_NOTIFY_WEBHOOK_CLIENT_CERT.to_string(),
(NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_CLIENT_KEY) => ENV_NOTIFY_WEBHOOK_CLIENT_KEY.to_string(),
(NOTIFY_MQTT_SUB_SYS, ENABLE_KEY) => ENV_NOTIFY_MQTT_ENABLE.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_BROKER) => ENV_NOTIFY_MQTT_BROKER.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_TOPIC) => ENV_NOTIFY_MQTT_TOPIC.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_QOS) => ENV_NOTIFY_MQTT_QOS.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_USERNAME) => ENV_NOTIFY_MQTT_USERNAME.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_PASSWORD) => ENV_NOTIFY_MQTT_PASSWORD.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_RECONNECT_INTERVAL) => ENV_NOTIFY_MQTT_RECONNECT_INTERVAL.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_KEEP_ALIVE_INTERVAL) => ENV_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_QUEUE_DIR) => ENV_NOTIFY_MQTT_QUEUE_DIR.to_string(),
(NOTIFY_MQTT_SUB_SYS, MQTT_QUEUE_LIMIT) => ENV_NOTIFY_MQTT_QUEUE_LIMIT.to_string(),
(AUDIT_WEBHOOK_SUB_SYS, ENABLE_KEY) => ENV_AUDIT_WEBHOOK_ENABLE.to_string(),
(AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_ENDPOINT) => ENV_AUDIT_WEBHOOK_ENDPOINT.to_string(),
(AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_AUTH_TOKEN) => ENV_AUDIT_WEBHOOK_AUTH_TOKEN.to_string(),
(AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_QUEUE_LIMIT) => ENV_AUDIT_WEBHOOK_QUEUE_LIMIT.to_string(),
(AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_QUEUE_DIR) => ENV_AUDIT_WEBHOOK_QUEUE_DIR.to_string(),
(AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_CLIENT_CERT) => ENV_AUDIT_WEBHOOK_CLIENT_CERT.to_string(),
(AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_CLIENT_KEY) => ENV_AUDIT_WEBHOOK_CLIENT_KEY.to_string(),
(AUDIT_MQTT_SUB_SYS, ENABLE_KEY) => ENV_AUDIT_MQTT_ENABLE.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_BROKER) => ENV_AUDIT_MQTT_BROKER.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_TOPIC) => ENV_AUDIT_MQTT_TOPIC.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_QOS) => ENV_AUDIT_MQTT_QOS.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_USERNAME) => ENV_AUDIT_MQTT_USERNAME.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_PASSWORD) => ENV_AUDIT_MQTT_PASSWORD.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_RECONNECT_INTERVAL) => ENV_AUDIT_MQTT_RECONNECT_INTERVAL.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_KEEP_ALIVE_INTERVAL) => ENV_AUDIT_MQTT_KEEP_ALIVE_INTERVAL.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_QUEUE_DIR) => ENV_AUDIT_MQTT_QUEUE_DIR.to_string(),
(AUDIT_MQTT_SUB_SYS, MQTT_QUEUE_LIMIT) => ENV_AUDIT_MQTT_QUEUE_LIMIT.to_string(),
_ => format!("{ENV_PREFIX}{}_{}", sub_system.to_ascii_uppercase(), key.to_ascii_uppercase()),
}
}
fn default_help_postfix(sub_system: &str, key: &str) -> String {
if DEFAULT_KVS.get().is_none() {
crate::admin::storage_api::config::init_admin_config_defaults();
}
DEFAULT_KVS
.get()
.and_then(|defaults| defaults.get(sub_system))
.and_then(|kvs| kvs.lookup(key))
.filter(|value| !value.trim().is_empty())
.map(|value| format!(" (default: '{}')", value))
.unwrap_or_default()
}
fn help_description(sub_system: &str, key: &str, description: &str) -> String {
format!("{description}{}", default_help_postfix(sub_system, key))
}
fn build_top_level_help_response() -> ConfigHelpResponse {
ConfigHelpResponse {
sub_sys: String::new(),
description: String::new(),
multiple_targets: false,
keys_help: HELP_SUBSYSTEMS
.iter()
.map(|metadata| ConfigHelpEntry {
key: metadata.key.to_string(),
type_name: String::new(),
description: metadata.description.to_string(),
optional: false,
multiple_targets: metadata.multiple_targets,
})
.collect(),
}
}
fn build_help_entries(
metadata: &HelpSubSystemMetadata,
key_filter: Option<&str>,
env_only: bool,
) -> S3Result<Vec<ConfigHelpEntry>> {
let enable_entry = metadata.multiple_targets.then(|| ConfigHelpEntry {
key: if env_only {
env_help_key(metadata.key, ENABLE_KEY)
} else {
ENABLE_KEY.to_string()
},
type_name: "on|off".to_string(),
description: format!("enable {} target, default is 'off'", metadata.key),
optional: false,
multiple_targets: false,
});
let comment_entry = ConfigHelpEntry {
key: if env_only {
env_help_key(metadata.key, COMMENT_KEY)
} else {
COMMENT_KEY.to_string()
},
type_name: "sentence".to_string(),
description: DEFAULT_COMMENT_DESCRIPTION.to_string(),
optional: true,
multiple_targets: false,
};
let mut entries = if let Some(key_filter) = key_filter.filter(|value| !value.trim().is_empty()) {
if key_filter == COMMENT_KEY {
vec![comment_entry]
} else {
let entry = metadata
.keys
.iter()
.find(|entry| entry.key == key_filter)
.ok_or_else(|| s3_error!(InvalidRequest, "unknown key {} for sub-system {}", key_filter, metadata.key))?;
vec![ConfigHelpEntry {
key: if env_only {
env_help_key(metadata.key, entry.key)
} else {
entry.key.to_string()
},
type_name: entry.type_name.to_string(),
description: help_description(metadata.key, entry.key, entry.description),
optional: entry.optional,
multiple_targets: false,
}]
}
} else {
let mut entries = metadata
.keys
.iter()
.map(|entry| ConfigHelpEntry {
key: if env_only {
env_help_key(metadata.key, entry.key)
} else {
entry.key.to_string()
},
type_name: entry.type_name.to_string(),
description: help_description(metadata.key, entry.key, entry.description),
optional: entry.optional,
multiple_targets: false,
})
.collect::<Vec<_>>();
entries.push(comment_entry);
entries
};
if let Some(enable_entry) = enable_entry {
entries.insert(0, enable_entry);
}
Ok(entries)
}
fn build_help_response(sub_system: Option<&str>, key: Option<&str>, env_only: bool) -> S3Result<ConfigHelpResponse> {
let Some(sub_system) = sub_system.filter(|value| !value.trim().is_empty()) else {
return Ok(build_top_level_help_response());
};
let sub_system = normalize_help_subsystem(sub_system);
let metadata =
lookup_help_subsystem(sub_system).ok_or_else(|| s3_error!(InvalidRequest, "unknown sub-system {}", sub_system))?;
let entries = build_help_entries(metadata, key, env_only)?;
Ok(ConfigHelpResponse {
sub_sys: metadata.key.to_string(),
description: metadata.description.to_string(),
multiple_targets: metadata.multiple_targets,
keys_help: entries,
})
}
fn publish_prepared_config_snapshots(config: ServerConfig, prepared: PreparedRuntimeConfig) -> S3Result<()> {
prepared.publish_storage_class()?;
publish_server_config(config);
Ok(())
}
async fn commit_prepared_config(
config: ServerConfig,
prepared: PreparedRuntimeConfig,
persist: impl Future<Output = S3Result<()>>,
publish: impl FnOnce(ServerConfig, PreparedRuntimeConfig) -> S3Result<()>,
) -> S3Result<()> {
persist.await?;
publish(config, prepared)
}
/// Re-apply local mutable worker families after a full-config replacement.
/// Peers receive one full-snapshot signal after this returns; signaling each
/// family here as well would recreate audit/scanner targets twice per peer.
fn publish_notify_config_intent(
config: &ServerConfig,
sub_system: Option<&str>,
) -> Option<rustfs_notify::NotificationLifecycleTransition> {
(sub_system.is_none() || sub_system.is_some_and(|sub_system| NOTIFY_SUB_SYSTEMS.contains(&sub_system)))
.then(|| rustfs_notify::ensure_live_events().publish_config(config.clone()))
}
async fn preflight_notify_config_intent(sub_system: Option<&str>) -> S3Result<()> {
if sub_system.is_none() || sub_system.is_some_and(|sub_system| NOTIFY_SUB_SYSTEMS.contains(&sub_system)) {
preflight_dynamic_config_reload(sub_system.unwrap_or(NOTIFY_WEBHOOK_SUB_SYS)).await?;
}
Ok(())
}
async fn wait_notify_config_intent(transition: Option<rustfs_notify::NotificationLifecycleTransition>) -> S3Result<bool> {
let Some(transition) = transition else {
return Ok(false);
};
transition.wait().await.map_err(|err| {
warn!(error = %err, "Failed to apply local notification config");
s3_error!(InternalError, "failed to apply notification config")
})?;
Ok(true)
}
async fn apply_non_notify_dynamic_subsystems(config: &ServerConfig) -> Vec<String> {
let mut failures = Vec::new();
for sub_system in FULL_CONFIG_WORKER_SUBSYSTEMS {
if NOTIFY_SUB_SYSTEMS.contains(&sub_system) {
continue;
}
if apply_dynamic_config_for_subsystem(config, sub_system).await.is_err() {
failures.push(format!("local {sub_system}"));
warn!(
event = EVENT_CONFIG_WORKER_RELOAD_FAILED,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_CONFIG,
config_subsystem = sub_system,
state = CONFIG_WORKER_RELOAD_FAILURE_STATE,
reason = "apply_failed",
"Published server config but failed to reload a local worker subsystem"
);
}
}
failures
}
fn finish_config_reconciliation(errors: Vec<String>) -> S3Result<()> {
if errors.is_empty() {
Ok(())
} else {
Err(s3_error!(
InternalError,
"server config persisted but runtime convergence failed: {}",
errors.join("; ")
))
}
}
async fn reconcile_targeted_config(
config: ServerConfig,
sub_system: Option<String>,
storage_class_applied: bool,
notify_transition: Option<rustfs_notify::NotificationLifecycleTransition>,
) -> S3Result<bool> {
let mut errors = Vec::new();
let notify_applied = notify_transition.is_some();
if let Err(err) = wait_notify_config_intent(notify_transition).await {
warn!(error = %err, "Local notification config failed to converge");
errors.push("local notify".to_string());
}
let config_applied = if notify_applied {
if let Some(sub_system) = sub_system.as_deref()
&& let Err(err) = signal_dynamic_config_reload_checked(sub_system).await
{
warn!(config_subsystem = sub_system, error = %err, "Peer config reload failed");
errors.push(format!("peer {sub_system}"));
}
true
} else if storage_class_applied {
if let Err(err) = signal_dynamic_config_reload_checked(STORAGE_CLASS_SUB_SYS).await {
warn!(error = %err, "Peer storage-class reload failed");
errors.push(format!("peer {STORAGE_CLASS_SUB_SYS}"));
}
true
} else if let Some(sub_system) = sub_system.as_deref()
&& is_dynamic_config_subsystem(sub_system)
{
let config_applied = match apply_dynamic_config_for_subsystem(&config, sub_system).await {
Ok(applied) => applied,
Err(_) => {
warn!(config_subsystem = sub_system, reason = "apply_failed", "Local config reload failed");
errors.push(format!("local {sub_system}"));
false
}
};
if let Err(err) = signal_dynamic_config_reload_checked(sub_system).await {
warn!(config_subsystem = sub_system, error = %err, "Peer config reload failed");
errors.push(format!("peer {sub_system}"));
}
config_applied
} else {
if let Err(err) = signal_config_snapshot_reload_checked().await {
warn!(error = %err, "Peer config snapshot reload failed");
errors.push("peer config snapshot".to_string());
}
false
};
finish_config_reconciliation(errors)?;
Ok(config_applied)
}
async fn reconcile_full_config(
config: ServerConfig,
notify_transition: Option<rustfs_notify::NotificationLifecycleTransition>,
) -> S3Result<()> {
let mut errors = Vec::new();
if let Err(err) = wait_notify_config_intent(notify_transition).await {
warn!(error = %err, "Local notification config failed to converge");
errors.push("local notify".to_string());
}
if let Err(err) = signal_dynamic_config_reload_checked(STORAGE_CLASS_SUB_SYS).await {
warn!(error = %err, "Peer storage-class reload failed");
errors.push(format!("peer {STORAGE_CLASS_SUB_SYS}"));
}
errors.extend(apply_non_notify_dynamic_subsystems(&config).await);
if let Err(err) = signal_dynamic_config_reload_checked(NOTIFY_WEBHOOK_SUB_SYS).await {
warn!(error = %err, "Peer notification config reload failed");
errors.push("peer notify".to_string());
}
if let Err(err) = signal_config_snapshot_reload_checked().await {
warn!(error = %err, "Peer config snapshot reload failed");
errors.push("peer config snapshot".to_string());
}
finish_config_reconciliation(errors)
}
pub struct GetConfigKVHandler {}
#[async_trait::async_trait]
impl Operation for GetConfigKVHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let cred = validate_config_admin_request(&req).await?;
let queries = extract_query_params(&req.uri);
let selector = parse_config_selector(
queries
.get("key")
.ok_or_else(|| s3_error!(InvalidRequest, "missing config key selector"))?,
)?;
let config = load_active_server_config().await?;
let payload = render_selected_config(&config, &selector, true)?;
let (body, content_type) = encode_config_payload(req.uri.path(), &cred.secret_key, payload, TEXT_CONTENT_TYPE)?;
response_with_content_type(StatusCode::OK, body, &content_type)
}
}
pub struct SetConfigKVHandler {}
#[async_trait::async_trait]
impl Operation for SetConfigKVHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let cred = validate_config_admin_request(&req).await?;
let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;
let directives = parse_config_directives(std::str::from_utf8(&body).map_err(ApiError::other)?, false)?;
if directives.is_empty() {
return Err(s3_error!(InvalidRequest, "config update body is empty"));
}
validate_config_directives(&directives)?;
let sub_system = config_update_sub_system(&directives)?.map(str::to_owned);
let transaction_sub_system = sub_system.clone();
let config_store = object_store()?;
let config_applied = supervise_admin_mutation("config mutation", async move {
preflight_notify_config_intent(sub_system.as_deref()).await?;
let (config, storage_class_applied, notify_transition) =
with_admin_server_config_write_lock(config_store, move || async move {
let sub_system = transaction_sub_system.as_deref();
let mut config = load_server_config_from_store_locked().await?;
apply_set_directives(&mut config, &directives)?;
let prepared = prepare_server_config(&config, sub_system).await?;
save_server_config_history(&body).await?;
if sub_system == Some(STORAGE_CLASS_SUB_SYS) {
commit_prepared_config(
config.clone(),
prepared,
save_server_config_to_store_locked(&config),
publish_prepared_config_snapshots,
)
.await?;
} else {
save_server_config_to_store_locked(&config).await?;
publish_server_config(config.clone());
}
let notify_transition = publish_notify_config_intent(&config, sub_system);
Ok::<_, S3Error>((config, sub_system == Some(STORAGE_CLASS_SUB_SYS), notify_transition))
})
.await
.map_err(|err| s3_error!(InternalError, "failed to lock server config update: {}", err))??;
reconcile_targeted_config(config, sub_system, storage_class_applied, notify_transition).await
})
.await?;
success_response(config_applied)
}
}
pub struct DelConfigKVHandler {}
#[async_trait::async_trait]
impl Operation for DelConfigKVHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let cred = validate_config_admin_request(&req).await?;
let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;
let directives = parse_config_directives(std::str::from_utf8(&body).map_err(ApiError::other)?, true)?;
if directives.is_empty() {
return Err(s3_error!(InvalidRequest, "config delete body is empty"));
}
validate_config_directives(&directives)?;
let sub_system = config_update_sub_system(&directives)?.map(str::to_owned);
let transaction_sub_system = sub_system.clone();
let config_store = object_store()?;
let config_applied = supervise_admin_mutation("config mutation", async move {
preflight_notify_config_intent(sub_system.as_deref()).await?;
let (config, storage_class_applied, notify_transition) =
with_admin_server_config_write_lock(config_store, move || async move {
let sub_system = transaction_sub_system.as_deref();
let mut config = load_server_config_from_store_locked().await?;
apply_delete_directives(&mut config, &directives);
let prepared = prepare_server_config(&config, sub_system).await?;
save_server_config_history(&body).await?;
if sub_system == Some(STORAGE_CLASS_SUB_SYS) {
commit_prepared_config(
config.clone(),
prepared,
save_server_config_to_store_locked(&config),
publish_prepared_config_snapshots,
)
.await?;
} else {
save_server_config_to_store_locked(&config).await?;
publish_server_config(config.clone());
}
let notify_transition = publish_notify_config_intent(&config, sub_system);
Ok::<_, S3Error>((config, sub_system == Some(STORAGE_CLASS_SUB_SYS), notify_transition))
})
.await
.map_err(|err| s3_error!(InternalError, "failed to lock server config update: {}", err))??;
reconcile_targeted_config(config, sub_system, storage_class_applied, notify_transition).await
})
.await?;
success_response(config_applied)
}
}
pub struct HelpConfigKVHandler {}
#[async_trait::async_trait]
impl Operation for HelpConfigKVHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
validate_config_admin_request(&req).await?;
let queries = extract_query_params(&req.uri);
let response = build_help_response(
queries.get("subSys").map(String::as_str),
queries.get("key").map(String::as_str),
queries.contains_key("env"),
)?;
let body = serde_json::to_vec(&response).map_err(ApiError::other)?;
response_with_content_type(StatusCode::OK, body, JSON_CONTENT_TYPE)
}
}
pub struct ListConfigHistoryKVHandler {}
#[async_trait::async_trait]
impl Operation for ListConfigHistoryKVHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let cred = validate_config_admin_request(&req).await?;
let queries = extract_query_params(&req.uri);
let count = queries
.get("count")
.ok_or_else(|| s3_error!(InvalidRequest, "missing count query parameter"))?
.parse::<usize>()
.map_err(ApiError::other)
.map_err(S3Error::from)?;
let entries = list_server_config_history(true, Some(count)).await?;
let payload = serde_json::to_vec(&entries).map_err(ApiError::other).map_err(S3Error::from)?;
let (body, content_type) = encode_config_payload(req.uri.path(), &cred.secret_key, payload, JSON_CONTENT_TYPE)?;
response_with_content_type(StatusCode::OK, body, &content_type)
}
}
pub struct ClearConfigHistoryKVHandler {}
#[async_trait::async_trait]
impl Operation for ClearConfigHistoryKVHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
validate_config_admin_request(&req).await?;
let queries = extract_query_params(&req.uri);
let restore_id = queries
.get("restoreId")
.ok_or_else(|| s3_error!(InvalidRequest, "missing restoreId query parameter"))?;
if restore_id == "all" {
for entry in list_server_config_history(false, None).await? {
delete_server_config_history(&entry.restore_id).await?;
}
} else {
delete_server_config_history(restore_id).await?;
}
success_response(false)
}
}
pub struct RestoreConfigHistoryKVHandler {}
#[async_trait::async_trait]
impl Operation for RestoreConfigHistoryKVHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
validate_config_admin_request(&req).await?;
let queries = extract_query_params(&req.uri);
let restore_id = queries
.get("restoreId")
.ok_or_else(|| s3_error!(InvalidRequest, "missing restoreId query parameter"))?;
let history = read_server_config_history(restore_id).await?;
let directives = parse_config_directives(std::str::from_utf8(&history).map_err(ApiError::other)?, false)?;
if directives.is_empty() {
return Err(s3_error!(InvalidRequest, "history entry is empty"));
}
validate_config_directives(&directives)?;
let mut config = ServerConfig::new();
apply_set_directives(&mut config, &directives)?;
let prepared = prepare_server_config(&config, None).await?;
let config_store = object_store()?;
supervise_admin_mutation("config mutation", async move {
preflight_notify_config_intent(None).await?;
let persisted_config = config.clone();
let notify_transition = with_admin_server_config_write_lock(config_store, move || async move {
commit_prepared_config(
persisted_config.clone(),
prepared,
save_server_config_to_store_locked(&persisted_config),
publish_prepared_config_snapshots,
)
.await?;
Ok::<_, S3Error>(publish_notify_config_intent(&persisted_config, None))
})
.await
.map_err(|err| s3_error!(InternalError, "failed to lock server config restore: {}", err))??;
reconcile_full_config(config, notify_transition).await
})
.await?;
success_response(false)
}
}
pub struct GetConfigHandler {}
#[async_trait::async_trait]
impl Operation for GetConfigHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let cred = validate_config_admin_request(&req).await?;
let config = load_active_server_config().await?;
let payload = render_full_config(&config);
let (body, content_type) = encode_config_payload(req.uri.path(), &cred.secret_key, payload, TEXT_CONTENT_TYPE)?;
response_with_content_type(StatusCode::OK, body, &content_type)
}
}
pub struct SetConfigHandler {}
#[async_trait::async_trait]
impl Operation for SetConfigHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let cred = validate_config_admin_request(&req).await?;
let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;
let directives = parse_config_directives(std::str::from_utf8(&body).map_err(ApiError::other)?, false)?;
if directives.is_empty() {
return Err(s3_error!(InvalidRequest, "full config body is empty"));
}
validate_config_directives(&directives)?;
let mut config = ServerConfig::new();
apply_set_directives(&mut config, &directives)?;
let prepared = prepare_server_config(&config, None).await?;
let config_store = object_store()?;
supervise_admin_mutation("config mutation", async move {
preflight_notify_config_intent(None).await?;
save_server_config_history(&body).await?;
let persisted_config = config.clone();
let notify_transition = with_admin_server_config_write_lock(config_store, move || async move {
commit_prepared_config(
persisted_config.clone(),
prepared,
save_server_config_to_store_locked(&persisted_config),
publish_prepared_config_snapshots,
)
.await?;
Ok::<_, S3Error>(publish_notify_config_intent(&persisted_config, None))
})
.await
.map_err(|err| s3_error!(InternalError, "failed to lock full server config update: {}", err))??;
reconcile_full_config(config, notify_transition).await
})
.await?;
success_response(false)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Mutex};
#[tokio::test]
async fn prepared_config_commit_persists_before_publish() {
let events = Arc::new(Mutex::new(Vec::new()));
let persist_events = events.clone();
let publish_events = events.clone();
commit_prepared_config(
ServerConfig::new(),
PreparedRuntimeConfig::default(),
async move {
persist_events.lock().expect("persist events lock").push("persist");
Ok(())
},
move |_, _| {
publish_events.lock().expect("publish events lock").push("publish");
Ok(())
},
)
.await
.expect("prepared config commit");
assert_eq!(*events.lock().expect("result events lock"), ["persist", "publish"]);
}
#[tokio::test]
async fn prepared_config_commit_does_not_publish_after_persist_failure() {
let events = Arc::new(Mutex::new(Vec::new()));
let persist_events = events.clone();
let publish_events = events.clone();
let err = commit_prepared_config(
ServerConfig::new(),
PreparedRuntimeConfig::default(),
async move {
persist_events.lock().expect("persist events lock").push("persist");
Err(s3_error!(InternalError, "injected persistence failure"))
},
move |_, _| {
publish_events.lock().expect("publish events lock").push("publish");
Ok(())
},
)
.await
.expect_err("persistence failure must propagate");
assert_eq!(err.code(), &S3ErrorCode::InternalError);
assert_eq!(*events.lock().expect("result events lock"), ["persist"]);
}
#[test]
fn tokenize_config_line_handles_quotes_and_escapes() {
let tokens = tokenize_config_line(r#"identity_openid client_id="console app" client_secret="s3cr\"et" enable=on"#)
.expect("tokenize");
assert_eq!(
tokens,
vec![
"identity_openid".to_string(),
"client_id=console app".to_string(),
r#"client_secret=s3cr"et"#.to_string(),
"enable=on".to_string(),
]
);
}
#[test]
fn parse_selector_supports_all_targets_and_default_target() {
let all_targets = parse_config_selector("notify_webhook").expect("parse selector");
assert_eq!(
all_targets,
ConfigSelector {
sub_system: "notify_webhook".to_string(),
target: None,
}
);
let default_target = parse_config_selector("notify_webhook:").expect("parse selector");
assert_eq!(
default_target,
ConfigSelector {
sub_system: "notify_webhook".to_string(),
target: Some(DEFAULT_DELIMITER.to_string()),
}
);
}
#[test]
fn validate_config_directives_rejects_unknown_subsystem_and_key() {
let unsupported_subsystem =
parse_config_directives(r#"not_real key="value""#, false).expect("parse unsupported subsystem directive");
let err = validate_config_directives(&unsupported_subsystem).expect_err("unknown subsystem should fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
let unsupported_key =
parse_config_directives(r#"identity_openid not_real="value""#, false).expect("parse unsupported key directive");
let err = validate_config_directives(&unsupported_key).expect_err("unknown key should fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
}
#[test]
fn validate_config_directives_accepts_notify_kafka_sasl_keys() {
let directives = parse_config_directives(
r#"notify_kafka sasl_enable="on" sasl_mechanism="SCRAM-SHA-512" sasl_username="user" sasl_password="secret""#,
false,
)
.expect("parse notify kafka sasl directives");
validate_config_directives(&directives).expect("notify kafka SASL keys should be accepted");
}
#[test]
fn validate_config_directives_accepts_audit_kafka_sasl_keys() {
let directives = parse_config_directives(
r#"audit_kafka sasl_enable="on" sasl_mechanism="SCRAM-SHA-512" sasl_username="user" sasl_password="secret""#,
false,
)
.expect("parse audit kafka sasl directives");
validate_config_directives(&directives).expect("audit kafka SASL keys should be accepted");
}
#[test]
fn set_get_and_delete_config_kv_round_trip() {
let mut config = ServerConfig::new();
let directives = parse_config_directives(
r#"identity_openid config_url="https://issuer.example" client_id="console" client_secret="secret-value""#,
false,
)
.expect("parse directives");
apply_set_directives(&mut config, &directives).expect("apply directives");
let rendered = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "identity_openid".to_string(),
target: Some(DEFAULT_DELIMITER.to_string()),
},
true,
)
.expect("render config"),
)
.expect("utf8");
assert!(rendered.contains(r#"client_id="console""#));
assert!(rendered.contains(r#"client_secret="*redacted*""#));
let delete_directives = parse_config_directives("identity_openid client_secret", true).expect("parse delete directives");
apply_delete_directives(&mut config, &delete_directives);
let rendered_after_delete = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "identity_openid".to_string(),
target: Some(DEFAULT_DELIMITER.to_string()),
},
false,
)
.expect("render config"),
)
.expect("utf8");
assert!(!rendered_after_delete.contains("client_secret="));
}
#[test]
fn full_config_export_can_be_reapplied() {
crate::admin::storage_api::config::init_admin_config_defaults();
let mut original = ServerConfig::new();
apply_set_directives(
&mut original,
&parse_config_directives(
r#"storage_class standard="EC:2" rrs="EC:1"
identity_openid config_url="https://issuer.example" client_id="console""#,
false,
)
.expect("parse directives"),
)
.expect("apply original directives");
let exported = String::from_utf8(render_full_config(&original)).expect("utf8 export");
let mut restored = ServerConfig::new();
apply_set_directives(&mut restored, &parse_config_directives(&exported, false).expect("parse exported config"))
.expect("apply restored directives");
assert_eq!(render_full_config(&original), render_full_config(&restored));
}
#[test]
fn build_help_response_reports_known_keys() {
let response = build_help_response(Some("identity_openid"), Some("client_secret"), false).expect("help response");
assert_eq!(response.sub_sys, "identity_openid");
assert_eq!(response.description, "enable OpenID SSO support");
assert!(response.multiple_targets);
assert_eq!(response.keys_help.len(), 2);
assert_eq!(response.keys_help[0].key, "enable");
assert_eq!(response.keys_help[1].key, "client_secret");
assert_eq!(response.keys_help[1].type_name, "string");
}
#[test]
fn build_help_response_supports_env_only_keys() {
let response = build_help_response(Some("notify_webhook"), Some("endpoint"), true).expect("env help response");
assert_eq!(response.sub_sys, "notify_webhook");
assert_eq!(response.keys_help.len(), 2);
assert_eq!(response.keys_help[0].key, "RUSTFS_NOTIFY_WEBHOOK_ENABLE");
assert_eq!(response.keys_help[0].description, "enable notify_webhook target, default is 'off'");
assert_eq!(response.keys_help[1].key, "RUSTFS_NOTIFY_WEBHOOK_ENDPOINT");
}
#[test]
fn build_help_response_appends_default_value_postfix() {
crate::admin::storage_api::config::init_admin_config_defaults();
let response = build_help_response(Some("identity_openid"), Some("scopes"), false).expect("help response");
assert_eq!(response.keys_help.len(), 2);
assert_eq!(response.keys_help[1].type_name, "csv");
assert!(
response.keys_help[1]
.description
.contains("(default: 'openid,profile,email')")
);
}
#[test]
fn build_help_response_exposes_comment_key() {
let response = build_help_response(Some("notify_webhook"), Some("comment"), false).expect("comment help response");
assert_eq!(response.keys_help.len(), 2);
assert_eq!(response.keys_help[0].key, "enable");
assert_eq!(response.keys_help[1].key, "comment");
assert_eq!(response.keys_help[1].type_name, "sentence");
assert_eq!(response.keys_help[1].description, DEFAULT_COMMENT_DESCRIPTION);
}
#[test]
fn build_help_response_uses_target_specific_descriptions() {
let response = build_help_response(Some("notify_webhook"), Some("endpoint"), false).expect("webhook help response");
assert_eq!(response.keys_help.len(), 2);
assert_eq!(
response.keys_help[1].description,
"webhook server endpoint e.g. \"http://localhost:8080/rustfs/events\""
);
}
#[test]
fn build_help_response_ignores_target_suffix_in_subsystem_query() {
let response =
build_help_response(Some("notify_webhook:primary"), Some("endpoint"), false).expect("targeted help response");
assert_eq!(response.sub_sys, "notify_webhook");
assert_eq!(response.keys_help.len(), 2);
assert_eq!(response.keys_help[1].key, "endpoint");
}
#[test]
fn build_help_response_reports_scanner_keys() {
let response = build_help_response(Some("scanner"), Some("speed"), false).expect("scanner help response");
assert_eq!(response.sub_sys, "scanner");
assert!(!response.multiple_targets);
assert_eq!(response.keys_help.len(), 1);
assert_eq!(response.keys_help[0].key, "speed");
assert_eq!(response.keys_help[0].type_name, "fastest|fast|default|slow|slowest");
}
#[test]
fn build_help_response_reports_scanner_start_delay_legacy_cycle_behavior() {
let response = build_help_response(Some("scanner"), Some("start_delay"), false).expect("scanner help response");
assert_eq!(response.sub_sys, "scanner");
assert_eq!(response.keys_help.len(), 1);
assert_eq!(response.keys_help[0].key, "start_delay");
assert!(response.keys_help[0].description.contains("cycle"));
}
#[test]
fn validate_config_directives_accepts_scanner_pacing_keys() {
let input = format!("{SCANNER_SUB_SYS} {SCANNER_DELAY}=\"3.5\" {SCANNER_MAX_WAIT}=\"7\"");
let directives = parse_config_directives(&input, false).expect("parse scanner pacing directive");
validate_config_directives(&directives).expect("scanner pacing keys should be supported");
}
#[test]
fn build_help_response_reports_scanner_delay() {
let response = build_help_response(Some(SCANNER_SUB_SYS), Some(SCANNER_DELAY), false).expect("scanner help response");
assert_eq!(response.sub_sys, SCANNER_SUB_SYS);
assert_eq!(response.keys_help.len(), 1);
assert_eq!(response.keys_help[0].key, SCANNER_DELAY);
assert_eq!(response.keys_help[0].type_name, "float");
assert!(response.keys_help[0].description.contains("multiplier"));
}
#[test]
fn validate_config_directives_accepts_heal_bitrot_cycle() {
let input = format!("{HEAL_SUB_SYS} {HEAL_BITROT_CYCLE}=\"off\"");
let directives = parse_config_directives(&input, false).expect("parse heal directive");
validate_config_directives(&directives).expect("heal bitrot_cycle should be a supported config key");
}
#[test]
fn build_help_response_reports_heal_bitrot_cycle() {
let response = build_help_response(Some(HEAL_SUB_SYS), Some(HEAL_BITROT_CYCLE), false).expect("heal help response");
assert_eq!(response.sub_sys, HEAL_SUB_SYS);
assert!(!response.multiple_targets);
assert_eq!(response.keys_help.len(), 1);
assert_eq!(response.keys_help[0].key, HEAL_BITROT_CYCLE);
assert!(response.keys_help[0].description.contains("scanner"));
}
#[test]
fn build_top_level_help_response_uses_empty_type_names() {
let response = build_help_response(None, None, false).expect("top level help response");
assert!(!response.keys_help.is_empty());
assert!(response.keys_help.iter().all(|entry| entry.type_name.is_empty()));
}
#[test]
fn render_selected_config_includes_env_override_lines() {
crate::admin::storage_api::config::init_admin_config_defaults();
temp_env::with_vars(
[
("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("http://env.example")),
("RUSTFS_NOTIFY_WEBHOOK_AUTH_TOKEN_PRIMARY", Some("secret-token")),
],
|| {
let mut config = ServerConfig::new();
apply_set_directives(
&mut config,
&parse_config_directives(r#"notify_webhook:primary endpoint="http://file.example""#, false)
.expect("parse directives"),
)
.expect("apply directives");
let rendered = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "notify_webhook".to_string(),
target: None,
},
true,
)
.expect("render config"),
)
.expect("utf8");
assert!(rendered.contains("# RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY=http://env.example"));
assert!(!rendered.contains("RUSTFS_NOTIFY_WEBHOOK_AUTH_TOKEN_PRIMARY"));
},
);
}
#[test]
fn render_selected_config_lists_env_only_targets() {
crate::admin::storage_api::config::init_admin_config_defaults();
temp_env::with_vars([("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("http://env.example"))], || {
let config = ServerConfig::new();
let rendered = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "notify_webhook".to_string(),
target: None,
},
true,
)
.expect("render config"),
)
.expect("utf8");
assert!(rendered.contains("# RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY=http://env.example"));
assert!(rendered.contains("notify_webhook:primary"));
});
}
#[test]
fn render_selected_config_supports_specific_env_only_target_queries() {
crate::admin::storage_api::config::init_admin_config_defaults();
temp_env::with_vars([("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("http://env.example"))], || {
let config = ServerConfig::new();
let rendered = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "notify_webhook".to_string(),
target: Some("primary".to_string()),
},
true,
)
.expect("render config"),
)
.expect("utf8");
assert!(rendered.contains("# RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY=http://env.example"));
assert!(rendered.contains("notify_webhook:primary"));
});
}
#[test]
fn render_selected_config_orders_default_before_named_targets() {
crate::admin::storage_api::config::init_admin_config_defaults();
temp_env::with_vars([("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_ALPHA", Some("http://alpha.example"))], || {
let mut config = ServerConfig::new();
apply_set_directives(
&mut config,
&parse_config_directives(
r#"notify_webhook endpoint="http://default.example"
notify_webhook:beta endpoint="http://beta.example""#,
false,
)
.expect("parse directives"),
)
.expect("apply directives");
let rendered = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "notify_webhook".to_string(),
target: None,
},
true,
)
.expect("render config"),
)
.expect("utf8");
let default_index = rendered.find("notify_webhook ").expect("default target");
let alpha_index = rendered.find("notify_webhook:alpha").expect("alpha target");
let beta_index = rendered.find("notify_webhook:beta").expect("beta target");
assert!(default_index < alpha_index);
assert!(alpha_index < beta_index);
});
}
#[test]
fn render_scope_line_omits_enable_on_and_hidden_empty_values() {
let kvs = KVS(vec![
KV {
key: ENABLE_KEY.to_string(),
value: "on".to_string(),
hidden_if_empty: false,
},
KV {
key: WEBHOOK_ENDPOINT.to_string(),
value: "http://file.example".to_string(),
hidden_if_empty: false,
},
KV {
key: WEBHOOK_AUTH_TOKEN.to_string(),
value: String::new(),
hidden_if_empty: true,
},
]);
let rendered = render_scope_line("notify_webhook", "primary", &kvs, true).expect("render scope");
assert!(rendered.contains(r#"endpoint="http://file.example""#));
assert!(!rendered.contains("enable="));
assert!(!rendered.contains("auth_token="));
}
#[test]
fn history_object_name_round_trips_restore_id() {
let name = history_object_name("restore-123");
assert_eq!(name, "config/history/restore-123.kv");
assert_eq!(history_restore_id_from_name(&name).as_deref(), Some("restore-123"));
assert!(history_restore_id_from_name("config/history/restore-123.txt").is_none());
}
#[test]
fn trim_history_entries_keeps_most_recent_count_in_time_order() {
let entries = vec![
ConfigHistoryEntry {
restore_id: "oldest".to_string(),
create_time: OffsetDateTime::from_unix_timestamp(1).expect("timestamp"),
data: None,
},
ConfigHistoryEntry {
restore_id: "middle".to_string(),
create_time: OffsetDateTime::from_unix_timestamp(2).expect("timestamp"),
data: None,
},
ConfigHistoryEntry {
restore_id: "newest".to_string(),
create_time: OffsetDateTime::from_unix_timestamp(3).expect("timestamp"),
data: None,
},
];
let trimmed = trim_history_entries(entries, Some(2));
assert_eq!(trimmed.len(), 2);
assert_eq!(trimmed[0].restore_id, "middle");
assert_eq!(trimmed[1].restore_id, "newest");
}
#[test]
fn apply_set_directives_merge_into_existing_config() {
let mut config = ServerConfig::new();
apply_set_directives(
&mut config,
&parse_config_directives(
r#"storage_class standard="EC:2"
identity_openid client_id="existing-client""#,
false,
)
.expect("parse initial directives"),
)
.expect("apply initial directives");
let history_directives = parse_config_directives(
r#"identity_openid config_url="https://issuer.example" client_secret="restored-secret""#,
false,
)
.expect("parse history directives");
apply_set_directives(&mut config, &history_directives).expect("apply history directives");
let mut webhook = ServerConfig::new();
apply_set_directives(
&mut webhook,
&parse_config_directives(r#"notify_webhook client_key="s3cr3t-key""#, false).expect("parse client key directives"),
)
.expect("apply client key directives");
let webhook_rendered = String::from_utf8(
render_selected_config(
&webhook,
&ConfigSelector {
sub_system: "notify_webhook".to_string(),
target: Some(DEFAULT_DELIMITER.to_string()),
},
true,
)
.expect("render webhook config"),
)
.expect("utf8");
assert!(webhook_rendered.contains(r#"client_key="*redacted*""#));
let missing_value_directives = vec![ConfigDirective {
sub_system: "identity_openid".to_string(),
target: DEFAULT_DELIMITER.to_string(),
entries: vec![ConfigEntry {
key: "client_secret".to_string(),
value: None,
}],
}];
let err =
apply_set_directives(&mut ServerConfig::new(), &missing_value_directives).expect_err("missing value should fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
let storage_class = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "storage_class".to_string(),
target: Some(DEFAULT_DELIMITER.to_string()),
},
false,
)
.expect("render storage class"),
)
.expect("utf8");
assert!(storage_class.contains(r#"standard="EC:2""#));
let oidc = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "identity_openid".to_string(),
target: Some(DEFAULT_DELIMITER.to_string()),
},
false,
)
.expect("render oidc"),
)
.expect("utf8");
assert!(oidc.contains(r#"client_id="existing-client""#));
assert!(oidc.contains(r#"config_url="https://issuer.example""#));
assert!(oidc.contains(r#"client_secret="restored-secret""#));
}
#[test]
fn storage_class_get_target_none_matches_full_export() {
crate::admin::storage_api::config::init_admin_config_defaults();
let mut config = ServerConfig::new();
apply_set_directives(
&mut config,
&parse_config_directives(r#"storage_class standard="EC:4" rrs="EC:2""#, false).expect("parse"),
)
.expect("apply");
// Simulate "mc admin config export" (render_full_config)
let full_output = String::from_utf8(render_full_config(&config)).expect("utf8");
// Simulate "mc admin config get storage_class" (target=None, redact=true)
let selected_output = String::from_utf8(
render_selected_config(
&config,
&ConfigSelector {
sub_system: "storage_class".to_string(),
target: None,
},
true,
)
.expect("render"),
)
.expect("utf8");
assert!(full_output.contains("EC:4"), "full export should contain EC:4, got:\n{}", full_output);
assert!(
selected_output.contains("EC:4"),
"selected get should contain EC:4, got:\n{}",
selected_output
);
assert!(
selected_output.contains("EC:2"),
"selected get should contain EC:2, got:\n{}",
selected_output
);
let full_sc_line = full_output.lines().find(|l| l.starts_with("storage_class")).unwrap();
let selected_sc_line = selected_output.lines().find(|l| l.starts_with("storage_class")).unwrap();
assert_eq!(full_sc_line, selected_sc_line);
}
}
Reference in New Issue View Git Blame Copy Permalink