Start implementing device communication

src/command.rs

@@ -0,0 +1,21 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Command {
+    //
+}
+
+impl Command {
+    pub fn sub_type(device_type: u32) -> Self {
+        Self {}
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CommandResponse {
+    //
+}
+
+impl CommandResponse {
+    //
+}

src/device.rs

@@ -1,27 +1,53 @@
 use std::{
     io::{Read, Write},
     net::TcpStream,
+    sync::Mutex,
     thread,
     time::Duration,
 };
 
 use anyhow::{bail, Context, Error, Result};
+use serde_json::to_string;
 
 use crate::{
+    command::Command,
     devices::{e1::E1, DeviceBackend},
     hex,
+    packet_builder::PacketBuilder,
     security::{MsgType, Security},
     DeviceInfo,
 };
 
+#[repr(u8)]
+enum MessageType {
+    Set = 0x02,
+    Query = 0x03,
+    Notify1 = 0x04,
+    Notify2 = 0x05,
+    Exception = 0x06,
+    QuerySN = 0x07,
+    Exception2 = 0x0A,
+    QuerySubtype = 0xA0,
+}
+
+enum ParseMessage {
+    Success,
+    Padding,
+}
+
 pub struct Device {
     info: DeviceInfo,
 
-    socket: TcpStream,
+    socket: Mutex<TcpStream>,
     security: Security,
 
     device_backend: Box<dyn DeviceBackend>,
 
+    buffer: Vec<u8>,
+    sub_type: Option<String>,
+
+    updates: Vec<Box<dyn Fn(&[u8]) -> Result<()>>>,
+
     token: [u8; 64],
     key: [u8; 32],
 }
@@ -52,9 +78,14 @@ impl Device {
             }),
 
             info,
-            socket,
+            socket: Mutex::new(socket),
             security: Security::default(),
 
+            buffer: Vec::new(),
+            sub_type: None,
+
+            updates: Vec::new(),
+
             token: hex(token)?.try_into().unwrap(),
             key: hex(key)?.try_into().unwrap(),
         };
