timeout and proper chunk handling

Cargo.lock

@@ -1147,8 +1147,10 @@ version = "0.2.0"
 dependencies = [
  "clap",
  "env_logger",
+ "futures-util",
  "homekit-controller",
  "log",
+ "mdns",
  "rocket",
  "tokio",
 ]

homekit-controller/src/homekit_http.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashMap, time::Duration};
+use std::{
+    collections::{HashMap, VecDeque},
+    mem,
+    time::Duration,
+};
 
 use chacha20poly1305::{
     aead::generic_array::GenericArray, AeadInPlace, ChaCha20Poly1305, KeyInit, Nonce,
@@ -228,30 +232,30 @@ impl AccessorySocket {
         {
             if transfer_encoding.value == b"chunked" {
                 loop {
-                    packet.append(&mut self.get_next().await?);
+                    let chunked = ChunkedTransfer::clone_from(&packet).collect::<Vec<Vec<u8>>>();
 
-                    let utf8_decoded = String::from_utf8(packet.clone())?;
+                    if let Some(last) = chunked.chunks(2).last() {
+                        if last.len() == 2 && last[0] == b"0" {
+                            break;
+                        }
+                    }
 
-                    let split = utf8_decoded
-                        .split_terminator("\r\n")
-                        .map(String::from)
-                        .collect::<Vec<_>>();
-                    if let Some(last) = split.chunks(2).last() {
-                        if last.len() == 2 && last[0] == "0" {
+                    match tokio::time::timeout(Duration::from_secs(2), self.get_next()).await {
+                        Ok(next) => {
+                            packet.append(&mut next?);
+                        }
+                        Err(_) => {
+                            log::error!("timed out");
                             break;
                         }
                     }
                 }
 
-                let utf8_decoded = String::from_utf8(std::mem::take(&mut packet))?;
+                let mut chunked =
+                    ChunkedTransfer::from(std::mem::take(&mut packet)).collect::<Vec<Vec<u8>>>();
 
-                let split = utf8_decoded
-                    .split_terminator("\r\n")
-                    .map(String::from)
-                    .collect::<Vec<_>>();
-
-                for chunk in split.chunks_exact(2) {
-                    packet.extend_from_slice(chunk[1].as_bytes())
+                for chunk in chunked.chunks_exact_mut(2) {
+                    packet.append(&mut chunk[1]);
                 }
             }
         }
@@ -329,3 +333,52 @@ pub enum DiscoveryError {
     #[error("not found")]
     NotFound,
 }
+
+struct ChunkedTransfer {
+    data: Vec<u8>,
+}
+
+impl From<Vec<u8>> for ChunkedTransfer {
+    fn from(value: Vec<u8>) -> Self {
+        Self { data: value }
+    }
+}
+
+impl ChunkedTransfer {
+    fn clone_from(value: &[u8]) -> Self {
+        Self {
+            data: value.to_owned(),
+        }
+    }
+}
+
+impl Iterator for ChunkedTransfer {
+    type Item = Vec<u8>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.data.is_empty() {
+            None
+        } else {
+            let mut crlf_index = None;
+            for (i, val) in self.data.iter().enumerate() {
+                if let Some(next) = self.data.get(i + 1) {
+                    if *val == b'\r' && *next == b'\n' {
+                        crlf_index = Some(i);
+                        break;
+                    }
+                }
+            }
+
+            if let Some(i) = crlf_index {
+                let remainder = self.data.split_off(i);
+                let mut remainder = VecDeque::from(remainder);
+                let _ = remainder.pop_front().map(|v| v as char);
+                let _ = remainder.pop_front().map(|v| v as char);
+
+                Some(mem::replace(&mut self.data, Vec::from(remainder)))
+            } else {
+                Some(mem::take(&mut self.data))
+            }
+        }
+    }
+}

homekit-controller/src/lib.rs

@@ -60,6 +60,10 @@ impl PythonPairingData {
     }
 }
 
+// (id, (ip, port))
+// id - from TXT record
+// ip - from A record
+// port - from SRV record
 pub type MdnsDiscoveredList = Arc<RwLock<HashMap<String, (String, u16)>>>;
 
 pub fn spawn_discover_thread() -> Result<MdnsDiscoveredList, DiscoveryError> {
@@ -107,11 +111,18 @@ pub fn spawn_discover_thread() -> Result<MdnsDiscoveredList, DiscoveryError> {
                             None
                         }) {
                             let mut connections = discovered.write().await;
+                            if !connections.get(&id).is_some_and(|(old_ip, old_port)| {
+                                *old_ip == ip.to_string() && *old_port == port
+                            }) {
+                                log::info!(
+                                    "mdns: discovered {name} - id: {id}, ip: {ip}, port: {port}"
+                                );
                                 connections.insert(id, (ip.to_string(), port));
                             }
                         }
                     }
                 }
+            }
             tokio::time::sleep(Duration::from_secs(5 * 60)).await;
         }
     });

