diff --git a/Cargo.lock b/Cargo.lock index 3bc0528..706d78b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -193,6 +193,17 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core 0.10.1", +] + [[package]] name = "cmov" version = "0.5.4" @@ -217,6 +228,16 @@ dependencies = [ "thiserror", ] +[[package]] +name = "cobs" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd93fd2c1b27acd030440c9dbd9d14c1122aad622374fe05a670b67a4bc034be" +dependencies = [ + "heapless 0.9.3", + "thiserror", +] + [[package]] name = "cobs-acc" version = "0.1.0" @@ -584,7 +605,7 @@ dependencies = [ "portable-atomic", "postcard", "postcard-schema", - "rand", + "rand 0.9.5", "rand_core 0.9.5", "serde", "tokio", @@ -1313,6 +1334,17 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", +] + [[package]] name = "rand_chacha" version = "0.9.0" @@ -1823,6 +1855,7 @@ dependencies = [ name = "water-pod" version = "0.1.0" dependencies = [ + "aead", "bluer", "env_logger", "ergot", @@ -1831,7 +1864,9 @@ dependencies = [ "getrandom 0.4.3", "jiff", "log", - "rand", + "postcard", + "rand 0.10.2", + "serde", "tokio", "water-pod-common", "wharrgarbl", @@ -1844,6 +1879,7 @@ version = "0.1.0" dependencies = [ "aead", "bbqueue", + "cobs 0.5.1", "defmt", "ergot", "jiff", diff --git a/Cargo.toml b/Cargo.toml index f6238c3..6bc7294 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ authors.workspace = true edition.workspace = true [dependencies] +aead = { version = "0.6.0", features = ["alloc"] } water-pod-common = { path = "./water-pod-common" } tokio = { version = "1", features = ["io-std", "io-util", "net", "rt", "macros", "signal", "sync"] } wharrgarbl = { git = "https://tangled.org/sachy.dev/wharrgarbl", package = "wharrgarbl" } @@ -25,5 +26,7 @@ winnow = "1" futures-lite = "2.6.1" jiff = "0.2.32" ergot = { git = "https://github.com/jamesmunns/ergot", features = ["tokio-std"] } -rand = { version = "0.9" } +rand = { version = "0.10" } futures-concurrency = "7.7.1" +postcard = { version = "1", default-features = false, features = ["alloc"] } +serde = { version = "1", default-features = false, features = ["alloc"] } diff --git a/src/main.rs b/src/main.rs index 40964a0..3128363 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,50 +1,29 @@ mod scan; -use std::{sync::Arc, time::Duration}; +use std::{ + io::{Cursor, ErrorKind}, + time::Duration, +}; -use bluer::l2cap::{Socket, SocketAddr}; -use ergot::{ - exports::bbqueue::{BBQueue, traits::storage::BoxedSlice}, - interface_manager::utils::std::StdQueue, - toolkits::tokio_stream::{self, EdgeStack, register_controller_stream}, +use bluer::l2cap::{SocketAddr, Stream, stream::OwnedReadHalf}; +use postcard::accumulator::{CobsAccumulator, FeedResult}; +use rand::rng; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + time::sleep, +}; +use water_pod_common::{Request, Response}; +use wharrgarbl::{ + Neko128, NekoServerHandshake128, + transport::{RecvState, SendState}, }; -use water_pod_common::SyncEndpoint; const SPSM_L2CAP_WATER: u16 = 0x0089; -const MAX_ERGOT_PACKET_SIZE: u16 = 2048; -const TX_BUFFER_SIZE: usize = 8192; - -async fn endpoint_server(stack: &EdgeStack) { - let server = stack - .endpoints() - .heap_bounded_server::(16, None); - - let server = core::pin::pin!(server); - - let mut hdl = server.attach(); - - hdl.serve(async |a| { - let mut b = a.clone(); - - b.set_t1(); - tokio::time::sleep(Duration::from_millis(1)).await; - b.set_t2(); - - b - }) - .await - .ok(); -} - #[tokio::main(flavor = "current_thread")] async fn main() -> Result<(), std::io::Error> { env_logger::init(); - let stack_queue: StdQueue = - Arc::new(BBQueue::new_with_storage(BoxedSlice::new(TX_BUFFER_SIZE))); - let stack = tokio_stream::new_controller_stack(&stack_queue, MAX_ERGOT_PACKET_SIZE); - log::info!("Starting session"); let session = bluer::Session::new().await?; @@ -62,21 +41,106 @@ async fn main() -> Result<(), std::io::Error> { if let Some(device) = scan::scan_for_water(&adapter).await? { let addr_type = device.address_type().await?; let target_sa = SocketAddr::new(device.address(), addr_type, SPSM_L2CAP_WATER); - let conn = Socket::new_stream()?; - conn.set_recv_mtu(2510)?; - conn.set_recv_buffer(2510 * 10)?; - let conn = conn.connect(target_sa).await?; + log::info!("Connecting..."); + let conn = Stream::connect(target_sa).await?; + + let server = NekoServerHandshake128::new(None); + let send_mtu = conn.as_ref().send_mtu()?; + + log::info!("SETTINGS: {:?}", conn.as_ref().conn_info()); log::info!("Stream Recv MTU: {:?}", conn.as_ref().recv_mtu()); + log::info!("Stream Send MTU: {:?}", send_mtu); log::info!("Stream PHY: {:?}", conn.as_ref().phy()); - let (reader, writer) = conn.into_split(); + let (mut reader, mut writer) = conn.into_split(); + let mut raw_buf = vec![0u8; 251]; + + let mut pair = cobs_read(&mut reader, &mut raw_buf).await?; + + let mut rng = rng(); + + let (transport, pair) = Response::respond_to_pairing(server, &mut rng, &mut pair) + .map_err(std::io::Error::other)?; + + let mut pair = Cursor::new(pair); + + writer.write_all_buf(&mut pair).await?; + writer.flush().await?; + log::info!("SUCCESSFULLY PAIRED"); - register_controller_stream(stack.clone(), reader, writer, stack_queue, None, None) + let (mut send_state, mut recv_state) = transport.split(); + + let req = cobs_read(&mut reader, &mut raw_buf).await?; + + log::info!("RESPONDING!"); + + let response = req_handler(req, &mut send_state, &mut recv_state) .await - .unwrap(); + .map_err(std::io::Error::other)?; + + log::info!( + "RESPONSE SIZE: {}, VALID? {}", + response.len(), + response.len() <= send_mtu as usize + ); + + let mut response = Cursor::new(response); - endpoint_server(&stack).await; + writer.write_all_buf(&mut response).await?; + writer.flush().await?; + log::info!("SUCCESSFULLY SYNCED"); + + writer.shutdown().await?; } Ok(()) } + +async fn req_handler<'a>( + request: Vec, + send_state: &mut SendState<'a, Neko128>, + recv_state: &mut RecvState<'a, Neko128>, +) -> aead::Result> { + let request = Request::from_bytes(request, recv_state)?; + + match request { + Request::TimeSync(mut sync) => { + sync.set_t1(); + sleep(Duration::from_millis(1)).await; + sync.set_t2(); + + Response::TimeSync(sync).to_bytes(send_state) + } + _ => Response::Error.to_bytes(send_state), + } +} + +async fn cobs_read(input: &mut OwnedReadHalf, raw_buf: &mut [u8]) -> std::io::Result> { + let mut cobs_buf = Box::new(CobsAccumulator::<251>::new()); + + loop { + let ct = input.read(raw_buf).await?; + + if ct == 0 { + break; + } + + let buf = &raw_buf[..ct]; + let mut window = &buf[..]; + + 'cobs: while !window.is_empty() { + window = match cobs_buf.feed::>(&window) { + FeedResult::Consumed => break 'cobs, + FeedResult::OverFull(new_wind) => new_wind, + FeedResult::DeserError(new_wind) => new_wind, + FeedResult::Success { data, .. } => { + // Do something with `data: MyData` here. + + return Ok(data); + } + }; + } + } + + return Err(ErrorKind::InvalidData.into()); +} diff --git a/water-pod-common/Cargo.toml b/water-pod-common/Cargo.toml index 7bd2a53..3953bc7 100644 --- a/water-pod-common/Cargo.toml +++ b/water-pod-common/Cargo.toml @@ -16,7 +16,7 @@ tokio = { version = "1", default-features = false, optional = true } bbqueue = "0.7.0" postcard-schema = { version = "0.2.5", features = ["alloc"] } ergot = { git = "https://github.com/jamesmunns/ergot" } - +cobs = { version = "0.5.1", default-features = false, features = ["alloc"] } [features] default = ["server"] diff --git a/water-pod-common/src/lib.rs b/water-pod-common/src/lib.rs index 3379bce..af99772 100644 --- a/water-pod-common/src/lib.rs +++ b/water-pod-common/src/lib.rs @@ -2,7 +2,7 @@ extern crate alloc; -use alloc::{boxed::Box, vec::Vec}; +use alloc::vec::Vec; use ergot::{endpoint, topic}; use jiff::{SignedDuration, Timestamp}; use rand_core::CryptoRng; @@ -17,15 +17,13 @@ endpoint!(SyncEndpoint, TimeSyncMs, TimeSyncMs, "req/sync"); #[derive(Debug, serde::Serialize, serde::Deserialize, postcard_schema::Schema)] #[cfg_attr(feature = "defmt", derive(defmt::Format))] pub enum Request { - Pair(Vec), TimeSync(TimeSyncMs), FetchCommands, } -#[derive(Debug, serde::Serialize, serde::Deserialize)] +#[derive(Debug, serde::Serialize, serde::Deserialize, postcard_schema::Schema)] #[cfg_attr(feature = "defmt", derive(defmt::Format))] pub enum Response { - Pair(Box<[u8]>), TimeSync(TimeSyncMs), FetchCommands(Command), Error, @@ -36,8 +34,9 @@ impl Request { client: &mut NekoClientHandshake128, rng: &mut impl CryptoRng, buf: &mut Vec, - ) -> aead::Result<()> { - client.send(rng, buf) + ) -> aead::Result> { + client.send(rng, buf)?; + Ok(cobs::encode_vec_including_sentinels(buf)) } pub fn finish_pairing( @@ -52,7 +51,7 @@ impl Request { send.encrypt(&mut payload, b"water-pod")?; - Ok(payload) + Ok(cobs::encode_vec_including_sentinels(&payload)) } pub fn from_bytes<'a>( @@ -69,8 +68,10 @@ impl Response { mut server: NekoServerHandshake128, rng: &mut impl CryptoRng, buf: &mut Vec, - ) -> aead::Result> { - server.respond(rng, buf) + ) -> aead::Result<(AeadTransport, Vec)> { + let transport = server.respond(rng, buf)?; + let encoded = cobs::encode_vec_including_sentinels(buf); + Ok((transport, encoded)) } pub fn to_bytes<'b>(self, send: &mut SendState<'b, Neko128>) -> aead::Result> { @@ -78,7 +79,7 @@ impl Response { send.encrypt(&mut payload, b"water-pod")?; - Ok(payload) + Ok(cobs::encode_vec_including_sentinels(&payload)) } pub fn from_bytes<'a>( @@ -132,7 +133,16 @@ impl TimeSyncMs { } } -#[derive(Debug, serde::Serialize, serde::Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[derive( + Debug, + serde::Serialize, + serde::Deserialize, + PartialEq, + Eq, + PartialOrd, + Ord, + postcard_schema::Schema, +)] #[cfg_attr(feature = "defmt", derive(defmt::Format))] pub struct Command { #[serde(with = "postcard::fixint::le")]