This is an archive of the discontinued Mercurial Phabricator instance.

Differential D8381

rust-chg: add config validation and process returned instructions
ClosedPublic

Authored by yuja on Apr 6 2020, 10:19 AM.

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHG9ce613d648de: rust-chg: add config validation and process returned instructions
Summary

This is the reimplementation of runinstructions() and main() in chg.c.

In Rust version, we only pass in early arguments to the server as the locator
doesn't know the full arguments. This should be fine since these arguments
are just passed in to _earlyparseopts() and _parseconfig(), which means the
server doesn't need full arguments.

Another difference is the handling of the "exit <code>" instruction. In Rust
version, we can simply reuse the connection instead of "exit(code)" as the
command error isn't displayed yet. That's because the client-side stdio is not
attached until the connection is validated. This behavior is cleaner than C,
but it also means that the early server exception wouldn't be propagated to
client because stderr isn't attached. So we might have to reconsider when to
attach/detach the server stdio.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

yuja created this revision.Apr 6 2020, 10:19 AM
Alphare accepted this revision.Apr 8 2020, 4:36 AM
yuja added a commit: rHG9ce613d648de: rust-chg: add config validation and process returned instructions.Apr 9 2020, 3:26 PM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHG9ce613d648de: rust-chg: add config validation and process returned instructions (authored by yuja). · Explain Why
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mrust/chg/src/locator.rs (113 lines)

Diff 21019

rust/chg/src/locator.rs

