Initial http2 client rewrite

idevice/src/lib.rs

@@ -498,8 +498,16 @@ pub enum IdeviceError {
     InternalError(String),
 
     #[cfg(feature = "xpc")]
-    #[error("xpc message failed")]
+    #[error("unknown http frame type")]
-    Xpc(#[from] xpc::error::XPCError),
+    UnknownFrame(u8),
+
+    #[cfg(feature = "xpc")]
+    #[error("unknown http setting type")]
+    UnknownHttpSetting(u16),
+
+    #[cfg(feature = "xpc")]
+    #[error("Unintialized stream ID")]
+    UninitializedStreamId,
 
     #[cfg(feature = "dvt")]
     #[error("NSKeyedArchive error")]

idevice/src/services/xpc/error.rs

@@ -1,103 +0,0 @@
-// DebianArch
-
-use super::http2::error::Http2Error;
-use std::{
-    array::TryFromSliceError, error::Error, ffi::FromVecWithNulError, io, num::TryFromIntError,
-    str::Utf8Error,
-};
-
-#[derive(Debug)]
-pub enum XPCError {
-    Io(io::Error),
-    Http2Error(Http2Error),
-    ParseError(ParseError),
-    Custom(String),
-}
-
-#[derive(Debug)]
-pub enum ParseError {
-    TryFromSliceError(TryFromSliceError),
-    TryFromIntError(TryFromIntError),
-    FromVecWithNulError(FromVecWithNulError),
-    Utf8Error(Utf8Error),
-}
-
-impl From<TryFromSliceError> for XPCError {
-    fn from(value: TryFromSliceError) -> Self {
-        Self::ParseError(ParseError::TryFromSliceError(value))
-    }
-}
-
-impl From<TryFromIntError> for XPCError {
-    fn from(value: TryFromIntError) -> Self {
-        Self::ParseError(ParseError::TryFromIntError(value))
-    }
-}
-
-impl From<ParseError> for XPCError {
-    fn from(value: ParseError) -> Self {
-        Self::ParseError(value)
-    }
-}
-
-impl From<FromVecWithNulError> for XPCError {
-    fn from(value: FromVecWithNulError) -> Self {
-        Self::ParseError(ParseError::FromVecWithNulError(value))
-    }
-}
-
-impl From<Utf8Error> for XPCError {
-    fn from(value: Utf8Error) -> Self {
-        Self::ParseError(ParseError::Utf8Error(value))
-    }
-}
-
-impl From<io::Error> for XPCError {
-    fn from(value: io::Error) -> Self {
-        Self::Io(value)
-    }
-}
-
-impl From<&str> for XPCError {
-    fn from(value: &str) -> Self {
-        Self::Custom(value.to_string())
-    }
-}
-
-impl From<Http2Error> for XPCError {
-    fn from(value: Http2Error) -> Self {
-        Self::Http2Error(value)
-    }
-}
-
-impl std::fmt::Display for XPCError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "XPCError({})",
-            match self {
-                Self::Io(io) => io.to_string(),
-                Self::Http2Error(http2) => http2.to_string(),
-                Self::ParseError(e) => e.to_string(),
-                Self::Custom(s) => s.clone(),
-            }
-        )
-    }
-}
-
-impl std::fmt::Display for ParseError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "ParseError({})",
-            match self {
-                Self::TryFromSliceError(e) => e.to_string(),
-                Self::TryFromIntError(e) => e.to_string(),
-                Self::FromVecWithNulError(e) => e.to_string(),
-                Self::Utf8Error(e) => e.to_string(),
-            }
-        )
-    }
-}
-
-impl Error for XPCError {}

idevice/src/services/xpc/format.rs

