gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-12 16:16:55 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
a97f3a9c5218a4ce141b59a3c04e70bb100f5b9d
rustfs/crates/targets/src/config/target_args.rs
T
houseme 8cf11af649 fix(security): add shared outbound egress guard (#3567) 
* fix(security): block unsafe outbound webhook targets

* chore: keep issue-3557 plan local only
2026-06-18 15:30:09 +08:00

1161 lines
49 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 super::common::{
parse_target_bool, parse_url, validate_nats_server_config, validate_outbound_http_url, validate_pulsar_broker_config,
};
use crate::error::TargetError;
use crate::target::{
TargetType,
amqp::AMQPArgs,
kafka::{KAFKA_SASL_PLAIN, KafkaArgs},
mqtt::{MQTTArgs, MQTTTlsConfig, validate_mqtt_broker_url},
mysql::MySqlArgs,
nats::{NATSArgs, validate_nats_address},
postgres::{PostgresArgs, PostgresDsn, parse_postgres_format},
pulsar::{PulsarArgs, validate_pulsar_broker},
redis::{RedisArgs, RedisTlsConfig, validate_redis_url},
webhook::WebhookArgs,
};
use rumqttc::QoS;
use rustfs_config::server_config::KVS;
use rustfs_config::{
AMQP_EXCHANGE, AMQP_MANDATORY, AMQP_PASSWORD, AMQP_PERSISTENT, AMQP_QUEUE_DIR, AMQP_QUEUE_LIMIT, AMQP_ROUTING_KEY,
AMQP_TLS_CA, AMQP_TLS_CLIENT_CERT, AMQP_TLS_CLIENT_KEY, AMQP_URL, AMQP_USERNAME, DEFAULT_LIMIT, KAFKA_ACKS, KAFKA_BROKERS,
KAFKA_QUEUE_DIR, KAFKA_QUEUE_LIMIT, KAFKA_SASL_ENABLE, KAFKA_SASL_MECHANISM, KAFKA_SASL_PASSWORD, KAFKA_SASL_USERNAME,
KAFKA_TLS_CA, KAFKA_TLS_CLIENT_CERT, KAFKA_TLS_CLIENT_KEY, KAFKA_TLS_ENABLE, KAFKA_TOPIC, MQTT_BROKER,
MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA,
MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_POLICY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME,
MQTT_WS_PATH_ALLOWLIST, MYSQL_DSN_STRING, MYSQL_FORMAT, MYSQL_MAX_OPEN_CONNECTIONS, MYSQL_QUEUE_DIR, MYSQL_QUEUE_LIMIT,
MYSQL_TABLE, MYSQL_TLS_CA, MYSQL_TLS_CLIENT_CERT, MYSQL_TLS_CLIENT_KEY, NATS_ADDRESS, NATS_CREDENTIALS_FILE, NATS_PASSWORD,
NATS_QUEUE_DIR, NATS_QUEUE_LIMIT, NATS_SUBJECT, NATS_TLS_CA, NATS_TLS_CLIENT_CERT, NATS_TLS_CLIENT_KEY, NATS_TLS_REQUIRED,
NATS_TOKEN, NATS_USERNAME, POSTGRES_DSN_STRING, POSTGRES_FORMAT, POSTGRES_QUEUE_DIR, POSTGRES_QUEUE_LIMIT, POSTGRES_TABLE,
POSTGRES_TLS_CA, POSTGRES_TLS_CLIENT_CERT, POSTGRES_TLS_CLIENT_KEY, POSTGRES_TLS_REQUIRED, PULSAR_AUTH_TOKEN, PULSAR_BROKER,
PULSAR_PASSWORD, PULSAR_QUEUE_DIR, PULSAR_QUEUE_LIMIT, PULSAR_TLS_ALLOW_INSECURE, PULSAR_TLS_CA,
PULSAR_TLS_HOSTNAME_VERIFICATION, PULSAR_TOPIC, PULSAR_USERNAME, REDIS_CHANNEL, REDIS_CONNECTION_TIMEOUT,
REDIS_KEEP_ALIVE_INTERVAL, REDIS_MAX_RETRY_ATTEMPTS, REDIS_MAX_RETRY_DELAY, REDIS_MIN_RETRY_DELAY, REDIS_PASSWORD,
REDIS_PIPELINE_BUFFER_SIZE, REDIS_QUEUE_DIR, REDIS_QUEUE_LIMIT, REDIS_RECONNECT_RETRY_ATTEMPTS, REDIS_RESPONSE_TIMEOUT,
REDIS_TLS_ALLOW_INSECURE, REDIS_TLS_CA, REDIS_TLS_CLIENT_CERT, REDIS_TLS_CLIENT_KEY, REDIS_TLS_POLICY, REDIS_URL,
REDIS_USERNAME, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT, WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT,
WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY,
};
use std::path::Path;
use std::time::Duration;
fn parse_kafka_acks_value(value: Option<&str>) -> Result<i16, TargetError> {
let Some(value) = value else {
return Ok(1);
};
let normalized = value.trim();
if normalized.is_empty() {
return Err(TargetError::Configuration("Kafka acks must be one of: 0, 1, -1, all".to_string()));
}
match normalized.to_ascii_lowercase().as_str() {
"0" => Ok(0),
"1" => Ok(1),
"-1" | "all" => Ok(-1),
_ => Err(TargetError::Configuration("Kafka acks must be one of: 0, 1, -1, all".to_string())),
}
}
fn parse_kafka_sasl_enable(config: &KVS, has_sasl_fields: bool) -> Result<bool, TargetError> {
match config.lookup(KAFKA_SASL_ENABLE) {
Some(value) => {
if value.trim().is_empty() {
return Ok(has_sasl_fields);
}
parse_target_bool(Some(value.as_str()))
.ok_or_else(|| TargetError::Configuration(format!("Invalid Kafka {KAFKA_SASL_ENABLE} boolean value: {value}")))
}
None => Ok(has_sasl_fields),
}
}
fn parse_amqp_bool_value(field: &str, config: &KVS, default: bool) -> Result<bool, TargetError> {
match config.lookup(field) {
Some(value) => parse_target_bool(Some(value.as_str()))
.ok_or_else(|| TargetError::Configuration(format!("Invalid AMQP {field} boolean value: {value}"))),
None => Ok(default),
}
}
pub fn build_amqp_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<AMQPArgs, TargetError> {
let url = config
.lookup(AMQP_URL)
.ok_or_else(|| TargetError::Configuration("Missing AMQP url".to_string()))?;
let url = parse_url(url.trim(), "AMQP URL")?;
let exchange = config
.lookup(AMQP_EXCHANGE)
.ok_or_else(|| TargetError::Configuration("Missing AMQP exchange".to_string()))?;
let routing_key = config
.lookup(AMQP_ROUTING_KEY)
.ok_or_else(|| TargetError::Configuration("Missing AMQP routing_key".to_string()))?;
let args = AMQPArgs {
enable: true,
url,
exchange,
routing_key,
mandatory: parse_amqp_bool_value(AMQP_MANDATORY, config, false)?,
persistent: parse_amqp_bool_value(AMQP_PERSISTENT, config, true)?,
username: config.lookup(AMQP_USERNAME).unwrap_or_default(),
password: config.lookup(AMQP_PASSWORD).unwrap_or_default(),
tls_ca: config.lookup(AMQP_TLS_CA).unwrap_or_default(),
tls_client_cert: config.lookup(AMQP_TLS_CLIENT_CERT).unwrap_or_default(),
tls_client_key: config.lookup(AMQP_TLS_CLIENT_KEY).unwrap_or_default(),
queue_dir: config.lookup(AMQP_QUEUE_DIR).unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(AMQP_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
target_type,
};
args.validate()?;
Ok(args)
}
pub fn validate_amqp_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
let _ = build_amqp_args(config, default_queue_dir, TargetType::NotifyEvent)?;
Ok(())
}
pub fn build_webhook_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<WebhookArgs, TargetError> {
let endpoint = config
.lookup(WEBHOOK_ENDPOINT)
.ok_or_else(|| TargetError::Configuration("Missing webhook endpoint".to_string()))?;
let parsed_endpoint = endpoint.trim();
let endpoint_url = parse_url(parsed_endpoint, "endpoint URL")?;
validate_outbound_http_url(&endpoint_url, "endpoint URL")?;
Ok(WebhookArgs {
enable: true,
endpoint: endpoint_url,
auth_token: config.lookup(WEBHOOK_AUTH_TOKEN).unwrap_or_default(),
queue_dir: config
.lookup(WEBHOOK_QUEUE_DIR)
.unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(WEBHOOK_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
client_cert: config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default(),
client_key: config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default(),
client_ca: config.lookup(WEBHOOK_CLIENT_CA).unwrap_or_default(),
skip_tls_verify: config
.lookup(WEBHOOK_SKIP_TLS_VERIFY)
.and_then(|v| v.parse::<bool>().ok())
.unwrap_or(RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT),
target_type,
})
}
pub fn validate_webhook_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
let endpoint = config
.lookup(WEBHOOK_ENDPOINT)
.ok_or_else(|| TargetError::Configuration("Missing webhook endpoint".to_string()))?;
let parsed_endpoint = endpoint.trim();
let endpoint_url = parse_url(parsed_endpoint, "endpoint URL")?;
validate_outbound_http_url(&endpoint_url, "endpoint URL")?;
let client_cert = config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default();
let client_key = config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default();
if client_cert.is_empty() != client_key.is_empty() {
return Err(TargetError::Configuration(
"Both client_cert and client_key must be specified together".to_string(),
));
}
let queue_dir = config
.lookup(WEBHOOK_QUEUE_DIR)
.unwrap_or_else(|| default_queue_dir.to_string());
if !queue_dir.is_empty() && !Path::new(&queue_dir).is_absolute() {
return Err(TargetError::Configuration("Webhook queue directory must be an absolute path".to_string()));
}
Ok(())
}
pub fn build_mqtt_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<MQTTArgs, TargetError> {
let broker = config
.lookup(MQTT_BROKER)
.ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
let broker_url = parse_url(&broker, "broker URL")?;
let topic = config
.lookup(MQTT_TOPIC)
.ok_or_else(|| TargetError::Configuration("Missing MQTT topic".to_string()))?;
Ok(MQTTArgs {
enable: true,
broker: broker_url,
topic,
qos: config
.lookup(MQTT_QOS)
.and_then(|v| v.parse::<u8>().ok())
.map(|q| match q {
0 => QoS::AtMostOnce,
1 => QoS::AtLeastOnce,
2 => QoS::ExactlyOnce,
_ => QoS::AtLeastOnce,
})
.unwrap_or(QoS::AtLeastOnce),
username: config.lookup(MQTT_USERNAME).unwrap_or_default(),
password: config.lookup(MQTT_PASSWORD).unwrap_or_default(),
max_reconnect_interval: config
.lookup(MQTT_RECONNECT_INTERVAL)
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs)
.unwrap_or_else(|| Duration::from_secs(5)),
keep_alive: config
.lookup(MQTT_KEEP_ALIVE_INTERVAL)
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs)
.unwrap_or_else(|| Duration::from_secs(30)),
tls: MQTTTlsConfig::from_values(
config.lookup(MQTT_TLS_POLICY).as_deref(),
config.lookup(MQTT_TLS_CA).as_deref(),
config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(),
config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(),
config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(),
config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(),
)?,
queue_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(MQTT_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
target_type,
})
}
pub fn validate_mqtt_config(config: &KVS) -> Result<(), TargetError> {
let broker = config
.lookup(MQTT_BROKER)
.ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
let url = parse_url(&broker, "broker URL")?;
let tls = MQTTTlsConfig::from_values(
config.lookup(MQTT_TLS_POLICY).as_deref(),
config.lookup(MQTT_TLS_CA).as_deref(),
config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(),
config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(),
config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(),
config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(),
)?;
validate_mqtt_broker_url(&url, &tls)?;
if config.lookup(MQTT_TOPIC).is_none() {
return Err(TargetError::Configuration("Missing MQTT topic".to_string()));
}
if let Some(qos_str) = config.lookup(MQTT_QOS) {
let qos = qos_str
.parse::<u8>()
.map_err(|_| TargetError::Configuration("Invalid QoS value".to_string()))?;
if qos > 2 {
return Err(TargetError::Configuration("QoS must be 0, 1, or 2".to_string()));
}
}
let queue_dir = config.lookup(MQTT_QUEUE_DIR).unwrap_or_default();
if !queue_dir.is_empty() {
if !Path::new(&queue_dir).is_absolute() {
return Err(TargetError::Configuration("MQTT queue directory must be an absolute path".to_string()));
}
if let Some(qos_str) = config.lookup(MQTT_QOS)
&& qos_str == "0"
{
return Err(TargetError::Configuration(
"QoS should be AtLeastOnce (1) or ExactlyOnce (2) if queue_dir is set".to_string(),
));
}
}
Ok(())
}
pub fn build_nats_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<NATSArgs, TargetError> {
let address = config
.lookup(NATS_ADDRESS)
.ok_or_else(|| TargetError::Configuration("Missing NATS address".to_string()))?;
validate_nats_address(&address)?;
let subject = config
.lookup(NATS_SUBJECT)
.ok_or_else(|| TargetError::Configuration("Missing NATS subject".to_string()))?;
Ok(NATSArgs {
enable: true,
address,
subject,
username: config.lookup(NATS_USERNAME).unwrap_or_default(),
password: config.lookup(NATS_PASSWORD).unwrap_or_default(),
token: config.lookup(NATS_TOKEN).unwrap_or_default(),
credentials_file: config.lookup(NATS_CREDENTIALS_FILE).unwrap_or_default(),
tls_ca: config.lookup(NATS_TLS_CA).unwrap_or_default(),
tls_client_cert: config.lookup(NATS_TLS_CLIENT_CERT).unwrap_or_default(),
tls_client_key: config.lookup(NATS_TLS_CLIENT_KEY).unwrap_or_default(),
tls_required: parse_target_bool(config.lookup(NATS_TLS_REQUIRED).as_deref()).unwrap_or(false),
queue_dir: config.lookup(NATS_QUEUE_DIR).unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(NATS_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
target_type,
})
}
pub fn validate_nats_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
let address = config
.lookup(NATS_ADDRESS)
.ok_or_else(|| TargetError::Configuration("Missing NATS address".to_string()))?;
let server = validate_nats_address(&address)?;
validate_nats_server_config(&server, config, default_queue_dir)
}
pub fn build_pulsar_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<PulsarArgs, TargetError> {
let broker = config
.lookup(PULSAR_BROKER)
.ok_or_else(|| TargetError::Configuration("Missing Pulsar broker".to_string()))?;
validate_pulsar_broker(&broker)?;
let topic = config
.lookup(PULSAR_TOPIC)
.ok_or_else(|| TargetError::Configuration("Missing Pulsar topic".to_string()))?;
Ok(PulsarArgs {
enable: true,
broker,
topic,
auth_token: config.lookup(PULSAR_AUTH_TOKEN).unwrap_or_default(),
username: config.lookup(PULSAR_USERNAME).unwrap_or_default(),
password: config.lookup(PULSAR_PASSWORD).unwrap_or_default(),
tls_ca: config.lookup(PULSAR_TLS_CA).unwrap_or_default(),
tls_allow_insecure: parse_target_bool(config.lookup(PULSAR_TLS_ALLOW_INSECURE).as_deref()).unwrap_or(false),
tls_hostname_verification: parse_target_bool(config.lookup(PULSAR_TLS_HOSTNAME_VERIFICATION).as_deref()).unwrap_or(true),
queue_dir: config
.lookup(PULSAR_QUEUE_DIR)
.unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(PULSAR_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
target_type,
})
}
pub fn validate_pulsar_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
let broker = config
.lookup(PULSAR_BROKER)
.ok_or_else(|| TargetError::Configuration("Missing Pulsar broker".to_string()))?;
validate_pulsar_broker_config(&broker, config, default_queue_dir)
}
pub fn build_redis_args(
config: &KVS,
default_queue_dir: &str,
default_channel: &str,
target_type: TargetType,
) -> Result<RedisArgs, TargetError> {
let url = config
.lookup(REDIS_URL)
.ok_or_else(|| TargetError::Configuration("Missing Redis URL".to_string()))?;
let url = parse_url(&url, "Redis URL")?;
let channel = config
.lookup(REDIS_CHANNEL)
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| default_channel.to_string());
Ok(RedisArgs {
enable: true,
url,
channel,
username: config.lookup(REDIS_USERNAME).filter(|value| !value.trim().is_empty()),
password: config.lookup(REDIS_PASSWORD).filter(|value| !value.trim().is_empty()),
tls: RedisTlsConfig::from_values(
config.lookup(REDIS_TLS_POLICY).as_deref(),
config.lookup(REDIS_TLS_CA).as_deref(),
config.lookup(REDIS_TLS_CLIENT_CERT).as_deref(),
config.lookup(REDIS_TLS_CLIENT_KEY).as_deref(),
config.lookup(REDIS_TLS_ALLOW_INSECURE).as_deref(),
)?,
keep_alive: config
.lookup(REDIS_KEEP_ALIVE_INTERVAL)
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs)
.unwrap_or_else(|| Duration::from_secs(15)),
queue_dir: config
.lookup(REDIS_QUEUE_DIR)
.unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(REDIS_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
max_retry_attempts: config
.lookup(REDIS_MAX_RETRY_ATTEMPTS)
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(3),
reconnect_retry_attempts: config
.lookup(REDIS_RECONNECT_RETRY_ATTEMPTS)
.and_then(|v| v.parse::<usize>().ok()),
min_retry_delay: config
.lookup(REDIS_MIN_RETRY_DELAY)
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_millis),
max_retry_delay: config
.lookup(REDIS_MAX_RETRY_DELAY)
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_millis),
connection_timeout: config
.lookup(REDIS_CONNECTION_TIMEOUT)
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs),
response_timeout: config
.lookup(REDIS_RESPONSE_TIMEOUT)
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs),
pipeline_buffer_size: config
.lookup(REDIS_PIPELINE_BUFFER_SIZE)
.and_then(|v| v.parse::<usize>().ok()),
target_type,
})
}
pub fn build_postgres_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<PostgresArgs, TargetError> {
let dsn_string = config
.lookup(POSTGRES_DSN_STRING)
.ok_or_else(|| TargetError::Configuration("Missing PostgreSQL dsn_string".to_string()))?;
let table = config
.lookup(POSTGRES_TABLE)
.ok_or_else(|| TargetError::Configuration("Missing PostgreSQL table".to_string()))?;
let schema = PostgresDsn::parse(&dsn_string)?.schema;
let format = parse_postgres_format(config.lookup(POSTGRES_FORMAT).as_deref())?;
Ok(PostgresArgs {
enable: true,
dsn_string,
schema,
table,
format,
tls_required: parse_target_bool(config.lookup(POSTGRES_TLS_REQUIRED).as_deref()).unwrap_or(false),
tls_ca: config.lookup(POSTGRES_TLS_CA).unwrap_or_default(),
tls_client_cert: config.lookup(POSTGRES_TLS_CLIENT_CERT).unwrap_or_default(),
tls_client_key: config.lookup(POSTGRES_TLS_CLIENT_KEY).unwrap_or_default(),
queue_dir: config
.lookup(POSTGRES_QUEUE_DIR)
.unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(POSTGRES_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
target_type,
})
}
pub fn validate_redis_config(config: &KVS, default_queue_dir: &str, default_channel: &str) -> Result<(), TargetError> {
let url = config
.lookup(REDIS_URL)
.ok_or_else(|| TargetError::Configuration("Missing Redis URL".to_string()))?;
let url = parse_url(&url, "Redis URL")?;
validate_redis_url(&url)?;
let args = build_redis_args(config, default_queue_dir, default_channel, TargetType::NotifyEvent)?;
args.validate()
}
pub fn validate_postgres_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
let args = build_postgres_args(config, default_queue_dir, TargetType::NotifyEvent)?;
args.validate()
}
pub fn build_kafka_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<KafkaArgs, TargetError> {
let brokers_raw = config
.lookup(KAFKA_BROKERS)
.ok_or_else(|| TargetError::Configuration("Missing Kafka brokers".to_string()))?;
if brokers_raw.split(',').all(|s| s.trim().is_empty()) {
return Err(TargetError::Configuration("Kafka brokers cannot be empty".to_string()));
}
let brokers: Vec<String> = brokers_raw
.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect();
let topic = config
.lookup(KAFKA_TOPIC)
.ok_or_else(|| TargetError::Configuration("Missing Kafka topic".to_string()))?;
let sasl_mechanism = config.lookup(KAFKA_SASL_MECHANISM).unwrap_or_default();
let sasl_username = config.lookup(KAFKA_SASL_USERNAME).unwrap_or_default();
let sasl_password = config.lookup(KAFKA_SASL_PASSWORD).unwrap_or_default();
let has_sasl_fields = !sasl_mechanism.trim().is_empty() || !sasl_username.is_empty() || !sasl_password.is_empty();
let sasl_enable = parse_kafka_sasl_enable(config, has_sasl_fields)?;
let args = KafkaArgs {
enable: true,
brokers,
topic,
acks: parse_kafka_acks_value(config.lookup(KAFKA_ACKS).as_deref())?,
tls_enable: parse_target_bool(config.lookup(KAFKA_TLS_ENABLE).as_deref()).unwrap_or(false),
tls_ca: config.lookup(KAFKA_TLS_CA).unwrap_or_default(),
tls_client_cert: config.lookup(KAFKA_TLS_CLIENT_CERT).unwrap_or_default(),
tls_client_key: config.lookup(KAFKA_TLS_CLIENT_KEY).unwrap_or_default(),
sasl_enable,
sasl_mechanism: if sasl_enable && sasl_mechanism.trim().is_empty() {
KAFKA_SASL_PLAIN.to_string()
} else {
sasl_mechanism.trim().to_string()
},
sasl_username,
sasl_password,
queue_dir: config
.lookup(KAFKA_QUEUE_DIR)
.unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(KAFKA_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
target_type,
};
args.validate()?;
Ok(args)
}
pub fn validate_kafka_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
let _ = build_kafka_args(config, default_queue_dir, TargetType::NotifyEvent)?;
Ok(())
}
/// Builds `MySqlArgs` from a KVS configuration.
///
/// Parses all MySQL target configuration keys, applies defaults for
/// missing optional values, and validates that all required fields
/// are present and well-formed.
pub fn build_mysql_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<MySqlArgs, TargetError> {
let dsn_string = config
.lookup(MYSQL_DSN_STRING)
.ok_or_else(|| TargetError::Configuration("Missing MySQL dsn_string".to_string()))?;
let table = config
.lookup(MYSQL_TABLE)
.ok_or_else(|| TargetError::Configuration("Missing MySQL table".to_string()))?;
let args = MySqlArgs {
enable: true,
dsn_string,
table,
format: config.lookup(MYSQL_FORMAT).unwrap_or_else(|| "access".to_string()),
tls_ca: config.lookup(MYSQL_TLS_CA).unwrap_or_default(),
tls_client_cert: config.lookup(MYSQL_TLS_CLIENT_CERT).unwrap_or_default(),
tls_client_key: config.lookup(MYSQL_TLS_CLIENT_KEY).unwrap_or_default(),
queue_dir: config
.lookup(MYSQL_QUEUE_DIR)
.unwrap_or_else(|| default_queue_dir.to_string()),
queue_limit: config
.lookup(MYSQL_QUEUE_LIMIT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_LIMIT),
max_open_connections: config
.lookup(MYSQL_MAX_OPEN_CONNECTIONS)
.map(|value| {
value.trim().parse::<usize>().map_err(|_| {
TargetError::Configuration(format!("MySQL max_open_connections value '{}' is not a valid number", value))
})
})
.transpose()?
.unwrap_or(2),
target_type,
};
args.validate()?;
Ok(args)
}
/// Validates MySQL target configuration from a KVS without building args.
///
/// Performs the same checks as `build_mysql_args` but discards the result,
/// used for pre-validation before target creation.
pub fn validate_mysql_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
let _ = build_mysql_args(config, default_queue_dir, TargetType::NotifyEvent)?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::{
build_amqp_args, build_kafka_args, build_mysql_args, build_postgres_args, build_redis_args, build_webhook_args,
validate_amqp_config, validate_kafka_config, validate_mysql_config, validate_postgres_config, validate_redis_config,
validate_webhook_config,
};
use crate::target::{
TargetType,
kafka::{KAFKA_SASL_PLAIN, KAFKA_SASL_SCRAM_SHA_512},
postgres::PostgresFormat,
};
use rustfs_config::server_config::KVS;
use rustfs_config::{
AMQP_EXCHANGE, AMQP_MANDATORY, AMQP_PASSWORD, AMQP_PERSISTENT, AMQP_QUEUE_DIR, AMQP_ROUTING_KEY, AMQP_TLS_CLIENT_CERT,
AMQP_TLS_CLIENT_KEY, AMQP_URL, AMQP_USERNAME, KAFKA_ACKS, KAFKA_BROKERS, KAFKA_SASL_ENABLE, KAFKA_SASL_MECHANISM,
KAFKA_SASL_PASSWORD, KAFKA_SASL_USERNAME, KAFKA_TLS_ENABLE, KAFKA_TOPIC, MYSQL_DSN_STRING, MYSQL_MAX_OPEN_CONNECTIONS,
MYSQL_QUEUE_DIR, MYSQL_TABLE, MYSQL_TLS_CA, MYSQL_TLS_CLIENT_CERT, MYSQL_TLS_CLIENT_KEY, POSTGRES_DSN_STRING,
POSTGRES_FORMAT, POSTGRES_QUEUE_DIR, POSTGRES_TABLE, POSTGRES_TLS_CA, POSTGRES_TLS_CLIENT_CERT, POSTGRES_TLS_CLIENT_KEY,
REDIS_CHANNEL, REDIS_CONNECTION_TIMEOUT, REDIS_MAX_RETRY_DELAY, REDIS_MIN_RETRY_DELAY, REDIS_PIPELINE_BUFFER_SIZE,
REDIS_RECONNECT_RETRY_ATTEMPTS, REDIS_RESPONSE_TIMEOUT, REDIS_TLS_ALLOW_INSECURE, REDIS_URL, WEBHOOK_ENDPOINT,
};
fn absolute_test_path(path: &str) -> String {
std::env::temp_dir().join(path).to_string_lossy().into_owned()
}
fn amqp_base_config() -> KVS {
let mut config = KVS::new();
config.insert(AMQP_URL.to_string(), "amqp://127.0.0.1:5672/%2f".to_string());
config.insert(AMQP_EXCHANGE.to_string(), "rustfs.events".to_string());
config.insert(AMQP_ROUTING_KEY.to_string(), "objects".to_string());
config
}
fn webhook_base_config() -> KVS {
let mut config = KVS::new();
config.insert(WEBHOOK_ENDPOINT.to_string(), "https://example.com/hook".to_string());
config
}
fn kafka_base_config() -> KVS {
let mut config = KVS::new();
config.insert(KAFKA_BROKERS.to_string(), "127.0.0.1:9092".to_string());
config.insert(KAFKA_TOPIC.to_string(), "events".to_string());
config
}
fn mysql_base_config() -> KVS {
let mut config = KVS::new();
config.insert(
MYSQL_DSN_STRING.to_string(),
"rustfs:password@tcp(127.0.0.1:3306)/rustfs_events".to_string(),
);
config.insert(MYSQL_TABLE.to_string(), "rustfs_events".to_string());
config
}
#[test]
fn build_amqp_args_accepts_valid_config() {
let args = build_amqp_args(&amqp_base_config(), "", TargetType::NotifyEvent).expect("valid AMQP args");
assert_eq!(args.url.as_str(), "amqp://127.0.0.1:5672/%2f");
assert_eq!(args.exchange, "rustfs.events");
assert_eq!(args.routing_key, "objects");
assert!(!args.mandatory);
assert!(args.persistent);
}
#[test]
fn build_amqp_args_accepts_bool_aliases() {
let mut config = amqp_base_config();
config.insert(AMQP_MANDATORY.to_string(), "on".to_string());
config.insert(AMQP_PERSISTENT.to_string(), "no".to_string());
let args = build_amqp_args(&config, "", TargetType::NotifyEvent).expect("valid AMQP bool aliases");
assert!(args.mandatory);
assert!(!args.persistent);
}
#[test]
fn validate_amqp_config_rejects_invalid_bool() {
let mut config = amqp_base_config();
config.insert(AMQP_MANDATORY.to_string(), "sometimes".to_string());
let err = validate_amqp_config(&config, "").expect_err("invalid AMQP bool should fail");
assert!(err.to_string().contains("Invalid AMQP mandatory boolean"));
}
#[test]
fn validate_amqp_config_rejects_invalid_scheme() {
let mut config = amqp_base_config();
config.insert(AMQP_URL.to_string(), "http://127.0.0.1:5672".to_string());
let err = validate_amqp_config(&config, "").expect_err("invalid AMQP scheme should fail");
assert!(err.to_string().contains("only amqp and amqps"));
}
#[test]
fn validate_amqp_config_rejects_missing_url_host() {
let mut config = amqp_base_config();
config.insert(AMQP_URL.to_string(), "amqp:///objects".to_string());
let err = validate_amqp_config(&config, "").expect_err("missing AMQP host should fail");
assert!(err.to_string().contains("missing host"));
}
#[test]
fn validate_amqp_config_rejects_missing_exchange() {
let mut config = amqp_base_config();
config.0.retain(|kv| kv.key != AMQP_EXCHANGE);
let err = validate_amqp_config(&config, "").expect_err("missing AMQP exchange should fail");
assert!(err.to_string().contains("Missing AMQP exchange"));
}
#[test]
fn validate_amqp_config_rejects_missing_routing_key() {
let mut config = amqp_base_config();
config.0.retain(|kv| kv.key != AMQP_ROUTING_KEY);
let err = validate_amqp_config(&config, "").expect_err("missing AMQP routing_key should fail");
assert!(err.to_string().contains("Missing AMQP routing_key"));
}
#[test]
fn validate_amqp_config_rejects_relative_queue_dir() {
let mut config = amqp_base_config();
config.insert(AMQP_QUEUE_DIR.to_string(), "relative-queue".to_string());
let err = validate_amqp_config(&config, "").expect_err("relative queue_dir should fail");
assert!(err.to_string().contains("absolute path"));
}
#[test]
fn validate_amqp_config_rejects_unpaired_tls_client_cert_key() {
let mut config = amqp_base_config();
config.insert(AMQP_URL.to_string(), "amqps://127.0.0.1:5671/%2f".to_string());
config.insert(AMQP_TLS_CLIENT_CERT.to_string(), "/tmp/client.crt".to_string());
let err = validate_amqp_config(&config, "").expect_err("unpaired TLS cert should fail");
assert!(err.to_string().contains("tls_client_cert and tls_client_key"));
}
#[test]
fn validate_amqp_config_rejects_tls_paths_without_amqps() {
let mut config = amqp_base_config();
config.insert(AMQP_TLS_CLIENT_CERT.to_string(), "/tmp/client.crt".to_string());
config.insert(AMQP_TLS_CLIENT_KEY.to_string(), "/tmp/client.key".to_string());
let err = validate_amqp_config(&config, "").expect_err("TLS paths without amqps should fail");
assert!(err.to_string().contains("only allowed with amqps"));
}
#[test]
fn validate_amqp_config_rejects_ambiguous_credentials() {
let mut config = amqp_base_config();
config.insert(AMQP_URL.to_string(), "amqp://guest:guest@127.0.0.1:5672/%2f".to_string());
config.insert(AMQP_USERNAME.to_string(), "user".to_string());
config.insert(AMQP_PASSWORD.to_string(), "password".to_string());
let err = validate_amqp_config(&config, "").expect_err("ambiguous credentials should fail");
assert!(err.to_string().contains("either in url or username/password"));
}
#[test]
fn build_webhook_args_rejects_loopback_endpoint() {
let mut config = webhook_base_config();
config.insert(WEBHOOK_ENDPOINT.to_string(), "https://127.0.0.1/hook".to_string());
let err = build_webhook_args(&config, "/tmp/webhook-queue", TargetType::NotifyEvent)
.expect_err("loopback endpoint should be rejected");
assert!(err.to_string().contains("not allowed"));
}
#[test]
fn validate_webhook_config_rejects_loopback_endpoint() {
let mut config = webhook_base_config();
config.insert(WEBHOOK_ENDPOINT.to_string(), "https://127.0.0.1/hook".to_string());
let err = validate_webhook_config(&config, "/tmp/webhook-queue").expect_err("loopback endpoint should be rejected");
assert!(err.to_string().contains("not allowed"));
}
#[test]
fn build_kafka_args_accepts_all_ack_alias() {
let mut config = kafka_base_config();
config.insert(KAFKA_ACKS.to_string(), "all".to_string());
let args = build_kafka_args(&config, "", TargetType::NotifyEvent).expect("valid kafka args");
assert_eq!(args.acks, -1);
}
#[test]
fn build_kafka_args_rejects_invalid_acks() {
let mut config = kafka_base_config();
config.insert(KAFKA_ACKS.to_string(), "leader".to_string());
let err = build_kafka_args(&config, "", TargetType::NotifyEvent).expect_err("invalid acks should fail");
assert!(err.to_string().contains("Kafka acks must be one of"));
}
#[test]
fn validate_kafka_config_rejects_invalid_acks() {
let mut config = kafka_base_config();
config.insert(KAFKA_ACKS.to_string(), "2".to_string());
let err = validate_kafka_config(&config, "").expect_err("invalid acks should fail");
assert!(err.to_string().contains("Kafka acks must be one of"));
}
#[test]
fn build_kafka_args_infers_sasl_enable_from_credentials() {
let mut config = kafka_base_config();
config.insert(KAFKA_TLS_ENABLE.to_string(), "on".to_string());
config.insert(KAFKA_SASL_USERNAME.to_string(), "user".to_string());
config.insert(KAFKA_SASL_PASSWORD.to_string(), "secret".to_string());
let args = build_kafka_args(&config, "", TargetType::NotifyEvent).expect("valid kafka SASL args");
assert!(args.sasl_enable);
assert_eq!(args.sasl_mechanism, KAFKA_SASL_PLAIN);
assert_eq!(args.sasl_username, "user");
assert_eq!(args.sasl_password, "secret");
}
#[test]
fn build_kafka_args_infers_sasl_enable_when_enable_is_blank() {
let mut config = kafka_base_config();
config.insert(KAFKA_TLS_ENABLE.to_string(), "on".to_string());
config.insert(KAFKA_SASL_ENABLE.to_string(), "".to_string());
config.insert(KAFKA_SASL_USERNAME.to_string(), "user".to_string());
config.insert(KAFKA_SASL_PASSWORD.to_string(), "secret".to_string());
let args =
build_kafka_args(&config, "", TargetType::NotifyEvent).expect("blank sasl_enable should infer from SASL fields");
assert!(args.sasl_enable);
assert_eq!(args.sasl_mechanism, KAFKA_SASL_PLAIN);
}
#[test]
fn build_kafka_args_accepts_scram_sha_512() {
let mut config = kafka_base_config();
config.insert(KAFKA_TLS_ENABLE.to_string(), "true".to_string());
config.insert(KAFKA_SASL_ENABLE.to_string(), "true".to_string());
config.insert(KAFKA_SASL_MECHANISM.to_string(), "SCRAM-SHA-512".to_string());
config.insert(KAFKA_SASL_USERNAME.to_string(), "user".to_string());
config.insert(KAFKA_SASL_PASSWORD.to_string(), "secret".to_string());
let args = build_kafka_args(&config, "", TargetType::NotifyEvent).expect("valid kafka SCRAM args");
assert!(args.sasl_enable);
assert_eq!(args.sasl_mechanism, KAFKA_SASL_SCRAM_SHA_512);
}
#[test]
fn validate_kafka_config_rejects_sasl_without_tls() {
let mut config = kafka_base_config();
config.insert(KAFKA_SASL_ENABLE.to_string(), "on".to_string());
config.insert(KAFKA_SASL_USERNAME.to_string(), "user".to_string());
config.insert(KAFKA_SASL_PASSWORD.to_string(), "secret".to_string());
let err = validate_kafka_config(&config, "").expect_err("SASL without TLS should fail");
assert!(err.to_string().contains("requires tls_enable"));
}
#[test]
fn validate_kafka_config_rejects_sasl_fields_when_disabled() {
let mut config = kafka_base_config();
config.insert(KAFKA_SASL_ENABLE.to_string(), "off".to_string());
config.insert(KAFKA_SASL_USERNAME.to_string(), "user".to_string());
config.insert(KAFKA_SASL_PASSWORD.to_string(), "secret".to_string());
let err = validate_kafka_config(&config, "").expect_err("SASL fields with disabled SASL should fail");
assert!(err.to_string().contains("sasl_enable must be true"));
}
#[test]
fn build_mysql_args_accepts_minimal_config() {
let args = build_mysql_args(&mysql_base_config(), "", TargetType::NotifyEvent).expect("valid mysql args");
assert!(args.enable);
assert_eq!(args.dsn_string, "rustfs:password@tcp(127.0.0.1:3306)/rustfs_events");
assert_eq!(args.table, "rustfs_events");
assert_eq!(args.format, "access");
assert_eq!(args.max_open_connections, 2);
assert_eq!(args.queue_limit, rustfs_config::DEFAULT_LIMIT);
}
#[test]
fn build_mysql_args_applies_defaults() {
let queue_dir = absolute_test_path("custom-queue");
let args = build_mysql_args(&mysql_base_config(), &queue_dir, TargetType::NotifyEvent).expect("valid mysql args");
assert_eq!(args.queue_dir, queue_dir);
assert_eq!(args.queue_limit, 100000);
assert_eq!(args.max_open_connections, 2);
}
#[test]
fn build_mysql_args_rejects_missing_dsn() {
let mut config = KVS::new();
config.insert(MYSQL_TABLE.to_string(), "events".to_string());
let err = build_mysql_args(&config, "", TargetType::NotifyEvent).expect_err("missing dsn should fail");
assert!(err.to_string().contains("dsn_string"));
}
#[test]
fn build_mysql_args_rejects_relative_queue_dir() {
let mut config = mysql_base_config();
config.insert(MYSQL_QUEUE_DIR.to_string(), "relative/path".to_string());
let err = build_mysql_args(&config, "", TargetType::NotifyEvent).expect_err("relative path should fail");
assert!(err.to_string().contains("absolute"));
}
#[test]
fn validate_mysql_config_rejects_invalid_max_open_connections() {
let mut config = mysql_base_config();
config.insert(MYSQL_MAX_OPEN_CONNECTIONS.to_string(), "not-a-number".to_string());
let err = validate_mysql_config(&config, "").expect_err("invalid max_open_connections should be rejected");
assert!(err.to_string().contains("max_open_connections"));
}
#[test]
fn validate_mysql_config_rejects_empty_dsn() {
let mut config = mysql_base_config();
config.insert(MYSQL_DSN_STRING.to_string(), "".to_string());
let err = validate_mysql_config(&config, "").expect_err("empty dsn should fail");
assert!(err.to_string().contains("empty"));
}
#[test]
fn validate_mysql_config_rejects_unpaired_tls_client_fields() {
let mut config = mysql_base_config();
config.insert(MYSQL_TLS_CLIENT_CERT.to_string(), "/etc/ssl/mysql/client.pem".to_string());
let err = validate_mysql_config(&config, "").expect_err("unpaired mysql TLS client cert should fail");
assert!(err.to_string().contains("must be specified together"));
}
#[test]
fn validate_mysql_config_rejects_relative_tls_paths() {
let mut config = mysql_base_config();
config.insert(MYSQL_TLS_CA.to_string(), "ca.pem".to_string());
let err = validate_mysql_config(&config, "").expect_err("relative tls_ca should fail");
assert!(err.to_string().contains("tls_ca must be an absolute path"));
config.insert(MYSQL_TLS_CA.to_string(), absolute_test_path("mysql-ca.pem"));
config.insert(MYSQL_TLS_CLIENT_CERT.to_string(), "client.pem".to_string());
config.insert(MYSQL_TLS_CLIENT_KEY.to_string(), "client.key".to_string());
let err = validate_mysql_config(&config, "").expect_err("relative tls client paths should fail");
assert!(err.to_string().contains("absolute path"));
}
#[test]
fn build_mysql_args_accepts_absolute_tls_paths() {
let mut config = mysql_base_config();
let tls_ca = absolute_test_path("mysql-ca.pem");
let tls_client_cert = absolute_test_path("mysql-client.pem");
let tls_client_key = absolute_test_path("mysql-client.key");
config.insert(MYSQL_TLS_CA.to_string(), tls_ca.clone());
config.insert(MYSQL_TLS_CLIENT_CERT.to_string(), tls_client_cert.clone());
config.insert(MYSQL_TLS_CLIENT_KEY.to_string(), tls_client_key.clone());
let args = build_mysql_args(&config, "", TargetType::NotifyEvent).expect("absolute mysql TLS paths should pass");
assert_eq!(args.tls_ca, tls_ca);
assert_eq!(args.tls_client_cert, tls_client_cert);
assert_eq!(args.tls_client_key, tls_client_key);
}
fn redis_base_config() -> KVS {
let mut config = KVS::new();
config.insert(REDIS_URL.to_string(), "redis://127.0.0.1:6379/0".to_string());
config.insert(REDIS_CHANNEL.to_string(), "events".to_string());
config
}
fn postgres_base_config() -> KVS {
let mut config = KVS::new();
config.insert(
POSTGRES_DSN_STRING.to_string(),
"postgres://postgres:rustfs@localhost:5432/rustfs_events?search_path=public".to_string(),
);
config.insert(POSTGRES_TABLE.to_string(), "rustfs_events_namespace".to_string());
config
}
#[test]
fn build_redis_args_keeps_manager_tuning_fields_none_when_unset() {
let config = redis_base_config();
let args = build_redis_args(&config, "/tmp/queue", "default-channel", TargetType::NotifyEvent).expect("valid redis args");
assert_eq!(args.channel, "events");
assert_eq!(args.reconnect_retry_attempts, None);
assert_eq!(args.min_retry_delay, None);
assert_eq!(args.max_retry_delay, None);
assert_eq!(args.connection_timeout, None);
assert_eq!(args.response_timeout, None);
assert_eq!(args.pipeline_buffer_size, None);
}
#[test]
fn build_redis_args_uses_default_channel_when_missing() {
let mut config = KVS::new();
config.insert(REDIS_URL.to_string(), "redis://127.0.0.1:6379/0".to_string());
let args =
build_redis_args(&config, "/tmp/queue", "fallback-channel", TargetType::NotifyEvent).expect("valid redis args");
assert_eq!(args.channel, "fallback-channel");
}
#[test]
fn build_redis_args_uses_default_channel_when_empty() {
let mut config = KVS::new();
config.insert(REDIS_URL.to_string(), "redis://127.0.0.1:6379/0".to_string());
config.insert(REDIS_CHANNEL.to_string(), " ".to_string());
let args =
build_redis_args(&config, "/tmp/queue", "fallback-channel", TargetType::NotifyEvent).expect("valid redis args");
assert_eq!(args.channel, "fallback-channel");
}
#[test]
fn build_redis_args_parses_optional_tuning_values_when_present() {
let mut config = redis_base_config();
config.insert(REDIS_RECONNECT_RETRY_ATTEMPTS.to_string(), "9".to_string());
config.insert(REDIS_MIN_RETRY_DELAY.to_string(), "250".to_string());
config.insert(REDIS_MAX_RETRY_DELAY.to_string(), "5000".to_string());
config.insert(REDIS_CONNECTION_TIMEOUT.to_string(), "7".to_string());
config.insert(REDIS_RESPONSE_TIMEOUT.to_string(), "11".to_string());
config.insert(REDIS_PIPELINE_BUFFER_SIZE.to_string(), "64".to_string());
let args = build_redis_args(&config, "/tmp/queue", "default-channel", TargetType::NotifyEvent).expect("valid redis args");
assert_eq!(args.reconnect_retry_attempts, Some(9));
assert_eq!(args.min_retry_delay, Some(std::time::Duration::from_millis(250)));
assert_eq!(args.max_retry_delay, Some(std::time::Duration::from_millis(5000)));
assert_eq!(args.connection_timeout, Some(std::time::Duration::from_secs(7)));
assert_eq!(args.response_timeout, Some(std::time::Duration::from_secs(11)));
assert_eq!(args.pipeline_buffer_size, Some(64));
}
#[test]
fn build_redis_args_parses_tls_allow_insecure_when_present() {
let mut config = redis_base_config();
config.insert(REDIS_URL.to_string(), "rediss://127.0.0.1:6379/0".to_string());
config.insert(REDIS_TLS_ALLOW_INSECURE.to_string(), "on".to_string());
let args = build_redis_args(&config, "/tmp/queue", "default-channel", TargetType::NotifyEvent).expect("valid redis args");
assert!(args.tls.allow_insecure);
}
#[test]
fn validate_redis_config_rejects_missing_url() {
let config = KVS::new();
let err = validate_redis_config(&config, "/tmp/queue", "default-channel").expect_err("missing redis url should fail");
assert!(err.to_string().contains("Missing Redis URL"));
}
#[test]
fn build_postgres_args_accepts_minimal_config() {
let config = postgres_base_config();
let args = build_postgres_args(&config, "", TargetType::NotifyEvent).expect("valid postgres args");
assert_eq!(
args.dsn_string,
"postgres://postgres:rustfs@localhost:5432/rustfs_events?search_path=public"
);
assert_eq!(args.format, PostgresFormat::Namespace);
}
#[test]
fn build_postgres_args_parses_access_format() {
let mut config = postgres_base_config();
config.insert(POSTGRES_FORMAT.to_string(), "access".to_string());
let args = build_postgres_args(&config, "", TargetType::NotifyEvent).expect("valid postgres args");
assert_eq!(args.format, PostgresFormat::Access);
}
#[test]
fn validate_postgres_config_rejects_missing_dsn_string() {
let mut config = postgres_base_config();
config.0.retain(|kv| kv.key != POSTGRES_DSN_STRING);
let err = validate_postgres_config(&config, "").expect_err("missing dsn_string should fail");
assert!(err.to_string().contains("Missing PostgreSQL dsn_string"));
}
#[test]
fn validate_postgres_config_rejects_empty_dsn_string() {
let mut config = postgres_base_config();
config.insert(POSTGRES_DSN_STRING.to_string(), "".to_string());
let err = validate_postgres_config(&config, "").expect_err("empty dsn_string should fail");
assert!(err.to_string().contains("dsn_string cannot be empty"));
}
#[test]
fn validate_postgres_config_rejects_missing_table() {
let mut config = postgres_base_config();
config.0.retain(|kv| kv.key != POSTGRES_TABLE);
let err = validate_postgres_config(&config, "").expect_err("missing table should fail");
assert!(err.to_string().contains("Missing PostgreSQL table"));
}
#[test]
fn validate_postgres_config_rejects_invalid_dsn() {
let mut config = postgres_base_config();
config.insert(POSTGRES_DSN_STRING.to_string(), "postgres://".to_string());
let err = validate_postgres_config(&config, "").expect_err("invalid dsn should fail");
assert!(err.to_string().contains("invalid PostgreSQL dsn_string"));
}
#[test]
fn validate_postgres_config_rejects_relative_queue_dir() {
let mut config = postgres_base_config();
config.insert(POSTGRES_QUEUE_DIR.to_string(), "relative/path".to_string());
let err = validate_postgres_config(&config, "").expect_err("relative queue_dir should fail");
assert!(err.to_string().contains("absolute path"));
}
#[test]
fn validate_postgres_config_rejects_mtls_without_key() {
let mut config = postgres_base_config();
config.insert(POSTGRES_TLS_CLIENT_CERT.to_string(), "/etc/ssl/cert.pem".to_string());
let err = validate_postgres_config(&config, "").expect_err("missing key should fail");
assert!(err.to_string().contains("must be specified together"));
}
#[test]
fn validate_postgres_config_rejects_relative_tls_ca() {
let mut config = postgres_base_config();
config.insert(POSTGRES_TLS_CA.to_string(), "relative/ca.pem".to_string());
let err = validate_postgres_config(&config, "").expect_err("relative tls_ca should fail");
assert!(err.to_string().contains("must be an absolute path"));
}
#[test]
fn validate_postgres_config_rejects_relative_tls_client_cert() {
let mut config = postgres_base_config();
config.insert(POSTGRES_TLS_CLIENT_CERT.to_string(), "relative/client.pem".to_string());
config.insert(POSTGRES_TLS_CLIENT_KEY.to_string(), "/etc/ssl/client.key".to_string());
let err = validate_postgres_config(&config, "").expect_err("relative tls_client_cert should fail");
assert!(err.to_string().contains("must be an absolute path"));
}
}
Reference in New Issue View Git Blame Copy Permalink