appmgr: mdns controller

appmgr/src/context/rpc.rs

@@ -11,7 +11,7 @@ use serde::Deserialize;
 use sqlx::SqlitePool;
 use tokio::fs::File;
 
-use crate::net::mdns::{enable_lan, LanHandle};
+use crate::net::mdns::MdnsController;
 use crate::net::tor::TorController;
 use crate::util::{from_toml_async_reader, AsyncFileExt, Container};
 use crate::{Error, ResultExt};
@@ -32,8 +32,8 @@ pub struct RpcContextSeed {
     pub db: PatchDb,
     pub secret_store: SqlitePool,
     pub docker: Docker,
-    pub lan_handle: Container<LanHandle>,
     pub tor_controller: TorController,
+    pub mdns_controller: MdnsController,
 }
 
 #[derive(Clone)]
@@ -65,22 +65,21 @@ impl RpcContext {
         ))
         .await?;
         let mut db_handle = db.handle();
-        let lan_handle = Container::new();
-        lan_handle.set(enable_lan(&mut db_handle).await?).await;
         let tor_controller = TorController::init(
             base.tor_control.unwrap_or(([127, 0, 0, 1], 9051).into()),
             &mut db_handle,
             &mut secret_store.acquire().await?,
         )
         .await?;
+        let mdns_controller = MdnsController::init(&mut db_handle).await?;
         let seed = Arc::new(RpcContextSeed {
             bind_rpc: base.bind_rpc.unwrap_or(([127, 0, 0, 1], 5959).into()),
             bind_ws: base.bind_ws.unwrap_or(([127, 0, 0, 1], 5960).into()),
             db,
             secret_store,
             docker: Docker::connect_with_unix_defaults()?,
-            lan_handle,
             tor_controller,
+            mdns_controller,
         });
         Ok(Self(seed))
     }

appmgr/src/install/mod.rs

@@ -349,10 +349,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
     log::info!("Install {}@{}: Installed main", pkg_id, version);
 
     log::info!("Install {}@{}: Installing interfaces", pkg_id, version);
-    let interface_info = manifest
+    let interface_info = manifest.interfaces.install(&mut sql_tx, pkg_id, ip).await?;
-        .interfaces
-        .install(&ctx, &mut sql_tx, pkg_id, ip)
-        .await?;
     log::info!("Install {}@{}: Installed interfaces", pkg_id, version);
 
     log::info!("Install {}@{}: Complete", pkg_id, version);
@@ -466,6 +463,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
     }
 
     ctx.tor_controller.sync(&mut tx, &mut sql_tx).await?;
+    ctx.mdns_controller.sync(&mut tx).await?;
 
     tx.commit(None).await?;
 

appmgr/src/net/interface.rs

@@ -2,12 +2,10 @@ use std::net::Ipv4Addr;
 use std::path::Path;
 
 use indexmap::IndexMap;
-use patch_db::DbHandle;
 use serde::{Deserialize, Deserializer, Serialize};
 use sqlx::{Executor, Sqlite};
 use torut::onion::TorSecretKeyV3;
 
-use crate::context::RpcContext;
 use crate::db::model::{InterfaceAddressMap, InterfaceAddresses, InterfaceInfo};
 use crate::id::Id;
 use crate::s9pk::manifest::PackageId;
