git.delta.rocks / remowt / refs/commits / 751b8f1ca8aa

--- a/Cargo.lock
+++ b/Cargo.lock
@@ -308,9 +308,9 @@
 
 [[package]]
 name = "bifrostlink"
-version = "0.2.2"
+version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4bd6d4c7cf73270cbf6846a42361d409df23262d276ff15110e787067d89ad1d"
+checksum = "704657867d2107831c57edd363726440c86675464b5fc113702854e17062cafc"
 dependencies = [
  "async-trait",
  "async_fn_traits",
@@ -327,9 +327,9 @@
 
 [[package]]
 name = "bifrostlink-macros"
-version = "0.2.2"
+version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "329813d3e34c7e65638480cefbb09d1e1dc47fa5dbf0f5f0cf8d9eee424f4f19"
+checksum = "158308eb569b467c0116680f79d0ecc389f4d540f6d5a0c9279bfe79b1cd5bdb"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -338,9 +338,9 @@
 
 [[package]]
 name = "bifrostlink-ports"
-version = "0.2.2"
+version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6683eb0d4366b762fc45c0eb14e384d3554a23bc1ba3f1246368900d41e3700"
+checksum = "7612993f0bd8bc6a71867461266567212a35a716b2a5aef5f9967ab08c891782"
 dependencies = [
  "bifrostlink",
  "bytes",
@@ -1835,7 +1835,7 @@
 
 [[package]]
 name = "polkit-backend"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "anyhow",
  "clap",
@@ -2055,7 +2055,7 @@
 
 [[package]]
 name = "remowt-agent"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "anyhow",
  "bifrostlink",
@@ -2083,13 +2083,14 @@
 
 [[package]]
 name = "remowt-client"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "anyhow",
  "bifrostlink",
  "bifrostlink-ports",
  "bytes",
  "camino",
+ "futures",
  "remowt-endpoints",
  "remowt-link-shared",
  "russh",
@@ -2098,13 +2099,14 @@
  "serde_json",
  "tempfile",
  "tokio",
+ "tokio-util",
  "tracing",
  "uuid",
 ]
 
 [[package]]
 name = "remowt-endpoints"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "anyhow",
  "bifrostlink",
@@ -2122,7 +2124,7 @@
 
 [[package]]
 name = "remowt-link-shared"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "bifrostlink",
  "bytes",
@@ -2136,7 +2138,7 @@
 
 [[package]]
 name = "remowt-plugin"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "anyhow",
  "bifrostlink",
@@ -2150,7 +2152,7 @@
 
 [[package]]
 name = "remowt-polkit-shared"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "nix",
  "serde",
@@ -2159,7 +2161,7 @@
 
 [[package]]
 name = "remowt-ssh"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -2187,7 +2189,7 @@
 
 [[package]]
 name = "remowt-ui-prompt"
-version = "0.1.1"
+version = "0.1.3"
 dependencies = [
  "bifrostlink",
  "bifrostlink-macros",

--- a/Cargo.toml
+++ b/Cargo.toml
@@ -3,18 +3,18 @@
 resolver = "2"
 
 [workspace.package]
-version = "0.1.1"
+version = "0.1.3"
 license = "MIT"
 edition = "2021"
-repository = "https://gitlab.delta.directory/iam/remowt"
+repository = "https://git.delta.rocks/r/remowt"
 
 [workspace.dependencies]
-remowt-client = { version = "0.1.1", path = "crates/remowt-client" }
-remowt-polkit-shared = { version = "0.1.1", path = "crates/polkit-shared" }
-remowt-link-shared = { version = "0.1.1", path = "crates/remowt-link-shared" }
-remowt-plugin = { version = "0.1.1", path = "crates/remowt-plugin" }
-remowt-ui-prompt = { version = "0.1.1", path = "crates/remowt-ui-prompt" }
-remowt-endpoints = { version = "0.1.1", path = "crates/remowt-endpoints" }
+remowt-client = { version = "0.1.3", path = "crates/remowt-client" }
+remowt-polkit-shared = { version = "0.1.3", path = "crates/polkit-shared" }
+remowt-link-shared = { version = "0.1.3", path = "crates/remowt-link-shared" }
+remowt-plugin = { version = "0.1.3", path = "crates/remowt-plugin" }
+remowt-ui-prompt = { version = "0.1.3", path = "crates/remowt-ui-prompt" }
+remowt-endpoints = { version = "0.1.3", path = "crates/remowt-endpoints" }
 
 bifrostlink = "0.2.0"
 bifrostlink-macros = "0.2.0"

--- a/cmds/remowt-agent/src/main.rs
+++ b/cmds/remowt-agent/src/main.rs
@@ -11,7 +11,7 @@
 use bifrostlink_ports::stdio::from_stdio;
 use bifrostlink_ports::unix_socket::from_socket;
 use clap::Parser;
-use remowt_endpoints::{fs::Fs, pty::Pty, systemd::Systemd};
+use remowt_endpoints::{fs::Fs, nix_daemon::NixDaemon, pty::Pty, subprocess::Subprocess, systemd::Systemd};
 use remowt_link_shared::{editor::EditorEndpointsClient, Address, BifConfig};
 use remowt_polkit_shared::{emphasize, BackendRequest, Identity, PidDisplay};
 use remowt_ui_prompt::bifrost::PromptEndpointsClient;
@@ -21,7 +21,7 @@
 use tokio::net::UnixStream;
 use tokio::runtime::Builder;
 use tokio::task::AbortHandle;
-use tracing::{debug, info, trace};
+use tracing::{debug, trace};
 use zbus::fdo;
 use zbus::zvariant::{OwnedValue, Str};
 use zbus::{interface, proxy, Connection};
@@ -85,7 +85,7 @@
 		identities: Vec<Identity>,
 	) -> zbus::fdo::Result<()> {
 		use std::fmt::Write;
-		info!("begin auth");
+		debug!("begin auth");
 		let _cancel_guard = Arc::new(OnceLock::new());
 		let task = {
 			let helper = self.helper.clone();
@@ -220,6 +220,8 @@
 		/// Expect own address to be AgentPrivileged, skip installing polkit agent
 		#[arg(long)]
 		privileged: bool,
+		#[arg(long)]
+		local: bool,
 	},
 	LocalAgent,
 }
@@ -240,7 +242,11 @@
 		} => runtime.block_on(askpass::ask(&prompt, description)),
 		Opts::LocalAgent => runtime.block_on(main_real()),
 		Opts::Editor { path } => runtime.block_on(editor::edit(path)),
-		Opts::RealAgent { path, privileged } => runtime.block_on(main_real_agent(path, privileged)),
+		Opts::RealAgent {
+			path,
+			privileged,
+			local,
+		} => runtime.block_on(main_real_agent(path, privileged, local)),
 	}
 }
 async fn main_real() -> anyhow::Result<()> {
@@ -253,7 +259,11 @@
 	let _conn = conn;
 	pending().await
 }
