git.delta.rocks / remowt / refs/commits / 42e2f16609cb

difftreelog

feat basic plugin loading

mwqtzvovYaroslav Bolyukin2026-06-07parent: #69f690d.patch.diff
in: trunk

11 files changed

modifiedCargo.lockdiffbeforeafterboth
after · Cargo.lock
382 packageslockfile v4
modifiedCargo.tomldiffbeforeafterboth
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -10,6 +10,7 @@
 remowt-client = { path = "crates/remowt-client" }
 polkit-shared = { version = "0.1.0", path = "crates/polkit-shared" }
 remowt-link-shared = { version = "0.1.0", path = "crates/remowt-link-shared" }
+remowt-plugin = { version = "0.1.0", path = "crates/remowt-plugin" }
 ui-prompt = { version = "0.1.0", path = "crates/ui-prompt" }
 
 bifrostlink = "0.2.0"
modifiedcmds/remowt-agent/Cargo.tomldiffbeforeafterboth
--- a/cmds/remowt-agent/Cargo.toml
+++ b/cmds/remowt-agent/Cargo.toml
@@ -14,6 +14,7 @@
 polkit-shared.workspace = true
 rand.workspace = true
 remowt-link-shared.workspace = true
+remowt-plugin.workspace = true
 remowt-pty.workspace = true
 serde = { workspace = true, features = ["derive"] }
 tempfile.workspace = true
modifiedcmds/remowt-agent/src/main.rsdiffbeforeafterboth
--- a/cmds/remowt-agent/src/main.rs
+++ b/cmds/remowt-agent/src/main.rs
@@ -253,6 +253,8 @@
 	Systemd.register_endpoints(&mut rpc);
 	Pty::new().register_endpoints(&mut rpc);
 
+	remowt_plugin::host::serve(&mut rpc);
+
 	let user_prompter = PromptEndpointsClient::wrap(rpc.remote(Address::User));
 	let editor_client = EditorEndpointsClient::wrap(rpc.remote(Address::User));
modifiedcrates/remowt-client/src/lib.rsdiffbeforeafterboth
--- a/crates/remowt-client/src/lib.rs
+++ b/crates/remowt-client/src/lib.rs
@@ -9,6 +9,7 @@
 use bifrostlink_ports::unix_socket::from_socket;
 use bytes::{Bytes, BytesMut};
 use camino::{Utf8Path, Utf8PathBuf};