@@ -1,448 +0,0 @@
-// DebianArch
-
-use std::{
-    ffi::CString,
-    io::{BufRead, Cursor, Read},
-    ops::{BitOr, BitOrAssign},
-};
-
-use super::error::XPCError;
-use indexmap::IndexMap;
-use log::debug;
-use serde::{Deserialize, Serialize};
-
-#[derive(Clone, Copy, Debug)]
-#[repr(u32)]
-pub enum XPCFlag {
-    AlwaysSet,
-    DataFlag,
-    WantingReply,
-    InitHandshake,
-
-    Custom(u32),
-}
-
-impl From<XPCFlag> for u32 {
-    fn from(value: XPCFlag) -> Self {
-        match value {
-            XPCFlag::AlwaysSet => 0x00000001,
-            XPCFlag::DataFlag => 0x00000100,
-            XPCFlag::WantingReply => 0x00010000,
-            XPCFlag::InitHandshake => 0x00400000,
-            XPCFlag::Custom(inner) => inner,
-        }
-    }
-}
-
-impl BitOr for XPCFlag {
-    fn bitor(self, rhs: Self) -> Self::Output {
-        XPCFlag::Custom(u32::from(self) | u32::from(rhs))
-    }
-
-    type Output = XPCFlag;
-}
-
-impl BitOrAssign for XPCFlag {
-    fn bitor_assign(&mut self, rhs: Self) {
-        *self = self.bitor(rhs);
-    }
-}
-
-impl PartialEq for XPCFlag {
-    fn eq(&self, other: &Self) -> bool {
-        u32::from(*self) == u32::from(*other)
-    }
-}
-
-#[repr(u32)]
-pub enum XPCType {
-    Bool = 0x00002000,
-    Dictionary = 0x0000f000,
-    Array = 0x0000e000,
-
-    Int64 = 0x00003000,
-    UInt64 = 0x00004000,
-
-    String = 0x00009000,
-    Data = 0x00008000,
-    Uuid = 0x0000a000,
-}
-
-impl TryFrom<u32> for XPCType {
-    type Error = XPCError;
-
-    fn try_from(value: u32) -> Result<Self, Self::Error> {
-        match value {
-            0x00002000 => Ok(Self::Bool),
-            0x0000f000 => Ok(Self::Dictionary),
-            0x0000e000 => Ok(Self::Array),
-            0x00003000 => Ok(Self::Int64),
-            0x00004000 => Ok(Self::UInt64),
-            0x00009000 => Ok(Self::String),
-            0x00008000 => Ok(Self::Data),
-            0x0000a000 => Ok(Self::Uuid),
-            _ => Err("Invalid XPCType")?,
-        }
-    }
-}
-
-pub type Dictionary = IndexMap<String, XPCObject>;
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub enum XPCObject {
-    Bool(bool),
-    Dictionary(Dictionary),
-    Array(Vec<XPCObject>),
-
-    Int64(i64),
-    UInt64(u64),
-
-    String(String),
-    Data(Vec<u8>),
-    Uuid(uuid::Uuid),
-}
-
-impl From<plist::Value> for XPCObject {
-    fn from(value: plist::Value) -> Self {
-        match value {
-            plist::Value::Array(v) => {
-                XPCObject::Array(v.iter().map(|item| XPCObject::from(item.clone())).collect())
-            }
-            plist::Value::Dictionary(v) => {
-                let mut dict = Dictionary::new();
-                for (k, v) in v.into_iter() {
-                    dict.insert(k.clone(), XPCObject::from(v));
-                }
-                XPCObject::Dictionary(dict)
-            }
-            plist::Value::Boolean(v) => XPCObject::Bool(v),
-            plist::Value::Data(v) => XPCObject::Data(v),
-            plist::Value::Date(_) => todo!(),
-            plist::Value::Real(_) => todo!(),
-            plist::Value::Integer(v) => XPCObject::Int64(v.as_signed().unwrap()),
-            plist::Value::String(v) => XPCObject::String(v),
-            plist::Value::Uid(_) => todo!(),
-            _ => todo!(),
-        }
-    }
-}
-
-impl XPCObject {
-    pub fn to_plist(&self) -> plist::Value {
-        match self {
-            Self::Bool(v) => plist::Value::Boolean(*v),
-            Self::Uuid(uuid) => plist::Value::String(uuid.to_string()),
-            Self::UInt64(v) => plist::Value::Integer({ *v }.into()),
-            Self::Int64(v) => plist::Value::Integer({ *v }.into()),
-            Self::String(v) => plist::Value::String(v.clone()),
-            Self::Data(v) => plist::Value::Data(v.clone()),
-            Self::Array(v) => plist::Value::Array(v.iter().map(|item| item.to_plist()).collect()),
-            Self::Dictionary(v) => {
-                let mut dict = plist::Dictionary::new();
-                for (k, v) in v.into_iter() {
-                    dict.insert(k.clone(), v.to_plist());
-                }
-                plist::Value::Dictionary(dict)
-            }
-        }
-    }
-
-    pub fn to_value<T: Serialize>(value: &T) -> Self {
-        match plist::to_value(value) {
-            Ok(v) => Self::from(v),
-            Err(_) => panic!("oof"),
-        }
-    }
-
-    pub fn encode(&self) -> Result<Vec<u8>, XPCError> {
-        let mut buf = Vec::new();
-        buf.extend_from_slice(&0x42133742_u32.to_le_bytes());
-        buf.extend_from_slice(&0x00000005_u32.to_le_bytes());
-        self.encode_object(&mut buf)?;
-        Ok(buf)
-    }
-
-    fn encode_object(&self, buf: &mut Vec<u8>) -> Result<(), XPCError> {
-        match self {
-            XPCObject::Bool(val) => {
-                buf.extend_from_slice(&(XPCType::Bool as u32).to_le_bytes());
-                buf.push(if *val { 0 } else { 1 });
-                buf.extend_from_slice(&[0].repeat(3));
-            }
-            XPCObject::Dictionary(dict) => {
-                buf.extend_from_slice(&(XPCType::Dictionary as u32).to_le_bytes());
-                buf.extend_from_slice(&0_u32.to_le_bytes()); // represents l, no idea what this is.
-                buf.extend_from_slice(&(dict.len() as u32).to_le_bytes());
-                for (k, v) in dict {
-                    let padding = Self::calculate_padding(k.len() + 1);
-                    buf.extend_from_slice(k.as_bytes());
-                    buf.push(0);
-                    buf.extend_from_slice(&[0].repeat(padding));
-                    v.encode_object(buf)?;
-                }
-            }
-            XPCObject::Array(items) => {
-                buf.extend_from_slice(&(XPCType::Array as u32).to_le_bytes());
-                buf.extend_from_slice(&0_u32.to_le_bytes()); // represents l, no idea what this is.
-                buf.extend_from_slice(&(items.len() as u32).to_le_bytes());
-                for item in items {
-                    item.encode_object(buf)?;
-                }
-            }
-
-            XPCObject::Int64(num) => {
-                buf.extend_from_slice(&(XPCType::Int64 as u32).to_le_bytes());
-                buf.extend_from_slice(&num.to_le_bytes());
-            }
-            XPCObject::UInt64(num) => {
-                buf.extend_from_slice(&(XPCType::UInt64 as u32).to_le_bytes());
-                buf.extend_from_slice(&num.to_le_bytes());
-            }
-            XPCObject::String(item) => {
-                let l = item.len() + 1;
-                let padding = Self::calculate_padding(l);
-                buf.extend_from_slice(&(XPCType::String as u32).to_le_bytes());
-                buf.extend_from_slice(&(l as u32).to_le_bytes());
-                buf.extend_from_slice(item.as_bytes());
-                buf.push(0);
-                buf.extend_from_slice(&[0].repeat(padding));
-            }
-            XPCObject::Data(data) => {
-                let l = data.len();
-                let padding = Self::calculate_padding(l);
-                buf.extend_from_slice(&(XPCType::Data as u32).to_le_bytes());
-                buf.extend_from_slice(&(l as u32).to_le_bytes());
-                buf.extend_from_slice(data);
-                buf.extend_from_slice(&[0].repeat(padding));
-            }
-            XPCObject::Uuid(uuid) => {
-                buf.extend_from_slice(&(XPCType::Uuid as u32).to_le_bytes());
-                buf.extend_from_slice(&16_u32.to_le_bytes());
-                buf.extend_from_slice(uuid.as_bytes());
-            }
-        }
-        Ok(())
-    }
-
-    pub fn decode(buf: &[u8]) -> Result<Self, XPCError> {
-        let magic = u32::from_le_bytes(buf[0..4].try_into()?);
-        if magic != 0x42133742 {
-            Err("Invalid magic for XPCObject")?
-        }
-
-        let version = u32::from_le_bytes(buf[4..8].try_into()?);
-        if version != 0x00000005 {
-            Err("Unexpected version for XPCObject")?
-        }
-
-        Self::decode_object(&mut Cursor::new(&buf[8..]))
-    }
-
-    fn decode_object(mut cursor: &mut Cursor<&[u8]>) -> Result<Self, XPCError> {
-        let mut buf_32: [u8; 4] = Default::default();
-        cursor.read_exact(&mut buf_32)?;
-        let xpc_type = u32::from_le_bytes(buf_32);
-        let xpc_type: XPCType = xpc_type.try_into()?;
-        match xpc_type {
-            XPCType::Dictionary => {
-                let mut ret = IndexMap::new();
-
-                cursor.read_exact(&mut buf_32)?;
-                let _l = u32::from_le_bytes(buf_32);
-                cursor.read_exact(&mut buf_32)?;
-                let num_entries = u32::from_le_bytes(buf_32);
-                for _i in 0..num_entries {
-                    let mut key_buf = Vec::new();
-                    BufRead::read_until(&mut cursor, 0, &mut key_buf)?;
-                    let key = CString::from_vec_with_nul(key_buf)?.to_str()?.to_string();
-                    let padding = Self::calculate_padding(key.len() + 1);
-
-                    BufRead::consume(&mut cursor, padding);
-                    ret.insert(key, Self::decode_object(cursor)?);
-                }
-                Ok(XPCObject::Dictionary(ret))
-            }
-            XPCType::Array => {
-                cursor.read_exact(&mut buf_32)?;
-                let _l = u32::from_le_bytes(buf_32);
-                cursor.read_exact(&mut buf_32)?;
-                let num_entries = u32::from_le_bytes(buf_32);
-
-                let mut ret = Vec::new();
-                for _i in 0..num_entries {
-                    ret.push(Self::decode_object(cursor)?);
-                }
-                Ok(XPCObject::Array(ret))
-            }
-            XPCType::Int64 => {
-                let mut buf: [u8; 8] = Default::default();
-                cursor.read_exact(&mut buf)?;
-                Ok(XPCObject::Int64(i64::from_le_bytes(buf)))
-            }
-            XPCType::UInt64 => {
-                let mut buf: [u8; 8] = Default::default();
-                cursor.read_exact(&mut buf)?;
-                Ok(XPCObject::UInt64(u64::from_le_bytes(buf)))
-            }
-            XPCType::String => {
-                // 'l' includes utf8 '\0' character.
-                cursor.read_exact(&mut buf_32)?;
-                let l = u32::from_le_bytes(buf_32) as usize;
-                let padding = Self::calculate_padding(l);
-
-                let mut key_buf = vec![0; l];
-                cursor.read_exact(&mut key_buf)?;
-                let key = CString::from_vec_with_nul(key_buf)?.to_str()?.to_string();
-                BufRead::consume(&mut cursor, padding);
-                Ok(XPCObject::String(key))
-            }
-            XPCType::Bool => {
-                let mut buf: [u8; 4] = Default::default();
-                cursor.read_exact(&mut buf)?;
-                Ok(XPCObject::Bool(buf[0] != 0))
-            }
-            XPCType::Data => {
-                cursor.read_exact(&mut buf_32)?;
-                let l = u32::from_le_bytes(buf_32) as usize;
-                let padding = Self::calculate_padding(l);
-
-                let mut data = vec![0; l];
-                cursor.read_exact(&mut data)?;
-                BufRead::consume(&mut cursor, padding);
-                Ok(XPCObject::Data(data))
-            }
-            XPCType::Uuid => {
-                let mut data: [u8; 16] = Default::default();
-                cursor.read_exact(&mut data)?;
-                Ok(XPCObject::Uuid(uuid::Builder::from_bytes(data).into_uuid()))
-            }
-        }
-    }
-
-    pub fn as_dictionary(&self) -> Option<&Dictionary> {
-        match self {
-            XPCObject::Dictionary(dict) => Some(dict),
-            _ => None,
-        }
-    }
-
-    pub fn as_array(&self) -> Option<&Vec<Self>> {
-        match self {
-            XPCObject::Array(array) => Some(array),
-            _ => None,
-        }
-    }
-
-    pub fn as_string(&self) -> Option<&str> {
-        match self {
-            XPCObject::String(s) => Some(s),
-            _ => None,
-        }
-    }
-
-    pub fn as_bool(&self) -> Option<&bool> {
-        match self {
-            XPCObject::Bool(b) => Some(b),
-            _ => None,
-        }
-    }
-
-    pub fn as_signed_integer(&self) -> Option<i64> {
-        match self {
-            XPCObject::String(s) => s.parse().ok(),
-            XPCObject::Int64(v) => Some(*v),
-            _ => None,
-        }
-    }
-
-    pub fn as_unsigned_integer(&self) -> Option<u64> {
-        match self {
-            XPCObject::String(s) => s.parse().ok(),
-            XPCObject::UInt64(v) => Some(*v),
-            _ => None,
-        }
-    }
-
-    fn calculate_padding(len: usize) -> usize {
-        let c = ((len as f64) / 4.0).ceil();
-        (c * 4.0 - (len as f64)) as usize
-    }
-}
-
-impl From<Dictionary> for XPCObject {
-    fn from(value: Dictionary) -> Self {
-        XPCObject::Dictionary(value)
-    }
-}
-
-#[derive(Debug)]
-pub struct XPCMessage {
-    pub flags: u32,
-    pub message: Option<XPCObject>,
-    pub message_id: Option<u64>,
-}
-
-impl XPCMessage {
-    pub fn new(
-        flags: Option<XPCFlag>,
-        message: Option<XPCObject>,
-        message_id: Option<u64>,
-    ) -> XPCMessage {
-        XPCMessage {
-            flags: flags.unwrap_or(XPCFlag::AlwaysSet).into(),
-            message,
-            message_id,
-        }
-    }
-
-    pub fn decode(data: &[u8]) -> Result<XPCMessage, XPCError> {
-        if data.len() < 24 {
-            Err("XPCMessage must be at least 24 bytes.")?
-        }
-
-        let magic = u32::from_le_bytes(data[0..4].try_into()?);
-        if magic != 0x29b00b92_u32 {
-            Err("XPCMessage magic is invalid.")?
-        }
-
-        let flags = u32::from_le_bytes(data[4..8].try_into()?);
-        let body_len = u64::from_le_bytes(data[8..16].try_into()?);
-        let message_id = u64::from_le_bytes(data[16..24].try_into()?);
-        if body_len + 24 > data.len().try_into()? {
-            Err("XPCMessage body length given is incorrect.")?
-        }
-
-        // for some reason the above if check doesn't work ???
-        debug!("Body length {} : {}", body_len, data.len());
-        if body_len == 0 {
-            return Ok(XPCMessage {
-                flags,
-                message: None,
-                message_id: Some(message_id),
-            });
-        }
-        Ok(XPCMessage {
-            flags,
-            message: Some(XPCObject::decode(&data[24..24 + body_len as usize])?),
-            message_id: Some(message_id),
-        })
-    }
-
-    pub fn encode(self, message_id: u64) -> Result<Vec<u8>, XPCError> {
-        let mut out = 0x29b00b92_u32.to_le_bytes().to_vec();
-        out.extend_from_slice(&self.flags.to_le_bytes());
-        match self.message {
-            Some(message) => {
-                let body = message.encode()?;
-                out.extend_from_slice(&(body.len() as u64).to_le_bytes()); // body length
-                out.extend_from_slice(&message_id.to_le_bytes()); // messageId
-                out.extend_from_slice(&body);
-            }
-            _ => {
-                out.extend_from_slice(&0_u64.to_le_bytes());
-                out.extend_from_slice(&message_id.to_le_bytes());
-            }
-        }
-        Ok(out)
-    }
-}

