git.delta.rocks / remowt / refs/commits / 5f64fb1ffdf0

difftreelog

feat socket polkit helper

vpnuwsmwYaroslav Bolyukin2026-01-25parent: #4516008.patch.diff
in: trunk

5 files changed

modifiedcmds/remowt-agent/src/helper/dbus.rsdiffbeforeafterboth
--- a/cmds/remowt-agent/src/helper/dbus.rs
+++ b/cmds/remowt-agent/src/helper/dbus.rs
@@ -2,7 +2,6 @@
 use std::marker::PhantomData;
 
 use polkit_shared::{BackendRequest, Identity};
-use tokio::runtime::Handle;
 use ui_prompt::dbus::DbusPrompterInterface;
 use ui_prompt::Prompter;
 use zbus::Connection;
@@ -10,70 +9,73 @@
 use crate::PolkitHelperProxy;
 
 use super::Helper;
-
 
 struct TemporaryPrompterInterface<P: Prompter + 'static> {
-    connection: Connection,
-    path: String,
-    _marker: PhantomData<P>,
+	connection: Connection,
+	path: String,
+	_marker: PhantomData<P>,
 }
 impl<P: Prompter + 'static> TemporaryPrompterInterface<P> {
-    async fn new(connection: Connection, prompter: P) -> Self {
-        let path = format!(
-            "/remowt/prompters/{}",
-            uuid::Uuid::new_v4().to_string().replace("-", "_")
-        );
-        let _ = connection
-            .object_server()
-            .at(path.clone(), DbusPrompterInterface(prompter))
-            .await;
-        Self {
-            connection,
-            path,
-            _marker: PhantomData,
-        }
-    }
+	async fn new(connection: Connection, prompter: P) -> Self {
+		let path = format!(
+			"/remowt/prompters/{}",
+			uuid::Uuid::new_v4().to_string().replace("-", "_")
+		);
+		let _ = connection
+			.object_server()
+			.at(path.clone(), DbusPrompterInterface(prompter))
+			.await;
+		Self {
+			connection,
+			path,
+			_marker: PhantomData,
+		}
+	}
 }
 impl<P: Prompter + Send + Sync + 'static> Drop for TemporaryPrompterInterface<P> {
-    fn drop(&mut self) {
-        // FIXME: block_in_place prevents to moving to current_thread runtime
-        // There should be a blocking way to remove ObjectServer listener.
-        // As far as I can see, it is only async because of async RwLock, shouldn't it be
-        // just a sync lock?
-        tokio::task::block_in_place(move || {
-            Handle::current().block_on(async {
-                let _ = self
-                    .connection
-                    .object_server()
-                    .remove::<DbusPrompterInterface<P>, String>(self.path.clone())
-                    .await;
-            });
-        });
-    }
+	fn drop(&mut self) {
+		// Removal is async because of async RwLock used inside...
+		// We should not care about its reuse
+		let connection = self.connection.clone();
+		let path = std::mem::take(&mut self.path);
+		tokio::spawn(async move {
+			let _ = connection
+				.object_server()
+				.remove::<DbusPrompterInterface<P>, String>(path)
+				.await;
+		});
+	}
 }
 
+#[derive(Clone)]
 pub struct DbusHelper {
-    connection: Connection,
-    helper: PolkitHelperProxy<'static>,
+	connection: Connection,
+	helper: PolkitHelperProxy<'static>,
 }
+impl DbusHelper {
+	pub async fn new(connection: Connection) -> zbus::Result<Self> {
+		let helper = PolkitHelperProxy::new(&connection).await?;
+		Ok(Self { connection, helper })
+	}
+}
 impl Helper for DbusHelper {
-    async fn help_me<P: Prompter + Send + Sync + 'static>(
-        &self,
-        cookie: &str,
-        prompter: P,
-        identity: Identity,
-    ) -> anyhow::Result<()> {
-        let prompter = TemporaryPrompterInterface::new(self.connection.clone(), prompter).await;
-        self.helper
-            .init_conversation(
-                BackendRequest {
-                    cookie: cookie.to_owned(),
-                    environment: HashMap::new(),
-                    prompter_path: prompter.path.clone(),
-                    identity,
-                }, // cookie.to_owned(), HashMap::new(), prompter.path.clone()
-            )
-            .await?;
-        Ok(())
-    }
+	async fn help_me<P: Prompter + Send + Sync + 'static>(
+		&self,
+		cookie: &str,
+		prompter: P,
+		identity: Identity,
+	) -> anyhow::Result<()> {
+		let prompter = TemporaryPrompterInterface::new(self.connection.clone(), prompter).await;
+		self.helper
+			.init_conversation(
+				BackendRequest {
+					cookie: cookie.to_owned(),
+					environment: HashMap::new(),
+					prompter_path: prompter.path.clone(),
+					identity,
+				}, // cookie.to_owned(), HashMap::new(), prompter.path.clone()
+			)
+			.await?;
+		Ok(())
+	}
 }