-async fn main_real_agent(path: Option<PathBuf>, privileged: bool) -> anyhow::Result<()> {
+async fn main_real_agent(
+	path: Option<PathBuf>,
+	privileged: bool,
+	local: bool,
+) -> anyhow::Result<()> {
 	let address = if privileged {
 		Address::AgentPrivileged
 	} else {
@@ -264,6 +274,8 @@
 	Fs::new().register_endpoints(&mut rpc);
 	Systemd.register_endpoints(&mut rpc);
 	Pty::new().register_endpoints(&mut rpc);
+	Subprocess::new().register_endpoints(&mut rpc);
+	NixDaemon.register_endpoints(&mut rpc);
 
 	remowt_plugin::host::serve(&mut rpc);
 
@@ -312,7 +324,7 @@
 	};
 	rpc.add_direct(Address::User, port, bifrostlink::Rtt(0));
 
-	let polkit_conn = if !privileged {
+	let polkit_conn = if !privileged && !local {
 		// The unprivileged agent doubles as a polkit authentication agent so
 		// `run0` (e.g. our own elevation) routes its prompt to the User over
 		// bifrost instead of failing on a tty-less session.

--- a/cmds/remowt-ssh/Cargo.toml
+++ b/cmds/remowt-ssh/Cargo.toml
@@ -11,7 +11,7 @@
 tracing-subscriber.workspace = true
 bifrostlink.workspace = true
 remowt-link-shared.workspace = true
-remowt-client = { workspace = true, features = ["shell"] }
+remowt-client.workspace = true
 tokio = { workspace = true, features = [
 	"macros",
 	"fs",

--- a/cmds/remowt-ssh/src/main.rs
+++ b/cmds/remowt-ssh/src/main.rs
@@ -24,9 +24,16 @@
 #[derive(Parser)]
 enum Opts {
 	/// Connect to remote host with remowt agent.
-	Ssh { host: String },
+	Ssh {
+		host: String,
+		#[arg(long)]
+		escalate: bool,
+	},
 	/// Connect to local host for testing the connectivity.
-	Local,
+	Local {
+		#[arg(long)]
+		escalate: bool,
+	},
 }
 
 fn agents_dir() -> anyhow::Result<PathBuf> {
@@ -45,9 +52,9 @@
 	let opts = Opts::parse();
 
 	let bundle = AgentBundle::from_dir(agents_dir()?)?;
-	let conn = match &opts {
-		Opts::Ssh { host } => Remowt::connect(host, &bundle).await?,
-		Opts::Local => Remowt::connect_local(&bundle).await?,
+	let (conn, escalate) = match &opts {
+		Opts::Ssh { host, escalate } => (Remowt::connect(host, &bundle).await?, *escalate),
+		Opts::Local { escalate } => (Remowt::connect_local(&bundle).await?, *escalate),
 	};
 	let mut rpc = conn.rpc();
 
@@ -56,8 +63,8 @@
 		PrependSourcePrompter {
 			prompter: RofiPrompter,
 			source: match opts {
-				Opts::Ssh { host } => vec![Source(Cow::Owned(format!("ssh host: {}", host)))],
-				Opts::Local => vec![],
+				Opts::Ssh { host, .. } => vec![Source(Cow::Owned(format!("ssh host: {}", host)))],
+				Opts::Local { .. } => vec![],
 			},
 			description: "".to_owned(),
 		},
@@ -67,13 +74,13 @@
 	}
 
 	debug!("entering shell");
-	run_shell(&conn).await?;
+	run_shell(&conn, escalate).await?;
 	debug!("shell ended");
 
 	Ok(())
 }
 
-async fn run_shell(conn: &Remowt) -> anyhow::Result<()> {
+async fn run_shell(conn: &Remowt, escalate: bool) -> anyhow::Result<()> {
 	let term = match std::env::var("TERM") {
 		Ok(v) => v,
 		Err(VarError::NotPresent) => "xterm-256color".to_owned(),
@@ -81,7 +88,7 @@
 	};
 	let (cols, rows) = term_size().unwrap_or((80, 24));
 
-	let shell = conn.open_shell(&term, cols, rows).await?;
+	let shell = conn.open_shell(&term, cols, rows, escalate).await?;
 	let resizer = shell.resizer();
 	let stream = shell.stream;
 

--- a/crates/remowt-client/Cargo.toml
+++ b/crates/remowt-client/Cargo.toml
@@ -27,7 +27,6 @@
 ] }
 tracing.workspace = true
 uuid = { workspace = true, features = ["v4"] }
-remowt-endpoints = { workspace = true, optional = true }
-
-[features]
-shell = ["dep:remowt-endpoints"]
+remowt-endpoints.workspace = true
+tokio-util = { workspace = true, features = ["codec"] }
+futures.workspace = true

6use anyhow::{anyhow, bail, ensure, Context as _, Result};6use anyhow::{anyhow, bail, ensure, Context as _, Result};
7use bifrostlink::declarative::RemoteEndpoints;7use bifrostlink::declarative::RemoteEndpoints;
8use bifrostlink::{Remote, Rpc, Rtt};8use bifrostlink::{Remote, Rpc, Rtt};
9use bifrostlink_ports::unix_socket::from_socket;
10use camino::{Utf8Path, Utf8PathBuf};9use camino::{Utf8Path, Utf8PathBuf};
11use remowt_link_shared::plugin::PluginEndpointsClient;10use remowt_link_shared::plugin::PluginEndpointsClient;
12use remowt_link_shared::port::child_port;11use remowt_link_shared::port::child_port;
19use russh::Channel;18use russh::Channel;
20use tempfile::TempDir;19use tempfile::TempDir;
21use tokio::net::UnixListener;20use tokio::net::UnixListener;
22use tokio::sync::oneshot::{self, channel};21use tokio::sync::oneshot;
23use tokio::{22use tokio::{
24	fs,23	fs,
25	io::{AsyncReadExt as _, AsyncWriteExt as _},24	io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _, BufReader, DuplexStream},
26};25};
27use tracing::{debug, error};26use tracing::{debug, warn};
28use uuid::Uuid;27use uuid::Uuid;
2928
30use self::port::channel_port;
31use self::subprocess::RemowtChild;
32
33pub mod editor;29pub mod editor;
34mod forwarded;30mod forwarded;
35mod port;
36#[cfg(feature = "shell")]
37mod shell;31mod shell;
32mod ssh_exec;
38mod subprocess;33mod subprocess;
3934
35use self::ssh_exec::SshExecChild;
36pub use self::subprocess::{RemowtChild, SpawnOptions, StderrMode, StdioMode};
40pub use forwarded::{RemowtListener, RemowtStream};37pub use forwarded::{RemowtListener, RemowtStream};
41#[cfg(feature = "shell")]
42pub use shell::{RemowtShell, RemowtShellResizer};38pub use shell::{RemowtShell, RemowtShellResizer};
4339
44type Subs = Arc<Mutex<HashMap<Utf8PathBuf, oneshot::Sender<Channel<Msg>>>>>;40type Subs = Arc<Mutex<HashMap<Utf8PathBuf, oneshot::Sender<Channel<Msg>>>>>;
99	let ch = sess.channel_open_session().await?;95	let ch = sess.channel_open_session().await?;
100	ch.exec(true, cmd).await?;96	ch.exec(true, cmd).await?;
10197
102	let mut child = RemowtChild::from_exec(ch);98	let mut child = SshExecChild::from_exec(ch);
103	drop(child.stdin);99	drop(child.stdin);
100	drain_stderr(child.stderr, cmd.to_owned());
104101
105	let mut out = Vec::new();102	let mut out = Vec::new();
106	let mut err = Vec::new();103	child.stdout.read_to_end(&mut out).await?;
107	tokio::try_join!(
108		child.stdout.read_to_end(&mut out),
109		child.stderr.read_to_end(&mut err),
110	)?;
111	if !err.is_empty() {
112		error!("remote stderr: {}", String::from_utf8_lossy(&err).trim());
113	}
114	let code = child.exit.await.ok().flatten();104	let code = child.exit.await.ok().flatten();
115	Ok((code, out))105	Ok((code, out))
116}106}
140		.ok_or_else(|| anyhow!("no remowt-agent build for remote arch {arch:?}"))?;130		.ok_or_else(|| anyhow!("no remowt-agent build for remote arch {arch:?}"))?;
141131
142	debug!("get dir");132	debug!("get dir");
143	let cache = run_string_ok(sess, "echo \"$XDG_CACHE_HOME\"")133	let cache = run_string_ok(sess, "echo \"$XDG_CACHE_HOME\"").await?;
144		.await?
145		.to_owned();
146	let dir = if cache.is_empty() {134	let dir = if cache.is_empty() {
147		let home = run_string_ok(sess, "echo \"$HOME\"").await?;135		let home = run_string_ok(sess, "echo \"$HOME\"").await?;
148		ensure!(136		ensure!(
183	debug!("cat");171	debug!("cat");
184	ch.exec(true, format!("cat > {}", sh_quote(&tmp))).await?;172	ch.exec(true, format!("cat > {}", sh_quote(&tmp))).await?;
185173
186	let mut child = RemowtChild::from_exec(ch);174	let mut child = SshExecChild::from_exec(ch);
187	child175	child
188		.stdin176		.stdin
189		.write_all(&bytes)177		.write_all(&bytes)
207	Ok(())195	Ok(())
208}196}
209197
210async fn detect_escalation(
211	sess: &Handle<SshHandler>,
212) -> Result<(&'static str, &'static [&'static str])> {
213	for (tool, flags) in ESCALATORS {
214		// `tool` is a fixed identifier (no metacharacters), safe to interpolate.
215		let (code, _) = run(sess, &format!("command -v {tool}")).await?;
216		if code == Some(0) {
217			return Ok((tool, flags));
218		}
219	}
220	bail!("no escalation tool found on remote")
221}
222
223fn privileged_cmd(tool: &str, flags: &[&str], agent_path: &Utf8Path, path: Option<&str>) -> String {
224	let mut parts = vec![tool.to_owned()];
225	parts.extend(flags.iter().map(|f| f.to_string()));
226	parts.push(sh_quote(agent_path));
227	parts.push("real-agent".to_owned());
228	parts.push("--privileged".to_owned());
229	if let Some(p) = path {
230		parts.push("--path".to_owned());
231		parts.push(sh_quote(p));
232	}
233	parts.join(" ")
234}
235
236fn find_in_path(name: &str) -> Option<std::path::PathBuf> {
237	let path = env::var_os("PATH")?;
238	env::split_paths(&path)
239		.map(|dir| dir.join(name))
240		.find(|p| p.is_file())
241}
242
243pub struct SshHandler {198pub struct SshHandler {
244	host: String,199	host: String,
245	port: u16,200	port: u16,
285	},240	},
286}241}
287242
288pub struct Remowt {243struct RemowtInner {
289	transport: Transport,244	transport: Transport,
290	rpc: Rpc<BifConfig>,245	rpc: Rpc<BifConfig>,
291	elevated: tokio::sync::OnceCell<()>,246	elevated: tokio::sync::OnceCell<()>,
247	#[allow(dead_code)]
292	children: Mutex<Vec<tokio::process::Child>>,248	children: Mutex<Vec<tokio::process::Child>>,
293	_runtime_tmp: Option<TempDir>,249	_runtime_tmp: Option<TempDir>,
294}250}
295251
252#[derive(Clone)]
253pub struct Remowt(Arc<RemowtInner>);
254
296pub type RemowtRemote = Remote<BifConfig>;255pub type RemowtRemote = Remote<BifConfig>;
297256
298impl Remowt {257impl Remowt {
258	/// Connect to the remote host over ssh, detect the architecture and deploy the required
259	/// agent binary.
299	pub async fn connect(host: &str, bundle: &AgentBundle) -> Result<Self> {260	pub async fn connect(host: &str, bundle: &AgentBundle) -> Result<Self> {
300		let conf = russh_config::parse_home(host)?;261		let conf = russh_config::parse_home(host)?;
301		let port = conf.host_config.port.or(conf.port).unwrap_or(22);262		let port = conf.host_config.port.or(conf.port).unwrap_or(22);
346307
347		debug!("runtime dir");308		debug!("runtime dir");
348		let runtime_dir = remote_runtime_dir(&sess).await?;309		let runtime_dir = remote_runtime_dir(&sess).await?;
349		let primary = runtime_dir.join(format!("remowt-{}.sock", Uuid::new_v4()));
350310
351		let (onetx, onerx) = channel();
352		subs.lock().expect("lock").insert(primary.clone(), onetx);
353		sess.streamlocal_forward(primary.clone()).await?;
354
355		let rpc = Rpc::<BifConfig>::new(Address::User);311		let rpc = Rpc::<BifConfig>::new(Address::User);
356312
357		// TODO: ensure no injection is possible in the socket path.
358		let cmd_chan = sess.channel_open_session().await?;313		let cmd_chan = sess.channel_open_session().await?;
359		debug!("starting agent");314		debug!("starting agent");
360		cmd_chan315		cmd_chan
361			.exec(316			.exec(true, format!("{} real-agent", sh_quote(&agent_path)))
362				true,
363				format!(
364					"{} real-agent --path={}",
365					sh_quote(&agent_path),
366					sh_quote(&primary)
367				),
368			)
369			.await?;317			.await?;
370318
371		let port = channel_port(319		let child = SshExecChild::from_exec(cmd_chan);
372			onerx320		drain_stderr(child.stderr, "agent".to_owned());
321		rpc.add_direct(
373				.await322			Address::Agent,
323			child_port(child.stdout, child.stdin),
374				.map_err(|_| anyhow!("agent never opened its channel"))?,324			Rtt(0),
375		);325		);
376		rpc.add_direct(Address::Agent, port, Rtt(0));
377326
378		Ok(Self {327		Ok(Self(Arc::new(RemowtInner {
379			transport: Transport::Ssh {328			transport: Transport::Ssh {
380				sess,329				sess,
381				subs,330				subs,
386			elevated: tokio::sync::OnceCell::new(),335			elevated: tokio::sync::OnceCell::new(),
387			children: Mutex::new(Vec::new()),336			children: Mutex::new(Vec::new()),
388			_runtime_tmp: None,337			_runtime_tmp: None,
389		})338		})))
390	}339	}
391340
341	/// "Connect" to the local machine's agent, by starting the agent binary locally.
392	pub async fn connect_local(bundle: &AgentBundle) -> Result<Self> {342	pub async fn connect_local(bundle: &AgentBundle) -> Result<Self> {
393		let agent_path = bundle.local_binary()?;343		let agent_path = bundle.local_binary()?;
394		let mut child = tokio::process::Command::new(&agent_path)344		let mut child = tokio::process::Command::new(&agent_path)
395			.arg("real-agent")345			.arg("real-agent")
346			.arg("--local")
396			.stdin(std::process::Stdio::piped())347			.stdin(std::process::Stdio::piped())
397			.stdout(std::process::Stdio::piped())348			.stdout(std::process::Stdio::piped())
398			.kill_on_drop(true)349			.kill_on_drop(true)
406357
407		let (runtime_dir, runtime_tmp) = local_runtime_dir()?;358		let (runtime_dir, runtime_tmp) = local_runtime_dir()?;
408359
409		Ok(Self {360		Ok(Self(Arc::new(RemowtInner {
410			transport: Transport::Local {361			transport: Transport::Local {
411				agent_path,362				agent_path,
412				runtime_dir,363				runtime_dir,
415			elevated: tokio::sync::OnceCell::new(),366			elevated: tokio::sync::OnceCell::new(),
416			children: Mutex::new(vec![child]),367			children: Mutex::new(vec![child]),
417			_runtime_tmp: runtime_tmp,368			_runtime_tmp: runtime_tmp,
418		})369		})))
419	}370	}
420371
372	/// Get the handle to the underlying russh session handle.
421	pub fn ssh(&self) -> Option<Arc<Handle<SshHandler>>> {373	pub fn ssh(&self) -> Option<Arc<Handle<SshHandler>>> {
422		match &self.transport {374		match &self.0.transport {
423			Transport::Ssh { sess, .. } => Some(sess.clone()),375			Transport::Ssh { sess, .. } => Some(sess.clone()),
424			Transport::Local { .. } => None,376			Transport::Local { .. } => None,
425		}377		}
426	}378	}
427379
428	pub fn rpc(&self) -> Rpc<BifConfig> {380	pub fn rpc(&self) -> Rpc<BifConfig> {
429		self.rpc.clone()381		self.0.rpc.clone()
430	}382	}
431383
432	pub fn endpoints<R: RemoteEndpoints<BifConfig>>(&self) -> R {
433		R::wrap(self.rpc.remote(Address::Agent))
434	}
435
436	pub async fn load_plugin(&self, id: u16, name: &str) -> Result<()> {384	pub async fn load_plugin(&self, id: u16, name: &str) -> Result<()> {
437		let client: PluginEndpointsClient<BifConfig> = self.endpoints();385		let client: PluginEndpointsClient<BifConfig> = self.endpoints();
438		client386		client
441			.map_err(|e| anyhow!("agent failed to load plugin: {e}"))389			.map_err(|e| anyhow!("agent failed to load plugin: {e}"))
442	}390	}
443	pub async fn run0_load_plugin_path(&self, id: u16, path: &str) -> Result<()> {391	pub async fn run0_load_plugin_path(&self, id: u16, path: &str) -> Result<()> {
444		self.ensure_elevated().await?;392		self.ensure_escalated().await?;
445		let client: PluginEndpointsClient<BifConfig> =393		let client: PluginEndpointsClient<BifConfig> =
446			PluginEndpointsClient::wrap(self.rpc.remote(Address::AgentPrivileged));394			PluginEndpointsClient::wrap(self.0.rpc.remote(Address::AgentPrivileged));
447		client395		client
448			.load_plugin_path(id, path.to_owned())396			.load_plugin_path(id, path.to_owned())
449			.await?397			.await?
450			.map_err(|e| anyhow!("privileged agent failed to load plugin: {e}"))398			.map_err(|e| anyhow!("privileged agent failed to load plugin: {e}"))
451	}399	}
452	pub fn plugin_endpoints<R: RemoteEndpoints<BifConfig>>(&self, id: u16) -> R {400	pub fn plugin_endpoints<R: RemoteEndpoints<BifConfig>>(&self, id: u16) -> R {
453		R::wrap(self.rpc.remote(Address::Plugin(id)))401		R::wrap(self.0.rpc.remote(Address::Plugin(id)))
454	}402	}
403
404	pub fn endpoints<R: RemoteEndpoints<BifConfig>>(&self) -> R {
405		R::wrap(self.0.rpc.remote(Address::Agent))
406	}
455	pub async fn run0_endpoints<R: RemoteEndpoints<BifConfig>>(&self) -> Result<R> {407	pub async fn run0_endpoints<R: RemoteEndpoints<BifConfig>>(&self) -> Result<R> {
456		self.ensure_elevated().await?;408		self.ensure_escalated().await?;
457		Ok(R::wrap(self.rpc.remote(Address::AgentPrivileged)))409		Ok(R::wrap(self.0.rpc.remote(Address::AgentPrivileged)))
458	}410	}
459411
460	async fn ensure_elevated(&self) -> Result<()> {412	async fn ensure_escalated(&self) -> Result<()> {
461		self.elevated413		self.0
414			.elevated
462			.get_or_try_init(|| async {415			.get_or_try_init(|| async {
463				let port = match &self.transport {416				let (agent_path, local) = match &self.0.transport {
464					Transport::Ssh {417					Transport::Ssh { agent_path, .. } => (agent_path.as_str().to_owned(), false),
465						sess, agent_path, ..
466					} => {
467						let (tool, flags) = detect_escalation(sess).await?;
468						let ch = sess.channel_open_session().await?;
469						ch.exec(true, privileged_cmd(tool, flags, agent_path, None))
470							.await?;418					Transport::Local { agent_path, .. } => (
471						channel_port(ch)
472					}
473					Transport::Local { agent_path, .. } => {
474						let sock = self
475							.runtime_dir()
476							.join(format!("remowt-priv-{}.sock", Uuid::new_v4()));419						agent_path
477						let _ = std::fs::remove_file(&sock);420							.to_str()
478						let listener = UnixListener::bind(&sock)?;
479						let (tool, flags) = ESCALATORS
480							.iter()
481							.find(|(t, _)| find_in_path(t).is_some())421							.ok_or_else(|| anyhow!("local agent path is not utf-8"))?
482							.ok_or_else(|| anyhow!("no escalation tool found"))?;
483						let child = tokio::process::Command::new(tool)
484							.args(*flags)
485							.arg(agent_path)
486							.arg("real-agent")
487							.arg("--privileged")
488							.arg("--path")422							.to_owned(),
489							.arg(sock.as_str())423						true,
490							.kill_on_drop(true)
491							.spawn()?;
492						self.children.lock().expect("lock").push(child);
493						let (stream, _) = listener.accept().await?;
494						let _ = std::fs::remove_file(&sock);424					),
495						from_socket(stream)
496					}
497				};425				};
426
498				self.rpc.add_direct(Address::AgentPrivileged, port, Rtt(0));427				let (tool, flags) = self.detect_escalation().await?;
428				let mut args: Vec<String> = flags.iter().map(|f| (*f).to_owned()).collect();
429				args.push(agent_path);
430				args.push("real-agent".to_owned());
431				args.push("--privileged".to_owned());
432				if local {
433					args.push("--local".to_owned());
434				}
435
436				let child = self
437					.spawn(SpawnOptions {
438						program: tool.to_owned(),
439						args,
440						stdin: StdioMode::Pipe,
441						stdout: StdioMode::Pipe,
442						stderr: StderrMode::Inherit,
443						..Default::default()
444					})
445					.await
446					.context("spawning privileged agent")?;
447
448				let stdin = child
449					.stdin
450					.ok_or_else(|| anyhow!("privileged agent stdin missing"))?;
451				let stdout = child
452					.stdout
453					.ok_or_else(|| anyhow!("privileged agent stdout missing"))?;
454
455				let port = child_port(stdout, stdin);
456				self.0
457					.rpc
458					.add_direct(Address::AgentPrivileged, port, Rtt(0));
499				anyhow::Ok(())459				anyhow::Ok(())
500			})460			})
501			.await?;461			.await?;
502		Ok(())462		Ok(())
503	}463	}
504464
505	pub async fn exec(&self, command: String) -> Result<RemowtChild> {465	async fn detect_escalation(&self) -> Result<(&'static str, &'static [&'static str])> {
506		let Some(sess) = self.ssh() else {466		for (tool, flags) in ESCALATORS {
467			let probe = self
468				.spawn(SpawnOptions {
507			bail!("exec should not be called on local")469					program: (*tool).to_owned(),
470					args: vec!["--version".to_owned()],
471					stdout: StdioMode::Null,
472					stderr: StderrMode::Null,
473					..Default::default()
508		};474				})
475				.await;
509		let ch = sess.channel_open_session().await?;476			if let Ok(child) = probe {
477				let _ = child.wait().await;
510		ch.exec(true, command).await?;478				return Ok((tool, flags));
511		Ok(RemowtChild::from_exec(ch))479			}
480		}
481		bail!("no escalation tool found")
512	}482	}
513483
514	fn runtime_dir(&self) -> Utf8PathBuf {484	/// XDG_RUNTIME_DIR on the remote machine.
485	pub fn runtime_dir(&self) -> Utf8PathBuf {
515		match &self.transport {486		match &self.0.transport {
516			Transport::Ssh { runtime_dir, .. } => runtime_dir.clone(),487			Transport::Ssh { runtime_dir, .. } => runtime_dir.clone(),
517			Transport::Local { runtime_dir, .. } => runtime_dir.clone(),488			Transport::Local { runtime_dir, .. } => runtime_dir.clone(),
518		}489		}
519	}490	}
520491
521	pub async fn forward_socket(&self, path: &Utf8Path) -> Result<RemowtListener> {492	/// Bind unix listener socket on the remote machine with auto-generated path, returning the path.
493	pub async fn bind_runtime_unix(&self, hint: &str) -> Result<(RemowtListener, Utf8PathBuf)> {
494		let sock = self
495			.runtime_dir()
496			.join(format!("remowt-{hint}-{}.sock", Uuid::new_v4()));
497		let listener = self.bind_unix(&sock).await?;
498		Ok((listener, sock))
499	}
500
501	/// Bind unix listener socket on the remote machine on the specified path.
502	pub async fn bind_unix(&self, path: &Utf8Path) -> Result<RemowtListener> {
522		match &self.transport {503		match &self.0.transport {
523			Transport::Ssh { sess, subs, .. } => {504			Transport::Ssh { sess, subs, .. } => {
524				let (tx, rx) = oneshot::channel();505				let (tx, rx) = oneshot::channel();
525				subs.lock().expect("lock").insert(path.to_owned(), tx);506				subs.lock().expect("lock").insert(path.to_owned(), tx);
537	}518	}
538}519}
539520
521fn drain_stderr(stream: DuplexStream, context: String) {
522	tokio::spawn(async move {
523		let mut reader = BufReader::new(stream).lines();
524		loop {
525			match reader.next_line().await {
526				Ok(Some(line)) => warn!(context = %context, "{line}"),
527				Ok(None) => break,
528				Err(e) => {
529					warn!(context = %context, "stderr read failed: {e}");
530					break;
531				}
532			}
533		}
534	});
535}
536
540fn local_runtime_dir() -> Result<(Utf8PathBuf, Option<TempDir>)> {537fn local_runtime_dir() -> Result<(Utf8PathBuf, Option<TempDir>)> {
541	if let Ok(dir) = env::var("XDG_RUNTIME_DIR") {538	if let Ok(dir) = env::var("XDG_RUNTIME_DIR") {
542		if !dir.is_empty() {539		if !dir.is_empty() {
556	let dir = run_string_ok(sess, "echo \"$XDG_RUNTIME_DIR\"").await?;553	let dir = run_string_ok(sess, "echo \"$XDG_RUNTIME_DIR\"").await?;
557	let dir = dir.trim();554	let dir = dir.trim();
558	if dir.is_empty() {555	if dir.is_empty() {
559		remote_mktemp(sess).await556		let tmp = run_string_ok(sess, "mktemp -d remowt.XXXXXXXXXXXX --tmpdir").await?;
557		Ok(Utf8PathBuf::from(tmp))
560	} else {558	} else {
561		Ok(Utf8PathBuf::from(dir))559		Ok(Utf8PathBuf::from(dir))
562	}
563}
564
565async fn remote_mktemp(sess: &Handle<SshHandler>) -> Result<Utf8PathBuf> {
566	let mut cmd_chan = sess.channel_open_session().await?;
567	cmd_chan
568		.exec(true, "mktemp -d remowt.XXXXXXXXXXXX --tmpdir")
569		.await?;
570	let mut stdout = vec![];
571	loop {
572		let Some(msg) = cmd_chan.wait().await else {
573			bail!("unexpected channel end");
574		};
575		match msg {
576			russh::ChannelMsg::Data { data } => stdout.extend(data.as_ref()),
577			russh::ChannelMsg::ExitStatus { exit_status } => {
578				if exit_status != 0 {
579					bail!("mktemp failed");
580				}
581				break;
582			}
583			_ => {}
584		}
585	}560	}
586	ensure!(stdout.ends_with(b"\n"));
587	stdout.pop();
588	Ok(Utf8PathBuf::from(String::from_utf8(stdout)?))
589}561}
590562

--- a/crates/remowt-client/src/shell.rs
+++ b/crates/remowt-client/src/shell.rs
@@ -3,7 +3,6 @@
 use bifrostlink::Remote;
 use remowt_endpoints::pty::{PtyClient, ShellId};
 use remowt_link_shared::{Address, BifConfig};
-use uuid::Uuid;
 
 use crate::forwarded::RemowtStream;
 use crate::Remowt;
@@ -38,15 +37,21 @@
 }
 
 impl Remowt {
-	pub async fn open_shell(&self, term: &str, cols: u16, rows: u16) -> Result<RemowtShell> {
-		let sock = self
-			.runtime_dir()
-			.join(format!("remowt-shell-{}.sock", Uuid::new_v4()));
-
-		let forwarded = self.forward_socket(&sock).await?;
-		let client: PtyClient<BifConfig> = self.endpoints();
+	pub async fn open_shell(
+		&self,
+		term: &str,
+		cols: u16,
+		rows: u16,
+		escalate: bool,
+	) -> Result<RemowtShell> {
+		let (forwarded, path) = self.bind_runtime_unix("shell").await?;
+		let client: PtyClient<BifConfig> = if escalate {
+			self.run0_endpoints().await?
+		} else {
+			self.endpoints()
+		};
 		let id = client
-			.open_shell(sock, term.to_owned(), cols, rows)
+			.open_shell(path, term.to_owned(), cols, rows)
 			.await?
 			.map_err(|e| anyhow!("agent failed to open shell: {e}"))?;
 		let stream = forwarded.accept().await?;
@@ -54,7 +59,7 @@
 		Ok(RemowtShell {
 			id,
 			stream,
-			remote: self.rpc.remote(Address::Agent),
+			remote: self.0.rpc.remote(Address::Agent),
 		})
 	}
 }

--- a/crates/remowt-client/src/subprocess.rs
+++ b/crates/remowt-client/src/subprocess.rs
@@ -1,84 +1,430 @@
-use bytes::Bytes;
-use russh::client::Msg;
-use russh::{Channel, ChannelMsg};
-use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _, DuplexStream};
-use tokio::sync::oneshot;
+use std::pin::pin;
+
+use anyhow::{anyhow, bail, Result};
+use camino::Utf8PathBuf;
+use futures::StreamExt as _;
+use remowt_endpoints::subprocess::{ProcId, SpawnSpec, StderrSpec, StdioSpec, SubprocessClient};
+use remowt_link_shared::BifConfig;
+use serde::de::DeserializeOwned;
+use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader};
+use tokio::select;
+use tokio_util::codec::{BytesCodec, FramedRead, LinesCodec};
+use tracing::{debug, info, warn};
+
+use crate::forwarded::{RemowtListener, RemowtStream};
+use crate::Remowt;
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub enum StdioMode {
+	#[default]
+	Null,
+	Pipe,
+	Inherit,
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub enum StderrMode {
+	#[default]
+	Null,
+	Pipe,
+	Inherit,
+	MergeWithStdout,
+}
 
-const BUF: usize = 64 * 1024;
+#[derive(Default)]
+pub struct SpawnOptions {
+	pub program: String,
+	pub args: Vec<String>,
+	pub env: Vec<(String, String)>,
+	pub env_clear: bool,
+	pub cwd: Option<Utf8PathBuf>,
+	pub escalated: bool,
+	pub stdin: StdioMode,
+	pub stdout: StdioMode,
+	pub stderr: StderrMode,
+}
 
 pub struct RemowtChild {
-	pub stdin: DuplexStream,
-	pub stdout: DuplexStream,
-	pub stderr: DuplexStream,
-	pub exit: oneshot::Receiver<Option<u32>>,
+	pub stdin: Option<RemowtStream>,
+	pub stdout: Option<RemowtStream>,
+	pub stderr: Option<RemowtStream>,
+	id: ProcId,
+	client: SubprocessClient<BifConfig>,
 }
 
 impl RemowtChild {
-	/// Manage channel returned by russh exec().
-	pub(crate) fn from_exec(ch: Channel<Msg>) -> Self {
-		let (stdin, mut stdin_r) = tokio::io::duplex(BUF);
-		let (mut out_w, stdout) = tokio::io::duplex(BUF);
-		let (mut err_w, stderr) = tokio::io::duplex(BUF);
-		let (exit_tx, exit) = oneshot::channel();
+	pub async fn wait(self) -> Result<Option<i32>> {
+		let RemowtChild {
+			stdin,
+			stdout,
+			stderr,
+			id,
+			client,
+		} = self;
+		drop(stdin);
+		drop(stdout);
+		drop(stderr);
+		client
+			.wait(id)
+			.await?
+			.map_err(|e| anyhow!("agent wait failed: {e}"))
+	}
+
+	pub async fn kill(&self, signal: i32) -> Result<()> {
+		self.client
+			.kill(self.id, signal)
+			.await?
+			.map_err(|e| anyhow!("agent kill failed: {e}"))
+	}
+}
+
+fn needs_socket(m: StdioMode) -> bool {
+	matches!(m, StdioMode::Pipe | StdioMode::Inherit)
+}
+
+fn stderr_needs_socket(m: StderrMode) -> bool {
+	matches!(m, StderrMode::Pipe | StderrMode::Inherit)
+}
+
+impl Remowt {
+	pub async fn spawn(&self, opts: SpawnOptions) -> Result<RemowtChild> {
+		let SpawnOptions {
+			program,
+			args,
+			env,
+			env_clear,
+			cwd,
+			escalated,
+			stdin,
+			stdout,
+			stderr,
+		} = opts;
+
+		if matches!(stderr, StderrMode::MergeWithStdout) && !needs_socket(stdout) {
+			bail!("stderr=MergeWithStdout requires stdout=Pipe or Inherit");
+		}
+
+		let stdin_bound = if needs_socket(stdin) {
+			Some(self.bind_runtime_unix("proc-stdin").await?)
+		} else {
+			None
+		};
+		let stdout_bound = if needs_socket(stdout) {
+			Some(self.bind_runtime_unix("proc-stdout").await?)
+		} else {
+			None
+		};
+		let stderr_bound = if stderr_needs_socket(stderr) {
+			Some(self.bind_runtime_unix("proc-stderr").await?)
+		} else {
+			None
+		};
+
+		let stdin_spec = match &stdin_bound {
+			Some((_, p)) => StdioSpec::Socket(p.clone()),
+			None => StdioSpec::Null,
+		};
+		let stdout_spec = match &stdout_bound {
+			Some((_, p)) => StdioSpec::Socket(p.clone()),
+			None => StdioSpec::Null,
+		};
+		let stderr_spec = match (&stderr, &stderr_bound) {
+			(StderrMode::Pipe | StderrMode::Inherit, Some((_, p))) => StderrSpec::Socket(p.clone()),
+			(StderrMode::MergeWithStdout, _) => StderrSpec::MergeWithStdout,
+			_ => StderrSpec::Null,
+		};
 
-		tokio::spawn(async move {
-			let (mut read, write) = ch.split();
+		let client: SubprocessClient<BifConfig> = if escalated {
+			// Boxed to break the async-fn type cycle
+			Box::pin(self.run0_endpoints::<SubprocessClient<BifConfig>>()).await?
+		} else {
+			self.endpoints()
+		};
 
-			// Forward our stdin to the channel, signalling EOF when it closes.
-			let stdin_pump = tokio::spawn(async move {
-				let mut buf = vec![0u8; BUF];
-				loop {
-					match stdin_r.read(&mut buf).await {
-						Ok(0) | Err(_) => break,
-						Ok(n) => {
-							if write
-								.data_bytes(Bytes::copy_from_slice(&buf[..n]))
-								.await
-								.is_err()
-							{
-								return;
-							}
-						}
+		let spec = SpawnSpec {
+			program: program.clone(),
+			args,
+			env,
+			env_clear,
+			cwd,
+			stdin: stdin_spec,
+			stdout: stdout_spec,
+			stderr: stderr_spec,
+		};
+		let id = client
+			.spawn(spec)
+			.await?
+			.map_err(|e| anyhow!("agent spawn failed: {e}"))?;
+
+		let (stdin_res, stdout_res, stderr_res) = tokio::join!(
+			accept(stdin_bound),
+			accept(stdout_bound),
+			accept(stderr_bound),
+		);
+
+		let stdin_stream = handle_stdin(stdin, stdin_res?, &program);
+		let stdout_stream = handle_output(stdout, stdout_res?, &program, false);
+		let stderr_stream = handle_output_err(stderr, stderr_res?, &program);
+
+		Ok(RemowtChild {
+			stdin: stdin_stream,
+			stdout: stdout_stream,
+			stderr: stderr_stream,
+			id,
+			client,
+		})
+	}
+
+	pub fn cmd(&self, program: impl AsRef<str>) -> RemowtCommand {
+		let program = program.as_ref().to_owned();
+		RemowtCommand {
+			program,
+			args: vec![],
+			env: vec![],
+			remowt: self.clone(),
+			escalated: false,
+		}
+	}
+}
+
+async fn accept(b: Option<(RemowtListener, Utf8PathBuf)>) -> Result<Option<RemowtStream>> {
+	match b {
+		Some((l, _)) => Ok(Some(l.accept().await?)),
+		None => Ok(None),
+	}
+}
+
+fn handle_stdin(mode: StdioMode, s: Option<RemowtStream>, program: &str) -> Option<RemowtStream> {
+	match mode {
+		StdioMode::Pipe => s,
+		StdioMode::Inherit => {
+			if let Some(s) = s {
+				let program = program.to_owned();
+				tokio::spawn(async move {
+					let mut stdin = tokio::io::stdin();
+					let mut s = s;
+					if let Err(e) = tokio::io::copy(&mut stdin, &mut s).await {
+						warn!(program, "stdin forward ended: {e}");
 					}
-				}
-				let _ = write.eof().await;
-			});
+					let _ = s.shutdown().await;
+				});
+			}
+			None
+		}
+		StdioMode::Null => None,
+	}
+}
+
+fn handle_output(
+	mode: StdioMode,
+	s: Option<RemowtStream>,
+	program: &str,
+	is_stderr: bool,
+) -> Option<RemowtStream> {
+	match mode {
+		StdioMode::Pipe => s,
+		StdioMode::Inherit => {
+			if let Some(s) = s {
+				let program = program.to_owned();
+				tokio::spawn(pump_to_tracing(s, program, is_stderr));
+			}
+			None
+		}
+		StdioMode::Null => None,
+	}
+}
+
+fn handle_output_err(
+	mode: StderrMode,
+	s: Option<RemowtStream>,
+	program: &str,
+) -> Option<RemowtStream> {
+	match mode {
+		StderrMode::Pipe => s,
+		StderrMode::Inherit => {
+			if let Some(s) = s {
+				let program = program.to_owned();
+				tokio::spawn(pump_to_tracing(s, program, true));
+			}
+			None
+		}
+		StderrMode::MergeWithStdout | StderrMode::Null => None,
+	}
+}
 
-			let mut code = None;
-			while let Some(msg) = read.wait().await {
-				match msg {
-					ChannelMsg::Data { data } => {
-						if out_w.write_all(&data).await.is_err() {
-							break;
-						}
-					}
-					ChannelMsg::ExtendedData { data, .. } => {
-						if err_w.write_all(&data).await.is_err() {
-							break;
-						}
-					}
-					ChannelMsg::ExitStatus { exit_status } => code = Some(exit_status),
-					_ => {}
+async fn pump_to_tracing(stream: RemowtStream, program: String, is_stderr: bool) {
+	let mut reader = BufReader::new(stream).lines();
+	loop {
+		match reader.next_line().await {
+			Ok(Some(line)) => {
+				if is_stderr {
+					warn!(program, "{line}");
+				} else {
+					info!(program, "{line}");
 				}
 			}
+			Ok(None) => break,
+			Err(e) => {
+				warn!(program, "child log stream error: {e}");
+				break;
+			}
+		}
+	}
+}
+
+fn escape_bash(input: &str, out: &mut String) {
+	const TO_ESCAPE: &str = "$ !\"#&'()*,;<>?[\\]^`{|}";
+	if input.chars().all(|c| !TO_ESCAPE.contains(c)) {
+		out.push_str(input);
+		return;
+	}
+	out.push('\'');
+	for (i, v) in input.split('\'').enumerate() {
+		if i != 0 {
+			out.push_str("'\"'\"'");
+		}
+		out.push_str(v);
+	}
+	out.push('\'');
+}
+
+#[derive(Clone)]
+pub struct RemowtCommand {
+	program: String,
+	args: Vec<String>,
+	env: Vec<(String, String)>,
+	remowt: Remowt,
+	escalated: bool,
+}
+
+impl RemowtCommand {
+	pub fn arg(&mut self, arg: impl AsRef<str>) -> &mut Self {
+		self.args.push(arg.as_ref().to_owned());
+		self
+	}
+	pub fn args<V: AsRef<str>>(&mut self, args: impl IntoIterator<Item = V>) -> &mut Self {
+		for arg in args {
+			self.args.push(arg.as_ref().to_owned());
+		}
+		self
+	}
+	pub fn eqarg(&mut self, key: impl AsRef<str>, value: impl AsRef<str>) -> &mut Self {
+		self.args
+			.push(format!("{}={}", key.as_ref(), value.as_ref()));
+		self
+	}
+	pub fn comparg(&mut self, key: impl AsRef<str>, value: impl AsRef<str>) -> &mut Self {
+		self.args.push(key.as_ref().to_owned());
+		self.args.push(value.as_ref().to_owned());
+		self
+	}
+	pub fn env(&mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> &mut Self {
+		self.env
+			.push((name.as_ref().to_owned(), value.as_ref().to_owned()));
+		self
+	}
 
-			// The process is gone; stop waiting on stdin we'll never forward.
-			stdin_pump.abort();
-			let _ = out_w.shutdown().await;
-			let _ = err_w.shutdown().await;
-			let _ = exit_tx.send(code);
-		});
+	pub fn sudo(mut self) -> Self {
+		self.escalated = true;
+		self
+	}
 
-		RemowtChild {
-			stdin,
-			stdout,
-			stderr,
-			exit,
+	/// Only for display.
+	fn shell_line(&self) -> String {
+		let mut out = String::new();
+		if self.escalated {
+			out.push_str("run0 ");
+		}
+		if !self.env.is_empty() {
+			out.push_str("env");
+			for (k, v) in &self.env {
+				out.push(' ');
+				assert!(!k.contains('='));
+				escape_bash(k, &mut out);
+				out.push('=');
+				escape_bash(v, &mut out);
+			}
+			out.push(' ');
 		}
+		escape_bash(&self.program, &mut out);
+		for arg in &self.args {
+			out.push(' ');
+			escape_bash(arg, &mut out);
+		}
+		out
 	}
 
-	/// Wait for the process to finish, returning its exit status.
-	pub async fn wait(self) -> Option<u32> {
-		self.exit.await.ok().flatten()
+	fn into_spawn_options(self) -> (Remowt, SpawnOptions, String) {
+		let line = self.shell_line();
+		let opts = SpawnOptions {
+			program: self.program,
+			args: self.args,
+			env: self.env,
+			env_clear: false,
+			cwd: None,
+			escalated: self.escalated,
+			stdin: StdioMode::Null,
+			stdout: StdioMode::Pipe,
+			stderr: StderrMode::Pipe,
+		};
+		(self.remowt, opts, line)
+	}
+
+	pub async fn run(self) -> Result<()> {
+		run_inner(self, false).await.map(|_| ())
+	}
+	pub async fn run_string(self) -> Result<String> {
+		let bytes = run_inner(self, true).await?.expect("want_stdout");
+		Ok(String::from_utf8(bytes)?)
+	}
+	pub async fn run_value<T: DeserializeOwned>(self) -> Result<T> {
+		let s = self.run_string().await?;
+		Ok(serde_json::from_str(&s)?)
+	}
+}
+
+async fn run_inner(cmd: RemowtCommand, want_stdout: bool) -> Result<Option<Vec<u8>>> {
+	let (remowt, opts, line) = cmd.into_spawn_options();
+	debug!("running command {line:?} over remowt");
+	let program = opts.program.clone();
+	let mut child = remowt.spawn(opts).await?;
+	let stderr = child.stderr.take().expect("stderr=Pipe");
+	let stdout = child.stdout.take().expect("stdout=Pipe");
+
+	let mut err = FramedRead::new(stderr, LinesCodec::new());
+	let (mut out_bytes, mut out_lines) = if want_stdout {
+		(Some(FramedRead::new(stdout, BytesCodec::new())), None)
+	} else {
+		(None, Some(FramedRead::new(stdout, LinesCodec::new())))
+	};
+
+	let mut buf = if want_stdout { Some(Vec::new()) } else { None };
+
+	let mut wait = pin!(child.wait());
+	let exit = loop {
+		select! {
+			biased;
+
+			Some(e) = err.next() => {
+				let e = e?;
+				warn!(program = %program, "{e}");
+			}
+			Some(o) = async { out_bytes.as_mut()?.next().await }, if want_stdout => {
+				buf.as_mut().expect("want_stdout").extend_from_slice(&o?);
+			}
+			Some(o) = async { out_lines.as_mut()?.next().await }, if !want_stdout => {
+				let o = o?;
+				info!(program = %program, "{o}");
+			}
+			res = &mut wait => {
+				break res?;
+			}
+		}
+	};
+
+	match exit {
+		Some(0) => Ok(buf),
+		Some(c) => bail!("command '{line}' failed with status {c}"),
+		None => Err(anyhow!("command '{line}' ended without an exit status")),
 	}
 }

--- a/crates/remowt-endpoints/Cargo.toml
+++ b/crates/remowt-endpoints/Cargo.toml
@@ -16,5 +16,5 @@
 tokio = { workspace = true, features = ["net", "io-util", "rt", "process"] }
 tracing.workspace = true
 uuid.workspace = true
-nix = { workspace = true, features = ["process", "term"] }
+nix = { workspace = true, features = ["process", "signal", "term"] }
 zbus.workspace = true

--- a/crates/remowt-endpoints/src/lib.rs
+++ b/crates/remowt-endpoints/src/lib.rs
@@ -1,4 +1,5 @@
 pub mod fs;
 pub mod nix_daemon;
 pub mod pty;
+pub mod subprocess;
 pub mod systemd;

--- a/crates/remowt-endpoints/src/pty.rs
+++ b/crates/remowt-endpoints/src/pty.rs
@@ -16,7 +16,7 @@
 use tokio::io::unix::AsyncFd;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::net::UnixStream;
-use tracing::{info, warn};
+use tracing::{debug, info, warn};
 
 pub type ShellId = u64;
 
@@ -117,7 +117,7 @@
 		};
 		let pty = AsyncPty::new(master)?;
 
-		info!(id, shell, "shell opened");
+		debug!(id, shell, "shell opened");
 		let shells = self.shells.clone();
 		tokio::spawn(async move {
 			let mut pty = pty;