gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-27 08:38:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Console Server Architecture (#685)

Browse Source 
* todo

* fix console server

* fix console server

* fix console server

* fix console server

* fix console server
This commit is contained in:
weisd
2025-10-23 00:06:09 +08:00
committed by GitHub
parent 7dcf01f127
commit f30698ec7f
8 changed files with 342 additions and 681 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/console.rs
+270 -154
View File
@@ -14,43 +14,34 @@
use crate::config::build;
use crate::license::get_license;
use axum::Json;
use axum::body::Body;
use axum::response::{IntoResponse, Response};
use axum::{Router, extract::Request, middleware, routing::get};
use axum_extra::extract::Host;
use axum_server::tls_rustls::RustlsConfig;
use http::{HeaderMap, HeaderName, StatusCode, Uri};
use http::{HeaderValue, Method};
use mime_guess::from_path;
use rust_embed::RustEmbed;
use rustfs_config::{RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
use serde::Serialize;
use serde_json::json;
use std::io::Result;
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::sync::OnceLock;
// use axum::response::Redirect;
// use axum::routing::get;
// use axum::{
// body::Body,
// http::{Response, StatusCode},
// response::IntoResponse,
// Router,
// };
// use axum_extra::extract::Host;
// use axum_server::tls_rustls::RustlsConfig;
// use http::{header, HeaderMap, HeaderName, Uri};
// use io::Error;
// use mime_guess::from_path;
// use rust_embed::RustEmbed;
// use rustfs_config::{RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
// use rustfs_utils::parse_and_resolve_address;
// use serde::Serialize;
// use std::io;
// use std::net::{IpAddr, SocketAddr};
// use std::sync::OnceLock;
// use std::time::Duration;
// use tokio::signal;
// use tower_http::cors::{Any, CorsLayer};
// use tower_http::trace::TraceLayer;
use tracing::{error, instrument};
use std::time::Duration;
use tokio_rustls::rustls::ServerConfig;
use tower_http::catch_panic::CatchPanicLayer;
use tower_http::cors::{AllowOrigin, Any, CorsLayer};
use tower_http::limit::RequestBodyLimitLayer;
use tower_http::timeout::TimeoutLayer;
use tower_http::trace::TraceLayer;
use tracing::{debug, error, info, instrument, warn};
// shadow!(build);
pub(crate) const CONSOLE_PREFIX: &str = "/rustfs/console";
const RUSTFS_ADMIN_PREFIX: &str = "/rustfs/admin/v3";
#[derive(RustEmbed)]
@@ -270,131 +261,256 @@ pub async fn config_handler(uri: Uri, Host(host): Host, headers: HeaderMap) -> i
.unwrap()
}
// pub fn register_router() -> Router {
// Router::new()
// .route("/license", get(license_handler))
// .route("/config.json", get(config_handler))
// .fallback_service(get(static_handler))
// }
//
// #[allow(dead_code)]
// pub async fn start_static_file_server(addrs: &str, tls_path: Option<String>) {
// // Configure CORS
// let cors = CorsLayer::new()
// .allow_origin(Any) // In the production environment, we recommend that you specify a specific domain name
// .allow_methods([http::Method::GET, http::Method::POST])
// .allow_headers([header::CONTENT_TYPE]);
//
// // Create a route
// let app = register_router()
// .layer(cors)
// .layer(tower_http::compression::CompressionLayer::new().gzip(true).deflate(true))
// .layer(TraceLayer::new_for_http());
//
// // Check and start the HTTPS/HTTP server
// match start_server(addrs, tls_path, app).await {
// Ok(_) => info!("Console Server shutdown gracefully"),
// Err(e) => error!("Console Server error: {}", e),
// }
// }
//
// async fn start_server(addrs: &str, tls_path: Option<String>, app: Router) -> io::Result<()> {
// let server_addr = parse_and_resolve_address(addrs).expect("Console Failed to parse socket address");
// let server_port = server_addr.port();
// let server_address = server_addr.to_string();
//
// info!("Console WebUI: http://{} http://127.0.0.1:{} ", server_address, server_port);
//
// let tls_path = tls_path.unwrap_or_default();
// let key_path = format!("{tls_path}/{RUSTFS_TLS_KEY}");
// let cert_path = format!("{tls_path}/{RUSTFS_TLS_CERT}");
// let handle = axum_server::Handle::new();
// // create a signal off listening task
// let handle_clone = handle.clone();
// tokio::spawn(async move {
// shutdown_signal().await;
// info!("Console Initiating graceful shutdown...");
// handle_clone.graceful_shutdown(Some(Duration::from_secs(10)));
// });
//
// let has_tls_certs = tokio::try_join!(tokio::fs::metadata(&key_path), tokio::fs::metadata(&cert_path)).is_ok();
// info!("Console TLS certs: {:?}", has_tls_certs);
// if has_tls_certs {
// info!("Console Found TLS certificates, starting with HTTPS");
// match RustlsConfig::from_pem_file(cert_path, key_path).await {
// Ok(config) => {
// info!("Console Starting HTTPS server...");
// axum_server::bind_rustls(server_addr, config)
// .handle(handle.clone())
// .serve(app.into_make_service())
// .await
// .map_err(Error::other)?;
//
// info!("Console HTTPS server running on https://{}", server_addr);
//
// Ok(())
// }
// Err(e) => {
// error!("Console Failed to create TLS config: {}", e);
// start_http_server(server_addr, app, handle).await
// }
// }
// } else {
// info!("Console TLS certificates not found at {} and {}", key_path, cert_path);
// start_http_server(server_addr, app, handle).await
// }
// }
//
// #[allow(dead_code)]
// /// 308 redirect for HTTP to HTTPS
// fn redirect_to_https(https_port: u16) -> Router {
// Router::new().route(
// "/*path",
// get({
// move |uri: Uri, req: http::Request<Body>| async move {
// let host = req
// .headers()
// .get("host")
// .map_or("localhost", |h| h.to_str().unwrap_or("localhost"));
// let path = uri.path_and_query().map(|pq| pq.as_str()).unwrap_or("");
// let https_url = format!("https://{host}:{https_port}{path}");
// Redirect::permanent(&https_url)
// }
// }),
// )
// }
//
// async fn start_http_server(addr: SocketAddr, app: Router, handle: axum_server::Handle) -> io::Result<()> {
// info!("Console Starting HTTP server... {}", addr.to_string());
// axum_server::bind(addr)
// .handle(handle)
// .serve(app.into_make_service())
// .await
// .map_err(Error::other)
// }
//
// async fn shutdown_signal() {
// let ctrl_c = async {
// signal::ctrl_c().await.expect("Console failed to install Ctrl+C handler");
// };
//
// #[cfg(unix)]
// let terminate = async {
// signal::unix::signal(signal::unix::SignalKind::terminate())
// .expect("Console failed to install signal handler")
// .recv()
// .await;
// };
//
// #[cfg(not(unix))]
// let terminate = std::future::pending::<()>();
//
// tokio::select! {
// _ = ctrl_c => {
// info!("Console shutdown_signal ctrl_c")
// },
// _ = terminate => {
// info!("Console shutdown_signal terminate")
// },
// }
// }
/// Console access logging middleware
async fn console_logging_middleware(req: Request, next: axum::middleware::Next) -> axum::response::Response {
let method = req.method().clone();
let uri = req.uri().clone();
let start = std::time::Instant::now();
let response = next.run(req).await;
let duration = start.elapsed();
info!(
target: "rustfs::console::access",
method = %method,
uri = %uri,
status = %response.status(),
duration_ms = %duration.as_millis(),
"Console access"
);
response
}
/// Setup TLS configuration for console using axum-server, following endpoint TLS implementation logic
#[instrument(skip(tls_path))]
async fn _setup_console_tls_config(tls_path: Option<&String>) -> Result<Option<RustlsConfig>> {
let tls_path = match tls_path {
Some(path) if !path.is_empty() => path,
_ => {
debug!("TLS path is not provided, console starting with HTTP");
return Ok(None);
}
};
if tokio::fs::metadata(tls_path).await.is_err() {
debug!("TLS path does not exist, console starting with HTTP");
return Ok(None);
}
debug!("Found TLS directory for console, checking for certificates");
// Make sure to use a modern encryption suite
let _ = rustls::crypto::ring::default_provider().install_default();
// 1. Attempt to load all certificates in the directory (multi-certificate support, for SNI)
if let Ok(cert_key_pairs) = rustfs_utils::load_all_certs_from_directory(tls_path) {
if !cert_key_pairs.is_empty() {
debug!(
"Found {} certificates for console, creating SNI-aware multi-cert resolver",
cert_key_pairs.len()
);
// Create an SNI-enabled certificate resolver
let resolver = rustfs_utils::create_multi_cert_resolver(cert_key_pairs)?;
// Configure the server to enable SNI support
let mut server_config = ServerConfig::builder()
.with_no_client_auth()
.with_cert_resolver(Arc::new(resolver));
// Configure ALPN protocol priority
server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
// Log SNI requests
if rustfs_utils::tls_key_log() {
server_config.key_log = Arc::new(rustls::KeyLogFile::new());
}
info!(target: "rustfs::console::tls", "Console TLS enabled with multi-certificate SNI support");
return Ok(Some(RustlsConfig::from_config(Arc::new(server_config))));
}
}
// 2. Revert to the traditional single-certificate mode
let key_path = format!("{tls_path}/{RUSTFS_TLS_KEY}");
let cert_path = format!("{tls_path}/{RUSTFS_TLS_CERT}");
if tokio::try_join!(tokio::fs::metadata(&key_path), tokio::fs::metadata(&cert_path)).is_ok() {
debug!("Found legacy single TLS certificate for console, starting with HTTPS");
return match RustlsConfig::from_pem_file(cert_path, key_path).await {
Ok(config) => {
info!(target: "rustfs::console::tls", "Console TLS enabled with single certificate");
Ok(Some(config))
}
Err(e) => {
error!(target: "rustfs::console::error", error = %e, "Failed to create TLS config for console");
Err(std::io::Error::other(e))
}
};
}
debug!("No valid TLS certificates found in the directory for console, starting with HTTP");
Ok(None)
}
/// Get console configuration from environment variables
fn get_console_config_from_env() -> (bool, u32, u64, String) {
let rate_limit_enable = std::env::var(rustfs_config::ENV_CONSOLE_RATE_LIMIT_ENABLE)
.unwrap_or_else(|_| rustfs_config::DEFAULT_CONSOLE_RATE_LIMIT_ENABLE.to_string())
.parse::<bool>()
.unwrap_or(rustfs_config::DEFAULT_CONSOLE_RATE_LIMIT_ENABLE);
let rate_limit_rpm = std::env::var(rustfs_config::ENV_CONSOLE_RATE_LIMIT_RPM)
.unwrap_or_else(|_| rustfs_config::DEFAULT_CONSOLE_RATE_LIMIT_RPM.to_string())
.parse::<u32>()
.unwrap_or(rustfs_config::DEFAULT_CONSOLE_RATE_LIMIT_RPM);
let auth_timeout = std::env::var(rustfs_config::ENV_CONSOLE_AUTH_TIMEOUT)
.unwrap_or_else(|_| rustfs_config::DEFAULT_CONSOLE_AUTH_TIMEOUT.to_string())
.parse::<u64>()
.unwrap_or(rustfs_config::DEFAULT_CONSOLE_AUTH_TIMEOUT);
let cors_allowed_origins = std::env::var(rustfs_config::ENV_CONSOLE_CORS_ALLOWED_ORIGINS)
.unwrap_or_else(|_| rustfs_config::DEFAULT_CONSOLE_CORS_ALLOWED_ORIGINS.to_string())
.parse::<String>()
.unwrap_or(rustfs_config::DEFAULT_CONSOLE_CORS_ALLOWED_ORIGINS.to_string());
(rate_limit_enable, rate_limit_rpm, auth_timeout, cors_allowed_origins)
}
pub fn is_console_path(path: &str) -> bool {
path.starts_with(CONSOLE_PREFIX)
}
/// Setup comprehensive middleware stack with tower-http features
fn setup_console_middleware_stack(
cors_layer: CorsLayer,
rate_limit_enable: bool,
rate_limit_rpm: u32,
auth_timeout: u64,
) -> Router {
let mut app = Router::new()
.route(&format!("{CONSOLE_PREFIX}/license"), get(crate::admin::console::license_handler))
.route(&format!("{CONSOLE_PREFIX}/config.json"), get(crate::admin::console::config_handler))
.route(&format!("{CONSOLE_PREFIX}/health"), get(health_check))
.nest(CONSOLE_PREFIX, Router::new().fallback_service(get(static_handler)))
.fallback_service(get(static_handler));
// Add comprehensive middleware layers using tower-http features
app = app
.layer(CatchPanicLayer::new())
.layer(TraceLayer::new_for_http())
.layer(middleware::from_fn(console_logging_middleware))
.layer(cors_layer)
// Add timeout layer - convert auth_timeout from seconds to Duration
.layer(TimeoutLayer::new(Duration::from_secs(auth_timeout)))
// Add request body limit (10MB for console uploads)
.layer(RequestBodyLimitLayer::new(10 * 1024 * 1024));
// Add rate limiting if enabled
if rate_limit_enable {
info!("Console rate limiting enabled: {} requests per minute", rate_limit_rpm);
// Note: tower-http doesn't provide a built-in rate limiter, but we have the foundation
// For production, you would integrate with a rate limiting service like Redis
// For now, we log that it's configured and ready for integration
}
app
}
/// Console health check handler with comprehensive health information
async fn health_check() -> Json<serde_json::Value> {
use rustfs_ecstore::new_object_layer_fn;
let mut health_status = "ok";
let mut details = json!({});
// Check storage backend health
if let Some(_store) = new_object_layer_fn() {
details["storage"] = json!({"status": "connected"});
} else {
health_status = "degraded";
details["storage"] = json!({"status": "disconnected"});
}
// Check IAM system health
match rustfs_iam::get() {
Ok(_) => {
details["iam"] = json!({"status": "connected"});
}
Err(_) => {
health_status = "degraded";
details["iam"] = json!({"status": "disconnected"});
}
}
Json(json!({
"status": health_status,
"service": "rustfs-console",
"timestamp": chrono::Utc::now().to_rfc3339(),
"version": env!("CARGO_PKG_VERSION"),
"details": details,
"uptime": std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}))
}
/// Parse CORS allowed origins from configuration
pub fn parse_cors_origins(origins: Option<&String>) -> CorsLayer {
let cors_layer = CorsLayer::new()
.allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE, Method::OPTIONS])
.allow_headers(Any);
match origins {
Some(origins_str) if origins_str == "*" => cors_layer.allow_origin(Any).expose_headers(Any),
Some(origins_str) => {
let origins: Vec<&str> = origins_str.split(',').map(|s| s.trim()).collect();
if origins.is_empty() {
warn!("Empty CORS origins provided, using permissive CORS");
cors_layer.allow_origin(Any).expose_headers(Any)
} else {
// Parse origins with proper error handling
let mut valid_origins = Vec::new();
for origin in origins {
match origin.parse::<HeaderValue>() {
Ok(header_value) => {
valid_origins.push(header_value);
}
Err(e) => {
warn!("Invalid CORS origin '{}': {}", origin, e);
}
}
}
if valid_origins.is_empty() {
warn!("No valid CORS origins found, using permissive CORS");
cors_layer.allow_origin(Any).expose_headers(Any)
} else {
info!("Console CORS origins configured: {:?}", valid_origins);
cors_layer.allow_origin(AllowOrigin::list(valid_origins)).expose_headers(Any)
}
}
}
None => {
debug!("No CORS origins configured for console, using permissive CORS");
cors_layer.allow_origin(Any)
}
}
}
pub(crate) fn make_console_server() -> Router {
let (rate_limit_enable, rate_limit_rpm, auth_timeout, cors_allowed_origins) = get_console_config_from_env();
// String to Option<&String>
let cors_allowed_origins = if cors_allowed_origins.is_empty() {
None
} else {
Some(&cors_allowed_origins)
};
// Configure CORS based on settings
let cors_layer = parse_cors_origins(cors_allowed_origins);
// Build console router with enhanced middleware stack using tower-http features
setup_console_middleware_stack(cors_layer, rate_limit_enable, rate_limit_rpm, auth_timeout)
}
rustfs/src/server/console_test.rs → rustfs/src/admin/console_test.rs
+1 -31
View File
@@ -15,42 +15,12 @@
#[cfg(test)]
mod tests {
use crate::config::Opt;
use crate::server::start_console_server;
use clap::Parser;
use tokio::time::{Duration, timeout};
#[tokio::test]
async fn test_console_server_can_start_and_stop() {
// Test that console server can be started and shut down gracefully
let args = vec!["rustfs", "/tmp/test", "--console-address", ":0"]; // Use port 0 for auto-assignment
let opt = Opt::parse_from(args);
let (tx, rx) = tokio::sync::broadcast::channel(1);
// Start console server in a background task
let handle = tokio::spawn(async move { start_console_server(&opt, rx).await });
// Give it a moment to start
tokio::time::sleep(Duration::from_millis(100)).await;
// Send shutdown signal
let _ = tx.send(());
// Wait for server to shut down
let result = timeout(Duration::from_secs(5), handle).await;
assert!(result.is_ok(), "Console server should shutdown gracefully");
let server_result = result.unwrap();
assert!(server_result.is_ok(), "Console server should not have errors");
let final_result = server_result.unwrap();
assert!(final_result.is_ok(), "Console server should complete successfully");
}
#[tokio::test]
async fn test_console_cors_configuration() {
// Test CORS configuration parsing
use crate::server::console::parse_cors_origins;
use crate::admin::console::parse_cors_origins;
// Test wildcard origin
let cors_wildcard = Some("*".to_string());
let _layer1 = parse_cors_origins(cors_wildcard.as_ref());
rustfs/src/admin/mod.rs
+3
View File
@@ -19,6 +19,9 @@ pub mod router;
mod rpc;
pub mod utils;
#[cfg(test)]
mod console_test;
// use ecstore::global::{is_dist_erasure, is_erasure};
use handlers::{
GetReplicationMetricsHandler, HealthCheckHandler, ListRemoteTargetHandler, RemoveRemoteTargetHandler, SetRemoteTargetHandler,
rustfs/src/admin/router.rs
+6 -12
View File
@@ -12,7 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use axum::routing::get;
use hyper::HeaderMap;
use hyper::Method;
use hyper::StatusCode;
@@ -32,11 +31,11 @@ use tower::Service;
use tracing::error;
use crate::admin::ADMIN_PREFIX;
use crate::admin::console;
use crate::admin::console::CONSOLE_PREFIX;
use crate::admin::console::is_console_path;
use crate::admin::console::make_console_server;
use crate::admin::rpc::RPC_PREFIX;
const CONSOLE_PREFIX: &str = "/rustfs/console";
pub struct S3Router<T> {
router: Router<T>,
console_enabled: bool,
@@ -48,12 +47,7 @@ impl<T: Operation> S3Router<T> {
let router = Router::new();
let console_router = if console_enabled {
Some(
axum::Router::new()
.nest(CONSOLE_PREFIX, axum::Router::new().fallback_service(get(console::static_handler)))
.fallback_service(get(console::static_handler))
.into_service::<Body>(),
)
Some(make_console_server().into_service::<Body>())
} else {
None
};
@@ -115,7 +109,7 @@ where
return Ok(());
}
// Allow unauthenticated access to console static files if console is enabled
if self.console_enabled && req.uri.path().starts_with(CONSOLE_PREFIX) {
if self.console_enabled && is_console_path(req.uri.path()) {
return Ok(());
}
@@ -139,7 +133,7 @@ where
}
async fn call(&self, req: S3Request<Body>) -> S3Result<S3Response<Body>> {
if self.console_enabled && req.uri.path().starts_with(CONSOLE_PREFIX) {
if self.console_enabled && is_console_path(req.uri.path()) {
if let Some(console_router) = &self.console_router {
let mut console_router = console_router.clone();
let req = convert_request(req);
rustfs/src/main.rs
+24 -51
View File
@@ -25,10 +25,9 @@ mod storage;
mod update;
mod version;
use crate::admin::console::init_console_cfg;
use crate::server::{
SHUTDOWN_TIMEOUT, ServiceState, ServiceStateManager, ShutdownSignal, init_event_notifier, shutdown_event_notifier,
start_audit_system, start_console_server, start_http_server, stop_audit_system, wait_for_shutdown,
start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown,
};
use crate::storage::ecfs::{process_lambda_configurations, process_queue_configurations, process_topic_configurations};
use chrono::Datelike;
@@ -224,53 +223,21 @@ async fn run(opt: config::Opt) -> Result<()> {
// Update service status to Starting
state_manager.update(ServiceState::Starting);
let shutdown_tx = start_http_server(&opt, state_manager.clone()).await?;
// Start console server if enabled
let console_shutdown_tx = shutdown_tx.clone();
if opt.console_enable && !opt.console_address.is_empty() {
// Deal with port mapping issues for virtual machines like docker
let (external_addr, external_port) = if !opt.external_address.is_empty() {
let external_addr = parse_and_resolve_address(opt.external_address.as_str()).map_err(Error::other)?;
let external_port = external_addr.port();
if external_port != server_port {
warn!(
"External port {} is different from server port {}, ensure your firewall allows access to the external port if needed.",
external_port, server_port
);
}
info!(
target: "rustfs::main::run",
external_address = %external_addr,
external_port = %external_port,
"Using external address {} for endpoint access", external_addr
);
rustfs_ecstore::global::set_global_rustfs_external_port(external_port);
set_global_addr(&opt.external_address).await;
(external_addr.ip(), external_port)
} else {
(server_addr.ip(), server_port)
};
warn!("Starting console server on address: '{}', port: '{}'", external_addr, external_port);
// init console configuration
init_console_cfg(external_addr, external_port);
let s3_shutdown_tx = {
let mut s3_opt = opt.clone();
s3_opt.console_enable = false;
let s3_shutdown_tx = start_http_server(&s3_opt, state_manager.clone()).await?;
Some(s3_shutdown_tx)
};
let opt_clone = opt.clone();
tokio::spawn(async move {
let console_shutdown_rx = console_shutdown_tx.subscribe();
if let Err(e) = start_console_server(&opt_clone, console_shutdown_rx).await {
error!("Console server failed to start: {}", e);
}
});
let console_shutdown_tx = if opt.console_enable && !opt.console_address.is_empty() {
let mut console_opt = opt.clone();
console_opt.address = console_opt.console_address.clone();
let console_shutdown_tx = start_http_server(&console_opt, state_manager.clone()).await?;
Some(console_shutdown_tx)
} else {
info!("Console server is disabled.");
info!("You can access the RustFS API at {}", &opt.address);
info!("For more information, visit https://rustfs.com/docs/");
info!("To enable the console, restart the server with --console-enable and a valid --console-address.");
info!(
"Current console address is set to: '{}' ,console enable is set to: '{}'",
&opt.console_address, &opt.console_enable
);
}
None
};
set_global_endpoints(endpoint_pools.as_ref().clone());
update_erasure_type(setup_type).await;
@@ -379,11 +346,11 @@ async fn run(opt: config::Opt) -> Result<()> {
match wait_for_shutdown().await {
#[cfg(unix)]
ShutdownSignal::CtrlC | ShutdownSignal::Sigint | ShutdownSignal::Sigterm => {
handle_shutdown(&state_manager, &shutdown_tx, ctx.clone()).await;
handle_shutdown(&state_manager, s3_shutdown_tx, console_shutdown_tx, ctx.clone()).await;
}
#[cfg(not(unix))]
ShutdownSignal::CtrlC => {
handle_shutdown(&state_manager, &shutdown_tx, ctx.clone()).await;
handle_shutdown(&state_manager, s3_shutdown_tx, console_shutdown_tx, ctx.clone()).await;
}
}
@@ -405,7 +372,8 @@ fn parse_bool_env_var(var_name: &str, default: bool) -> bool {
/// Handles the shutdown process of the server
async fn handle_shutdown(
state_manager: &ServiceStateManager,
shutdown_tx: &tokio::sync::broadcast::Sender<()>,
s3_shutdown_tx: Option<tokio::sync::broadcast::Sender<()>>,
console_shutdown_tx: Option<tokio::sync::broadcast::Sender<()>>,
ctx: CancellationToken,
) {
ctx.cancel();
@@ -462,7 +430,12 @@ async fn handle_shutdown(
target: "rustfs::main::handle_shutdown",
"Server is stopping..."
);
let _ = shutdown_tx.send(());
if let Some(s3_shutdown_tx) = s3_shutdown_tx {
let _ = s3_shutdown_tx.send(());
}
if let Some(console_shutdown_tx) = console_shutdown_tx {
let _ = console_shutdown_tx.send(());
}
// Wait for the worker thread to complete the cleaning work
tokio::time::sleep(SHUTDOWN_TIMEOUT).await;
rustfs/src/server/console.rs
-410
View File
@@ -1,410 +0,0 @@
// 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::console::static_handler;
use crate::config::Opt;
use axum::{Router, extract::Request, middleware, response::Json, routing::get};
use axum_server::tls_rustls::RustlsConfig;
use http::{HeaderValue, Method};
use rustfs_config::{RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
use rustfs_utils::net::parse_and_resolve_address;
use serde_json::json;
use std::io::Result;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio_rustls::rustls::ServerConfig;
use tower_http::catch_panic::CatchPanicLayer;
use tower_http::cors::{AllowOrigin, Any, CorsLayer};
use tower_http::limit::RequestBodyLimitLayer;
use tower_http::timeout::TimeoutLayer;
use tower_http::trace::TraceLayer;
use tracing::{debug, error, info, instrument, warn};
const CONSOLE_PREFIX: &str = "/rustfs/console";
/// Console access logging middleware
async fn console_logging_middleware(req: Request, next: axum::middleware::Next) -> axum::response::Response {
let method = req.method().clone();
let uri = req.uri().clone();
let start = std::time::Instant::now();
let response = next.run(req).await;
let duration = start.elapsed();
info!(
target: "rustfs::console::access",
method = %method,
uri = %uri,
status = %response.status(),
duration_ms = %duration.as_millis(),
"Console access"
);
response
}
/// Setup TLS configuration for console using axum-server, following endpoint TLS implementation logic
#[instrument(skip(tls_path))]
async fn setup_console_tls_config(tls_path: Option<&String>) -> Result<Option<RustlsConfig>> {
let tls_path = match tls_path {
Some(path) if !path.is_empty() => path,
_ => {
debug!("TLS path is not provided, console starting with HTTP");
return Ok(None);
}
};
if tokio::fs::metadata(tls_path).await.is_err() {
debug!("TLS path does not exist, console starting with HTTP");
return Ok(None);
}
debug!("Found TLS directory for console, checking for certificates");
// Make sure to use a modern encryption suite
let _ = rustls::crypto::ring::default_provider().install_default();
// 1. Attempt to load all certificates in the directory (multi-certificate support, for SNI)
if let Ok(cert_key_pairs) = rustfs_utils::load_all_certs_from_directory(tls_path) {
if !cert_key_pairs.is_empty() {
debug!(
"Found {} certificates for console, creating SNI-aware multi-cert resolver",
cert_key_pairs.len()
);
// Create an SNI-enabled certificate resolver
let resolver = rustfs_utils::create_multi_cert_resolver(cert_key_pairs)?;
// Configure the server to enable SNI support
let mut server_config = ServerConfig::builder()
.with_no_client_auth()
.with_cert_resolver(Arc::new(resolver));
// Configure ALPN protocol priority
server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
// Log SNI requests
if rustfs_utils::tls_key_log() {
server_config.key_log = Arc::new(rustls::KeyLogFile::new());
}
info!(target: "rustfs::console::tls", "Console TLS enabled with multi-certificate SNI support");
return Ok(Some(RustlsConfig::from_config(Arc::new(server_config))));
}
}
// 2. Revert to the traditional single-certificate mode
let key_path = format!("{tls_path}/{RUSTFS_TLS_KEY}");
let cert_path = format!("{tls_path}/{RUSTFS_TLS_CERT}");
if tokio::try_join!(tokio::fs::metadata(&key_path), tokio::fs::metadata(&cert_path)).is_ok() {
debug!("Found legacy single TLS certificate for console, starting with HTTPS");
return match RustlsConfig::from_pem_file(cert_path, key_path).await {
Ok(config) => {
info!(target: "rustfs::console::tls", "Console TLS enabled with single certificate");
Ok(Some(config))
}
Err(e) => {
error!(target: "rustfs::console::error", error = %e, "Failed to create TLS config for console");
Err(std::io::Error::other(e))
}
};
}
debug!("No valid TLS certificates found in the directory for console, starting with HTTP");
Ok(None)
}
/// Get console configuration from environment variables
fn get_console_config_from_env() -> (bool, u32, u64, String) {
let rate_limit_enable = std::env::var(rustfs_config::ENV_CONSOLE_RATE_LIMIT_ENABLE)
.unwrap_or_else(|_| rustfs_config::DEFAULT_CONSOLE_RATE_LIMIT_ENABLE.to_string())
.parse::<bool>()
.unwrap_or(rustfs_config::DEFAULT_CONSOLE_RATE_LIMIT_ENABLE);
let rate_limit_rpm = std::env::var(rustfs_config::ENV_CONSOLE_RATE_LIMIT_RPM)
.unwrap_or_else(|_| rustfs_config::DEFAULT_CONSOLE_RATE_LIMIT_RPM.to_string())
.parse::<u32>()
.unwrap_or(rustfs_config::DEFAULT_CONSOLE_RATE_LIMIT_RPM);
let auth_timeout = std::env::var(rustfs_config::ENV_CONSOLE_AUTH_TIMEOUT)
.unwrap_or_else(|_| rustfs_config::DEFAULT_CONSOLE_AUTH_TIMEOUT.to_string())
.parse::<u64>()
.unwrap_or(rustfs_config::DEFAULT_CONSOLE_AUTH_TIMEOUT);
let cors_allowed_origins = std::env::var(rustfs_config::ENV_CONSOLE_CORS_ALLOWED_ORIGINS)
.unwrap_or_else(|_| rustfs_config::DEFAULT_CONSOLE_CORS_ALLOWED_ORIGINS.to_string())
.parse::<String>()
.unwrap_or(rustfs_config::DEFAULT_CONSOLE_CORS_ALLOWED_ORIGINS.to_string());
(rate_limit_enable, rate_limit_rpm, auth_timeout, cors_allowed_origins)
}
/// Setup comprehensive middleware stack with tower-http features
fn setup_console_middleware_stack(
cors_layer: CorsLayer,
rate_limit_enable: bool,
rate_limit_rpm: u32,
auth_timeout: u64,
) -> Router {
let mut app = Router::new()
.route("/license", get(crate::admin::console::license_handler))
.route("/config.json", get(crate::admin::console::config_handler))
.route("/health", get(health_check))
.nest(CONSOLE_PREFIX, Router::new().fallback_service(get(static_handler)))
.fallback_service(get(static_handler));
// Add comprehensive middleware layers using tower-http features
app = app
.layer(CatchPanicLayer::new())
.layer(TraceLayer::new_for_http())
.layer(middleware::from_fn(console_logging_middleware))
.layer(cors_layer)
// Add timeout layer - convert auth_timeout from seconds to Duration
.layer(TimeoutLayer::new(Duration::from_secs(auth_timeout)))
// Add request body limit (10MB for console uploads)
.layer(RequestBodyLimitLayer::new(10 * 1024 * 1024));
// Add rate limiting if enabled
if rate_limit_enable {
info!("Console rate limiting enabled: {} requests per minute", rate_limit_rpm);
// Note: tower-http doesn't provide a built-in rate limiter, but we have the foundation
// For production, you would integrate with a rate limiting service like Redis
// For now, we log that it's configured and ready for integration
}
app
}
/// Console health check handler with comprehensive health information
async fn health_check() -> Json<serde_json::Value> {
use rustfs_ecstore::new_object_layer_fn;
let mut health_status = "ok";
let mut details = json!({});
// Check storage backend health
if let Some(_store) = new_object_layer_fn() {
details["storage"] = json!({"status": "connected"});
} else {
health_status = "degraded";
details["storage"] = json!({"status": "disconnected"});
}
// Check IAM system health
match rustfs_iam::get() {
Ok(_) => {
details["iam"] = json!({"status": "connected"});
}
Err(_) => {
health_status = "degraded";
details["iam"] = json!({"status": "disconnected"});
}
}
Json(json!({
"status": health_status,
"service": "rustfs-console",
"timestamp": chrono::Utc::now().to_rfc3339(),
"version": env!("CARGO_PKG_VERSION"),
"details": details,
"uptime": std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}))
}
/// Parse CORS allowed origins from configuration
pub fn parse_cors_origins(origins: Option<&String>) -> CorsLayer {
let cors_layer = CorsLayer::new()
.allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE, Method::OPTIONS])
.allow_headers(Any);
match origins {
Some(origins_str) if origins_str == "*" => cors_layer.allow_origin(Any).expose_headers(Any),
Some(origins_str) => {
let origins: Vec<&str> = origins_str.split(',').map(|s| s.trim()).collect();
if origins.is_empty() {
warn!("Empty CORS origins provided, using permissive CORS");
cors_layer.allow_origin(Any).expose_headers(Any)
} else {
// Parse origins with proper error handling
let mut valid_origins = Vec::new();
for origin in origins {
match origin.parse::<HeaderValue>() {
Ok(header_value) => {
valid_origins.push(header_value);
}
Err(e) => {
warn!("Invalid CORS origin '{}': {}", origin, e);
}
}
}
if valid_origins.is_empty() {
warn!("No valid CORS origins found, using permissive CORS");
cors_layer.allow_origin(Any).expose_headers(Any)
} else {
info!("Console CORS origins configured: {:?}", valid_origins);
cors_layer.allow_origin(AllowOrigin::list(valid_origins)).expose_headers(Any)
}
}
}
None => {
debug!("No CORS origins configured for console, using permissive CORS");
cors_layer.allow_origin(Any)
}
}
}
/// Start the standalone console server with enhanced security and monitoring
#[instrument(skip(opt, shutdown_rx))]
pub async fn start_console_server(opt: &Opt, shutdown_rx: tokio::sync::broadcast::Receiver<()>) -> Result<()> {
if !opt.console_enable {
debug!("Console server is disabled");
return Ok(());
}
let console_addr = parse_and_resolve_address(&opt.console_address)?;
// Get configuration from environment variables
let (rate_limit_enable, rate_limit_rpm, auth_timeout, cors_allowed_origins) = get_console_config_from_env();
// Setup TLS configuration if certificates are available
let tls_config = setup_console_tls_config(opt.tls_path.as_ref()).await?;
let tls_enabled = tls_config.is_some();
info!(
target: "rustfs::console::startup",
address = %console_addr,
tls_enabled = tls_enabled,
rate_limit_enabled = rate_limit_enable,
rate_limit_rpm = rate_limit_rpm,
auth_timeout_seconds = auth_timeout,
cors_allowed_origins = %cors_allowed_origins,
"Starting console server"
);
// String to Option<&String>
let cors_allowed_origins = if cors_allowed_origins.is_empty() {
None
} else {
Some(&cors_allowed_origins)
};
// Configure CORS based on settings
let cors_layer = parse_cors_origins(cors_allowed_origins);
// Build console router with enhanced middleware stack using tower-http features
let app = setup_console_middleware_stack(cors_layer, rate_limit_enable, rate_limit_rpm, auth_timeout);
let local_ip = rustfs_utils::get_local_ip().unwrap_or_else(|| "127.0.0.1".parse().unwrap());
let protocol = if tls_enabled { "https" } else { "http" };
info!(
target: "rustfs::console::startup",
"Console WebUI available at: {}://{}:{}/rustfs/console/index.html",
protocol, local_ip, console_addr.port()
);
info!(
target: "rustfs::console::startup",
"Console WebUI (localhost): {}://127.0.0.1:{}/rustfs/console/index.html",
protocol, console_addr.port()
);
println!(
"Console WebUI available at: {}://{}:{}/rustfs/console/index.html",
protocol,
local_ip,
console_addr.port()
);
println!(
"Console WebUI (localhost): {}://127.0.0.1:{}/rustfs/console/index.html",
protocol,
console_addr.port()
);
// Handle connections based on TLS availability using axum-server
if let Some(tls_config) = tls_config {
handle_tls_connections(console_addr, app, tls_config, shutdown_rx).await
} else {
handle_plain_connections(console_addr, app, shutdown_rx).await
}
}
/// Handle TLS connections for console using axum-server with proper TLS support
async fn handle_tls_connections(
server_addr: SocketAddr,
app: Router,
tls_config: RustlsConfig,
mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<()> {
info!(target: "rustfs::console::tls", "Starting Console HTTPS server on {}", server_addr);
let handle = axum_server::Handle::new();
let handle_clone = handle.clone();
// Spawn shutdown signal handler
tokio::spawn(async move {
let _ = shutdown_rx.recv().await;
info!(target: "rustfs::console::shutdown", "Console TLS server shutdown signal received");
handle_clone.graceful_shutdown(Some(Duration::from_secs(10)));
});
// Start the HTTPS server using axum-server with RustlsConfig
if let Err(e) = axum_server::bind_rustls(server_addr, tls_config)
.handle(handle)
.serve(app.into_make_service())
.await
{
error!(target: "rustfs::console::error", error = %e, "Console TLS server error");
return Err(std::io::Error::other(e));
}
info!(target: "rustfs::console::shutdown", "Console TLS server stopped");
Ok(())
}
/// Handle plain HTTP connections using axum-server
async fn handle_plain_connections(
server_addr: SocketAddr,
app: Router,
mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<()> {
info!(target: "rustfs::console::startup", "Starting Console HTTP server on {}", server_addr);
let handle = axum_server::Handle::new();
let handle_clone = handle.clone();
// Spawn shutdown signal handler
tokio::spawn(async move {
let _ = shutdown_rx.recv().await;
info!(target: "rustfs::console::shutdown", "Console server shutdown signal received");
handle_clone.graceful_shutdown(Some(Duration::from_secs(10)));
});
// Start the HTTP server using axum-server
if let Err(e) = axum_server::bind(server_addr)
.handle(handle)
.serve(app.into_make_service())
.await
{
error!(target: "rustfs::console::error", error = %e, "Console server error");
return Err(std::io::Error::other(e));
}
info!(target: "rustfs::console::shutdown", "Console server stopped");
Ok(())
}
rustfs/src/server/http.rs
+38 -19
View File
@@ -144,12 +144,8 @@ pub async fn start_http_server(
// Obtain the listener address
let local_addr: SocketAddr = listener.local_addr()?;
debug!("Listening on: {}", local_addr);
let local_ip = match rustfs_utils::get_local_ip() {
Some(ip) => {
debug!("Obtained local IP address: {}", ip);
ip
}
Some(ip) => ip,
None => {
warn!("Unable to obtain local IP address, using fallback IP: {}", local_addr.ip());
local_addr.ip()
@@ -159,15 +155,37 @@ pub async fn start_http_server(
// Detailed endpoint information (showing all API endpoints)
let api_endpoints = format!("http://{local_ip}:{server_port}");
let localhost_endpoint = format!("http://127.0.0.1:{server_port}");
info!(" API: {} {}", api_endpoints, localhost_endpoint);
println!(" API: {api_endpoints} {localhost_endpoint}");
info!(" RootUser: {}", opt.access_key.clone());
info!(" RootPass: {}", opt.secret_key.clone());
if DEFAULT_ACCESS_KEY.eq(&opt.access_key) && DEFAULT_SECRET_KEY.eq(&opt.secret_key) {
warn!(
"Detected default credentials '{}:{}', we recommend that you change these values with 'RUSTFS_ACCESS_KEY' and 'RUSTFS_SECRET_KEY' environment variables",
DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY
let tls_acceptor = setup_tls_acceptor(opt.tls_path.as_deref().unwrap_or_default()).await?;
let tls_enabled = tls_acceptor.is_some();
if opt.console_enable {
admin::console::init_console_cfg(local_ip, server_port);
let protocol = if tls_enabled { "https" } else { "http" };
info!(
target: "rustfs::console::startup",
"Console WebUI available at: {protocol}://{local_ip}:{server_port}/rustfs/console/index.html"
);
info!(
target: "rustfs::console::startup",
"Console WebUI (localhost): {protocol}://127.0.0.1:{server_port}/rustfs/console/index.html",
);
println!("Console WebUI available at: {protocol}://{local_ip}:{server_port}/rustfs/console/index.html");
println!("Console WebUI (localhost): {protocol}://127.0.0.1:{server_port}/rustfs/console/index.html",);
} else {
info!(" API: {} {}", api_endpoints, localhost_endpoint);
println!(" API: {api_endpoints} {localhost_endpoint}");
if DEFAULT_ACCESS_KEY.eq(&opt.access_key) && DEFAULT_SECRET_KEY.eq(&opt.secret_key) {
warn!(
"Detected default credentials '{}:{}', we recommend that you change these values with 'RUSTFS_ACCESS_KEY' and 'RUSTFS_SECRET_KEY' environment variables",
DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY
);
}
info!("For more information, visit https://rustfs.com/docs/");
info!("To enable the console, restart the server with --console-enable and a valid --console-address.");
}
// Setup S3 service
@@ -178,13 +196,10 @@ pub async fn start_http_server(
let access_key = opt.access_key.clone();
let secret_key = opt.secret_key.clone();
debug!("authentication is enabled {}, {}", &access_key, &secret_key);
b.set_auth(IAMAuth::new(access_key, secret_key));
b.set_access(store.clone());
// When console runs on separate port, disable console routes on main endpoint
let console_on_endpoint = opt.console_enable; // Console will run separately
b.set_route(admin::make_admin_route(console_on_endpoint)?);
b.set_route(admin::make_admin_route(opt.console_enable)?);
if !opt.server_domains.is_empty() {
MultiDomain::new(&opt.server_domains).map_err(Error::other)?; // validate domains
@@ -223,7 +238,6 @@ pub async fn start_http_server(
}
});
let tls_acceptor = setup_tls_acceptor(opt.tls_path.as_deref().unwrap_or_default()).await?;
// Create shutdown channel
let (shutdown_tx, mut shutdown_rx) = tokio::sync::broadcast::channel(1);
let shutdown_tx_clone = shutdown_tx.clone();
@@ -235,6 +249,8 @@ pub async fn start_http_server(
} else {
Some(cors_allowed_origins)
};
let is_console = opt.console_enable;
tokio::spawn(async move {
// Create CORS layer inside the server loop closure
let cors_layer = parse_cors_origins(cors_allowed_origins.as_ref());
@@ -332,6 +348,7 @@ pub async fn start_http_server(
s3_service.clone(),
graceful.clone(),
cors_layer.clone(),
is_console,
);
}
@@ -440,6 +457,7 @@ fn process_connection(
s3_service: S3Service,
graceful: Arc<GracefulShutdown>,
cors_layer: CorsLayer,
is_console: bool,
) {
tokio::spawn(async move {
// Build services inside each connected task to avoid passing complex service types across tasks,
@@ -492,8 +510,9 @@ fn process_connection(
}),
)
.layer(cors_layer)
.layer(RedirectLayer)
.option_layer(if is_console { Some(RedirectLayer) } else { None })
.service(service);
let hybrid_service = TowerToHyperService::new(hybrid_service);
// Decide whether to handle HTTPS or HTTP connections based on the existence of TLS Acceptor
rustfs/src/server/mod.rs
-4
View File
@@ -13,18 +13,14 @@
// limitations under the License.
mod audit;
mod console;
mod http;
mod hybrid;
mod layer;
mod service_state;
#[cfg(test)]
mod console_test;
mod event;
pub(crate) use audit::{start_audit_system, stop_audit_system};
pub(crate) use console::start_console_server;
pub(crate) use event::{init_event_notifier, shutdown_event_notifier};
pub(crate) use http::{get_tokio_runtime_builder, print_tokio_thread_enable, start_http_server};
pub(crate) use service_state::SHUTDOWN_TIMEOUT;