+use remowt_link_shared::plugin::PluginEndpointsClient;
 use remowt_link_shared::{
 	Address, BifConfig, ElevateEndpoints, ElevateError, Elevator, Fs, Pty, PtyClient, ShellId,
 	Systemd,
@@ -508,6 +509,26 @@
 	pub fn endpoints<R: RemoteEndpoints<BifConfig>>(&self) -> R {
 		R::wrap(self.rpc.remote(Address::Agent))
 	}
+
+	pub async fn load_plugin(&self, id: u16, name: &str) -> Result<()> {
+		let client: PluginEndpointsClient<BifConfig> = self.endpoints();
+		client
+			.load_plugin(id, name.to_owned())
+			.await?
+			.map_err(|e| anyhow!("agent failed to load plugin: {e}"))
+	}
+	pub async fn run0_load_plugin_path(&self, id: u16, path: &str) -> Result<()> {
+		self.ensure_elevated().await?;
+		let client: PluginEndpointsClient<BifConfig> =
+			PluginEndpointsClient::wrap(self.rpc.remote(Address::AgentPrivileged));
+		client
+			.load_plugin_path(id, path.to_owned())
+			.await?
+			.map_err(|e| anyhow!("privileged agent failed to load plugin: {e}"))
+	}
+	pub fn plugin_endpoints<R: RemoteEndpoints<BifConfig>>(&self, id: u16) -> R {
+		R::wrap(self.rpc.remote(Address::Plugin(id)))
+	}
 	pub async fn run0_endpoints<R: RemoteEndpoints<BifConfig>>(&self) -> Result<R> {
 		self.ensure_elevated().await?;
 		Ok(R::wrap(self.rpc.remote(Address::AgentPrivileged)))
modifiedcrates/remowt-link-shared/src/lib.rsdiffbeforeafterboth
--- a/crates/remowt-link-shared/src/lib.rs
+++ b/crates/remowt-link-shared/src/lib.rs
@@ -13,9 +13,12 @@
 	User,
 	Agent,
 	AgentPrivileged,
+	Plugin(u16),
 }
 impl AddressT for Address {}
 
+pub mod plugin;
+
 pub use remowt_fs::{Error as FsError, Fs, FsClient};
 pub use remowt_pty::{Error as PtyError, Pty, PtyClient, ShellId};
 pub use remowt_systemd::{Error as SystemdError, Systemd, SystemdClient};
addedcrates/remowt-link-shared/src/plugin.rsdiffbeforeafterboth
--- /dev/null
+++ b/crates/remowt-link-shared/src/plugin.rs
@@ -0,0 +1,39 @@
+use std::future::Future;
+
+use bifrostlink::declarative::endpoints;
+use bifrostlink::Config;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, thiserror::Error)]
+pub enum Error {
+	#[error("plugin name must be a bare file name")]
+	BadName,
+	#[error("spawning plugin failed: {0}")]
+	Spawn(String),
+	#[error("agent is shutting down")]
+	Gone,
+}
+
+pub trait PluginHost: Send + Sync {
+	fn load_plugin(&self, id: u16, name: String) -> impl Future<Output = Result<(), Error>> + Send;
+
+	fn load_plugin_path(
+		&self,
+		id: u16,
+		path: String,
+	) -> impl Future<Output = Result<(), Error>> + Send;
+}
+
+pub struct PluginEndpoints<H>(pub H);
+
+#[endpoints(ns = 9)]
+impl<H: PluginHost + 'static> PluginEndpoints<H> {
+	#[endpoints(id = 1)]
+	async fn load_plugin(&self, id: u16, name: String) -> Result<(), Error> {
+		self.0.load_plugin(id, name).await
+	}
+	#[endpoints(id = 2)]
+	async fn load_plugin_path(&self, id: u16, path: String) -> Result<(), Error> {
+		self.0.load_plugin_path(id, path).await
+	}
+}
modifiedcrates/remowt-nix-daemon/Cargo.tomldiffbeforeafterboth
--- a/crates/remowt-nix-daemon/Cargo.toml
+++ b/crates/remowt-nix-daemon/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
-name = "fleet-nix-daemon"
-description = "Nix daemon proxy endpoint + connection logic for fleet"
+name = "remowt-nix-daemon"
+description = "Nix daemon proxy"
 version.workspace = true
 edition = "2021"