@@ -73,10 +104,12 @@ impl Device {
             .security
             .encode_8370(&self.token, MsgType::HANDSHAKE_REQUEST)?;
 
-        self.socket.write(&request)?;
+        let mut socket = self.socket.lock().unwrap();
+
+        socket.write(&request)?;
 
         let mut buffer = [0; 512];
-        let bytes_read = self.socket.read(&mut buffer)?;
+        let bytes_read = socket.read(&mut buffer)?;
 
         if bytes_read < 20 {
             bail!(
@@ -90,9 +123,109 @@ impl Device {
         Ok(())
     }
 
-    pub fn refresh_status(&self) -> Result<()> {
+    pub fn refresh_status(&mut self) -> Result<()> {
+        let mut cmds = self.device_backend.build_query()?;
+
+        if self.sub_type.is_none() {
+            cmds.insert(0, Command::sub_type(self.info.device_type));
+        }
+
+        for cmd in cmds {
+            self.build_send(cmd)?;
+
+            loop {
+                let mut buf = [0; 512];
+                let bytes_read = self.socket.lock().unwrap().read(&mut buf)?;
+
+                if bytes_read == 0 {
+                    bail!("socket error");
+                }
+
+                match self.parse_message(&buf[..bytes_read])? {
+                    ParseMessage::Success => break,
+                    ParseMessage::Padding => continue,
+                }
+            }
+        }
+
         Ok(())
     }
+
+    pub fn register_update<F>(&mut self, f: F)
+    where
+        F: Fn(&[u8]) -> Result<()> + 'static,
+    {
+        self.updates.push(Box::new(f));
+    }
+
+    fn parse_message(&mut self, msg: &[u8]) -> Result<ParseMessage> {
+        let (messages, buffer) = self.security.decode_8370(&self.buffer, msg)?;
+        self.buffer = buffer;
+
+        if messages.is_empty() {
+            return Ok(ParseMessage::Padding);
+        }
+
+        for mut message in messages {
+            if message == b"ERROR" {
+                bail!("parse message error");
+            }
+
+            let payload_len = message[4] as u16 + ((message[5] as u16) << 8) - 56;
+            let payload_type = message[2] as u16 + ((message[3] as u16) << 8);
+
+            // heartbeat
+            if payload_type == 0x1001 || payload_type == 0x0001 {
+                continue;
+            }
+
+            if message.len() > 56 && payload_len % 16 == 0 {
+                let len = message.len();
+                let crypt = &mut message[40..len - 16];
+
+                Security::aes_decrypt(crypt);
+
+                if self.pre_process_message(crypt)? {
+                    let status = self.device_backend.process_message(crypt)?;
+
+                    if status.len() > 0 {
+                        for update in self.updates.iter() {
+                            update(&status)?;
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(ParseMessage::Success)
+    }
+
+    fn pre_process_message(&self, msg: &[u8]) -> Result<bool> {
+        if msg[9] == MessageType::QuerySubtype as u8 {
+            let message = todo!();
+
+            Ok(false)
+        } else {
+            Ok(true)
+        }
+    }
+
+    fn send_message(&self, msg: &[u8]) -> Result<()> {
+        let data = self.security.encode_8370(msg, MsgType::ENCRYPTED_REQUEST)?;
+        self.socket.lock().unwrap().write(&data)?;
+
+        Ok(())
+    }
+
+    fn send_heartbeat(&self) -> Result<()> {
+        let msg = PacketBuilder::builder(self.info.id, &[0x00]).finalize(0);
+        self.send_message(&msg)
+    }
+
+    fn build_send(&self, cmd: Command) -> Result<()> {
+        let data = PacketBuilder::builder(self.info.id, to_string(&cmd)?.as_bytes()).finalize(1);
+        self.send_message(&data)
+    }
 }
 
 #[cfg(test)]

src/devices/e1.rs

@@ -2,6 +2,8 @@ use std::collections::HashMap;
 
 use anyhow::Result;
 
+use crate::command::Command;
+
 use super::DeviceBackend;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
@@ -189,11 +191,11 @@ impl E1 {
 }
 
 impl DeviceBackend for E1 {
-    fn build_query(&self) -> () {
+    fn build_query(&self) -> Result<Vec<Command>> {
         todo!()
     }
 
-    fn process_message(&self, msg: &str) -> () {
+    fn process_message(&self, msg: &[u8]) -> Result<Vec<u8>> {
         todo!()
     }
 

src/devices/mod.rs

@@ -1,7 +1,9 @@
+use crate::command::Command;
+
 pub mod e1;
 
 pub trait DeviceBackend {
-    fn build_query(&self) -> ();
-    fn process_message(&self, msg: &str) -> ();
+    fn build_query(&self) -> anyhow::Result<Vec<Command>>;
+    fn process_message(&self, msg: &[u8]) -> anyhow::Result<Vec<u8>>;
     fn set_attribute(&self, attribute: &str, value: &str) -> ();
 }

src/discover.rs

@@ -72,7 +72,8 @@ impl Startup {
                     }
 
                     let len = bytes.len();
-                    let encrypt_data = Security::aes_decrypt(&mut bytes[40..(len - 16)]);
+                    let encrypt_data = &mut bytes[40..(len - 16)];
+                    Security::aes_decrypt(encrypt_data);
 
                     let start = 41;
                     let upper = start + encrypt_data[40] as usize;

src/lib.rs

@@ -2,9 +2,11 @@ use std::num::ParseIntError;
 
 mod cloud;
 mod cloud_security;
+mod command;
 mod device;
 mod devices;
 mod discover;
+mod packet_builder;
 mod security;
 
 fn hex(s: &str) -> Result<Vec<u8>, ParseIntError> {

src/packet_builder.rs

@@ -0,0 +1,13 @@
+pub struct PacketBuilder {
+    //
+}
+
+impl PacketBuilder {
+    pub fn builder(device_id: u64, msg: &[u8]) -> Self {
+        Self {}
+    }
+
+    pub fn finalize(self, msg_type: u8) -> Vec<u8> {
+        todo!()
+    }
+}

src/security.rs

@@ -1,3 +1,5 @@
+use std::sync::atomic::{AtomicU16, Ordering::SeqCst};
+
 use aes::{
     cipher::{
         block_padding::NoPadding, generic_array::GenericArray, BlockDecrypt, BlockDecryptMut,
@@ -22,8 +24,8 @@ pub enum MsgType {
 
 #[derive(Debug, Default)]
 pub struct Security {
-    request_count: u16,
-    response_count: u16,
+    request_count: AtomicU16,
+    response_count: AtomicU16,
 
     tcp_key: Option<[u8; 32]>,
 }
@@ -33,7 +35,7 @@ impl Security {
     const KEY: [u8; 16] = Self::N.to_be_bytes();
     const IV: [u8; 16] = [b'\0'; 16];
 
-    pub fn aes_decrypt(data: &mut [u8]) -> &[u8] {
+    pub fn aes_decrypt(data: &mut [u8]) {
         let array = GenericArray::from(Self::KEY);
         let cipher = Aes128::new(&array);
 
@@ -41,8 +43,6 @@ impl Security {
             let mut block = GenericArray::from_mut_slice(chunk);
             cipher.decrypt_block(&mut block);
         }
-
-        data
     }
 
     pub fn aes_cbc_encrypt(&self, raw: [u8; 32], key: &[u8; 32]) -> [u8; 32] {
@@ -84,8 +84,8 @@ impl Security {
         }
 
         self.tcp_key = Some(Self::xorstr(&result, key).try_into().unwrap());
-        self.request_count = 0;
-        self.response_count = 0;
+        self.request_count.store(0, SeqCst);
+        self.response_count.store(0, SeqCst);
 
         Ok(())
     }
@@ -96,7 +96,7 @@ impl Security {
         lhs.iter().zip(rhs.iter()).map(|(&l, &r)| l ^ r).collect()
     }
 
-    pub fn encode_8370(&mut self, msg: &[u8], msg_type: MsgType) -> Result<Vec<u8>> {
+    pub fn encode_8370(&self, msg: &[u8], msg_type: MsgType) -> Result<Vec<u8>> {
         let mut header = hex("8370")?;
         let mut data: Vec<u8> = msg.to_vec();
 
@@ -120,13 +120,15 @@ impl Security {
         header.extend([0x20, (padding << 4) as u8 | msg_type as u8]);
 
         data = {
-            let mut b = self.request_count.to_be_bytes().to_vec();
+            let mut b = self
+                .request_count
+                .fetch_add(1, SeqCst)
+                .to_be_bytes()
+                .to_vec();
             b.extend(data);
             b
         };
 
-        (self.request_count, _) = self.request_count.overflowing_add(1);
-
         if msg_type == MsgType::ENCRYPTED_RESPONSE || msg_type == MsgType::ENCRYPTED_REQUEST {
             let sign: Vec<u8> = Sha256::digest(Self::add_bytes(header.clone(), data.clone()))
                 .into_iter()
@@ -146,6 +148,10 @@ impl Security {
         lhs.extend(rhs);
         lhs
     }
+
+    pub fn decode_8370(&self, buffer: &[u8], msg: &[u8]) -> Result<(Vec<Vec<u8>>, Vec<u8>)> {
+        todo!()
+    }
 }
 
 #[cfg(test)]