Implement proper message channeling

idevice/src/dvt/message.rs

@@ -21,7 +21,7 @@ pub struct MessageHeader {
     length: u32, // Length of of the payload
     identifier: u32,
     conversation_index: u32,
-    channel: u32,
+    pub channel: u32,
     expects_reply: bool,
 }
 
@@ -194,7 +194,7 @@ impl MessageHeader {
         expects_reply: bool,
     ) -> Self {
         Self {
-            magic: 0x795b3d1f,
+            magic: 0x1F3D5B79,
             header_len: 32,
             fragment_id,
             fragment_count,
@@ -236,8 +236,7 @@ impl PayloadHeader {
         res
     }
 
-    // from pymobiledevice3
-    pub fn instruments_message_type() -> Self {
+    pub fn method_invocation() -> Self {
         Self {
             flags: 2,
             ..Default::default()
@@ -369,7 +368,7 @@ mod tests {
 
     #[tokio::test]
     async fn t1() {
-        let test = "/Users/jacksoncoxson/Desktop/try1";
+        let test = "/Users/jacksoncoxson/Desktop/try2";
         let mut bytes = tokio::fs::File::open(test).await.unwrap();
 
         let message = Message::from_reader(&mut bytes).await.unwrap();

idevice/src/dvt/process_control.rs

@@ -1,5 +1,19 @@
 // Jackson Coxson
 
-pub struct ProcessControlClient {
-    client: super::remote_server::RemoteServerClient,
+use crate::IdeviceError;
+
+use super::remote_server::{Channel, RemoteServerClient};
+
+const IDENTIFIER: &str = "com.apple.instruments.server.services.processcontrol";
+
+pub struct ProcessControlClient<'a> {
+    channel: Channel<'a>,
+}
+
+impl<'a> ProcessControlClient<'a> {
+    pub async fn new(client: &'a mut RemoteServerClient) -> Result<Self, IdeviceError> {
+        let channel = client.make_channel(IDENTIFIER).await?; // Drop `&mut client` before continuing
+
+        Ok(Self { channel })
+    }
 }

idevice/src/dvt/remote_server.rs

@@ -1,7 +1,8 @@
 // Jackson Coxson
 
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 
+use log::{debug, warn};
 use tokio::io::AsyncWriteExt;
 
 use crate::{
@@ -17,7 +18,12 @@ pub struct RemoteServerClient {
     idevice: Box<dyn ReadWrite>,
     current_message: u32,
     new_channel: u32,
-    channels: HashMap<u8, Vec<Message>>,
+    channels: HashMap<u32, VecDeque<Message>>,
+}
+
+pub struct Channel<'a> {
+    client: &'a mut RemoteServerClient,
+    channel: u32,
 }
 
 impl RemoteServerClient {
@@ -33,9 +39,17 @@ impl RemoteServerClient {
     pub async fn make_channel(
         &mut self,
         identifier: impl Into<String>,
-    ) -> Result<(), IdeviceError> {
+    ) -> Result<Channel, IdeviceError> {
         let code = self.new_channel;
-        let args = vec![AuxValue::U32(code), AuxValue::String(identifier.into())];
+        self.new_channel += 1;
+
+        let args = vec![
+            AuxValue::U32(code),
+            AuxValue::Array(
+                ns_keyed_archive::encode::encode_to_bytes(plist::Value::String(identifier.into()))
+                    .expect("Failed to encode"),
+            ),
+        ];
         self.send_message(
             0,
             Some("_requestChannelWithCode:identifier:"),
@@ -44,12 +58,19 @@ impl RemoteServerClient {
         )
         .await?;
 
-        let res = self.read_message().await?;
+        let res = self.read_message(0).await?;
         if res.data.is_some() {
             return Err(IdeviceError::UnexpectedResponse);
         }
 
-        Ok(())
+        self.build_channel(code)
+    }
+
+    fn build_channel(&mut self, code: u32) -> Result<Channel, IdeviceError> {
+        Ok(Channel {
+            client: self,
+            channel: code,
+        })
     }
 
     pub async fn send_message(
@@ -62,35 +83,64 @@ impl RemoteServerClient {
         self.current_message += 1;
 
         let mheader = MessageHeader::new(0, 1, self.current_message, 0, channel, expect_reply);
-        let mut pheader = PayloadHeader::instruments_message_type();
-        if expect_reply {
-            pheader.apply_expects_reply_map();
-        }
+        let pheader = PayloadHeader::method_invocation();
         let aux = args.map(Aux::from_values);
         let data: Option<plist::Value> = data.map(Into::into);
 
         let message = Message::new(mheader, pheader, aux, data);
+        debug!("Sending message: {message:#?}");
+        let bytes = message.serialize();
+        debug!(
+            "Re serde: {:#?}",
+            Message::from_reader(&mut std::io::Cursor::new(bytes)).await
+        );
         self.idevice.write_all(&message.serialize()).await?;
 
         Ok(())
     }
 
+    pub async fn read_message(&mut self, channel: u32) -> Result<Message, IdeviceError> {
+        // Determine if we already have a message cached
+        let cache = match self.channels.get_mut(&channel) {
+            Some(c) => c,
+            None => return Err(IdeviceError::UnknownChannel(channel)),
+        };
+
+        if let Some(msg) = cache.pop_front() {
+            return Ok(msg);
+        }
+
+        loop {
+            let msg = Message::from_reader(&mut self.idevice).await?;
+            debug!("Read message: {msg:#?}");
+
+            if msg.message_header.channel == channel {
+                return Ok(msg);
+            } else if let Some(cache) = self.channels.get_mut(&msg.message_header.channel) {
+                cache.push_back(msg);
+            } else {
+                warn!(
+                    "Received message for unknown channel: {}",
+                    msg.message_header.channel
+                );
+            }
+        }
+    }
+}
+
+impl Channel<'_> {
     pub async fn read_message(&mut self) -> Result<Message, IdeviceError> {
-        Message::from_reader(&mut self.idevice).await
-    }
+        self.client.read_message(self.channel).await
     }
 
-pub struct Channel {}
-
-#[cfg(test)]
-mod tests {
-    use crate::util::plist_to_archived_bytes;
-
-    #[test]
-    fn t1() {
-        let selector: plist::Value = "asdf".into();
-        let selector = plist_to_archived_bytes(selector);
-
-        std::fs::write("/Users/jacksoncoxson/code/test/test-rs.plist", selector).unwrap();
+    pub async fn send_message(
+        &mut self,
+        data: Option<impl Into<plist::Value>>,
+        args: Option<Vec<AuxValue>>,
+        expect_reply: bool,
+    ) -> Result<(), IdeviceError> {
+        self.client
+            .send_message(self.channel, data, args, expect_reply)
+            .await
     }
 }

idevice/src/lib.rs

@@ -308,6 +308,10 @@ pub enum IdeviceError {
     #[error("Unknown aux value type")]
     UnknownAuxValueType(u32),
 
+    #[cfg(feature = "dvt")]
+    #[error("unknown channel")]
+    UnknownChannel(u32),
+
     #[error("not enough bytes, expected {1}, got {0}")]
     NotEnoughBytes(usize, usize),
 

tools/src/process_control.rs

@@ -1,13 +1,12 @@
 // Jackson Coxson
 
 use std::{
-    io::Write,
     net::{IpAddr, SocketAddr},
     str::FromStr,
 };
 
 use clap::{Arg, Command};
-use idevice::{debug_proxy::DebugProxyClient, tunneld::get_tunneld_devices, xpc::XPCDevice};
+use idevice::{tunneld::get_tunneld_devices, xpc::XPCDevice};
 use tokio::net::TcpStream;
 
 mod common;
@@ -78,5 +77,10 @@ async fn main() {
     .await
     .expect("Failed to connect");
 
-    let rs_client = idevice::dvt::remote_server::RemoteServerClient::new(Box::new(stream));
+    let mut rs_client =
+        idevice::dvt::remote_server::RemoteServerClient::new(Box::new(stream)).unwrap();
+    rs_client.read_message(0).await.expect("no read??");
+    let pc_client = idevice::dvt::process_control::ProcessControlClient::new(&mut rs_client)
+        .await
+        .unwrap();
 }