idevice/src/services/xpc/http2/error.rs

@@ -1,62 +0,0 @@
-// DebianArch
-
-use std::{array::TryFromSliceError, error::Error, io, num::TryFromIntError};
-
-use tokio::sync::mpsc::error::SendError;
-
-#[derive(Debug)]
-pub enum Http2Error {
-    Io(io::Error),
-    SendError,
-    TryFromIntError(TryFromIntError),
-    TryFromSliceError(TryFromSliceError),
-    Custom(String),
-}
-
-impl From<io::Error> for Http2Error {
-    fn from(value: io::Error) -> Self {
-        Self::Io(value)
-    }
-}
-
-impl<T> From<SendError<T>> for Http2Error {
-    fn from(_: SendError<T>) -> Self {
-        Self::SendError
-    }
-}
-
-impl From<&str> for Http2Error {
-    fn from(value: &str) -> Self {
-        Self::Custom(value.to_string())
-    }
-}
-
-impl From<TryFromIntError> for Http2Error {
-    fn from(value: TryFromIntError) -> Self {
-        Self::TryFromIntError(value)
-    }
-}
-
-impl From<TryFromSliceError> for Http2Error {
-    fn from(value: TryFromSliceError) -> Self {
-        Self::TryFromSliceError(value)
-    }
-}
-
-impl std::fmt::Display for Http2Error {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "Http2Error({})",
-            match self {
-                Self::Io(io) => io.to_string(),
-                Self::SendError => "SendError".to_string(),
-                Self::TryFromIntError(e) => e.to_string(),
-                Self::TryFromSliceError(e) => e.to_string(),
-                Self::Custom(s) => s.clone(),
-            }
-        )
-    }
-}
-
-impl Error for Http2Error {}

