https://github.com/bolcom/unFTP unftp源码分析
// starts the FTP server as a Tokio task.
fn start_ftp(
log: &Logger,
root_log: &Logger,
m: &clap::ArgMatches,
shutdown: tokio::sync::broadcast::Receiver<()>,
done: tokio::sync::mpsc::Sender<()>,
) -> Result<(), String> {
let event_dispatcher =
notify::create_event_dispatcher(Arc::new(log.new(o!("module" => "storage"))), m)?;
match m.value_of(args::STORAGE_BACKEND_TYPE) {
None | Some("filesystem") => start_ftp_with_storage(
log,
root_log,
m,
fs_storage_backend(root_log, m),
event_dispatcher,
shutdown,
done,
),
Some("gcs") => start_ftp_with_storage(
log,
root_log,
m,
gcs_storage_backend(root_log, m)?,
event_dispatcher,
shutdown,
done,
),
Some(x) => Err(format!("unknown storage back-end type {}", x)),
}
}type VfsProducer = Box<
dyn (Fn() -> storage::RooterVfs<storage::RestrictingVfs, auth::User, storage::SbeMeta>)
+ Send
+ Sync,
>;
// Creates the filesystem storage back-end
fn fs_storage_backend(log: &Logger, m: &clap::ArgMatches) -> VfsProducer {
let p: PathBuf = m.value_of(args::ROOT_DIR).unwrap().into();
let sub_log = Arc::new(log.new(o!("module" => "storage")));
Box::new(move || {
storage::RooterVfs::new(storage::RestrictingVfs {
delegate: storage::ChoosingVfs {
inner: storage::InnerVfs::File(unftp_sbe_fs::Filesystem::new(p.clone())),
log: sub_log.clone(),
},
})
})
}/**
* A virtual file system that represents either a Cloud or file system back-end.
*/
#[derive(Debug)]
pub struct ChoosingVfs {
pub inner: InnerVfs,
pub log: Arc<slog::Logger>,
}
#[derive(Debug)]
#[allow(clippy::large_enum_variant)]
pub enum InnerVfs {
Cloud(unftp_sbe_gcs::CloudStorage),
File(unftp_sbe_fs::Filesystem),
}// Given a storage back-end, starts the FTP server as a Tokio task.
fn start_ftp_with_storage<S>(
log: &Logger,
root_log: &Logger,
arg_matches: &ArgMatches,
storage_backend: Box<dyn (Fn() -> S) + Send + Sync>,
event_dispatcher: Arc<dyn EventDispatcher<FTPEvent>>,
mut shutdown: tokio::sync::broadcast::Receiver<()>,
done: tokio::sync::mpsc::Sender<()>,
) -> Result<(), String>
where
S: StorageBackend<auth::User> + Send + Sync + 'static,
S::Metadata: Sync + Send,
{
let addr = String::from(arg_matches.value_of(args::BIND_ADDRESS).unwrap());
...
}// Holds the options the libunftp user opted for.
pub struct OptionsHolder<Storage, User>
where
Storage: StorageBackend<User>,
User: UserDetail,
{
pub storage: Arc<dyn (Fn() -> Storage) + Send + Sync>,
pub greeting: &'static str,
pub authenticator: Arc<dyn Authenticator<User>>,
pub passive_ports: Range<u16>,
pub passive_host: PassiveHost,
pub ftps_config: FtpsConfig,
pub collect_metrics: bool,
pub idle_session_timeout: Duration,
pub logger: slog::Logger,
pub ftps_required_control_chan: FtpsRequired,
pub ftps_required_data_chan: FtpsRequired,
pub site_md5: SiteMd5,
pub data_listener: Arc<dyn DataListener>,
pub presence_listener: Arc<dyn PresenceListener>,
}
impl<Storage, User> From<&OptionsHolder<Storage, User>> for controlchan::LoopConfig<Storage, User>
where
User: UserDetail + 'static,
Storage: StorageBackend<User> + 'static,
Storage::Metadata: Metadata,
{
fn from(server: &OptionsHolder<Storage, User>) -> Self {
controlchan::LoopConfig {
authenticator: server.authenticator.clone(),
storage: (server.storage)(),
ftps_config: server.ftps_config.clone(),
collect_metrics: server.collect_metrics,
greeting: server.greeting,
idle_session_timeout: server.idle_session_timeout,
passive_ports: server.passive_ports.clone(),
passive_host: server.passive_host.clone(),
logger: server.logger.new(slog::o!()),
ftps_required_control_chan: server.ftps_required_control_chan,
ftps_required_data_chan: server.ftps_required_data_chan,
site_md5: server.site_md5,
data_listener: server.data_listener.clone(),
presence_listener: server.presence_listener.clone(),
}
}
}// Listener listens for control channel connections on a TCP port and spawns a control channel loop
// in a new task for each incoming connection.
pub struct Listener<Storage, User>
where
Storage: StorageBackend<User>,
User: UserDetail,
{
pub bind_address: SocketAddr,
pub logger: slog::Logger,
pub options: OptionsHolder<Storage, User>,
pub shutdown_topic: Arc<shutdown::Notifier>,
pub failed_logins: Option<Arc<FailedLoginsCache>>,
}
impl<Storage, User> Listener<Storage, User>
where
Storage: StorageBackend<User> + 'static,
User: UserDetail + 'static,
{
// Starts listening, returning an error if the TCP address could not be bound to.
pub async fn listen(self) -> std::result::Result<(), ServerError> {
let Listener {
logger,
bind_address,
options,
shutdown_topic,
failed_logins,
} = self;
let listener = TcpListener::bind(bind_address).await?;
loop {
let shutdown_listener = shutdown_topic.subscribe().await;
match listener.accept().await {
Ok((tcp_stream, socket_addr)) => {
slog::info!(logger, "Incoming control connection from {:?}", socket_addr);
let result =
controlchan::spawn_loop::<Storage, User>((&options).into(), tcp_stream, None, None, shutdown_listener, failed_logins.clone()).await;
if let Err(err) = result {
slog::error!(logger, "Could not spawn control channel loop for connection from {:?}: {:?}", socket_addr, err)
}
}
Err(err) => {
slog::error!(logger, "Error accepting incoming control connection {:?}", err);
}
}
}
}
}fn make_auth(
m: &clap::ArgMatches,
) -> Result<Arc<dyn auth_spi::Authenticator<auth::User> + Send + Sync + 'static>, String> {
let mut auth: LookupAuthenticator = match m.value_of(args::AUTH_TYPE) {
None | Some("anonymous") => make_anon_auth(),
Some("pam") => make_pam_auth(m),
Some("rest") => make_rest_auth(m),
Some("json") => make_json_auth(m),
unknown_type => Err(format!("unknown auth type: {}", unknown_type.unwrap())),
}?;
auth.set_usr_detail(match m.value_of(args::USR_JSON_PATH) {
Some(path) => {
let json: String = fs::read_to_string(path)
.map_err(|e| format!("could not load user file '{}': {}", path, e))?;
Box::new(JsonUserProvider::from_json(json.as_str())?)
}
None => Box::new(DefaultUserProvider {}),
});
Ok(Arc::new(auth))
}