@@ -19,7 +17,6 @@ pub struct Interfaces(pub IndexMap<InterfaceId, Interface>); // TODO
 impl Interfaces {
     pub async fn install<Ex>(
         &self,
-        ctx: &RpcContext,
         secrets: &mut Ex,
         package_id: &PackageId,
         ip: Ipv4Addr,
@@ -36,7 +33,7 @@ impl Interfaces {
                 tor_address: None,
                 lan_address: None,
             };
-            if iface.tor_config.is_some() {
+            if iface.tor_config.is_some() || iface.lan_config.is_some() {
                 let key = TorSecretKeyV3::generate();
                 let key_vec = key.as_bytes().to_vec();
                 sqlx::query!(
@@ -47,11 +44,18 @@ impl Interfaces {
                 )
                 .execute(&mut *secrets)
                 .await?;
-                addrs.tor_address = Some(key.public().get_onion_address().to_string());
+                let onion = key.public().get_onion_address();
+                if iface.tor_config.is_some() {
+                    addrs.tor_address = Some(onion.to_string());
+                }
+                if iface.lan_config.is_some() {
+                    addrs.lan_address =
+                        Some(format!("{}.local", onion.get_address_without_dot_onion()));
+                }
             }
             interface_info.addresses.0.insert(id.clone(), addrs);
         }
-        todo!()
+        Ok(interface_info)
     }
 }
 

appmgr/src/net/mdns.rs

@@ -1,93 +1,148 @@
-use avahi_sys;
+use avahi_sys::{
-use futures::future::pending;
+    self, avahi_client_free, avahi_entry_group_commit, avahi_entry_group_free,
+    avahi_entry_group_reset, avahi_free, AvahiClient, AvahiEntryGroup,
+};
+use libc::c_void;
 use patch_db::{DbHandle, OptionModel};
+use tokio::sync::RwLock;
 
-use crate::db::model::{InterfaceAddressesModel, InterfaceInfoModel};
 use crate::util::Apply;
 use crate::Error;
 
-pub async fn enable_lan<Db: DbHandle>(mut db: Db) -> Result<LanHandle, Error> {
+const HOSTNAME_LEN: usize = 1 + 15 + 1 + 5; // leading byte, main address, dot, "local"
-    unsafe {
-        // let app_list = crate::apps::list_info().await?;
 
-        let simple_poll = avahi_sys::avahi_simple_poll_new();
+pub struct MdnsController(RwLock<MdnsControllerInner>);
-        let poll = avahi_sys::avahi_simple_poll_get(simple_poll);
+impl MdnsController {
-        let mut stack_err = 0;
+    pub async fn init<Db: DbHandle>(db: &mut Db) -> Result<Self, Error> {
-        let err_c: *mut i32 = &mut stack_err;
+        Ok(MdnsController(RwLock::new(
-        let avahi_client = avahi_sys::avahi_client_new(
+            MdnsControllerInner::init(db).await?,
-            poll,
+        )))
-            avahi_sys::AvahiClientFlags::AVAHI_CLIENT_NO_FAIL,
+    }
-            None,
+    pub async fn sync<Db: DbHandle>(&self, db: &mut Db) -> Result<(), Error> {
-            std::ptr::null_mut(),
+        self.0.write().await.sync(db).await
-            err_c,
+    }
-        );
+}
-        let group =
-            avahi_sys::avahi_entry_group_new(avahi_client, Some(noop), std::ptr::null_mut());
-        let hostname_raw = avahi_sys::avahi_client_get_host_name_fqdn(avahi_client);
-        let hostname_bytes = std::ffi::CStr::from_ptr(hostname_raw).to_bytes_with_nul();
-        const HOSTNAME_LEN: usize = 1 + 15 + 1 + 5; // leading byte, main address, dot, "local"
-        debug_assert_eq!(hostname_bytes.len(), HOSTNAME_LEN);
-        let mut hostname_buf = [0; HOSTNAME_LEN + 1];
-        hostname_buf[1..].copy_from_slice(hostname_bytes);
-        // assume fixed length prefix on hostname due to local address
-        hostname_buf[0] = 15; // set the prefix length to 15 for the main address
-        hostname_buf[16] = 5; // set the prefix length to 5 for "local"
 
-        for app_id in crate::db::DatabaseModel::new()
+pub struct MdnsControllerInner {
-            .package_data()
+    hostname: [u8; HOSTNAME_LEN + 1],
-            .keys(&mut db)
+    client: *mut AvahiClient,
-            .await?
+    entry_group: *mut AvahiEntryGroup,
-        {
+}
-            let iface_model = if let Some(model) = crate::db::DatabaseModel::new()
+unsafe impl Send for MdnsControllerInner {}
+unsafe impl Sync for MdnsControllerInner {}
+
+impl MdnsControllerInner {
+    async fn load_services<Db: DbHandle>(&mut self, db: &mut Db) -> Result<(), Error> {
+        unsafe {
+            for app_id in crate::db::DatabaseModel::new()
                 .package_data()
-                .idx_model(&app_id)
+                .keys(db)
-                .expect(&mut db)
-                .await?
-                .installed()
-                .map(|i| i.interface_info().addresses())
-                .apply(OptionModel::from)
-                .check(&mut db)
                 .await?
             {
-                model
+                let iface_model = if let Some(model) = crate::db::DatabaseModel::new()
-            } else {
+                    .package_data()
-                continue;
+                    .idx_model(&app_id)
-            };
+                    .expect(db)
-            for iface in iface_model.keys(&mut db).await? {
-                let lan_address = if let Some(addr) = iface_model
-                    .clone()
-                    .idx_model(&iface)
-                    .expect(&mut db)
                     .await?
-                    .lan_address()
+                    .installed()
-                    .get(&mut db)
+                    .map(|i| i.interface_info().addresses())
+                    .apply(OptionModel::from)
+                    .check(db)
                     .await?
-                    .to_owned()
                 {
-                    addr
+                    model
                 } else {
                     continue;
                 };
-                let lan_address_ptr = std::ffi::CString::new(lan_address)
+                for iface in iface_model.keys(db).await? {
-                    .expect("Could not cast lan address to c string");
+                    let lan_address = if let Some(addr) = iface_model
-                let _ = avahi_sys::avahi_entry_group_add_record(
+                        .clone()
-                    group,
+                        .idx_model(&iface)
-                    avahi_sys::AVAHI_IF_UNSPEC,
+                        .expect(db)
-                    avahi_sys::AVAHI_PROTO_UNSPEC,
+                        .await?
-                    avahi_sys::AvahiPublishFlags_AVAHI_PUBLISH_USE_MULTICAST
+                        .lan_address()
-                        | avahi_sys::AvahiPublishFlags_AVAHI_PUBLISH_ALLOW_MULTIPLE,
+                        .get(db)
-                    lan_address_ptr.as_ptr(),
+                        .await?
-                    avahi_sys::AVAHI_DNS_CLASS_IN as u16,
+                        .to_owned()
-                    avahi_sys::AVAHI_DNS_TYPE_CNAME as u16,
+                    {
-                    avahi_sys::AVAHI_DEFAULT_TTL,
+                        addr
-                    hostname_buf.as_ptr().cast(),
+                    } else {
-                    hostname_buf.len(),
+                        continue;
-                );
+                    };
-                log::info!("Published {:?}", lan_address_ptr);
+                    let lan_address_ptr = std::ffi::CString::new(lan_address)
+                        .expect("Could not cast lan address to c string");
+                    let _ = avahi_sys::avahi_entry_group_add_record(
+                        self.entry_group,
+                        avahi_sys::AVAHI_IF_UNSPEC,
+                        avahi_sys::AVAHI_PROTO_UNSPEC,
+                        avahi_sys::AvahiPublishFlags_AVAHI_PUBLISH_USE_MULTICAST
+                            | avahi_sys::AvahiPublishFlags_AVAHI_PUBLISH_ALLOW_MULTIPLE,
+                        lan_address_ptr.as_ptr(),
+                        avahi_sys::AVAHI_DNS_CLASS_IN as u16,
+                        avahi_sys::AVAHI_DNS_TYPE_CNAME as u16,
+                        avahi_sys::AVAHI_DEFAULT_TTL,
+                        self.hostname.as_ptr().cast(),
+                        self.hostname.len(),
+                    );
+                    log::info!("Published {:?}", lan_address_ptr);
+                }
             }
         }
-        avahi_sys::avahi_entry_group_commit(group);
+        Ok(())
-        Ok(LanHandle(group))
+    }
+    async fn init<Db: DbHandle>(db: &mut Db) -> Result<Self, Error> {
+        unsafe {
+            // let app_list = crate::apps::list_info().await?;
+
+            let simple_poll = avahi_sys::avahi_simple_poll_new();
+            let poll = avahi_sys::avahi_simple_poll_get(simple_poll);
+            let mut stack_err = 0;
+            let err_c: *mut i32 = &mut stack_err;
+            let avahi_client = avahi_sys::avahi_client_new(
+                poll,
+                avahi_sys::AvahiClientFlags::AVAHI_CLIENT_NO_FAIL,
+                None,
+                std::ptr::null_mut(),
+                err_c,
+            );
+            let group =
+                avahi_sys::avahi_entry_group_new(avahi_client, Some(noop), std::ptr::null_mut());
+            let mut hostname_buf = [0; HOSTNAME_LEN + 1];
+            {
+                let hostname_raw = avahi_sys::avahi_client_get_host_name_fqdn(avahi_client);
+                hostname_buf[1..]
+                    .copy_from_slice(std::ffi::CStr::from_ptr(hostname_raw).to_bytes_with_nul());
+                avahi_free(hostname_raw as *mut c_void);
+            }
+            // assume fixed length prefix on hostname due to local address
+            hostname_buf[0] = 15; // set the prefix length to 15 for the main address
+            hostname_buf[16] = 5; // set the prefix length to 5 for "local"
+
+            let mut ctrl = MdnsControllerInner {
+                hostname: hostname_buf,
+                client: avahi_client,
+                entry_group: group,
+            };
+            avahi_entry_group_commit(group);
+            ctrl.load_services(db).await?;
+            Ok(ctrl)
+        }
+    }
+    async fn sync<Db: DbHandle>(&mut self, db: &mut Db) -> Result<(), Error> {
+        unsafe {
+            avahi_entry_group_reset(self.entry_group);
+            self.load_services(db).await?;
+            avahi_entry_group_commit(self.entry_group);
+        }
+        Ok(())
+    }
+}
+impl Drop for MdnsControllerInner {
+    fn drop(&mut self) {
+        unsafe {
+            avahi_entry_group_free(self.entry_group);
+            avahi_client_free(self.client);
+        }
     }
 }
 
@@ -97,15 +152,3 @@ unsafe extern "C" fn noop(
     _userdata: *mut core::ffi::c_void,
 ) {
 }
-
-pub struct LanHandle(*mut avahi_sys::AvahiEntryGroup);
-impl Drop for LanHandle {
-    fn drop(&mut self) {
-        unsafe {
-            avahi_sys::avahi_entry_group_reset(self.0);
-            avahi_sys::avahi_entry_group_free(self.0);
-        }
-    }
-}
-unsafe impl Send for LanHandle {}
-unsafe impl Sync for LanHandle {}

appmgr/src/net/tor.rs

@@ -19,8 +19,7 @@ fn event_handler(event: AsyncEvent<'static>) -> BoxFuture<'static, Result<(), Co
     async move { Ok(()) }.boxed()
 }
 
-#[derive(Clone)]
+pub struct TorController(RwLock<TorControllerInner>);
-pub struct TorController(Arc<RwLock<TorControllerInner>>);
 impl TorController {
     pub async fn init<Db: DbHandle, Ex>(
         tor_cp: SocketAddr,
@@ -30,9 +29,9 @@ impl TorController {
     where
         for<'a> &'a mut Ex: Executor<'a, Database = Sqlite>,
     {
-        Ok(TorController(Arc::new(RwLock::new(
+        Ok(TorController(RwLock::new(
             TorControllerInner::init(tor_cp, db, secrets).await?,
-        ))))
+        )))
     }
 
     pub async fn sync<Db: DbHandle, Ex>(&self, db: &mut Db, secrets: &mut Ex) -> Result<(), Error>