idevice/src/services/xpc/http2/frame.rs

@@ -0,0 +1,186 @@
+// Jackson Coxson
+
+use crate::{IdeviceError, ReadWrite};
+use tokio::io::AsyncReadExt;
+
+pub trait HttpFrame {
+    fn serialize(&self) -> Vec<u8>;
+}
+
+pub enum Frame {
+    Settings(SettingsFrame),
+    WindowUpdate(WindowUpdateFrame),
+    Headers(HeadersFrame),
+    Data(DataFrame),
+}
+
+impl Frame {
+    pub async fn next(socket: &mut impl ReadWrite) -> Result<Self, IdeviceError> {
+        // Read the len of the frame
+        let mut buf = [0u8; 3];
+        socket.read_exact(&mut buf).await?;
+        let frame_len = u32::from_be_bytes([0x00, buf[0], buf[1], buf[2]]);
+
+        // Read the fields
+        let frame_type = socket.read_u8().await?;
+        let flags = socket.read_u8().await?;
+        let stream_id = socket.read_u32().await?;
+
+        let body = vec![0; frame_len as usize];
+        socket.read_exact(&mut buf).await?;
+
+        Ok(match frame_type {
+            0x00 => {
+                // data
+                Self::Data(DataFrame {
+                    stream_id,
+                    payload: body,
+                })
+            }
+            0x01 => {
+                // headers
+                Self::Headers(HeadersFrame { stream_id })
+            }
+            0x04 => {
+                // settings
+                let mut body = std::io::Cursor::new(body);
+                let mut settings = Vec::new();
+
+                while let Ok(setting_type) = body.read_u16().await {
+                    settings.push(match setting_type {
+                        0x03 => {
+                            let max_streams = body.read_u32().await?;
+                            Setting::MaxConcurrentStreams(max_streams)
+                        }
+                        0x04 => {
+                            let window_size = body.read_u32().await?;
+                            Setting::InitialWindowSize(window_size)
+                        }
+                        _ => {
+                            return Err(IdeviceError::UnknownHttpSetting(setting_type));
+                        }
+                    });
+                }
+                Self::Settings(SettingsFrame {
+                    settings,
+                    stream_id,
+                    flags,
+                })
+            }
+            0x08 => {
+                // window update
+                if body.len() != 4 {
+                    return Err(IdeviceError::UnexpectedResponse);
+                }
+
+                let window = u32::from_be_bytes([body[0], body[1], body[2], body[3]]);
+                Self::WindowUpdate(WindowUpdateFrame {
+                    increment_size: window,
+                    stream_id,
+                })
+            }
+            _ => {
+                return Err(IdeviceError::UnknownFrame(frame_type));
+            }
+        })
+    }
+}
+
+pub struct SettingsFrame {
+    pub settings: Vec<Setting>,
+    pub stream_id: u32,
+    pub flags: u8,
+}
+
+pub enum Setting {
+    MaxConcurrentStreams(u32),
+    InitialWindowSize(u32),
+}
+
+impl SettingsFrame {
+    pub fn ack(&mut self) {
+        self.flags = 1; // this seems to be the only http flag used
+    }
+}
+
+impl Setting {
+    fn serialize(&self) -> Vec<u8> {
+        match self {
+            Setting::MaxConcurrentStreams(m) => {
+                let mut res = vec![0x00, 0x03];
+                res.extend(m.to_be_bytes());
+                res
+            }
+            Setting::InitialWindowSize(s) => {
+                let mut res = vec![0x00, 0x04];
+                res.extend(s.to_be_bytes());
+                res
+            }
+        }
+    }
+}
+
+impl HttpFrame for SettingsFrame {
+    fn serialize(&self) -> Vec<u8> {
+        let settings = self
+            .settings
+            .iter()
+            .map(|x| x.serialize())
+            .collect::<Vec<Vec<u8>>>()
+            .concat();
+        let settings_len = (settings.len() as u32).to_be_bytes();
+        let mut res = vec![
+            settings_len[1],
+            settings_len[2],
+            settings_len[3],
+            0x04,
+            self.flags,
+        ];
+        res.extend(self.stream_id.to_be_bytes());
+        res.extend(settings);
+        res
+    }
+}
+
+pub struct WindowUpdateFrame {
+    pub increment_size: u32,
+    pub stream_id: u32,
+}
+
+impl HttpFrame for WindowUpdateFrame {
+    fn serialize(&self) -> Vec<u8> {
+        let mut res = vec![0x00, 0x00, 0x04, 0x08, 0x00]; // size, frame ID, flags
+        res.extend(self.stream_id.to_be_bytes());
+        res.extend(self.increment_size.to_be_bytes());
+        res
+    }
+}
+
+/// We don't actually care about this frame according to spec. This is just to open new channels.
+pub struct HeadersFrame {
+    pub stream_id: u32,
+}
+
+impl HttpFrame for HeadersFrame {
+    fn serialize(&self) -> Vec<u8> {
+        let mut res = vec![0x00, 0x00, 0x00, 0x01, 0x04];
+        res.extend(self.stream_id.to_be_bytes());
+        res
+    }
+}
+
+pub struct DataFrame {
+    pub stream_id: u32,
+    pub payload: Vec<u8>,
+}
+
+impl HttpFrame for DataFrame {
+    fn serialize(&self) -> Vec<u8> {
+        let mut res = (self.payload.len() as u32).to_be_bytes().to_vec();
+        res.remove(0); // only 3 significant bytes
+        res.extend([0x00, 0x00]); // frame type, flags
+        res.extend(self.stream_id.to_be_bytes());
+        res.extend(self.payload.clone());
+        res
+    }
+}

