Implement Linux proxy controller, subscription state, and CLI/TUI

1 files changed, 357 insertions, 0 deletions

diff --git a/src/bus.rs b/src/bus.rs
new file mode 100644
index 0000000..fd26f1d
--- /dev/null
+++ b/src/bus.rs
@@ -0,0 +1,357 @@
+use crate::controller::{Caller, Controller, Operation};
+use anyhow::{Result, anyhow};
+use std::{collections::HashMap, sync::Arc};
+use tokio::sync::Mutex;
+use zbus::{Connection, message::Header, zvariant::Value};
+
+pub const NAME: &str = "org.roentgen.Controller1";
+pub const PATH: &str = "/org/roentgen/Controller1";
+pub const INTERFACE: &str = "org.roentgen.Controller1";
+pub type SharedController = Arc<Mutex<Controller>>;
+
+pub struct Service {
+    pub controller: SharedController,
+    pub cancellation: Arc<tokio::sync::Notify>,
+}
+
+async fn authorize(
+    connection: &Connection,
+    header: &Header<'_>,
+    interactive: bool,
+) -> zbus::fdo::Result<Caller> {
+    let sender = header
+        .sender()
+        .ok_or_else(|| zbus::fdo::Error::AccessDenied("Missing bus sender".into()))?;
+    let bus = zbus::fdo::DBusProxy::new(connection).await?;
+    let uid = bus.get_connection_unix_user(sender.clone().into()).await?;
+    if uid == 0 {
+        return Ok(Caller::authorized());
+    }
+    let authority = zbus::Proxy::new(
+        connection,
+        "org.freedesktop.PolicyKit1",
+        "/org/freedesktop/PolicyKit1/Authority",
+        "org.freedesktop.PolicyKit1.Authority",
+    )
+    .await
+    .map_err(|_| zbus::fdo::Error::AccessDenied("Polkit is unavailable".into()))?;
+    let mut subject = HashMap::new();
+    subject.insert("name", Value::from(sender.as_str()));
+    let result: (bool, bool, HashMap<String, String>) = authority
+        .call(
+            "CheckAuthorization",
+            &(
+                ("system-bus-name", subject),
+                "org.roentgen.control",
+                HashMap::<String, String>::new(),
+                u32::from(interactive),
+                "",
+            ),
+        )
+        .await
+        .map_err(|_| zbus::fdo::Error::AccessDenied("Authorization denied or cancelled".into()))?;
+    if !result.0 {
+        return Err(zbus::fdo::Error::AccessDenied(
+            "Administrator authorization required; no state changed".into(),
+        ));
+    }
+    Ok(Caller::authorized())
+}
+
+fn failure(error: anyhow::Error) -> zbus::fdo::Error {
+    zbus::fdo::Error::Failed(error.to_string())
+}
+fn json<T: serde::Serialize>(value: T) -> zbus::fdo::Result<String> {
+    serde_json::to_string(&value)
+        .map_err(|_| zbus::fdo::Error::Failed("Could not encode response".into()))
+}
+
+impl Service {
+    async fn change(
+        &self,
+        operation: Operation,
+        interactive: bool,
+        connection: &Connection,
+        header: &Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        let caller = authorize(connection, header, interactive).await?;
+        self.controller
+            .lock()
+            .await
+            .execute(&caller, operation)
+            .await
+            .map_err(failure)
+    }
+}
+
+#[zbus::interface(name = "org.roentgen.Controller1")]
+impl Service {
+    async fn status(&self) -> zbus::fdo::Result<String> {
+        json(self.controller.lock().await.status().map_err(failure)?)
+    }
+    async fn subscriptions(&self) -> zbus::fdo::Result<String> {
+        json(self.controller.lock().await.subscriptions())
+    }
+    async fn configurations(&self, subscription: &str) -> zbus::fdo::Result<String> {
+        json(
+            self.controller
+                .lock()
+                .await
+                .configurations((!subscription.is_empty()).then_some(subscription))
+                .map_err(failure)?,
+        )
+    }
+    async fn logs(&self) -> zbus::fdo::Result<String> {
+        json(self.controller.lock().await.logs())
+    }
+    async fn diagnostic(
+        &self,
+        id: &str,
+        speed: bool,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<String> {
+        let caller = authorize(connection, &header, interactive).await?;
+        let mut controller = self.controller.lock().await;
+        let cancellation = self.cancellation.notified();
+        tokio::pin!(cancellation);
+        cancellation.as_mut().enable();
+        tokio::select! {
+            result = controller.diagnostic(&caller, (!id.is_empty()).then_some(id), speed) => json(result.map_err(failure)?),
+            _ = cancellation => Err(zbus::fdo::Error::Failed("Diagnostic cancelled; proxy intent is unchanged".into())),
+        }
+    }
+    async fn cancel_diagnostic(
+        &self,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        let _caller = authorize(connection, &header, interactive).await?;
+        self.cancellation.notify_waiters();
+        Ok(())
+    }
+    async fn settings(&self) -> zbus::fdo::Result<String> {
+        json(self.controller.lock().await.settings())
+    }
+    async fn add_subscription(
+        &self,
+        name: &str,
+        url: &str,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        self.change(
+            Operation::AddSubscription {
+                name: name.into(),
+                url: url.into(),
+            },
+            interactive,
+            connection,
+            &header,
+        )
+        .await
+    }
+    async fn edit_subscription(
+        &self,
+        name: &str,
+        url: &str,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        self.change(
+            Operation::EditSubscription {
+                name: name.into(),
+                url: url.into(),
+            },
+            interactive,
+            connection,
+            &header,
+        )
+        .await
+    }
+    async fn delete_subscription(
+        &self,
+        name: &str,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        self.change(
+            Operation::DeleteSubscription { name: name.into() },
+            interactive,
+            connection,
+            &header,
+        )
+        .await
+    }
+    async fn refresh(
+        &self,
+        name: &str,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        self.change(
+            Operation::Refresh {
+                name: (!name.is_empty()).then(|| name.into()),
+            },
+            interactive,
+            connection,
+            &header,
+        )
+        .await
+    }
+    async fn select(
+        &self,
+        id: &str,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        self.change(
+            Operation::Select { id: id.into() },
+            interactive,
+            connection,
+            &header,
+        )
+        .await
+    }
+    async fn unset(
+        &self,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        self.change(Operation::Unset, interactive, connection, &header)
+            .await
+    }
+    async fn enable(
+        &self,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        self.change(Operation::Enable, interactive, connection, &header)
+            .await
+    }
+    async fn disable(
+        &self,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        self.change(Operation::Disable, interactive, connection, &header)
+            .await
+    }
+    async fn save_settings(
+        &self,
+        value: &str,
+        interactive: bool,
+        #[zbus(connection)] connection: &Connection,
+        #[zbus(header)] header: Header<'_>,
+    ) -> zbus::fdo::Result<()> {
+        let settings = serde_json::from_str(value)
+            .map_err(|_| zbus::fdo::Error::InvalidArgs("Invalid settings".into()))?;
+        self.change(
+            Operation::SaveSettings { settings },
+            interactive,
+            connection,
+            &header,
+        )
+        .await
+    }
+}
+
+pub struct Client {
+    connection: Connection,
+}
+impl Client {
+    pub async fn system() -> Result<Self> {
+        Ok(Self {
+            connection: Connection::system()
+                .await
+                .map_err(|_| anyhow!("Cannot connect to the system bus"))?,
+        })
+    }
+    pub fn from_connection(connection: Connection) -> Self {
+        Self { connection }
+    }
+    async fn proxy(&self) -> Result<zbus::Proxy<'_>> {
+        zbus::Proxy::new(&self.connection, NAME, PATH, INTERFACE)
+            .await
+            .map_err(|_| anyhow!("Cannot contact Roentgen; install and start roentgen.service"))
+    }
+    pub async fn read<T: serde::de::DeserializeOwned>(&self, method: &str) -> Result<T> {
+        let value: String = self.proxy().await?.call(method, &()).await?;
+        Ok(serde_json::from_str(&value)?)
+    }
+    pub async fn configurations(
+        &self,
+        name: Option<&str>,
+    ) -> Result<Vec<crate::model::ConfigurationSummary>> {
+        let value: String = self
+            .proxy()
+            .await?
+            .call("Configurations", &(name.unwrap_or(""),))
+            .await?;
+        Ok(serde_json::from_str(&value)?)
+    }
+    pub async fn diagnostic(
+        &self,
+        id: Option<&str>,
+        speed: bool,
+        interactive: bool,
+    ) -> Result<String> {
+        let proxy = self.proxy().await?;
+        let arguments = (id.unwrap_or(""), speed, interactive);
+        tokio::select! {
+            result = proxy.call("Diagnostic", &arguments) => Ok(result?),
+            _ = tokio::signal::ctrl_c() => {
+                let _: () = proxy.call("CancelDiagnostic", &(interactive,)).await?;
+                anyhow::bail!("Diagnostic cancelled; proxy intent is unchanged")
+            }
+        }
+    }
+    pub async fn change(&self, operation: Operation, interactive: bool) -> Result<()> {
+        let proxy = self.proxy().await?;
+        match operation {
+            Operation::AddSubscription { name, url } => {
+                proxy
+                    .call("AddSubscription", &(name, url, interactive))
+                    .await?
+            }
+            Operation::EditSubscription { name, url } => {
+                proxy
+                    .call("EditSubscription", &(name, url, interactive))
+                    .await?
+            }
+            Operation::DeleteSubscription { name } => {
+                proxy
+                    .call("DeleteSubscription", &(name, interactive))
+                    .await?
+            }
+            Operation::Refresh { name } => {
+                proxy
+                    .call("Refresh", &(name.unwrap_or_default(), interactive))
+                    .await?
+            }
+            Operation::Select { id } => proxy.call("Select", &(id, interactive)).await?,
+            Operation::Unset => proxy.call("Unset", &(interactive,)).await?,
+            Operation::Enable => proxy.call("Enable", &(interactive,)).await?,
+            Operation::Disable => proxy.call("Disable", &(interactive,)).await?,
+            Operation::SaveSettings { settings } => {
+                proxy
+                    .call(
+                        "SaveSettings",
+                        &(serde_json::to_string(&settings)?, interactive),
+                    )
+                    .await?
+            }
+        }
+        Ok(())
+    }
+}