use std::fmt::Debug;
use std::fmt::Formatter;
use std::io::SeekFrom;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use bb8::PooledConnection;
use bb8::RunError;
use log::debug;
use openssh::KnownHosts;
use openssh::SessionBuilder;
use openssh_sftp_client::Sftp;
use openssh_sftp_client::SftpOptions;
use tokio::io::AsyncSeekExt;
use tokio::sync::OnceCell;
use super::error::is_not_found;
use super::error::is_sftp_protocol_error;
use super::error::parse_sftp_error;
use super::error::parse_ssh_error;
use super::lister::SftpLister;
use super::reader::SftpReader;
use super::writer::SftpWriter;
use crate::raw::*;
use crate::services::SftpConfig;
use crate::*;
impl Configurator for SftpConfig {
type Builder = SftpBuilder;
fn into_builder(self) -> Self::Builder {
SftpBuilder { config: self }
}
}
#[doc = include_str!("docs.md")]
#[derive(Default)]
pub struct SftpBuilder {
config: SftpConfig,
}
impl Debug for SftpBuilder {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SftpBuilder")
.field("config", &self.config)
.finish()
}
}
impl SftpBuilder {
pub fn endpoint(mut self, endpoint: &str) -> Self {
self.config.endpoint = if endpoint.is_empty() {
None
} else {
Some(endpoint.to_string())
};
self
}
pub fn root(mut self, root: &str) -> Self {
self.config.root = if root.is_empty() {
None
} else {
Some(root.to_string())
};
self
}
pub fn user(mut self, user: &str) -> Self {
self.config.user = if user.is_empty() {
None
} else {
Some(user.to_string())
};
self
}
pub fn key(mut self, key: &str) -> Self {
self.config.key = if key.is_empty() {
None
} else {
Some(key.to_string())
};
self
}
pub fn known_hosts_strategy(mut self, strategy: &str) -> Self {
self.config.known_hosts_strategy = if strategy.is_empty() {
None
} else {
Some(strategy.to_string())
};
self
}
pub fn enable_copy(mut self, enable_copy: bool) -> Self {
self.config.enable_copy = enable_copy;
self
}
}
impl Builder for SftpBuilder {
const SCHEME: Scheme = Scheme::Sftp;
type Config = SftpConfig;
fn build(self) -> Result<impl Access> {
debug!("sftp backend build started: {:?}", &self);
let endpoint = match self.config.endpoint.clone() {
Some(v) => v,
None => return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")),
};
let user = self.config.user.clone();
let root = self
.config
.root
.clone()
.map(|r| normalize_root(r.as_str()))
.unwrap_or_default();
let known_hosts_strategy = match &self.config.known_hosts_strategy {
Some(v) => {
let v = v.to_lowercase();
if v == "strict" {
KnownHosts::Strict
} else if v == "accept" {
KnownHosts::Accept
} else if v == "add" {
KnownHosts::Add
} else {
return Err(Error::new(
ErrorKind::ConfigInvalid,
format!("unknown known_hosts strategy: {}", v).as_str(),
));
}
}
None => KnownHosts::Strict,
};
debug!("sftp backend finished: {:?}", &self);
Ok(SftpBackend {
endpoint,
root,
user,
key: self.config.key.clone(),
known_hosts_strategy,
copyable: self.config.enable_copy,
client: OnceCell::new(),
})
}
}
#[derive(Clone)]
pub struct SftpBackend {
copyable: bool,
endpoint: String,
root: String,
user: Option<String>,
key: Option<String>,
known_hosts_strategy: KnownHosts,
client: OnceCell<bb8::Pool<Manager>>,
}
pub struct Manager {
endpoint: String,
root: String,
user: Option<String>,
key: Option<String>,
known_hosts_strategy: KnownHosts,
}
#[async_trait::async_trait]
impl bb8::ManageConnection for Manager {
type Connection = Sftp;
type Error = Error;
async fn connect(&self) -> Result<Self::Connection, Self::Error> {
let mut session = SessionBuilder::default();
if let Some(user) = &self.user {
session.user(user.clone());
}
if let Some(key) = &self.key {
session.keyfile(key);
}
session.known_hosts_check(self.known_hosts_strategy.clone());
let session = session
.connect(&self.endpoint)
.await
.map_err(parse_ssh_error)?;
let sftp = Sftp::from_session(session, SftpOptions::default())
.await
.map_err(parse_sftp_error)?;
if !self.root.is_empty() {
let mut fs = sftp.fs();
let paths = Path::new(&self.root).components();
let mut current = PathBuf::new();
for p in paths {
current.push(p);
let res = fs.create_dir(p).await;
if let Err(e) = res {
if !is_sftp_protocol_error(&e) {
return Err(parse_sftp_error(e));
}
}
fs.set_cwd(¤t);
}
}
debug!("sftp connection created at {}", self.root);
Ok(sftp)
}
async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
let _ = conn.fs().metadata("./").await.map_err(parse_sftp_error)?;
Ok(())
}
fn has_broken(&self, _: &mut Self::Connection) -> bool {
false
}
}
impl Debug for SftpBackend {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Backend").finish()
}
}
impl SftpBackend {
pub async fn connect(&self) -> Result<PooledConnection<'static, Manager>> {
let client = self
.client
.get_or_try_init(|| async {
bb8::Pool::builder()
.max_size(64)
.build(Manager {
endpoint: self.endpoint.clone(),
root: self.root.clone(),
user: self.user.clone(),
key: self.key.clone(),
known_hosts_strategy: self.known_hosts_strategy.clone(),
})
.await
})
.await?;
client.get_owned().await.map_err(|err| match err {
RunError::User(err) => err,
RunError::TimedOut => {
Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary()
}
})
}
}
impl Access for SftpBackend {
type Reader = SftpReader;
type Writer = SftpWriter;
type Lister = Option<SftpLister>;
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
am.set_root(self.root.as_str())
.set_scheme(Scheme::Sftp)
.set_native_capability(Capability {
stat: true,
read: true,
write: true,
write_can_multi: true,
create_dir: true,
delete: true,
list: true,
list_with_limit: true,
copy: self.copyable,
rename: true,
shared: true,
..Default::default()
});
am.into()
}
async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
let client = self.connect().await?;
let mut fs = client.fs();
fs.set_cwd(&self.root);
let paths = Path::new(&path).components();
let mut current = PathBuf::from(&self.root);
for p in paths {
current = current.join(p);
let res = fs.create_dir(p).await;
if let Err(e) = res {
if !is_sftp_protocol_error(&e) {
return Err(parse_sftp_error(e));
}
}
fs.set_cwd(¤t);
}
Ok(RpCreateDir::default())
}
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
let client = self.connect().await?;
let mut fs = client.fs();
fs.set_cwd(&self.root);
let meta: Metadata = fs.metadata(path).await.map_err(parse_sftp_error)?.into();
Ok(RpStat::new(meta))
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let client = self.connect().await?;
let mut fs = client.fs();
fs.set_cwd(&self.root);
let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
let mut f = client
.open(path.as_path())
.await
.map_err(parse_sftp_error)?;
if args.range().offset() != 0 {
f.seek(SeekFrom::Start(args.range().offset()))
.await
.map_err(new_std_io_error)?;
}
Ok((
RpRead::default(),
SftpReader::new(client, f, args.range().size()),
))
}
async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::Writer)> {
if let Some((dir, _)) = path.rsplit_once('/') {
self.create_dir(dir, OpCreateDir::default()).await?;
}
let client = self.connect().await?;
let mut fs = client.fs();
fs.set_cwd(&self.root);
let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
let mut option = client.options();
option.create(true);
if op.append() {
option.append(true);
} else {
option.write(true).truncate(true);
}
let file = option.open(path).await.map_err(parse_sftp_error)?;
Ok((RpWrite::new(), SftpWriter::new(file)))
}
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let client = self.connect().await?;
let mut fs = client.fs();
fs.set_cwd(&self.root);
let res = if path.ends_with('/') {
fs.remove_dir(path).await
} else {
fs.remove_file(path).await
};
match res {
Ok(()) => Ok(RpDelete::default()),
Err(e) if is_not_found(&e) => Ok(RpDelete::default()),
Err(e) => Err(parse_sftp_error(e)),
}
}
async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
let client = self.connect().await?;
let mut fs = client.fs();
fs.set_cwd(&self.root);
let file_path = format!("./{}", path);
let dir = match fs.open_dir(&file_path).await {
Ok(dir) => dir,
Err(e) => {
if is_not_found(&e) {
return Ok((RpList::default(), None));
} else {
return Err(parse_sftp_error(e));
}
}
}
.read_dir();
Ok((
RpList::default(),
Some(SftpLister::new(dir, path.to_owned())),
))
}
async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
let client = self.connect().await?;
let mut fs = client.fs();
fs.set_cwd(&self.root);
if let Some((dir, _)) = to.rsplit_once('/') {
self.create_dir(dir, OpCreateDir::default()).await?;
}
let src = fs.canonicalize(from).await.map_err(parse_sftp_error)?;
let dst = fs.canonicalize(to).await.map_err(parse_sftp_error)?;
let mut src_file = client.open(&src).await.map_err(parse_sftp_error)?;
let mut dst_file = client.create(dst).await.map_err(parse_sftp_error)?;
src_file
.copy_all_to(&mut dst_file)
.await
.map_err(parse_sftp_error)?;
Ok(RpCopy::default())
}
async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
let client = self.connect().await?;
let mut fs = client.fs();
fs.set_cwd(&self.root);
if let Some((dir, _)) = to.rsplit_once('/') {
self.create_dir(dir, OpCreateDir::default()).await?;
}
fs.rename(from, to).await.map_err(parse_sftp_error)?;
Ok(RpRename::default())
}
}