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

--- 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(())
+	}
 }

--- 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;
 }

--- a/cmds/remowt-agent/src/helper/suid.rs
+++ b/cmds/remowt-agent/src/helper/suid.rs
@@ -1,83 +1,46 @@
-use std::pin::pin;
 use std::process::Stdio;
 
-use anyhow::{bail, anyhow};
-use futures::stream::Peekable;
-use futures::StreamExt as _;
+use anyhow::{anyhow, bail};
 use nix::unistd::User;
 use polkit_shared::Identity;
 use tokio::io::AsyncWriteExt as _;
 use tokio::process::Command;
-use tokio::select;
-use tokio_util::codec::{FramedRead, LinesCodec};
 use ui_prompt::Prompter;
 
+use super::protocol::run_conversation;
 use super::Helper;
 
 #[derive(Clone)]
 pub struct SuidHelper;
 impl Helper for SuidHelper {
-    async fn help_me<P: Prompter + 'static>(
-        &self,
-        cookie: &str,
-        prompt: P,
-        identity: Identity,
-    ) -> anyhow::Result<()> {
-        let Some(uid) = dbg!(identity.uid()) else {
-            bail!("can't process identity");
-        };
-        let user = User::from_uid(dbg!(uid))
-            .map_err(|e| anyhow!("error querying user: {e}"))?
-            .ok_or_else(|| anyhow!("user not found"))?;
-
-        let mut cmd = Command::new("polkit-agent-helper-1");
-        cmd.arg(user.name);
-        cmd.stdin(Stdio::piped());
-        cmd.stdout(Stdio::piped());
-        cmd.kill_on_drop(true);
-        let mut child = cmd.spawn()?;
-        let mut stdin = child.stdin.take().expect("piped");
-        let mut stdout =
-            pin!(
-                FramedRead::new(child.stdout.take().expect("piped"), LinesCodec::new()).peekable()
-            );
-
-        assert!(!cookie.contains("\n"));
-        stdin.write_all(cookie.as_bytes()).await?;
-        stdin.write_all(b"\n").await?;
+	async fn help_me<P: Prompter + 'static>(
+		&self,
+		cookie: &str,
+		prompt: P,
+		identity: Identity,
+	) -> anyhow::Result<()> {
+		let Some(uid) = identity.uid() else {
+			bail!("can't process identity");
+		};
+		let user = User::from_uid(uid)
+			.map_err(|e| anyhow!("error querying user: {e}"))?
+			.ok_or_else(|| anyhow!("user not found"))?;
 
-        while let Some(line) = stdout.next().await {
-            let line = dbg!(line?);
-            // TODO: Dedicated codec?
-            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(stdout.as_mut()) => {},
-                    r = prompt.display_text(false, msg_text, &[]) => {r?}
-                }
-                String::new()
-            } else if line == "SUCCESS" {
-                return Ok(());
-            } else if line == "FAILURE" {
-                bail!("helper binary reported failure")
-            } else {
-                // TODO: Success/failure handling
-                bail!("unknown agent request");
-            };
+		let mut cmd = Command::new("polkit-agent-helper-1");
+		cmd.arg(user.name);
+		cmd.stdin(Stdio::piped());
+		cmd.stdout(Stdio::piped());
+		cmd.kill_on_drop(true);
+		let mut child = cmd.spawn()?;
+		let mut stdin = child.stdin.take().expect("piped");
+		let stdout = child.stdout.take().expect("piped");
 
-            if res.contains("\n") {
-                bail!("response should not include newline")
-            }
+		assert!(!cookie.contains('\n'));
+		stdin.write_all(cookie.as_bytes()).await?;
+		stdin.write_all(b"\n").await?;
 
-            stdin.write_all(res.as_bytes()).await?;
-            stdin.write_all(b"\n").await?;
-        }
-        bail!("agent finished unexpectedly")
-    }
+		let res = run_conversation(stdout, stdin, prompt).await;
+		drop(child);
+		res
+	}
 }