feat: migrate FTP/SFTP to protocols crate and update dependencies (#1580)

Signed-off-by: yxrxy <yxrxytrigger@gmail.com>
Signed-off-by: houseme <housemecn@gmail.com>
Signed-off-by: heihutu <30542132+heihutu@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <30542132+heihutu@users.noreply.github.com>
Co-authored-by: loverustfs <hello@rustfs.com>
This commit is contained in:
yxrxy
2026-02-06 20:58:42 +08:00
committed by GitHub
parent a2b88a79ec
commit 5c2eda356e
38 changed files with 1997 additions and 4756 deletions
+17
View File
@@ -0,0 +1,17 @@
// 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.
pub mod s3;
pub use s3::StorageBackend as S3StorageBackend;
+74
View File
@@ -0,0 +1,74 @@
// 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 async_trait::async_trait;
use s3s::dto::*;
#[async_trait]
pub trait StorageBackend: Send + Sync {
/// Error type for this storage backend
type Error: std::error::Error + Send + Sync + 'static;
/// Get object content and metadata
async fn get_object(
&self,
bucket: &str,
key: &str,
access_key: &str,
secret_key: &str,
start_pos: Option<u64>,
) -> Result<GetObjectOutput, Self::Error>;
async fn get_object_range(
&self,
bucket: &str,
key: &str,
access_key: &str,
secret_key: &str,
start_pos: u64,
length: u64,
) -> Result<GetObjectOutput, Self::Error>;
/// Put object content with metadata
async fn put_object(&self, input: PutObjectInput, access_key: &str, secret_key: &str)
-> Result<PutObjectOutput, Self::Error>;
/// Delete an object
async fn delete_object(
&self,
bucket: &str,
key: &str,
access_key: &str,
secret_key: &str,
) -> Result<DeleteObjectOutput, Self::Error>;
/// Get object metadata without content
async fn head_object(
&self,
bucket: &str,
key: &str,
access_key: &str,
secret_key: &str,
) -> Result<HeadObjectOutput, Self::Error>;
/// Check if bucket exists and get metadata
async fn head_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<HeadBucketOutput, Self::Error>;
/// List objects in a bucket with pagination
async fn list_objects_v2(
&self,
input: ListObjectsV2Input,
access_key: &str,
secret_key: &str,
) -> Result<ListObjectsV2Output, Self::Error>;
/// List all buckets (requires authentication)
async fn list_buckets(&self, access_key: &str, secret_key: &str) -> Result<ListBucketsOutput, Self::Error>;
/// Create a new bucket
async fn create_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<CreateBucketOutput, Self::Error>;
/// Delete a bucket (must be empty)
async fn delete_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<DeleteBucketOutput, Self::Error>;
}
+220
View File
@@ -0,0 +1,220 @@
// 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 rustfs_credentials;
use rustfs_policy::policy::action::S3Action as PolicyS3Action;
use serde_json;
use std::collections::HashMap;
use thiserror::Error;
use tracing::error;
use super::session::SessionContext;
/// Authorization errors
#[derive(Debug, Error)]
pub enum AuthorizationError {
#[error("Access denied")]
AccessDenied,
}
/// S3 actions that can be performed through the gateway
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum S3Action {
// Bucket operations
CreateBucket,
DeleteBucket,
ListBucket,
ListBuckets,
HeadBucket,
// Object operations
GetObject,
PutObject,
DeleteObject,
HeadObject,
CopyObject,
// Multipart operations
CreateMultipartUpload,
UploadPart,
CompleteMultipartUpload,
AbortMultipartUpload,
ListMultipartUploads,
ListParts,
// ACL operations
GetBucketAcl,
PutBucketAcl,
GetObjectAcl,
PutObjectAcl,
}
impl From<S3Action> for PolicyS3Action {
fn from(action: S3Action) -> Self {
match action {
S3Action::CreateBucket => PolicyS3Action::CreateBucketAction,
S3Action::DeleteBucket => PolicyS3Action::DeleteBucketAction,
S3Action::ListBucket => PolicyS3Action::ListBucketAction,
S3Action::ListBuckets => PolicyS3Action::ListAllMyBucketsAction,
S3Action::HeadBucket => PolicyS3Action::HeadBucketAction,
S3Action::GetObject => PolicyS3Action::GetObjectAction,
S3Action::PutObject => PolicyS3Action::PutObjectAction,
S3Action::DeleteObject => PolicyS3Action::DeleteObjectAction,
S3Action::HeadObject => PolicyS3Action::GetObjectAction,
S3Action::CreateMultipartUpload => PolicyS3Action::PutObjectAction,
S3Action::UploadPart => PolicyS3Action::PutObjectAction,
S3Action::CompleteMultipartUpload => PolicyS3Action::PutObjectAction,
S3Action::AbortMultipartUpload => PolicyS3Action::AbortMultipartUploadAction,
S3Action::ListMultipartUploads => PolicyS3Action::ListBucketMultipartUploadsAction,
S3Action::ListParts => PolicyS3Action::ListMultipartUploadPartsAction,
S3Action::GetBucketAcl => PolicyS3Action::GetBucketPolicyAction,
S3Action::PutBucketAcl => PolicyS3Action::PutBucketPolicyAction,
S3Action::GetObjectAcl => PolicyS3Action::GetObjectAction,
S3Action::PutObjectAcl => PolicyS3Action::PutObjectAction,
S3Action::CopyObject => PolicyS3Action::PutObjectAction,
}
}
}
impl From<S3Action> for rustfs_policy::policy::action::Action {
fn from(action: S3Action) -> Self {
rustfs_policy::policy::action::Action::S3Action(action.into())
}
}
impl S3Action {
/// Get the string representation of the action
pub fn as_str(&self) -> &'static str {
match self {
S3Action::CreateBucket => "s3:CreateBucket",
S3Action::DeleteBucket => "s3:DeleteBucket",
S3Action::ListBucket => "s3:ListBucket",
S3Action::ListBuckets => "s3:ListAllMyBuckets",
S3Action::HeadBucket => "s3:ListBucket",
S3Action::GetObject => "s3:GetObject",
S3Action::PutObject => "s3:PutObject",
S3Action::DeleteObject => "s3:DeleteObject",
S3Action::HeadObject => "s3:GetObject",
S3Action::CreateMultipartUpload => "s3:PutObject",
S3Action::UploadPart => "s3:PutObject",
S3Action::CompleteMultipartUpload => "s3:PutObject",
S3Action::AbortMultipartUpload => "s3:AbortMultipartUpload",
S3Action::ListMultipartUploads => "s3:ListBucketMultipartUploads",
S3Action::ListParts => "s3:ListMultipartUploadParts",
S3Action::GetBucketAcl => "s3:GetBucketAcl",
S3Action::PutBucketAcl => "s3:PutBucketAcl",
S3Action::GetObjectAcl => "s3:GetObjectAcl",
S3Action::PutObjectAcl => "s3:PutObjectAcl",
S3Action::CopyObject => "s3:PutObject",
}
}
}
/// Check if operation is supported for the given protocol
pub fn is_operation_supported(protocol: super::session::Protocol, action: &S3Action) -> bool {
match protocol {
super::session::Protocol::Ftps => match action {
// Bucket operations
S3Action::CreateBucket => true,
S3Action::DeleteBucket => true,
// Object operations
S3Action::GetObject => true, // RETR command
S3Action::PutObject => true, // STOR and APPE commands both map to PutObject
S3Action::DeleteObject => true, // DELE command
S3Action::HeadObject => true, // SIZE command
// Multipart operations
S3Action::CreateMultipartUpload => false,
S3Action::UploadPart => false,
S3Action::CompleteMultipartUpload => false,
S3Action::AbortMultipartUpload => false,
S3Action::ListMultipartUploads => false,
S3Action::ListParts => false,
// ACL operations
S3Action::GetBucketAcl => false,
S3Action::PutBucketAcl => false,
S3Action::GetObjectAcl => false,
S3Action::PutObjectAcl => false,
// Other operations
S3Action::CopyObject => false, // No native copy support in FTPS
S3Action::ListBucket => true, // LIST command
S3Action::ListBuckets => true, // LIST at root level
S3Action::HeadBucket => true, // Can check if directory exists
},
}
}
/// Check if a principal is allowed to perform an S3 action
pub async fn is_authorized(session_context: &SessionContext, action: &S3Action, bucket: &str, object: Option<&str>) -> bool {
let iam_sys = match rustfs_iam::get() {
Ok(sys) => sys,
Err(e) => {
error!("IAM system unavailable: {}", e);
return false;
}
};
// Create policy arguments
let mut claims = HashMap::new();
claims.insert(
"principal".to_string(),
serde_json::Value::String(session_context.principal.access_key().to_string()),
);
let policy_action: rustfs_policy::policy::action::Action = action.clone().into();
// Check if user is the owner (admin)
let is_owner = if let Some(global_cred) = rustfs_credentials::get_global_action_cred() {
session_context.principal.access_key() == global_cred.access_key
} else {
false
};
let args = rustfs_policy::policy::Args {
account: session_context.principal.access_key(),
groups: &session_context.principal.user_identity.credentials.groups,
action: policy_action,
bucket,
conditions: &HashMap::new(),
is_owner,
object: object.unwrap_or(""),
claims: &claims,
deny_only: false,
};
iam_sys.is_allowed(&args).await
}
/// Authorize an operation and return an error if not authorized
pub async fn authorize_operation(
session_context: &SessionContext,
action: &S3Action,
bucket: &str,
object: Option<&str>,
) -> Result<(), AuthorizationError> {
// check if the operation is supported
if !is_operation_supported(session_context.protocol, action) {
return Err(AuthorizationError::AccessDenied);
}
// check IAM authorization
if is_authorized(session_context, action, bucket, object).await {
Ok(())
} else {
Err(AuthorizationError::AccessDenied)
}
}
+21
View File
@@ -0,0 +1,21 @@
// 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.
pub mod client;
pub mod gateway;
pub mod session;
pub use client::s3::StorageBackend as S3StorageBackend;
pub use gateway::{AuthorizationError, S3Action, authorize_operation, is_operation_supported};
pub use session::{ProtocolPrincipal, SessionContext};
+66
View File
@@ -0,0 +1,66 @@
// 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 rustfs_policy::auth::UserIdentity;
use std::net::IpAddr;
use std::sync::Arc;
/// Protocol types
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Protocol {
Ftps,
}
/// Protocol principal representing an authenticated user
#[derive(Debug, Clone)]
pub struct ProtocolPrincipal {
/// User identity from IAM system
pub user_identity: Arc<UserIdentity>,
}
impl ProtocolPrincipal {
pub fn new(user_identity: Arc<UserIdentity>) -> Self {
Self { user_identity }
}
pub fn access_key(&self) -> &str {
&self.user_identity.credentials.access_key
}
}
/// Session context for protocol operations
#[derive(Debug, Clone)]
pub struct SessionContext {
/// The protocol principal (authenticated user)
pub principal: ProtocolPrincipal,
/// The protocol type
pub protocol: Protocol,
/// The source IP address
pub source_ip: IpAddr,
}
impl SessionContext {
/// Create a new session context
pub fn new(principal: ProtocolPrincipal, protocol: Protocol, source_ip: IpAddr) -> Self {
Self {
principal,
protocol,
source_ip,
}
}
/// Get the access key for this session
pub fn access_key(&self) -> &str {
self.principal.access_key()
}
}
+58
View File
@@ -0,0 +1,58 @@
// 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.
/// Path and file system constants
pub mod paths {
/// Universal path constants
pub const ROOT_PATH: &str = "/";
pub const CURRENT_DIR: &str = ".";
pub const PARENT_DIR: &str = "..";
pub const PATH_SEPARATOR: &str = "/";
/// File mode and permission constants
pub const DIR_MODE: u32 = 0o040000;
pub const FILE_MODE: u32 = 0o100000;
pub const DIR_PERMISSIONS: u32 = 0o755;
pub const FILE_PERMISSIONS: u32 = 0o644;
}
/// Network constants
pub mod network {
/// Default network addresses
pub const DEFAULT_SOURCE_IP: &str = "0.0.0.0";
pub const DEFAULT_ADDR: &str = "0.0.0.0:0";
/// Authentication constants
pub const AUTH_SUFFIX_SVC: &str = "=svc";
pub const AUTH_SUFFIX_LDAP: &str = "=ldap";
pub const AUTH_FAILURE_DELAY_MS: u64 = 300;
}
/// FTPS constants
#[cfg(feature = "ftps")]
pub mod ftps {
pub const PORT_RANGE_SEPARATOR: &str = "-";
pub const PASSIVE_PORTS_PART_COUNT: usize = 2;
}
/// Default configuration values
pub mod defaults {
/// Default protocol addresses
#[cfg(feature = "ftps")]
pub const DEFAULT_FTPS_ADDRESS: &str = "0.0.0.0:8021";
/// Default FTPS passive port range
#[cfg(feature = "ftps")]
pub const DEFAULT_FTPS_PASSIVE_PORTS: &str = "40000-50000";
}
+121
View File
@@ -0,0 +1,121 @@
// 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::constants::ftps;
use std::fmt::Debug;
use std::net::SocketAddr;
use thiserror::Error;
/// FTPS server initialization error
#[derive(Debug, Error)]
pub enum FtpsInitError {
#[error("failed to bind address {0}")]
Bind(#[from] std::io::Error),
#[error("server error: {0}")]
Server(#[from] libunftp::ServerError),
#[error("invalid FTPS configuration: {0}")]
InvalidConfig(String),
}
/// FTPS server configuration
#[derive(Debug, Clone)]
pub struct FtpsConfig {
/// Server bind address
pub bind_addr: SocketAddr,
/// Passive port range (e.g., "40000-50000")
pub passive_ports: Option<String>,
/// External IP address for passive mode
pub external_ip: Option<String>,
/// Whether FTPS is required
pub ftps_required: bool,
/// Whether TLS is enabled (default: true)
pub tls_enabled: bool,
/// Certificate directory path (supports multiple certificates)
pub cert_dir: Option<String>,
/// CA certificate file path for client certificate verification
pub ca_file: Option<String>,
}
impl FtpsConfig {
/// Validates the configuration
pub async fn validate(&self) -> Result<(), FtpsInitError> {
if self.ftps_required && self.cert_dir.is_none() {
return Err(FtpsInitError::InvalidConfig(
"FTPS is required but certificate directory is missing".to_string(),
));
}
if let Some(path) = &self.cert_dir
&& !tokio::fs::try_exists(path).await.unwrap_or(false)
{
return Err(FtpsInitError::InvalidConfig(format!("Certificate directory not found: {}", path)));
}
// Validate CA file exists if specified
if let Some(path) = &self.ca_file
&& !tokio::fs::try_exists(path).await.unwrap_or(false)
{
return Err(FtpsInitError::InvalidConfig(format!("CA file not found: {}", path)));
}
// Validate passive ports format
if self.passive_ports.is_some() {
self.parse_passive_ports()?;
}
Ok(())
}
/// Parse passive ports range from string format "start-end"
pub fn parse_passive_ports(&self) -> Result<std::ops::RangeInclusive<u16>, FtpsInitError> {
match &self.passive_ports {
Some(ports) => {
let parts: Vec<&str> = ports.split(ftps::PORT_RANGE_SEPARATOR).collect();
if parts.len() != ftps::PASSIVE_PORTS_PART_COUNT {
return Err(FtpsInitError::InvalidConfig(format!(
"Invalid passive ports format: {}, expected 'start-end'",
ports
)));
}
let start = parts[0]
.parse::<u16>()
.map_err(|e| FtpsInitError::InvalidConfig(format!("Invalid start port: {}", e)))?;
let end = parts[1]
.parse::<u16>()
.map_err(|e| FtpsInitError::InvalidConfig(format!("Invalid end port: {}", e)))?;
if start > end {
return Err(FtpsInitError::InvalidConfig("Start port cannot be greater than end port".to_string()));
}
Ok(start..=end)
}
None => Err(FtpsInitError::InvalidConfig("No passive ports configured".to_string())),
}
}
}
impl Default for FtpsConfig {
fn default() -> Self {
Self {
bind_addr: crate::constants::defaults::DEFAULT_FTPS_ADDRESS.parse().unwrap(),
passive_ports: Some(crate::constants::defaults::DEFAULT_FTPS_PASSIVE_PORTS.to_string()),
external_ip: None,
ftps_required: false,
tls_enabled: true,
cert_dir: None,
ca_file: None,
}
}
}
+573
View File
@@ -0,0 +1,573 @@
// 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::common::client::s3::StorageBackend as S3StorageBackend;
use crate::common::gateway::S3Action;
use crate::common::gateway::authorize_operation;
use async_trait::async_trait;
use futures_util::stream;
use libunftp::storage::{Error, ErrorKind, Fileinfo, Metadata, Result, StorageBackend};
use rustfs_utils::path;
use s3s::dto::*;
use std::fmt::Debug;
use std::path::{Path, PathBuf};
use tokio::io::AsyncRead;
use tracing::{debug, error};
/// FTPS metadata implementation
#[derive(Debug, Clone)]
pub struct FtpsMetadata {
/// File size in bytes
pub size: u64,
/// Modification time
pub modified: Option<std::time::SystemTime>,
/// Whether this is a directory
pub is_dir: bool,
}
impl Metadata for FtpsMetadata {
fn len(&self) -> u64 {
self.size
}
fn is_dir(&self) -> bool {
self.is_dir
}
fn is_file(&self) -> bool {
!self.is_dir
}
fn is_symlink(&self) -> bool {
false
}
fn modified(&self) -> Result<std::time::SystemTime> {
self.modified
.ok_or_else(|| Error::new(ErrorKind::PermanentFileNotAvailable, "No modification time available"))
}
fn gid(&self) -> u32 {
0
}
fn uid(&self) -> u32 {
0
}
}
/// FTPS storage driver implementation
pub struct FtpsDriver<S> {
/// Storage backend for S3 operations
storage: S,
}
impl<S> Debug for FtpsDriver<S>
where
S: S3StorageBackend + Debug,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FtpsDriver").field("storage", &"StorageBackend").finish()
}
}
impl<S> FtpsDriver<S>
where
S: S3StorageBackend + Debug,
{
/// Create a new FTPS driver with the given storage backend
pub fn new(storage: S) -> Self {
Self { storage }
}
/// List all buckets (for root path)
async fn list_buckets(
&self,
session_context: &crate::common::session::SessionContext,
) -> Result<Vec<Fileinfo<PathBuf, <FtpsDriver<S> as libunftp::storage::StorageBackend<super::server::FtpsUser>>::Metadata>>>
{
match authorize_operation(session_context, &S3Action::ListBuckets, "", None).await {
Ok(_) => {}
Err(_e) => {
return Err(Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"));
}
}
let mut list_result = Vec::new();
match self
.storage
.list_buckets(
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(output) => {
if let Some(buckets) = output.buckets {
for bucket in buckets {
if let Some(ref bucket_name) = bucket.name {
let metadata = FtpsMetadata {
size: 0,
modified: bucket.creation_date.map(|dt| {
let offset_dt: time::OffsetDateTime = dt.into();
std::time::SystemTime::from(offset_dt)
}),
is_dir: true,
};
list_result.push(Fileinfo {
path: PathBuf::from(bucket_name),
metadata,
});
}
}
}
Ok(list_result)
}
Err(_) => Err(Error::new(ErrorKind::PermanentFileNotAvailable, "List failed")),
}
}
fn parse_s3_path(&self, path: &str) -> std::result::Result<(String, Option<String>), String> {
let cleaned_path = path::clean(path);
let (bucket, object) = path::path_to_bucket_object(&cleaned_path);
let key = if object.is_empty() { None } else { Some(object) };
Ok((bucket, key))
}
}
#[async_trait]
impl<S> StorageBackend<super::server::FtpsUser> for FtpsDriver<S>
where
S: S3StorageBackend + Debug,
{
type Metadata = FtpsMetadata;
async fn metadata<P: AsRef<Path> + Send>(&self, user: &super::server::FtpsUser, path: P) -> Result<Self::Metadata> {
let path_str = path.as_ref().to_string_lossy();
let session_context = &user.session_context;
let (bucket, key) = self
.parse_s3_path(&path_str)
.map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
if let Some(key) = key {
match self
.storage
.head_object(
&bucket,
&key,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(output) => {
let size = output.content_length.unwrap_or(0) as u64;
let modified = output.last_modified.map(|dt| {
// Convert s3s Timestamp to SystemTime
let offset_dt: time::OffsetDateTime = dt.into();
std::time::SystemTime::from(offset_dt)
});
Ok(FtpsMetadata {
size,
modified,
is_dir: false,
})
}
Err(e) => {
error!("Failed to get metadata for '{}': {}", path_str, e);
Err(Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Metadata failed", e)))
}
}
} else {
// Directory metadata - use HeadBucket
let bucket_clone = bucket.clone();
match self
.storage
.head_bucket(
&bucket,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(_) => Ok(FtpsMetadata {
size: 0,
modified: Some(std::time::SystemTime::now()),
is_dir: true,
}),
Err(e) => {
error!("Failed to get bucket metadata for '{}': {}", bucket_clone, e);
Err(Error::new(
ErrorKind::PermanentFileNotAvailable,
format!("{}: {}", "Bucket metadata failed", e),
))
}
}
}
}
async fn list<P: AsRef<Path> + Send>(
&self,
user: &super::server::FtpsUser,
path: P,
) -> Result<Vec<Fileinfo<PathBuf, Self::Metadata>>> {
let path_str = path.as_ref().to_string_lossy();
// Get session context from user
let session_context = &user.session_context;
// Check if this is root path listing
if path_str == "/" || path_str == "/." {
return self.list_buckets(session_context).await;
}
let (bucket, prefix) = self
.parse_s3_path(&path_str)
.map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
// Authorize the operation
authorize_operation(session_context, &S3Action::ListBucket, &bucket, prefix.as_deref())
.await
.map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
let list_input = ListObjectsV2Input::builder()
.bucket(bucket)
.prefix(prefix.map(|p| p.to_string()))
.delimiter(Some("/".to_string()))
.build()
.map_err(|e| {
Error::new(ErrorKind::PermanentFileNotAvailable, format!("Failed to build ListObjectsV2Input: {}", e))
})?;
match self
.storage
.list_objects_v2(
list_input,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(output) => {
let mut fileinfos = Vec::new();
// Add files (objects)
if let Some(objects) = output.contents {
for obj in objects {
if let Some(key) = obj.key {
let filename = PathBuf::from(key.as_str());
let size = obj.size.unwrap_or(0) as u64;
let modified = obj.last_modified.map(|dt: s3s::dto::Timestamp| {
// Convert s3s Timestamp to SystemTime
let offset_dt: time::OffsetDateTime = dt.into();
std::time::SystemTime::from(offset_dt)
});
let metadata = FtpsMetadata {
size,
modified,
is_dir: false,
};
fileinfos.push(Fileinfo {
path: filename,
metadata,
});
}
}
}
// Add directories (common prefixes)
if let Some(common_prefixes) = output.common_prefixes {
for prefix in common_prefixes {
if let Some(prefix_str) = prefix.prefix {
let dir_name = PathBuf::from(prefix_str.as_str().trim_end_matches('/'));
let metadata = FtpsMetadata {
size: 0,
modified: Some(std::time::SystemTime::now()),
is_dir: true,
};
fileinfos.push(Fileinfo {
path: dir_name,
metadata,
});
}
}
}
Ok(fileinfos)
}
Err(e) => {
error!("Failed to list '{}': {}", path_str, e);
Err(Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "List failed", e)))
}
}
}
async fn get<P: AsRef<Path> + Send>(
&self,
user: &super::server::FtpsUser,
path: P,
start_pos: u64,
) -> Result<Box<dyn AsyncRead + Send + Sync + Unpin>> {
let path_str = path.as_ref().to_string_lossy();
let session_context = &user.session_context;
let (bucket, key) = self
.parse_s3_path(&path_str)
.map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
let key = key.ok_or_else(|| Error::new(ErrorKind::PermanentFileNotAvailable, "Cannot get directory"))?;
match self
.storage
.get_object(
&bucket,
&key,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
Some(start_pos), // Pass start_pos for range request
)
.await
{
Ok(output) => {
let body = output
.body
.ok_or_else(|| Error::new(ErrorKind::PermanentFileNotAvailable, "No body in response"))?;
use futures_util::StreamExt;
let mut data = Vec::new();
let mut stream = body;
while let Some(chunk_result) = stream.next().await {
match chunk_result {
Ok(bytes) => data.extend_from_slice(&bytes),
Err(e) => {
error!("Error reading stream: {}", e);
return Err(Error::new(ErrorKind::PermanentFileNotAvailable, format!("Stream error: {}", e)));
}
}
}
Ok(Box::new(std::io::Cursor::new(data)))
}
Err(e) => {
error!("Failed to get '{}': {}", path_str, e);
Err(Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Get failed", e)))
}
}
}
async fn put<P: AsRef<Path> + Send + Debug, R: tokio::io::AsyncRead + Send + Sync + Unpin + 'static>(
&self,
user: &super::server::FtpsUser,
bytes: R,
path: P,
start_pos: u64,
) -> Result<u64> {
let path_str = path.as_ref().to_string_lossy();
let session_context = &user.session_context;
let (bucket, key) = self
.parse_s3_path(&path_str)
.map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
let key = key.ok_or_else(|| Error::new(ErrorKind::PermanentFileNotAvailable, "Cannot put to directory"))?;
// Check if this is an append operation (start_pos > 0)
if start_pos > 0 {
return Err(Error::new(
ErrorKind::CommandNotImplemented,
"Append operations (start_pos > 0) are not supported with S3 backend",
));
}
// Authorize the operation
authorize_operation(session_context, &S3Action::PutObject, &bucket, Some(&key))
.await
.map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
// Convert AsyncRead to bytes
let bytes_vec = {
let mut buffer = Vec::new();
let mut reader = bytes;
tokio::io::copy(&mut reader, &mut buffer)
.await
.map_err(|e| Error::new(ErrorKind::TransientFileNotAvailable, e.to_string()))?;
buffer
};
let file_size = bytes_vec.len();
let mut put_builder = PutObjectInput::builder();
put_builder.set_bucket(bucket.clone());
put_builder.set_key(key.clone());
put_builder.set_content_length(Some(file_size as i64));
// Create StreamingBlob with known size
let data_bytes = bytes::Bytes::from(bytes_vec);
let stream = stream::once(async move { Ok::<bytes::Bytes, std::io::Error>(data_bytes) });
let streaming_blob = s3s::dto::StreamingBlob::wrap(stream);
put_builder.set_body(Some(streaming_blob));
let put_input = put_builder
.build()
.map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Failed to build PutObjectInput"))?;
match self
.storage
.put_object(
put_input,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(_output) => {
Ok(file_size as u64) // Return the size of the uploaded object
}
Err(e) => {
error!("FTPS put - S3 error details: {:?}", e);
Err(Error::new(
ErrorKind::PermanentFileNotAvailable,
format!("Failed to upload object: {:?}", e),
))
}
}
}
async fn del<P: AsRef<Path> + Send>(&self, user: &super::server::FtpsUser, path: P) -> Result<()> {
let path_str = path.as_ref().to_string_lossy();
let session_context = &user.session_context;
debug!("FTPS delete request for user '{}' path '{}'", user.username, path_str);
let (bucket, key) = self
.parse_s3_path(&path_str)
.map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
if let Some(key) = key {
// Delete file
match self
.storage
.delete_object(
&bucket,
&key,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(_) => Ok(()),
Err(e) => {
error!("Failed to delete file '{}': {}", path_str, e);
Err(Error::new(ErrorKind::PermanentFileNotAvailable, format!("Delete failed: {}", e)))
}
}
} else {
// Delete directory (bucket) - not supported in typical FTP
Err(Error::new(ErrorKind::PermanentFileNotAvailable, "Directory deletion not supported"))
}
}
async fn mkd<P: AsRef<Path> + Send>(&self, user: &super::server::FtpsUser, path: P) -> Result<()> {
let path_str = path.as_ref().to_string_lossy();
let session_context = &user.session_context;
debug!("FTPS mkdir request for user '{}' path '{}'", user.username, path_str);
let (bucket, _key) = self
.parse_s3_path(&path_str)
.map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
// Create bucket for directory
match self
.storage
.create_bucket(
&bucket,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(_) => {
debug!("Successfully created directory/bucket '{}'", path_str);
Ok(())
}
Err(e) => {
error!("Failed to create directory/bucket '{}': {}", path_str, e);
Err(Error::new(ErrorKind::PermanentFileNotAvailable, format!("Mkdir failed: {}", e)))
}
}
}
async fn rmd<P: AsRef<Path> + Send>(&self, user: &super::server::FtpsUser, path: P) -> Result<()> {
let path_str = path.as_ref().to_string_lossy();
let session_context = &user.session_context;
let (bucket, _key) = self
.parse_s3_path(&path_str)
.map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
// Delete bucket for directory
match self
.storage
.delete_bucket(
&bucket,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(_) => {
debug!("Successfully removed directory/bucket '{}'", path_str);
Ok(())
}
Err(e) => {
error!("Failed to remove directory/bucket '{}': {}", path_str, e);
Err(Error::new(ErrorKind::PermanentFileNotAvailable, format!("Rmdir failed: {}", e)))
}
}
}
async fn cwd<P: AsRef<Path> + Send>(&self, user: &super::server::FtpsUser, path: P) -> Result<()> {
let path_str = path.as_ref().to_string_lossy();
let session_context = &user.session_context;
let (bucket, _key) = self
.parse_s3_path(&path_str)
.map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
// Check if bucket exists
match self
.storage
.head_bucket(
&bucket,
&session_context.principal.user_identity.credentials.access_key,
&session_context.principal.user_identity.credentials.secret_key,
)
.await
{
Ok(_) => Ok(()),
Err(e) => {
error!("CWD to '{}' failed: {}", path_str, e);
Err(Error::new(ErrorKind::PermanentFileNotAvailable, format!("CWD failed: {}", e)))
}
}
}
async fn rename<P: AsRef<Path> + Send>(&self, user: &super::server::FtpsUser, from: P, to: P) -> Result<()> {
let from_str = from.as_ref().to_string_lossy();
let to_str = to.as_ref().to_string_lossy();
debug!("FTPS rename request for user '{}' from '{}' to '{}'", user.username, from_str, to_str);
Err(Error::new(ErrorKind::PermanentFileNotAvailable, "Atomic rename not supported in S3"))
}
}
+17
View File
@@ -0,0 +1,17 @@
// 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.
pub mod config;
pub mod driver;
pub mod server;
+268
View File
@@ -0,0 +1,268 @@
// 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::config::{FtpsConfig, FtpsInitError};
use super::driver::FtpsDriver;
use crate::common::client::s3::StorageBackend;
use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext};
use crate::constants::{network::DEFAULT_SOURCE_IP, paths::ROOT_PATH};
use libunftp::auth::{AuthenticationError, UserDetail};
use libunftp::options::FtpsRequired;
use std::fmt::{Debug, Display, Formatter};
use std::net::IpAddr;
use std::path::Path;
use std::sync::Arc;
use tokio::sync::broadcast;
use tracing::{debug, error, info, warn};
/// FTPS user implementation
#[derive(Debug, Clone)]
pub struct FtpsUser {
/// Username for the FTP session
pub username: String,
/// User's display name
pub name: Option<String>,
/// Session context for this user
pub session_context: SessionContext,
}
impl UserDetail for FtpsUser {
fn home(&self) -> Option<&Path> {
Some(Path::new(ROOT_PATH))
}
}
impl Display for FtpsUser {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match &self.name {
Some(display_name) => write!(f, "FtpsUser({} - {})", self.username, display_name),
None => write!(f, "FtpsUser({})", self.username),
}
}
}
/// FTPS server implementation
pub struct FtpsServer<S> {
/// Server configuration
config: FtpsConfig,
/// S3 storage backend
storage: S,
}
impl<S> FtpsServer<S>
where
S: StorageBackend + Clone + Send + Sync + 'static + std::fmt::Debug,
{
/// Create a new FTPS server
pub async fn new(config: FtpsConfig, storage: S) -> Result<Self, FtpsInitError> {
config.validate().await?;
Ok(Self { config, storage })
}
/// Start the FTPS server
///
/// This method binds the listener first to ensure the port is available,
/// then spawns the server loop in a background task.
pub async fn start(&self, mut shutdown_rx: broadcast::Receiver<()>) -> Result<(), FtpsInitError> {
info!("Initializing FTPS server on {}", self.config.bind_addr);
let storage_clone = self.storage.clone();
let mut server_builder = libunftp::ServerBuilder::with_authenticator(
Box::new(move || FtpsDriver::new(storage_clone.clone())),
Arc::new(FtpsAuthenticator::new()),
);
// Configure passive ports for data connections
if let Some(passive_ports) = &self.config.passive_ports {
let range = self.config.parse_passive_ports()?;
info!("Configuring FTPS passive ports range: {:?} ({})", range, passive_ports);
server_builder = server_builder.passive_ports(range);
} else {
warn!("No passive ports configured, using system-assigned ports");
}
// Configure external IP address for passive mode
if let Some(ref external_ip) = self.config.external_ip {
info!("Configuring FTPS external IP for passive mode: {}", external_ip);
server_builder = server_builder.passive_host(external_ip.as_str());
}
// Configure both active and passive mode support
use libunftp::options::ActivePassiveMode;
server_builder = server_builder.active_passive_mode(ActivePassiveMode::ActiveAndPassive);
info!("FTPS server configured for both active and passive mode support");
// Configure FTPS / TLS
if self.config.tls_enabled {
if let Some(cert_dir) = &self.config.cert_dir {
debug!("Enabling FTPS with multi-certificate support from directory: {}", cert_dir);
// Load all certificates from directory
let cert_key_pairs = rustfs_utils::load_all_certs_from_directory(cert_dir)
.map_err(|e| FtpsInitError::InvalidConfig(format!("Failed to load certificates: {}", e)))?;
if cert_key_pairs.is_empty() {
return Err(FtpsInitError::InvalidConfig("No valid certificates found in directory".into()));
}
debug!("Loaded {} certificates for FTPS", cert_key_pairs.len());
// Create multi-certificate resolver with SNI support
let resolver = rustfs_utils::create_multi_cert_resolver(cert_key_pairs)
.map_err(|e| FtpsInitError::InvalidConfig(format!("Failed to create certificate resolver: {}", e)))?;
// Build ServerConfig with SNI support
let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
let server_config = rustls::ServerConfig::builder()
.with_no_client_auth()
.with_cert_resolver(std::sync::Arc::new(resolver));
server_builder = server_builder.ftps_manual::<std::path::PathBuf>(std::sync::Arc::new(server_config));
if self.config.ftps_required {
info!("FTPS is explicitly required for all connections");
server_builder = server_builder.ftps_required(FtpsRequired::All, FtpsRequired::All);
}
} else if self.config.ftps_required {
return Err(FtpsInitError::InvalidConfig(
"FTPS required but certificate directory not provided".into(),
));
}
} else {
info!("TLS disabled, running in plain FTP mode");
}
// Build the server instance
let server = server_builder.build().map_err(FtpsInitError::Server)?;
// libunftp's listen() binds to the address and runs the loop
let bind_addr = self.config.bind_addr.to_string();
let server_handle = tokio::spawn(async move {
if let Err(e) = server.listen(bind_addr).await {
error!("FTPS server runtime error: {}", e);
return Err(FtpsInitError::Server(e));
}
Ok(())
});
// Wait for shutdown signal or server failure
tokio::select! {
result = server_handle => {
match result {
Ok(Ok(())) => {
info!("FTPS server stopped normally");
Ok(())
}
Ok(Err(e)) => {
error!("FTPS server internal error: {}", e);
Err(e)
}
Err(e) => {
error!("FTPS server panic or task cancellation: {}", e);
Err(FtpsInitError::Bind(std::io::Error::other(e.to_string())))
}
}
}
_ = shutdown_rx.recv() => {
info!("FTPS server received shutdown signal");
// libunftp listen() is not easily cancellable gracefully without dropping the future.
// The select! dropping server_handle will close the listener.
Ok(())
}
}
}
/// Get server configuration
pub fn config(&self) -> &FtpsConfig {
&self.config
}
/// Get storage backend
pub fn storage(&self) -> &S {
&self.storage
}
}
/// FTPS authenticator implementation
#[derive(Debug, Default)]
pub struct FtpsAuthenticator;
impl FtpsAuthenticator {
/// Create a new FTPS authenticator
pub fn new() -> Self {
Self
}
}
#[async_trait::async_trait]
impl libunftp::auth::Authenticator<FtpsUser> for FtpsAuthenticator {
/// Authenticate FTP user against RustFS IAM system
async fn authenticate(&self, username: &str, creds: &libunftp::auth::Credentials) -> Result<FtpsUser, AuthenticationError> {
use rustfs_credentials::Credentials as S3Credentials;
use rustfs_iam::get;
// Access IAM system
let iam_sys = get().map_err(|e| {
error!("IAM system unavailable during FTPS auth: {}", e);
AuthenticationError::ImplPropagated("Internal authentication service unavailable".to_string(), Some(Box::new(e)))
})?;
let s3_creds = S3Credentials {
access_key: username.to_string(),
secret_key: creds.password.clone().unwrap_or_default(),
session_token: String::new(),
expiration: None,
status: String::new(),
parent_user: String::new(),
groups: None,
claims: None,
name: None,
description: None,
};
let (user_identity, is_valid) = iam_sys.check_key(&s3_creds.access_key).await.map_err(|e| {
error!("IAM check_key failed for {}: {}", username, e);
AuthenticationError::ImplPropagated("Authentication verification failed".to_string(), Some(Box::new(e)))
})?;
if !is_valid {
warn!("FTPS login failed: Invalid access key '{}'", username);
return Err(AuthenticationError::BadUser);
}
let identity = user_identity.ok_or_else(|| {
error!("User identity missing despite valid key for {}", username);
AuthenticationError::BadUser
})?;
if !identity.credentials.secret_key.eq(&s3_creds.secret_key) {
warn!("FTPS login failed: Invalid secret key for '{}'", username);
return Err(AuthenticationError::BadPassword);
}
let source_ip: IpAddr = DEFAULT_SOURCE_IP.parse().unwrap();
let session_context = SessionContext::new(ProtocolPrincipal::new(Arc::new(identity.clone())), Protocol::Ftps, source_ip);
let ftps_user = FtpsUser {
username: username.to_string(),
name: identity.credentials.name.clone(),
session_context,
};
info!("FTPS user '{}' authenticated successfully", username);
Ok(ftps_user)
}
}
+27
View File
@@ -0,0 +1,27 @@
// 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.
#![deny(unsafe_code)]
pub mod common;
pub mod constants;
#[cfg(feature = "ftps")]
pub mod ftps;
pub use common::session::Protocol;
pub use common::{AuthorizationError, ProtocolPrincipal, S3Action, SessionContext, authorize_operation};
#[cfg(feature = "ftps")]
pub use ftps::{config::FtpsConfig, server::FtpsServer};