idevice/src/services/xpc/http2/h2.rs

@@ -1,256 +0,0 @@
-// DebianArch
-
-use std::collections::HashMap;
-
-use super::error::Http2Error;
-
-pub const HTTP2_MAGIC: &[u8; 24] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
-
-#[derive(Debug)]
-pub struct Frame {
-    pub stream_id: u32,
-    pub flags: u8,
-    pub frame_type: FrameType,
-
-    pub body: Vec<u8>,
-}
-
-impl Frame {
-    pub fn deserialize(buf: &[u8]) -> Result<Self, Http2Error> {
-        let mut len_buf = buf[0..3].to_vec();
-        len_buf.insert(0, 0);
-
-        let body_len = u32::from_be_bytes(len_buf.try_into().unwrap()) as usize;
-        let frame_type = buf[3];
-        let flags = buf[4];
-        let stream_id = u32::from_be_bytes(buf[5..9].try_into()?);
-        let body = buf[9..9 + body_len].to_vec();
-        Ok(Self {
-            stream_id,
-            flags,
-            frame_type: frame_type.into(),
-            body,
-        })
-    }
-}
-
-impl Framable for Frame {
-    fn serialize(&self) -> Vec<u8> {
-        let mut res = Vec::new();
-
-        let body_len = (self.body.len() as u32).to_be_bytes();
-        res.extend_from_slice(&[body_len[1], body_len[2], body_len[3]]); // [0..3]
-        res.push(self.frame_type.into()); // [3]
-        res.push(self.flags); // flag mask [4]
-        res.extend_from_slice(&self.stream_id.to_be_bytes()); // [4..8]
-        res.extend_from_slice(&self.body); // [9..9+len]
-        res
-    }
-}
-
-pub trait Framable: From<Frame> {
-    fn serialize(&self) -> Vec<u8>;
-}
-
-// Frame implementations:
-pub struct SettingsFrame {
-    frame: Frame,
-    pub settings: HashMap<u16, u32>,
-}
-
-impl SettingsFrame {
-    pub const MAX_CONCURRENT_STREAMS: u16 = 0x03;
-    pub const INITIAL_WINDOW_SIZE: u16 = 0x04;
-
-    pub const ACK: u8 = 0x01;
-    pub fn new(/*stream_id: u32, */ settings: HashMap<u16, u32>, flags: u8) -> Self {
-        let mut body = Vec::new();
-        for setting in settings.clone() {
-            body.extend_from_slice(&setting.0.to_be_bytes());
-            body.extend_from_slice(&setting.1.to_be_bytes());
-        }
-        Self {
-            frame: Frame {
-                /*stream_id*/ stream_id: 0,
-                flags,
-                frame_type: FrameType::Settings,
-                body,
-            },
-            settings,
-        }
-    }
-
-    pub fn ack(/*stream_id: u32*/) -> Self {
-        Self {
-            frame: Frame {
-                /*stream_id*/ stream_id: 0,
-                flags: Self::ACK,
-                frame_type: FrameType::Settings,
-                body: Vec::new(),
-            },
-            settings: HashMap::new(),
-        }
-    }
-}
-
-impl Framable for SettingsFrame {
-    fn serialize(&self) -> Vec<u8> {
-        self.frame.serialize()
-    }
-}
-
-impl From<Frame> for SettingsFrame {
-    fn from(value: Frame) -> Self {
-        let mut idx = 0;
-        let mut settings = HashMap::new();
-        while idx < value.body.len() {
-            let key = u16::from_be_bytes(value.body[idx..idx + 2].try_into().unwrap());
-            let value = u32::from_be_bytes(value.body[idx + 2..idx + 2 + 4].try_into().unwrap());
-            settings.insert(key, value);
-            idx += 2 + 4;
-        }
-        Self {
-            frame: value,
-            settings,
-        }
-    }
-}
-
-pub struct WindowUpdateFrame {
-    frame: Frame,
-}
-
-impl WindowUpdateFrame {
-    // the frame's stream identifier indicates the affected stream; in the latter, the value "0" indicates that the entire connection is the subject of the frame.
-    pub fn new(stream_id: u32, window_increment: u32) -> Self {
-        if window_increment == 0 {
-            panic!("PROTOCOL_ERROR");
-        }
-        Self {
-            frame: Frame {
-                stream_id,
-                flags: Default::default(),
-                frame_type: FrameType::WindowUpdate,
-                body: window_increment.to_be_bytes().to_vec(),
-            },
-        }
-    }
-}
-
-impl Framable for WindowUpdateFrame {
-    fn serialize(&self) -> Vec<u8> {
-        self.frame.serialize()
-    }
-}
-
-impl From<Frame> for WindowUpdateFrame {
-    fn from(value: Frame) -> Self {
-        Self { frame: value }
-    }
-}
-
-pub struct HeadersFrame {
-    frame: Frame,
-}
-
-impl HeadersFrame {
-    pub const END_HEADERS: u8 = 0x04;
-    pub const PADDED: u8 = 0x08;
-    pub const PRIORITY: u8 = 0x20;
-    pub fn new(stream_id: u32, flags: u8) -> Self {
-        Self {
-            frame: Frame {
-                stream_id,
-                flags,
-                frame_type: FrameType::Headers,
-                body: Vec::new(),
-            },
-        }
-    }
-}
-
-impl Framable for HeadersFrame {
-    fn serialize(&self) -> Vec<u8> {
-        if self.frame.flags & Self::PADDED == Self::PADDED {
-            unimplemented!("haven't added padding support !")
-        };
-
-        if self.frame.flags & Self::PRIORITY == Self::PRIORITY {
-            unimplemented!("haven't added priority support !")
-        };
-
-        //   let padding = 0; for 'PADDED' flag
-        // let priority_data = b""; // for PRIORITY flag
-        self.frame.serialize()
-    }
-}
-
-impl From<Frame> for HeadersFrame {
-    fn from(value: Frame) -> Self {
-        if value.flags & Self::PADDED == Self::PADDED {
-            unimplemented!("haven't added padding support !")
-        };
-
-        if value.flags & Self::PRIORITY == Self::PRIORITY {
-            unimplemented!("haven't added priority support !")
-        };
-        Self { frame: value }
-    }
-}
-
-pub struct DataFrame {
-    frame: Frame,
-}
-
-impl DataFrame {
-    // TODO: Handle padding flag.
-    pub fn new(stream_id: u32, data: Vec<u8>, flags: u8) -> Self {
-        Self {
-            frame: Frame {
-                stream_id,
-                flags,
-                frame_type: FrameType::Data,
-                body: data,
-            },
-        }
-    }
-}
-
-impl Framable for DataFrame {
-    fn serialize(&self) -> Vec<u8> {
-        self.frame.serialize()
-    }
-}
-
-impl From<Frame> for DataFrame {
-    fn from(value: Frame) -> Self {
-        Self { frame: value }
-    }
-}
-
-#[repr(u8)]
-#[derive(Clone, Copy, Debug)]
-pub enum FrameType {
-    Data = 0,
-    Headers = 1,
-    Priority = 2,
-    RstStream = 3,
-    Settings = 4,
-    PushPromise = 5,
-    Ping = 6,
-    GoAway = 7,
-    WindowUpdate = 8,
-    Continuation = 9,
-}
-
-impl From<FrameType> for u8 {
-    fn from(value: FrameType) -> Self {
-        value as u8
-    }
-}
-
-impl From<u8> for FrameType {
-    fn from(value: u8) -> Self {
-        unsafe { std::mem::transmute::<_, FrameType>(value) }
-    }
-}