modifiedcmds/remowt-agent/src/helper/mod.rsdiffbeforeafterboth
--- a/cmds/remowt-agent/src/helper/mod.rs
+++ b/cmds/remowt-agent/src/helper/mod.rs
@@ -2,17 +2,20 @@
 use polkit_shared::Identity;
 use ui_prompt::Prompter;
 
-mod suid;
 mod dbus;
+mod protocol;
+mod socket;
+mod suid;
 
+pub use dbus::DbusHelper;
+pub use socket::SocketHelper;
 pub use suid::SuidHelper;
-pub use dbus::DbusHelper;
 
 pub trait Helper {
-    fn help_me<P: Prompter + Send + Sync + 'static>(
-        &self,
-        cookie: &str,
-        prompt: P,
-        identity: Identity,
-    ) -> impl Future<Output = anyhow::Result<()>> + Send;
+	fn help_me<P: Prompter + Send + Sync + 'static>(
+		&self,
+		cookie: &str,
+		prompt: P,
+		identity: Identity,
+	) -> impl Future<Output = anyhow::Result<()>> + Send;
 }
addedcmds/remowt-agent/src/helper/protocol.rsdiffbeforeafterboth
--- /dev/null
+++ b/cmds/remowt-agent/src/helper/protocol.rs
@@ -0,0 +1,50 @@
+use std::pin::pin;
+
+use anyhow::bail;
+use futures::stream::Peekable;
+use futures::StreamExt as _;
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt as _};
+use tokio::select;
+use tokio_util::codec::{FramedRead, LinesCodec};
+use ui_prompt::Prompter;
+
+pub async fn run_conversation<R, W, P>(reader: R, mut writer: W, prompt: P) -> anyhow::Result<()>
+where
+	R: AsyncRead,
+	W: AsyncWrite + Unpin,
+	P: Prompter,
+{
+	let mut lines = pin!(FramedRead::new(reader, LinesCodec::new()).peekable());
+
+	while let Some(line) = lines.next().await {
+		let line = line?;
+		let res = if let Some(prompt_text) = line.strip_prefix("PAM_PROMPT_ECHO_OFF ") {
+			prompt.prompt_text(false, prompt_text, "", &[]).await?
+		} else if let Some(prompt_text) = line.strip_prefix("PAM_PROMPT_ECHO_ON ") {
+			prompt.prompt_text(true, prompt_text, "", &[]).await?
+		} else if let Some(msg_text) = line.strip_prefix("PAM_ERROR_MSG ") {
+			prompt.display_text(true, msg_text, &[]).await?;
+			String::new()
+		} else if let Some(msg_text) = line.strip_prefix("PAM_TEXT_INFO ") {
+			select! {
+				_ = Peekable::peek(lines.as_mut()) => {},
+				r = prompt.display_text(false, msg_text, &[]) => {r?}
+			}
+			String::new()
+		} else if line == "SUCCESS" {
+			return Ok(());
+		} else if line == "FAILURE" {
+			bail!("helper reported failure")
+		} else {
+			bail!("unknown agent request: {line}")
+		};
+
+		if res.contains('\n') {
+			bail!("response should not include newline")
+		}
+
+		writer.write_all(res.as_bytes()).await?;
+		writer.write_all(b"\n").await?;
+	}
+	bail!("agent finished unexpectedly")
+}
addedcmds/remowt-agent/src/helper/socket.rsdiffbeforeafterboth
--- /dev/null
+++ b/cmds/remowt-agent/src/helper/socket.rs
@@ -0,0 +1,53 @@
+use anyhow::{anyhow, bail};
+use nix::unistd::User;
+use polkit_shared::Identity;
+use tokio::io::AsyncWriteExt as _;
+use tokio::net::UnixStream;
+use tracing::debug;
+use ui_prompt::Prompter;
+
+use super::protocol::run_conversation;
+use super::Helper;
+
+/// Polkit 127 introduced an alternative backend similar to `lach.PolkitHelper`
+const SOCKET_PATH: &str = "/run/polkit/agent-helper.socket";
+
+#[derive(Clone)]
+pub struct SocketHelper<F> {
+	pub fallback: F,
+}
+
+impl<F: Helper + Sync> Helper for SocketHelper<F> {
+	async fn help_me<P: Prompter + Send + Sync + 'static>(
+		&self,
+		cookie: &str,
+		prompt: P,
+		identity: Identity,
+	) -> anyhow::Result<()> {
+		let Some(uid) = identity.uid() else {
+			bail!("can't process identity");
+		};
+
+		let stream = match UnixStream::connect(SOCKET_PATH).await {
+			Ok(stream) => stream,
+			Err(e) => {
+				debug!("agent-helper.socket unavailable ({e}), using fallback helper");
+				return self.fallback.help_me(cookie, prompt, identity).await;
+			}
+		};
+
+		let user = User::from_uid(uid)
+			.map_err(|e| anyhow!("error querying user: {e}"))?
+			.ok_or_else(|| anyhow!("user not found"))?;
+
+		assert!(!cookie.contains('\n'));
+		let (reader, mut writer) = stream.into_split();
+
+		writer.write_all(user.name.as_bytes()).await?;
+		writer.write_all(b"\n").await?;
+		writer.write_all(cookie.as_bytes()).await?;
+		writer.write_all(b"\n").await?;
+
+		run_conversation(reader, writer, prompt).await
+	}
+}
modifiedcmds/remowt-agent/src/helper/suid.rsdiffbeforeafterboth
1use std::pin::pin;
2use std::process::Stdio;1use std::process::Stdio;
32
4use anyhow::{bail, anyhow};3use anyhow::{anyhow, bail};
5use futures::stream::Peekable;
6use futures::StreamExt as _;
7use nix::unistd::User;4use nix::unistd::User;
8use polkit_shared::Identity;5use polkit_shared::Identity;
9use tokio::io::AsyncWriteExt as _;6use tokio::io::AsyncWriteExt as _;
10use tokio::process::Command;7use tokio::process::Command;
11use tokio::select;8use ui_prompt::Prompter;
9
12use tokio_util::codec::{FramedRead, LinesCodec};10use super::protocol::run_conversation;
13use ui_prompt::Prompter;
14
15use super::Helper;11use super::Helper;
1612
23 prompt: P,19 prompt: P,
24 identity: Identity,20 identity: Identity,
25 ) -> anyhow::Result<()> {21 ) -> anyhow::Result<()> {
26 let Some(uid) = dbg!(identity.uid()) else {22 let Some(uid) = identity.uid() else {
27 bail!("can't process identity");23 bail!("can't process identity");
28 };24 };
29 let user = User::from_uid(dbg!(uid))25 let user = User::from_uid(uid)
30 .map_err(|e| anyhow!("error querying user: {e}"))?26 .map_err(|e| anyhow!("error querying user: {e}"))?
31 .ok_or_else(|| anyhow!("user not found"))?;27 .ok_or_else(|| anyhow!("user not found"))?;
3228
37 cmd.kill_on_drop(true);33 cmd.kill_on_drop(true);
38 let mut child = cmd.spawn()?;34 let mut child = cmd.spawn()?;
39 let mut stdin = child.stdin.take().expect("piped");35 let mut stdin = child.stdin.take().expect("piped");
40 let mut stdout =36 let stdout = child.stdout.take().expect("piped");
41 pin!(
42 FramedRead::new(child.stdout.take().expect("piped"), LinesCodec::new()).peekable()
43 );
4437
45 assert!(!cookie.contains("\n"));38 assert!(!cookie.contains('\n'));
46 stdin.write_all(cookie.as_bytes()).await?;39 stdin.write_all(cookie.as_bytes()).await?;
47 stdin.write_all(b"\n").await?;40 stdin.write_all(b"\n").await?;
4841
49 while let Some(line) = stdout.next().await {42 let res = run_conversation(stdout, stdin, prompt).await;
50 let line = dbg!(line?);
51 // TODO: Dedicated codec?
52 let res = if let Some(prompt_text) = line.strip_prefix("PAM_PROMPT_ECHO_OFF ") {
53 prompt.prompt_text(false, prompt_text, "", &[]).await?
54 } else if let Some(prompt_text) = line.strip_prefix("PAM_PROMPT_ECHO_ON ") {
55 prompt.prompt_text(true, prompt_text, "", &[]).await?
56 } else if let Some(msg_text) = line.strip_prefix("PAM_ERROR_MSG ") {
57 prompt.display_text(true, msg_text, &[]).await?;
58 String::new()
59 } else if let Some(msg_text) = line.strip_prefix("PAM_TEXT_INFO ") {
60 select! {
61 _ = Peekable::peek(stdout.as_mut()) => {},
62 r = prompt.display_text(false, msg_text, &[]) => {r?}
63 }
64 String::new()
65 } else if line == "SUCCESS" {
66 return Ok(());
67 } else if line == "FAILURE" {
68 bail!("helper binary reported failure")
69 } else {
70 // TODO: Success/failure handling
71 bail!("unknown agent request");
72 };
73
74 if res.contains("\n") {
75 bail!("response should not include newline")
76 }
77
78 stdin.write_all(res.as_bytes()).await?;
79 stdin.write_all(b"\n").await?;
80 }
81 bail!("agent finished unexpectedly")43 drop(child);
44 res
82 }45 }
83}46}
8447