addedcrates/remowt-plugin/Cargo.tomldiffbeforeafterboth
--- /dev/null
+++ b/crates/remowt-plugin/Cargo.toml
@@ -0,0 +1,22 @@
+[package]
+name = "remowt-plugin"
+version.workspace = true
+edition = "2021"
+
+[dependencies]
+anyhow.workspace = true
+bifrostlink.workspace = true
+bifrostlink-ports.workspace = true
+bytes.workspace = true
+remowt-link-shared.workspace = true
+tokio = { workspace = true, features = [
+	"rt",
+	"net",
+	"io-std",
+	"io-util",
+	"macros",
+	"time",
+	"process",
+] }
+tracing.workspace = true
+tracing-subscriber.workspace = true
addedcrates/remowt-plugin/src/host.rsdiffbeforeafterboth
--- /dev/null
+++ b/crates/remowt-plugin/src/host.rs
@@ -0,0 +1,110 @@
+use std::ffi::OsStr;
+use std::io;
+use std::process::Stdio;
+use std::sync::Mutex;
+
+use bifrostlink::{Port, Rpc, Rtt, WeakRpc};
+use bytes::{Bytes, BytesMut};
+use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+use tokio::process::{Child, ChildStdin, ChildStdout, Command};
+
+use remowt_link_shared::plugin::{Error, PluginEndpoints, PluginHost};
+use remowt_link_shared::{Address, BifConfig};
+
+pub fn serve(rpc: &mut Rpc<BifConfig>) {
+	let host = Host {
+		rpc: rpc.clone().downgrade(),
+		children: Mutex::new(Vec::new()),
+	};
+	PluginEndpoints(host).register_endpoints(rpc);
+}
+
+struct Host {
+	rpc: WeakRpc<BifConfig>,
+	children: Mutex<Vec<Child>>,
+}
+
+impl Host {
+	fn spawn(&self, id: u16, path: impl AsRef<OsStr>) -> Result<(), Error> {
+		let rpc = self.rpc.clone().upgrade().ok_or(Error::Gone)?;
+
+		let mut child = Command::new(path)
+			.arg(id.to_string())
+			.stdin(Stdio::piped())
+			.stdout(Stdio::piped())
+			.kill_on_drop(true)
+			.spawn()
+			.map_err(|e| Error::Spawn(e.to_string()))?;
+		let stdin = child.stdin.take().expect("stdin piped");
+		let stdout = child.stdout.take().expect("stdout piped");
+
+		rpc.add_direct(Address::Plugin(id), child_port(stdout, stdin), Rtt(0));
+		self.children.lock().expect("not poisoned").push(child);
+		Ok(())
+	}
+}
+
+impl PluginHost for Host {
+	async fn load_plugin(&self, id: u16, name: String) -> Result<(), Error> {
+		// TODO: Right now loads plugin next to the binary...
+		// But with our CA addressed schema, the plugins should be located in content-addressed subdir...
+		// Maybe it should just be scrapped in favor of load_plugin_path.
+		if name.is_empty() || name == "." || name == ".." || name.contains(['/', '\0']) {
+			return Err(Error::BadName);
+		}
+		let exe = std::env::current_exe().map_err(|e| Error::Spawn(e.to_string()))?;
+		let dir = exe
+			.parent()
+			.ok_or_else(|| Error::Spawn("primary agent has no parent directory".to_owned()))?;
+		self.spawn(id, dir.join(&name))
+	}
+
+	async fn load_plugin_path(&self, id: u16, path: String) -> Result<(), Error> {
+		if path.is_empty() || path.contains('\0') {
+			return Err(Error::BadName);
+		}
+		self.spawn(id, path)
+	}
+}
+
+fn child_port(mut stdout: ChildStdout, mut stdin: ChildStdin) -> Port {
+	Port::new(|mut rx, tx| async move {
+		let reader = async move {
+			loop {
+				let len = match stdout.read_u32().await {
+					Ok(len) => len,
+					Err(e) => {
+						tracing::error!("plugin stdout read failed: {e}");
+						break;
+					}
+				};
+				let mut buf = BytesMut::zeroed(len as usize);
+				if let Err(e) = stdout.read_exact(&mut buf).await {
+					tracing::error!("plugin stdout read failed: {e}");
+					break;
+				}
+				if tx.send(buf.freeze()).is_err() {
+					break;
+				}
+			}
+		};
+		let writer = async move {
+			while let Some(msg) = rx.recv().await {
+				if let Err(e) = write_frame(&mut stdin, msg).await {
+					tracing::error!("plugin stdin write failed: {e}");
+					break;
+				}
+			}
+		};
+		tokio::join!(reader, writer);
+	})
+}
+
+async fn write_frame(stdin: &mut ChildStdin, msg: Bytes) -> io::Result<()> {
+	let len = u32::try_from(msg.len())
+		.map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "message larger than 4GB"))?;
+	stdin.write_u32(len).await?;
+	stdin.write_all(&msg).await?;
+	stdin.flush().await?;
+	Ok(())
+}
addedcrates/remowt-plugin/src/lib.rsdiffbeforeafterboth
--- /dev/null
+++ b/crates/remowt-plugin/src/lib.rs
@@ -0,0 +1,38 @@
+use std::future::pending;
+
+use anyhow::Result;
+use bifrostlink::{Rpc, Rtt};
+use bifrostlink_ports::stdio::from_stdio;
+use tokio::runtime::Builder;
+
+pub mod host;
+
+pub use bifrostlink;
+pub use remowt_link_shared::{self, Address, BifConfig, Fs, Pty, Systemd};
+
+pub fn plugin_index() -> Result<u16> {
+	let arg = std::env::args()
+		.nth(1)
+		.ok_or_else(|| anyhow::anyhow!("missing plugin index argument"))?;
+	arg.parse()
+		.map_err(|e| anyhow::anyhow!("invalid plugin index {arg:?}: {e}"))
+}
+
+pub fn run<F>(register: F) -> Result<()>
+where
+	F: FnOnce(&mut Rpc<BifConfig>),
+{
+	tracing_subscriber::fmt()
+		.with_writer(std::io::stderr)
+		.init();
+
+	let index = plugin_index()?;
+	let runtime = Builder::new_current_thread().enable_all().build()?;
+	runtime.block_on(async move {
+		let mut rpc = Rpc::<BifConfig>::new(Address::Plugin(index));
+		rpc.add_direct(Address::Agent, from_stdio(), Rtt(0));
+		register(&mut rpc);
+		let _rpc = rpc;
+		pending::<Result<()>>().await
+	})
+}