idevice/src/services/xpc/http2/mod.rs

@@ -1,143 +1,115 @@
-// DebianArch
+// Jackson Coxson
 
-use async_recursion::async_recursion;
+use frame::HttpFrame;
-use error::Http2Error;
+use log::warn;
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
+use tokio::io::AsyncWriteExt;
 
-use tokio::{
+use crate::{IdeviceError, ReadWrite};
-    io::{AsyncReadExt, AsyncWriteExt},
-    sync::mpsc::{self, Receiver, Sender},
-};
 
-pub mod error;
+mod frame;
-pub mod h2;
+pub use frame::Setting;
 
-use h2::{
+const HTTP2_MAGIC: &[u8] = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n".as_bytes();
-    DataFrame, Framable, Frame, FrameType, HeadersFrame, SettingsFrame, WindowUpdateFrame,
-    HTTP2_MAGIC,
-};
 
-use crate::ReadWrite;
+pub struct Http2Client<R: ReadWrite> {
-
+    inner: R,
-pub type Channels = HashMap<u32, (Sender<Vec<u8>>, Receiver<Vec<u8>>)>;
+    cache: HashMap<u32, VecDeque<Vec<u8>>>,
-
-pub const INIT_STREAM: u32 = 0;
-pub const ROOT_CHANNEL: u32 = 1;
-pub const REPLY_CHANNEL: u32 = 3;
-
-pub struct Connection<R: ReadWrite> {
-    pub stream: R,
-    channels: Channels,
-    window_size: u32,
 }
 
-impl<R: ReadWrite> Connection<R> {
+impl<R: ReadWrite> Http2Client<R> {
-    pub async fn new(mut stream: R) -> Result<Self, Http2Error> {
+    /// Writes the magic and inits the caches
-        stream.write_all(HTTP2_MAGIC).await?;
+    pub async fn new(mut inner: R) -> Result<Self, IdeviceError> {
+        inner.write_all(HTTP2_MAGIC).await?;
+        inner.flush().await?;
         Ok(Self {
-            stream,
+            inner,
-            channels: HashMap::new(),
+            cache: HashMap::new(),
-            window_size: 1048576,
         })
     }
 
-    pub async fn send_frame<A: Framable>(&mut self, frame: A) -> Result<(), Http2Error> {
+    pub async fn set_settings(
-        let body = &frame.serialize();
-        if body.len() > self.window_size as usize {
-            panic!("we need to chunk it :D")
-        }
-        self.stream.write_all(body).await?;
-        Ok(())
-    }
-
-    pub async fn read_data(&mut self) -> Result<Vec<u8>, Http2Error> {
-        loop {
-            let frame = self.read_frame().await?;
-            match frame.frame_type {
-                FrameType::Data => {
-                    if frame.stream_id % 2 == 0 && !frame.body.is_empty() {
-                        let frame_len: u32 = frame.body.len().try_into()?;
-                        self.send_frame(WindowUpdateFrame::new(0, frame_len))
-                            .await?;
-                        self.send_frame(WindowUpdateFrame::new(frame.stream_id, frame_len))
-                            .await?;
-                    }
-                    match self.channels.get_mut(&frame.stream_id) {
-                        Some((sender, _receiver)) => {
-                            sender.send(frame.body.clone()).await?;
-                        }
-                        None => {
-                            let chan = mpsc::channel(100);
-                            chan.0.send(frame.body.clone()).await?;
-                            self.channels.insert(frame.stream_id, chan);
-                        }
-                    }
-                    return Ok(frame.body);
-                }
-                FrameType::GoAway | FrameType::RstStream => {
-                    let _last_streamid = u32::from_be_bytes(frame.body[0..4].try_into().unwrap());
-                    return Err("connection closed, bye")?;
-                }
-                FrameType::Settings => {
-                    let flags = frame.flags;
-                    let settings_frame: SettingsFrame = frame.into();
-                    if flags & SettingsFrame::ACK != SettingsFrame::ACK {
-                        self.send_frame(SettingsFrame::ack()).await?;
-                    }
-                    if let Some(&window_size) = settings_frame
-                        .settings
-                        .get(&SettingsFrame::INITIAL_WINDOW_SIZE)
-                    {
-                        self.window_size = window_size;
-                    }
-                }
-                _ => continue,
-            }
-        }
-    }
-
-    pub async fn read_frame(&mut self) -> Result<Frame, Http2Error> {
-        let mut length_buf = vec![0; 3];
-        self.stream.read_exact(&mut length_buf).await?;
-        length_buf.insert(0, 0);
-        let len = u32::from_be_bytes(length_buf.clone().try_into().unwrap()) as usize;
-        let mut rest = vec![0; 9 - 3 + len];
-        self.stream.read_exact(&mut rest).await?;
-
-        let mut content = vec![length_buf[1], length_buf[2], length_buf[3]];
-        content.extend_from_slice(&rest);
-        Frame::deserialize(&content)
-    }
-
-    // pub async fn multiplex_write(&mut self, stream_id: u32) -> Result<()> {}
-
-    // gets a Reader + Writer for a channel.
-    pub async fn write_streamid(
         &mut self,
+        settings: Vec<frame::Setting>,
         stream_id: u32,
-        data: Vec<u8>,
+    ) -> Result<(), IdeviceError> {
-    ) -> Result<(), Http2Error> {
+        let frame = frame::SettingsFrame {
-        // TODO: If we ever allow concurrent writes we must not always send 'END_HEADERS'.
+            settings,
-        self.send_frame(HeadersFrame::new(stream_id, HeadersFrame::END_HEADERS))
+            stream_id,
-            .await?;
+            flags: 0,
-        self.send_frame(DataFrame::new(stream_id, data, Default::default()))
+        }
-            .await?;
+        .serialize();
+        self.inner.write_all(&frame).await?;
         Ok(())
     }
 
-    #[async_recursion]
+    pub async fn window_update(
-    pub async fn read_streamid(&mut self, stream_id: u32) -> Result<Vec<u8>, Http2Error> {
+        &mut self,
-        match self.channels.get_mut(&stream_id) {
+        increment_size: u32,
-            Some((_sender, receiver)) => match receiver.try_recv().ok() {
+        stream_id: u32,
-                Some(data) => Ok(data),
+    ) -> Result<(), IdeviceError> {
-                None => {
+        let frame = frame::WindowUpdateFrame {
-                    self.read_data().await?;
+            increment_size,
-                    self.read_streamid(stream_id).await
+            stream_id,
-                }
+        }
-            },
+        .serialize();
+        self.inner.write_all(&frame).await?;
+        Ok(())
+    }
+
+    pub async fn open_stream(&mut self, stream_id: u32) -> Result<(), IdeviceError> {
+        self.cache.insert(stream_id, VecDeque::new());
+        let frame = frame::HeadersFrame { stream_id }.serialize();
+        self.inner.write_all(&frame).await?;
+        Ok(())
+    }
+
+    pub async fn read(&mut self, stream_id: u32) -> Result<Vec<u8>, IdeviceError> {
+        // See if we already have a cached message from another read
+        let c = match self.cache.get_mut(&stream_id) {
+            Some(c) => c,
             None => {
-                self.read_data().await?;
+                warn!("Requested stream ID is not in cache");
-                self.read_streamid(stream_id).await
+                return Err(IdeviceError::UninitializedStreamId);
+            }
+        };
+        if let Some(d) = c.pop_front() {
+            return Ok(d);
+        }
+
+        // handle packets until we get what we want
+        loop {
+            let frame = frame::Frame::next(&mut self.inner).await?;
+            match frame {
+                frame::Frame::Settings(settings_frame) => {
+                    if settings_frame.flags != 1 {
+                        // ack that
+                        let frame = frame::SettingsFrame {
+                            settings: Vec::new(),
+                            stream_id: settings_frame.stream_id,
+                            flags: 1,
+                        }
+                        .serialize();
+                        self.inner.write_all(&frame).await?;
+                    }
+                }
+                frame::Frame::Data(data_frame) => {
+                    if data_frame.stream_id == stream_id {
+                        return Ok(data_frame.payload);
+                    } else {
+                        let c = match self.cache.get_mut(&data_frame.stream_id) {
+                            Some(c) => c,
+                            None => {
+                                warn!("Received message for stream ID not in cache");
+                                continue;
+                            }
+                        };
+                        c.push_back(data_frame.payload);
+                    }
+                }
+                _ => {
+                    // do nothing, we shouldn't receive these frames
+                }
             }
         }
     }

idevice/src/services/xpc/mod.rs

@@ -1,290 +1,5 @@
-//! XPC (Cross-Process Communication) Implementation
+// Jackson Coxson
-//!
-//! Provides functionality for interacting with Apple's XPC protocol over HTTP/2,
-//! which is used for inter-process communication between iOS/macOS components.
 
-use std::collections::HashMap;
 mod http2;
 
-use crate::{IdeviceError, ReadWrite};
+pub trait XpcBackend {}
-use error::XPCError;
-use format::{XPCFlag, XPCMessage, XPCObject};
-use http2::h2::{SettingsFrame, WindowUpdateFrame};
-use log::{debug, warn};
-use serde::Deserialize;
-
-pub mod error;
-mod format;
-
-/// Represents an XPC connection to a device with available services
-pub struct XPCDevice<R: ReadWrite> {
-    /// The underlying XPC connection
-    pub connection: XPCConnection<R>,
-    /// Map of available XPC services by name
-    pub services: HashMap<String, XPCService>,
-}
-
-/// Describes an available XPC service
-#[derive(Debug, Clone, Deserialize)]
-pub struct XPCService {
-    /// Required entitlement to access this service
-    pub entitlement: String,
-    /// Port number where the service is available
-    pub port: u16,
-    /// Whether the service uses remote XPC
-    pub uses_remote_xpc: bool,
-    /// Optional list of supported features
-    pub features: Option<Vec<String>>,
-    /// Optional service version number
-    pub service_version: Option<i64>,
-}
-
-/// Manages an active XPC connection over HTTP/2
-pub struct XPCConnection<R: ReadWrite> {
-    pub(crate) inner: http2::Connection<R>,
-    root_message_id: u64,
-    reply_message_id: u64,
-}
-
-impl<R: ReadWrite> XPCDevice<R> {
-    /// Creates a new XPC device connection
-    ///
-    /// # Arguments
-    /// * `stream` - The underlying transport stream
-    ///
-    /// # Returns
-    /// A connected XPCDevice instance with discovered services
-    ///
-    /// # Errors
-    /// Returns `IdeviceError` if:
-    /// - The connection fails
-    /// - The service discovery response is malformed
-    pub async fn new(stream: R) -> Result<Self, IdeviceError> {
-        let mut connection = XPCConnection::new(stream).await?;
-
-        // Read initial services message
-        let data = connection.read_message(http2::ROOT_CHANNEL).await?;
-
-        let data = match data.message {
-            Some(d) => match d
-                .as_dictionary()
-                .and_then(|x| x.get("Services"))
-                .and_then(|x| x.as_dictionary())
-            {
-                Some(d) => d.to_owned(),
-                None => return Err(IdeviceError::UnexpectedResponse),
-            },
-            None => return Err(IdeviceError::UnexpectedResponse),
-        };
-
-        // Parse available services
-        let mut services = HashMap::new();
-        for (name, service) in data.into_iter() {
-            match service.as_dictionary() {
-                Some(service) => {
-                    let entitlement = match service.get("Entitlement").and_then(|x| x.as_string()) {
-                        Some(e) => e.to_string(),
-                        None => {
-                            warn!("Service did not contain entitlement string");
-                            continue;
-                        }
-                    };
-                    let port = match service
-                        .get("Port")
-                        .and_then(|x| x.as_string())
-                        .and_then(|x| x.parse::<u16>().ok())
-                    {
-                        Some(e) => e,
-                        None => {
-                            warn!("Service did not contain port string");
-                            continue;
-                        }
-                    };
-                    let uses_remote_xpc = match service
-                        .get("Properties")
-                        .and_then(|x| x.as_dictionary())
-                        .and_then(|x| x.get("UsesRemoteXPC"))
-                        .and_then(|x| x.as_bool())
-                    {
-                        Some(e) => e.to_owned(),
-                        None => false, // default is false
-                    };
-
-                    let features = service
-                        .get("Properties")
-                        .and_then(|x| x.as_dictionary())
-                        .and_then(|x| x.get("Features"))
-                        .and_then(|x| x.as_array())
-                        .map(|f| {
-                            f.iter()
-                                .filter_map(|x| x.as_string())
-                                .map(|x| x.to_string())
-                                .collect::<Vec<String>>()
-                        });
-
-                    let service_version = service
-                        .get("Properties")
-                        .and_then(|x| x.as_dictionary())
-                        .and_then(|x| x.get("ServiceVersion"))
-                        .and_then(|x| x.as_signed_integer())
-                        .map(|e| e.to_owned());
-
-                    services.insert(
-                        name,
-                        XPCService {
-                            entitlement,
-                            port,
-                            uses_remote_xpc,
-                            features,
-                            service_version,
-                        },
-                    );
-                }
-                None => {
-                    warn!("Service is not a dictionary!");
-                    continue;
-                }
-            }
-        }
-
-        Ok(Self {
-            connection,
-            services,
-        })
-    }
-
-    /// Consumes the device and returns the underlying transport stream
-    pub fn into_inner(self) -> R {
-        self.connection.inner.stream
-    }
-}
-
-impl<R: ReadWrite> XPCConnection<R> {
-    /// Channel ID for root messages
-    pub const ROOT_CHANNEL: u32 = http2::ROOT_CHANNEL;
-    /// Channel ID for reply messages
-    pub const REPLY_CHANNEL: u32 = http2::REPLY_CHANNEL;
-    /// Initial stream ID for HTTP/2 connection
-    const INIT_STREAM: u32 = http2::INIT_STREAM;
-
-    /// Establishes a new XPC connection
-    ///
-    /// # Arguments
-    /// * `stream` - The underlying transport stream
-    ///
-    /// # Returns
-    /// A connected XPCConnection instance
-    ///
-    /// # Errors
-    /// Returns `XPCError` if the connection handshake fails
-    pub async fn new(stream: R) -> Result<Self, XPCError> {
-        let mut client = http2::Connection::new(stream).await?;
-
-        // Configure HTTP/2 settings
-        client
-            .send_frame(SettingsFrame::new(
-                [
-                    (SettingsFrame::MAX_CONCURRENT_STREAMS, 100),
-                    (SettingsFrame::INITIAL_WINDOW_SIZE, 1048576),
-                ]
-                .into_iter()
-                .collect(),
-                Default::default(),
-            ))
-            .await?;
-
-        // Update window size
-        client
-            .send_frame(WindowUpdateFrame::new(Self::INIT_STREAM, 983041))
-            .await?;
-
-        let mut xpc_client = Self {
-            inner: client,
-            root_message_id: 1,
-            reply_message_id: 1,
-        };
-
-        // Perform XPC handshake
-        xpc_client
-            .send_recv_message(
-                Self::ROOT_CHANNEL,
-                XPCMessage::new(
-                    Some(XPCFlag::AlwaysSet),
-                    Some(XPCObject::Dictionary(Default::default())),
-                    None,
-                ),
-            )
-            .await?;
-
-        xpc_client
-            .send_recv_message(
-                Self::REPLY_CHANNEL,
-                XPCMessage::new(
-                    Some(XPCFlag::InitHandshake | XPCFlag::AlwaysSet),
-                    None,
-                    None,
-                ),
-            )
-            .await?;
-
-        xpc_client
-            .send_recv_message(
-                Self::ROOT_CHANNEL,
-                XPCMessage::new(Some(XPCFlag::Custom(0x201)), None, None),
-            )
-            .await?;
-
-        Ok(xpc_client)
-    }
-
-    /// Sends a message and waits for the response
-    ///
-    /// # Arguments
-    /// * `stream_id` - The channel/stream to use
-    /// * `message` - The XPC message to send
-    ///
-    /// # Returns
-    /// The response message
-    pub async fn send_recv_message(
-        &mut self,
-        stream_id: u32,
-        message: XPCMessage,
-    ) -> Result<XPCMessage, XPCError> {
-        self.send_message(stream_id, message).await?;
-        self.read_message(stream_id).await
-    }
-
-    /// Sends an XPC message without waiting for a response
-    pub async fn send_message(
-        &mut self,
-        stream_id: u32,
-        message: XPCMessage,
-    ) -> Result<(), XPCError> {
-        self.inner
-            .write_streamid(stream_id, message.encode(self.root_message_id)?)
-            .await?;
-        Ok(())
-    }
-
-    /// Reads an XPC message from the specified stream
-    pub async fn read_message(&mut self, stream_id: u32) -> Result<XPCMessage, XPCError> {
-        let mut buf = self.inner.read_streamid(stream_id).await?;
-        loop {
-            match XPCMessage::decode(&buf) {
-                Ok(decoded) => {
-                    debug!("Decoded message: {:?}", decoded);
-                    match stream_id {
-                        1 => self.root_message_id += 1,
-                        3 => self.reply_message_id += 1,
-                        _ => {}
-                    }
-                    return Ok(decoded);
-                }
-                Err(err) => {
-                    log::warn!("Error decoding message: {:?}", err);
-                    buf.extend_from_slice(&self.inner.read_streamid(stream_id).await?);
-                }
-            }
-        }
-    }
-}