From d645bfa7eb77f74c8a236d348d55c82e1b9b5891 Mon Sep 17 00:00:00 2001 From: Sachymetsu Date: Thu, 23 Jul 2026 13:24:37 +0000 Subject: [PATCH] Custom stream with COBS & encryption --- Cargo.lock | 40 ++++++++++++++++++++++++++++++++++++++-- Cargo.toml | 5 ++++- src/main.rs | 154 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------------------------------- water-pod-common/Cargo.toml | 2 +- water-pod-common/src/lib.rs | 32 +++++++++++++++++++++----------- 5 file(s) changed, 173 insertion(s)(+), 60 deletion(s)(-) diff --git a/Cargo.lock b/Cargo.lock --- a/Cargo.lock +++ b/Cargo.lock @@ -194,6 +194,17 @@ 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" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -214,6 +225,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fea6d1b751c55bd9c0dda7d4ff752074e98f4765ae969664648bd193bb326d15" 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", ] @@ -584,7 +605,7 @@ "portable-atomic", "postcard", "postcard-schema", - "rand", + "rand 0.9.5", "rand_core 0.9.5", "serde", "tokio", @@ -1314,6 +1335,17 @@ ] [[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" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -1823,6 +1855,7 @@ name = "water-pod" version = "0.1.0" dependencies = [ + "aead", "bluer", "env_logger", "ergot", @@ -1831,7 +1864,9 @@ "getrandom 0.4.3", "jiff", "log", - "rand", + "postcard", + "rand 0.10.2", + "serde", "tokio", "water-pod-common", "wharrgarbl", @@ -1844,6 +1879,7 @@ dependencies = [ "aead", "bbqueue", + "cobs 0.5.1", "defmt", "ergot", "jiff", diff --git a/Cargo.toml b/Cargo.toml --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ 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 @@ 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 --- a/src/main.rs +++ b/src/main.rs @@ -1,49 +1,28 @@ mod scan; -use std::{sync::Arc, 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 std::{ + io::{Cursor, ErrorKind}, + time::Duration, }; -use water_pod_common::SyncEndpoint; + +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}, +}; 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 @@ 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]; - register_controller_stream(stack.clone(), reader, writer, stack_queue, None, None) + 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"); + + 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)?; - endpoint_server(&stack).await; + log::info!( + "RESPONSE SIZE: {}, VALID? {}", + response.len(), + response.len() <= send_mtu as usize + ); + + let mut response = Cursor::new(response); + + 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 --- a/water-pod-common/Cargo.toml +++ b/water-pod-common/Cargo.toml @@ -16,7 +16,7 @@ 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 --- 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 @@ #[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 @@ 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 @@ 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 @@ 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 @@ 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 @@ } } -#[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")] -- tangled.sh