use std::process::{self, Command}; use std::process::{self, Command};
use std::time::Duration; use std::time::Duration;
use tokio::prelude::*; use tokio::prelude::*;
use tokio_hglib::UnixClient; use tokio_hglib::UnixClient;
use tokio_process::{Child, CommandExt}; use tokio_process::{Child, CommandExt};
use tokio_timer; use tokio_timer;
use super::clientext::ChgClientExt; use super::clientext::ChgClientExt;
use super::message::ServerSpec; use super::message::{Instruction, ServerSpec};
use super::procutil; use super::procutil;
const REQUIRED_SERVER_CAPABILITIES: &[&str] = &["attachio", "chdir", "runcommand", "setenv"]; const REQUIRED_SERVER_CAPABILITIES: &[&str] =
&["attachio", "chdir", "runcommand", "setenv", "validate"];
/// Helper to connect to and spawn a server process. /// Helper to connect to and spawn a server process.
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct Locator { pub struct Locator {
hg_command: OsString, hg_command: OsString,
hg_early_args: Vec<OsString>, hg_early_args: Vec<OsString>,
current_dir: PathBuf, current_dir: PathBuf,
env_vars: Vec<(OsString, OsString)>, env_vars: Vec<(OsString, OsString)>,
process_id: u32, process_id: u32,
base_sock_path: PathBuf, base_sock_path: PathBuf,
redirect_sock_path: Option<PathBuf>,
timeout: Duration, timeout: Duration,
} }
impl Locator { impl Locator {
/// Creates locator capturing the current process environment. /// Creates locator capturing the current process environment.
/// ///
/// If no `$CHGSOCKNAME` is specified, the socket directory will be /// If no `$CHGSOCKNAME` is specified, the socket directory will be
/// created as necessary. /// created as necessary.
pub fn prepare_from_env() -> io::Result<Locator> { pub fn prepare_from_env() -> io::Result<Locator> {
Ok(Locator { Ok(Locator {
hg_command: default_hg_command(), hg_command: default_hg_command(),
hg_early_args: Vec::new(), hg_early_args: Vec::new(),
current_dir: env::current_dir()?, current_dir: env::current_dir()?,
env_vars: env::vars_os().collect(), env_vars: env::vars_os().collect(),
process_id: process::id(), process_id: process::id(),
base_sock_path: prepare_server_socket_path()?, base_sock_path: prepare_server_socket_path()?,
redirect_sock_path: None,
timeout: default_timeout(), timeout: default_timeout(),
}) })
} }
/// Temporary socket path for this client process. /// Temporary socket path for this client process.
fn temp_sock_path(&self) -> PathBuf { fn temp_sock_path(&self) -> PathBuf {
let src = self.base_sock_path.as_os_str().as_bytes(); let src = self.base_sock_path.as_os_str().as_bytes();
let mut buf = Vec::with_capacity(src.len() + 6); // "{src}.{pid}".len() let mut buf = Vec::with_capacity(src.len() + 6); // "{src}.{pid}".len()
{ {
self.hg_early_args = args.into_iter().map(|a| a.as_ref().to_owned()).collect(); self.hg_early_args = args.into_iter().map(|a| a.as_ref().to_owned()).collect();
} }
/// Connects to the server. /// Connects to the server.
/// ///
/// The server process will be spawned if not running. /// The server process will be spawned if not running.
pub fn connect(self) -> impl Future<Item = (Self, UnixClient), Error = io::Error> { pub fn connect(self) -> impl Future<Item = (Self, UnixClient), Error = io::Error> {
self.try_connect() future::loop_fn((self, 0), |(loc, cnt)| {
if cnt < 10 {
let fut = loc
.try_connect()
.and_then(|(loc, client)| {
client
.validate(&loc.hg_early_args)
.map(|(client, instructions)| (loc, client, instructions))
})
.and_then(move |(loc, client, instructions)| {
loc.run_instructions(client, instructions, cnt)
});
Either::A(fut)
} else {
let msg = format!(
concat!(
"too many redirections.\n",
"Please make sure {:?} is not a wrapper which ",
"changes sensitive environment variables ",
"before executing hg. If you have to use a ",
"wrapper, wrap chg instead of hg.",
),
loc.hg_command
);
Either::B(future::err(io::Error::new(io::ErrorKind::Other, msg)))
}
})
}
/// Runs instructions received from the server.
fn run_instructions(
mut self,
client: UnixClient,
instructions: Vec<Instruction>,
cnt: usize,
) -> io::Result<Loop<(Self, UnixClient), (Self, usize)>> {
let mut reconnect = false;
for inst in instructions {
debug!("instruction: {:?}", inst);
match inst {
Instruction::Exit(_) => {
// Just returns the current connection to run the
// unparsable command and report the error
return Ok(Loop::Break((self, client)));
}
Instruction::Reconnect => {
reconnect = true;
}
Instruction::Redirect(path) => {
if path.parent() != self.base_sock_path.parent() {
let msg = format!(
"insecure redirect instruction from server: {}",
path.display()
);
return Err(io::Error::new(io::ErrorKind::InvalidData, msg));
}
self.redirect_sock_path = Some(path);
reconnect = true;
}
Instruction::Unlink(path) => {
if path.parent() != self.base_sock_path.parent() {
let msg = format!(
"insecure unlink instruction from server: {}",
path.display()
);
return Err(io::Error::new(io::ErrorKind::InvalidData, msg));
}
fs::remove_file(path).unwrap_or(()); // may race
}
}
}
if reconnect {
Ok(Loop::Continue((self, cnt + 1)))
} else {
Ok(Loop::Break((self, client)))
}
} }
/// Tries to connect to the existing server, or spawns new if not running. /// Tries to connect to the existing server, or spawns new if not running.
fn try_connect(self) -> impl Future<Item = (Self, UnixClient), Error = io::Error> { fn try_connect(self) -> impl Future<Item = (Self, UnixClient), Error = io::Error> {
debug!("try connect to {}", self.base_sock_path.display()); let sock_path = self
UnixClient::connect(self.base_sock_path.clone()) .redirect_sock_path
.then(|res| match res { .as_ref()
.unwrap_or(&self.base_sock_path)
.clone();
debug!("try connect to {}", sock_path.display());
UnixClient::connect(sock_path)
.then(|res| {
match res {
Ok(client) => Either::A(future::ok((self, client))), Ok(client) => Either::A(future::ok((self, client))),
Err(_) => Either::B(self.spawn_connect()), Err(_) => {
// Prevent us from being re-connected to the outdated
// master server: We were told by the server to redirect
// to redirect_sock_path, which didn't work. We do not
// want to connect to the same master server again
// because it would probably tell us the same thing.
if self.redirect_sock_path.is_some() {
fs::remove_file(&self.base_sock_path).unwrap_or(());
// may race
}
Either::B(self.spawn_connect())
}
}
}) })
.and_then(|(loc, client)| { .and_then(|(loc, client)| {
check_server_capabilities(client.server_spec())?; check_server_capabilities(client.server_spec())?;
Ok((loc, client)) Ok((loc, client))
}) })
.and_then(|(loc, client)| { .and_then(|(loc, client)| {
client client
.set_current_dir(&loc.current_dir) .set_current_dir(&loc.current_dir)