use super::{parse_line, RemoteRef};
use crate::repository::Repository;
use crate::Error;
use byteorder::{BigEndian, ReadBytesExt};
use libpijul::pristine::{Base32, ChannelRef, Hash, Merkle, MutTxnT};
use libpijul::MutTxnTExt;
use regex::Regex;
use std::borrow::Cow;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use thrussh::client::Session;
pub struct Ssh {
pub h: thrussh::client::Handle,
pub c: thrussh::client::Channel,
pub channel: String,
pub remote_cmd: String,
pub path: String,
pub is_running: bool,
pub name: String,
}
lazy_static! {
static ref ADDRESS: Regex = Regex::new(
r#"((?P<user>[^@]+)@)?((?P<host>(\[([^\]]+)\])|([^:]+)))((:(?P<port>\d+)/)|:)(?P<path>.+)"#
)
.unwrap();
}
#[derive(Debug)]
pub struct Remote<'a> {
host: &'a str,
port: u16,
user: Cow<'a, str>,
path: &'a str,
}
pub fn ssh_remote<'a>(addr: &'a str) -> Option<Remote<'a>> {
let cap = if let Some(cap) = ADDRESS.captures(addr) {
cap
} else {
return None;
};
debug!("ssh_remote: {:?}", cap);
let user = if let Some(u) = cap.name("user") {
Cow::Borrowed(u.as_str())
} else {
Cow::Owned(whoami::username())
};
let host = cap.name("host").unwrap().as_str();
let port: u16 = cap
.name("port")
.map(|x| x.as_str().parse().unwrap())
.unwrap_or(22);
let path = cap.name("path").unwrap().as_str();
Some(Remote {
host,
port,
user,
path,
})
}
impl<'a> Remote<'a> {
pub async fn connect(&self, name: &str, channel: &str) -> Result<Ssh, anyhow::Error> {
let mut home = dirs::home_dir().unwrap();
home.push(".ssh");
home.push("known_hosts");
let client = SshClient {
addr: format!("{}:{}", self.host, self.port),
known_hosts: home,
last_window_adjustment: SystemTime::now(),
};
let config = Arc::new(thrussh::client::Config::default());
use std::net::ToSocketAddrs;
debug!("client: {:?}", client.addr);
debug!(
"socket: {:?}",
client.addr.to_socket_addrs().unwrap().next().unwrap()
);
let addr = client.addr.to_socket_addrs().unwrap().next().unwrap();
let mut h = thrussh::client::connect(config, &addr, client).await?;
let mut key_path = dirs::home_dir().unwrap().join(".ssh");
let authenticated = self.auth_agent(&mut h, &mut key_path).await
|| self.auth_pk(&mut h, &mut key_path).await
|| self.auth_password(&mut h).await?;
if !authenticated {
return Err(Error::NotAuthenticated.into());
}
let c = h.channel_open_session().await?;
let remote_cmd = if let Ok(cmd) = std::env::var("REMOTE_PIJUL") {
cmd
} else {
"pijul".to_string()
};
Ok(Ssh {
h,
c,
channel: channel.to_string(),
remote_cmd,
path: self.path.to_string(),
is_running: false,
name: name.to_string(),
})
}
async fn auth_agent(&self, h: &mut thrussh::client::Handle, key_path: &mut PathBuf) -> bool {
let mut authenticated = false;
match thrussh_keys::agent::client::AgentClient::connect_env().await {
Ok(agent) => {
let mut agent = Some(agent);
for k in &["id_ed25519.pub", "id_rsa.pub"] {
key_path.push(k);
if let Ok(key) = thrussh_keys::load_public_key(&key_path) {
debug!("key");
if let Some(a) = agent.take() {
debug!("authenticate future");
match h.authenticate_future(self.user.as_ref(), key, a).await {
Ok((a, auth)) => {
if !auth {
eprintln!("Key {:?} (with agent) rejected", k)
}
debug!("auth");
authenticated = auth;
agent = Some(a);
}
Err(e) => {
debug!("not auth {:?}", e);
if let Ok(thrussh_keys::Error::AgentFailure) = e.downcast() {
eprintln!("Failed to sign with agent");
}
}
}
}
}
key_path.pop();
if authenticated {
return true;
}
}
}
Err(e) => {
error!("{:?}", e);
}
}
false
}
async fn auth_pk(&self, h: &mut thrussh::client::Handle, key_path: &mut PathBuf) -> bool {
let mut authenticated = false;
for k in &["id_ed25519", "id_rsa"] {
key_path.push(k);
let k = if let Some(k) = load_secret_key(&key_path, k) {
k
} else {
key_path.pop();
continue;
};
if let Ok(auth) = h
.authenticate_publickey(self.user.as_ref(), Arc::new(k))
.await
{
authenticated = auth
}
key_path.pop();
if authenticated {
return true;
}
}
false
}
async fn auth_password(&self, h: &mut thrussh::client::Handle) -> Result<bool, anyhow::Error> {
let pass = rpassword::read_password_from_tty(Some(&format!(
"Password for {}@{}: ",
self.user, self.host
)))?;
h.authenticate_password(self.user.to_string(), &pass).await
}
}
pub fn load_secret_key(key_path: &Path, k: &str) -> Option<thrussh_keys::key::KeyPair> {
match thrussh_keys::load_secret_key(&key_path, None) {
Ok(k) => Some(k),
Err(e) => {
if let Ok(thrussh_keys::Error::KeyIsEncrypted) = e.downcast() {
let pass = if let Ok(pass) =
rpassword::read_password_from_tty(Some(&format!("Password for key {:?}: ", k)))
{
pass
} else {
return None;
};
if pass.is_empty() {
return None;
}
if let Ok(k) = thrussh_keys::load_secret_key(&key_path, Some(pass.as_bytes())) {
return Some(k);
}
}
None
}
}
}
pub struct SshClient {
addr: String,
known_hosts: PathBuf,
last_window_adjustment: SystemTime,
}
impl thrussh::client::Handler for SshClient {
type FutureBool = futures::future::Ready<Result<(Self, bool), anyhow::Error>>;
type FutureUnit = futures::future::Ready<Result<(Self, Session), anyhow::Error>>;
fn finished_bool(self, b: bool) -> Self::FutureBool {
futures::future::ready(Ok((self, b)))
}
fn finished(self, session: Session) -> Self::FutureUnit {
futures::future::ready(Ok((self, session)))
}
fn check_server_key(
self,
server_public_key: &thrussh_keys::key::PublicKey,
) -> Self::FutureBool {
let mut it = self.addr.split(':');
let addr = it.next().unwrap();
let port = it.next().unwrap_or("22").parse().unwrap();
match thrussh_keys::check_known_hosts_path(addr, port, server_public_key, &self.known_hosts)
{
Ok(e) => {
if e {
futures::future::ready(Ok((self, true)))
} else {
match learn(addr, port, server_public_key) {
Ok(x) => futures::future::ready(Ok((self, x))),
Err(e) => futures::future::ready(Err(e)),
}
}
}
Err(e) => {
error!("Key changed for {:?}", self.addr);
futures::future::ready(Err(e))
}
}
}
fn adjust_window(&mut self, _channel: thrussh::ChannelId, target: u32) -> u32 {
let elapsed = self.last_window_adjustment.elapsed().unwrap();
self.last_window_adjustment = SystemTime::now();
if target >= 10_000_000 {
return target;
}
if elapsed < Duration::from_secs(2) {
target * 2
} else if elapsed > Duration::from_secs(8) {
target / 2
} else {
target
}
}
}
fn learn(addr: &str, port: u16, pk: &thrussh_keys::key::PublicKey) -> Result<bool, anyhow::Error> {
if port == 22 {
print!(
"Unknown key for {:?}, fingerprint {:?}. Learn it (y/N)? ",
addr,
pk.fingerprint()
);
} else {
print!(
"Unknown key for {:?}:{}, fingerprint {:?}. Learn it (y/N)? ",
addr,
port,
pk.fingerprint()
);
}
std::io::stdout().flush()?;
let mut buffer = String::new();
std::io::stdin().read_line(&mut buffer)?;
let buffer = buffer.trim();
if buffer == "Y" || buffer == "y" {
thrussh_keys::learn_known_hosts(addr, port, pk)?;
Ok(true)
} else {
Ok(false)
}
}
impl Ssh {
pub async fn finish(&mut self) -> Result<(), anyhow::Error> {
self.c.eof().await?;
while let Some(msg) = self.c.wait().await {
debug!("msg = {:?}", msg);
match msg {
thrussh::ChannelMsg::Data { .. } => {}
thrussh::ChannelMsg::ExtendedData { data, ext } => {
debug!("{:?} {:?}", ext, std::str::from_utf8(&data[..]));
if let Ok(data) = std::str::from_utf8(&data) {
eprintln!("{}", data);
}
}
thrussh::ChannelMsg::WindowAdjusted { .. } => {}
thrussh::ChannelMsg::Eof => {}
thrussh::ChannelMsg::ExitStatus { exit_status } => {
if exit_status != 0 {
return Err((Error::RemoteExit {
status: exit_status,
})
.into());
}
}
msg => error!("wrong message {:?}", msg),
}
}
Ok(())
}
pub async fn get_state(
&mut self,
mid: Option<u64>,
) -> Result<Option<(u64, Merkle)>, anyhow::Error> {
self.run_protocol().await?;
if let Some(mid) = mid {
self.c
.data(format!("state {} {}\n", self.channel, mid).as_bytes())
.await?;
} else {
self.c
.data(format!("state {}\n", self.channel).as_bytes())
.await?;
}
while let Some(msg) = self.c.wait().await {
match msg {
thrussh::ChannelMsg::Data { data } => {
let mut s = std::str::from_utf8(&data)?.split(' ');
debug!("s = {:?}", s);
if let (Some(n), Some(m)) = (s.next(), s.next()) {
let n = n.parse().unwrap();
return Ok(Some((n, Merkle::from_base32(m.trim().as_bytes()).unwrap())));
} else {
break;
}
}
thrussh::ChannelMsg::ExtendedData { data, ext } => {
if ext == 1 {
debug!("{:?}", std::str::from_utf8(&data))
}
}
thrussh::ChannelMsg::Eof => {}
thrussh::ChannelMsg::ExitStatus { exit_status } => {
if exit_status != 0 {
return Err((Error::RemoteExit {
status: exit_status,
})
.into());
}
}
msg => panic!("wrong message {:?}", msg),
}
}
Ok(None)
}
pub async fn archive<W: std::io::Write>(
&mut self,
prefix: Option<String>,
state: Option<(Merkle, &[Hash])>,
mut w: W,
) -> Result<u64, anyhow::Error> {
self.run_protocol().await?;
if let Some((ref state, ref extra)) = state {
let mut cmd = format!("archive {} {}", self.channel, state.to_base32(),);
for e in extra.iter() {
cmd.push_str(&format!(" {}", e.to_base32()));
}
if let Some(ref p) = prefix {
cmd.push_str(" :");
cmd.push_str(p)
}
cmd.push('\n');
self.c.data(cmd.as_bytes()).await?;
} else {
self.c
.data(
format!(
"archive {}{}{}\n",
self.channel,
if prefix.is_some() { " :" } else { "" },
prefix.unwrap_or(String::new())
)
.as_bytes(),
)
.await?;
}
let mut len = 0;
let mut conflicts = 0;
let mut len_n = 0;
while let Some(msg) = self.c.wait().await {
match msg {
thrussh::ChannelMsg::Data { data } => {
let mut off = 0;
while len_n < 16 && off < data.len() {
if len_n < 8 {
len = (len << 8) | (data[off] as u64);
} else {
conflicts = (conflicts << 8) | (data[off] as u64);
}
len_n += 1;
off += 1;
}
if len_n >= 16 {
w.write_all(&data[off..])?;
len -= (data.len() - off) as u64;
if len == 0 {
break;
}
}
}
thrussh::ChannelMsg::ExtendedData { data, ext } => {
if ext == 1 {
debug!("{:?}", std::str::from_utf8(&data))
}
}
thrussh::ChannelMsg::Eof => {}
thrussh::ChannelMsg::ExitStatus { exit_status } => {
if exit_status != 0 {
return Err((Error::RemoteExit {
status: exit_status,
})
.into());
}
}
msg => panic!("wrong message {:?}", msg),
}
}
Ok(conflicts)
}
pub async fn run_protocol(&mut self) -> Result<(), anyhow::Error> {
if !self.is_running {
self.is_running = true;
debug!("run_protocol");
self.c
.exec(
true,
format!(
"{} protocol --version {} --repository {}",
self.remote_cmd,
crate::PROTOCOL_VERSION,
self.path
),
)
.await?;
while let Some(msg) = self.c.wait().await {
debug!("msg = {:?}", msg);
match msg {
thrussh::ChannelMsg::Success => break,
thrussh::ChannelMsg::WindowAdjusted { .. } => {}
thrussh::ChannelMsg::Eof => {}
thrussh::ChannelMsg::ExitStatus { exit_status } => {
if exit_status != 0 {
return Err((Error::RemoteExit {
status: exit_status,
})
.into());
}
}
_ => {}
}
}
debug!("run_protocol done");
}
Ok(())
}
pub async fn download_changelist<T: MutTxnT>(
&mut self,
txn: &mut T,
remote: &mut RemoteRef<T>,
from: u64,
paths: &[String],
) -> Result<(), anyhow::Error> {
self.run_protocol().await?;
debug!("download_changelist");
let mut command = Vec::new();
write!(command, "changelist {} {}", self.channel, from).unwrap();
for p in paths {
write!(command, " {}", p).unwrap()
}
command.push(b'\n');
self.c.data(&command[..]).await?;
debug!("waiting ssh");
'msg: while let Some(msg) = self.c.wait().await {
debug!("msg = {:?}", msg);
match msg {
thrussh::ChannelMsg::Data { data } => {
if &data[..] == b"\n" {
debug!("log done");
break;
} else if let Ok(data) = std::str::from_utf8(&data) {
for l in data.lines() {
if !l.is_empty() {
debug!("line = {:?}", l);
let (n, h, m) = parse_line(l)?;
txn.put_remote(remote, n, (h, m))?;
} else {
break 'msg;
}
}
}
}
thrussh::ChannelMsg::ExtendedData { data, ext } => {
debug!("{:?} {:?}", ext, std::str::from_utf8(&data[..]));
}
thrussh::ChannelMsg::WindowAdjusted { .. } => {}
thrussh::ChannelMsg::Eof => {}
thrussh::ChannelMsg::ExitStatus { exit_status } => {
if exit_status != 0 {
return Err((Error::RemoteExit {
status: exit_status,
})
.into());
}
}
msg => panic!("wrong message {:?}", msg),
}
}
debug!("no msg");
Ok(())
}
pub async fn upload_changes(
&mut self,
mut local: PathBuf,
to_channel: Option<&str>,
changes: &[Hash],
) -> Result<(), anyhow::Error> {
self.run_protocol().await?;
debug!("upload_changes");
for c in changes {
libpijul::changestore::filesystem::push_filename(&mut local, &c);
let mut change_file = std::fs::File::open(&local)?;
let change_len = change_file.metadata()?.len();
let mut change = cryptovec::CryptoVec::new_zeroed(change_len as usize);
use std::io::Read;
change_file.read_exact(&mut change[..])?;
let to_channel = if let Some(t) = to_channel {
t
} else {
self.channel.as_str()
};
self.c
.data(format!("apply {} {} {}\n", to_channel, c.to_base32(), change_len).as_bytes())
.await?;
self.c.data(&change[..]).await?;
libpijul::changestore::filesystem::pop_filename(&mut local);
}
Ok(())
}
pub async fn start_change_download(
&mut self,
c: libpijul::pristine::Hash,
full: bool,
) -> Result<(), anyhow::Error> {
self.run_protocol().await?;
debug!("download_change {:?}", full);
if full {
self.c
.data(format!("change {}\n", c.to_base32()).as_bytes())
.await?;
} else {
self.c
.data(format!("partial {}\n", c.to_base32()).as_bytes())
.await?;
}
Ok(())
}
pub async fn wait_downloads(
&mut self,
changes_dir: &Path,
hashes: &[libpijul::pristine::Hash],
send: &mut tokio::sync::mpsc::Sender<libpijul::pristine::Hash>,
) -> Result<(), anyhow::Error> {
debug!("wait_downloads");
if !self.is_running {
return Ok(());
}
let mut remaining_len = 0;
let mut current: usize = 0;
let mut path = changes_dir.to_path_buf();
libpijul::changestore::filesystem::push_filename(&mut path, &hashes[current]);
std::fs::create_dir_all(&path.parent().unwrap())?;
path.set_extension("");
let mut file = std::fs::File::create(&path)?;
'outer: while let Some(msg) = self.c.wait().await {
match msg {
thrussh::ChannelMsg::Data { data } => {
debug!("data = {:?}", &data[..]);
let mut p = 0;
while p < data.len() {
if remaining_len == 0 {
remaining_len = (&data[p..]).read_u64::<BigEndian>().unwrap() as usize;
p += 8;
debug!("remaining_len = {:?}", remaining_len);
}
if data.len() >= p + remaining_len {
file.write_all(&data[p..p + remaining_len])?;
p += remaining_len;
remaining_len = 0;
file.flush()?;
let mut final_path = path.clone();
final_path.set_extension("change");
debug!("moving {:?} to {:?}", path, final_path);
std::fs::rename(&path, &final_path)?;
debug!("sending");
send.send(hashes[current].clone()).await.unwrap();
debug!("sent");
current += 1;
if current < hashes.len() {
libpijul::changestore::filesystem::pop_filename(&mut path);
libpijul::changestore::filesystem::push_filename(
&mut path,
&hashes[current],
);
std::fs::create_dir_all(&path.parent().unwrap())?;
path.set_extension("");
file = std::fs::File::create(&path)?;
} else {
break 'outer;
}
} else {
file.write_all(&data[p..])?;
remaining_len -= data.len() - p;
break;
}
}
}
thrussh::ChannelMsg::ExitStatus { exit_status } => {
debug!("exit: {:?}", exit_status);
if exit_status != 0 {
error!("Remote command returned {:?}", exit_status)
}
self.is_running = false;
return Ok(());
}
msg => {
debug!("{:?}", msg);
}
}
}
debug!("done waiting for downloads");
Ok(())
}
pub async fn clone_channel<T: MutTxnTExt>(
&mut self,
repo: &mut Repository,
txn: &mut T,
channel: &mut ChannelRef<T>,
lazy: bool,
) -> Result<(), anyhow::Error> {
self.run_protocol().await?;
self.c
.data(format!("channel {}\n", self.channel).as_bytes())
.await?;
let from_dump_alive = {
let mut from_dump =
libpijul::pristine::channel_dump::ChannelFromDump::new(txn, channel.clone());
while let Some(msg) = self.c.wait().await {
match msg {
thrussh::ChannelMsg::Data { data } => {
debug!("data = {:?}", &data[..]);
if from_dump.read(&data)? {
break;
}
}
thrussh::ChannelMsg::ExtendedData { data, ext } => {
debug!("data = {:?}, ext = {:?}", &data[..], ext);
}
thrussh::ChannelMsg::ExitStatus { exit_status } => {
if exit_status != 0 {
error!("Remote command returned {:?}", exit_status)
}
self.is_running = false;
break;
}
msg => {
debug!("msg = {:?}", msg);
}
}
}
from_dump.alive
};
let channel_ = channel.borrow();
debug!("cloned, now downloading changes");
let mut hashes = Vec::new();
if lazy {
for &ch in from_dump_alive.iter() {
let h = txn.get_external(ch).unwrap();
self.c
.data(format!("change {}\n", h.to_base32()).as_bytes())
.await?;
hashes.push(h);
}
} else {
for (_, (ch, _)) in txn.changeid_log(&channel_, 0) {
let h = txn.get_external(ch).unwrap();
self.c
.data(format!("change {}\n", h.to_base32()).as_bytes())
.await?;
hashes.push(h);
}
}
std::mem::drop(channel_);
debug!("hashes = {:?}", hashes);
let (mut send, recv) = tokio::sync::mpsc::channel(100);
self.wait_downloads(&repo.changes_dir, &hashes, &mut send)
.await?;
txn.output_repository_no_pending(&mut repo.working_copy, &repo.changes, channel, "", true)?;
std::mem::drop(recv);
Ok(())
}
}