homekit-controller/src/pairing_data/characteristics.rs

@@ -3,15 +3,6 @@ use strum_macros::Display;
 
 // data from https://github.com/oznu/homebridge-gsh/blob/master/src/hap-types.ts
 
-pub(crate) fn deserialize_characteristic_type<'de, D>(
-    deserializer: D,
-) -> Result<CharacteristicType, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    CharacteristicTypeInner::deserialize(deserializer).map(|v| v.into())
-}
-
 #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Display)]
 #[strum(serialize_all = "snake_case")]
 
@@ -243,7 +234,7 @@ pub enum CharacteristicType {
 
 #[derive(Deserialize)]
 #[serde(field_identifier)]
-enum CharacteristicTypeInner {
+pub(super) enum CharacteristicTypeInner {
     #[serde(rename = "000000E5-0000-1000-8000-0026BB765291", alias = "E5")]
     AccessControlLevel,
     #[serde(rename = "000000A6-0000-1000-8000-0026BB765291", alias = "A6")]

homekit-controller/src/pairing_data/mod.rs

@@ -2,9 +2,11 @@ use std::collections::HashMap;
 
 use serde::{Deserialize, Deserializer, Serialize};
 
-use characteristics::{deserialize_characteristic_type, deserialize_service_type};
+use characteristics::deserialize_service_type;
 pub use characteristics::{CharacteristicType, ServiceType};
 
+use crate::pairing_data::characteristics::CharacteristicTypeInner;
+
 mod characteristics;
 
 #[derive(Deserialize, Debug, Clone)]
@@ -97,8 +99,8 @@ where
         iid: usize,
         #[serde(rename = "type", deserialize_with = "deserialize_service_type")]
         service_type: ServiceType,
-        primary: bool,
-        hidden: bool,
+        primary: Option<bool>,
+        hidden: Option<bool>,
         #[serde(deserialize_with = "deserialize_service_characteristics")]
         characteristics: HashMap<usize, ServiceCharacteristic>,
     }
@@ -107,8 +109,8 @@ where
             Self {
                 name: val.name,
                 service_type: val.service_type,
-                primary: val.primary,
-                hidden: val.hidden,
+                primary: val.primary.unwrap_or(false),
+                hidden: val.hidden.unwrap_or(false),
                 characteristics: val.characteristics,
                 name_for_accessory: val.name_for_accessory,
             }
@@ -209,9 +211,9 @@ where
     struct ServiceCharacteristicInner {
         #[serde(rename = "iid")]
         iid: usize,
-        #[serde(rename = "type", deserialize_with = "deserialize_characteristic_type")]
-        characteristic_type: CharacteristicType,
-        perms: Vec<CharacteristicPermissions>,
+        #[serde(rename = "type")]
+        characteristic_type: Option<CharacteristicTypeInner>,
+        perms: Option<Vec<CharacteristicPermissions>>,
         #[serde(flatten)]
         value: Option<Data>,
         #[serde(rename = "ev")]
@@ -227,11 +229,13 @@ where
         min_step: Option<f64>,
         unit: Option<Unit>,
     }
-    impl From<ServiceCharacteristicInner> for ServiceCharacteristic {
-        fn from(value: ServiceCharacteristicInner) -> Self {
-            Self {
-                characteristic_type: value.characteristic_type,
-                perms: value.perms,
+    impl ServiceCharacteristic {
+        fn maybe_from(value: ServiceCharacteristicInner) -> Option<Self> {
+            let characteristic_type = value.characteristic_type.map(|v| v.into())?;
+            let perms = value.perms?;
+            Some(Self {
+                characteristic_type,
+                perms,
                 value: value.value,
                 event_notifications_enabled: value.event_notifications_enabled,
                 enc: value.enc,
@@ -240,13 +244,16 @@ where
                 max_value: value.max_value,
                 min_step: value.min_step,
                 unit: value.unit,
-            }
+            })
         }
     }
     Vec::<ServiceCharacteristicInner>::deserialize(deserializer).map(|v| {
         v.into_iter()
             .fold(HashMap::new(), |mut map, characteristic| {
-                map.insert(characteristic.iid, characteristic.into());
+                let iid = characteristic.iid;
+                if let Some(c) = ServiceCharacteristic::maybe_from(characteristic) {
+                    map.insert(iid, c);
+                }
                 map
             })
     })

homekit-controller/src/tlv8/mod.rs

@@ -80,7 +80,10 @@ pub fn decode(data: &[u8]) -> Result<HashMap<u8, Vec<u8>>, TlvCodecError> {
         let tlv_len = (data[1] as usize) + 2;
         let current;
         if tlv_len > data.len() {
-            return Err(TlvCodecError::WrongLength);
+            return Err(TlvCodecError::WrongLength {
+                expected: tlv_len,
+                got: data.len(),
+            });
         }
 
         (current, data) = data.split_at(tlv_len);
@@ -90,7 +93,10 @@ pub fn decode(data: &[u8]) -> Result<HashMap<u8, Vec<u8>>, TlvCodecError> {
         }
 
         if current.len() < tlv_len {
-            return Err(TlvCodecError::WrongLength);
+            return Err(TlvCodecError::WrongLength {
+                expected: tlv_len,
+                got: current.len(),
+            });
         }
 
         let tlv_type = current[0];
@@ -110,7 +116,7 @@ pub enum TlvCodecError {
     #[error("too short")]
     TooShort,
     #[error("wrong length")]
-    WrongLength,
+    WrongLength { expected: usize, got: usize },
 }
 
 pub trait TlvEncode {

homekit-exporter/Cargo.toml

@@ -19,3 +19,5 @@ log = "0.4"
 clap = { version = "4.0", features = ["derive"] }
 homekit-controller = { path = "../homekit-controller" }
 rocket = { version = "0.5", features = ["json"] }
+mdns = "3.0.0"
+futures-util = "0.3.30"

homekit-exporter/packaging/service/service

@@ -9,7 +9,7 @@ Restart=always
 RestartSec=10s
 Environment="RUST_LOG=error,warn"
 Environment="LOG_TIMESTAMP=false"
-ExecStart=/usr/bin/homekit-exporter
+ExecStart=/usr/bin/homekit-exporter watch
 
 [Install]
 WantedBy=multi-user.target

homekit-exporter/src/main.rs

@@ -3,7 +3,8 @@ extern crate rocket;
 
 use std::{collections::HashMap, path::PathBuf, time::Duration};
 
-use clap::Parser;
+use clap::{Parser, Subcommand};
+use futures_util::{pin_mut, StreamExt};
 use homekit_controller::{spawn_discover_thread, DeviceConnection, HomekitError, ServiceType};
 use server::launch;
 
@@ -12,12 +13,22 @@ mod server;
 #[derive(Parser, Debug, Clone)]
 #[clap(author, version, about, long_about = None)]
 struct Args {
+    #[command(subcommand)]
+    command: Commands,
     #[clap(long, default_value = "/etc/homekit-exporter/pairing.json")]
     pairing_data: PathBuf,
     #[clap(long, default_value_t = 6666)]
     port: usize,
 }
 
+#[derive(Subcommand, Debug, Clone)]
+enum Commands {
+    /// Run logging server
+    Watch,
+    /// Discover devices
+    Discover,
+}
+
 const SENSORS: [ServiceType; 11] = [
     ServiceType::AirQualitySensor,
     ServiceType::CarbonDioxideSensor,
@@ -36,11 +47,31 @@ const SENSORS: [ServiceType; 11] = [
 async fn rocket() -> rocket::Rocket<rocket::Build> {
     env_logger::init();
     let args = Args::parse();
-    match init(args.pairing_data).await {
+    match args.command {
+        Commands::Watch => match init(args.pairing_data).await {
             Ok(paired) => launch(paired, args.port),
             Err(e) => panic!("Error {e:#?}"),
+        },
+        Commands::Discover => {
+            println!("discovering homekit devices via mdns");
+            match discover().await {
+                Ok(_) => todo!(),
+                Err(_) => todo!(),
             }
         }
+    }
+}
+
+async fn discover() -> Result<(), mdns::Error> {
+    let stream = mdns::discover::all("_hap._tcp.local", Duration::from_secs(1))?.listen();
+    pin_mut!(stream);
+
+    while let Some(Ok(response)) = stream.next().await {
+        println!("found device:\n\t{response:#?}");
+    }
+
+    Ok(())
+}
 
 async fn init(pairing_data: PathBuf) -> Result<HashMap<String, DeviceConnection>, HomekitError> {
     let discovered = spawn_discover_thread()?;
@@ -51,8 +82,9 @@ async fn init(pairing_data: PathBuf) -> Result<HashMap<String, DeviceConnection>
         for (k, v) in devices {
             let mut num = 0;
             let connected = loop {
-                if let Ok(v) = v.connect(&discovered).await {
-                    break Some(v);
+                match v.connect(&discovered).await {
+                    Ok(v) => break Some(v),
+                    Err(e) => log::error!("error connecting to {k}: {e:#?}"),
                 }
                 num += 1;
                 if num > 10 {