From 961431d1e83d5c93bd5fe6773464bbff0c332959 Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Mon, 26 Jan 2026 03:33:53 +0300 Subject: [PATCH] sockudo --- .gitignore | 2 +- Cargo.lock | 237 ++++++++++++++++++++++++++++++++++-------- tapped/Cargo.toml | 11 +- tapped/src/channel.rs | 76 +++++++++----- tapped/src/error.rs | 2 +- tapped/src/types.rs | 27 ++--- 6 files changed, 269 insertions(+), 86 deletions(-) diff --git a/.gitignore b/.gitignore index c5e6e9d..eb80a52 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,4 @@ documents.txt publications.txt target tap-example* -test.db* +test.db* \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 20ad7c6..68645e8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -134,6 +134,28 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "aws-lc-rs" +version = "1.15.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b7b6141e96a8c160799cc2d5adecd5cbbe5054cb8c7c4af53da0f83bb7ad256" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.37.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c34dda4df7017c8db52132f0f8a2e0f8161649d15723ed63fc00c82d0f2081a" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", +] + [[package]] name = "base-x" version = "0.2.11" @@ -168,6 +190,12 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + [[package]] name = "bitflags" version = "2.10.0" @@ -275,6 +303,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "755d2fce177175ffca841e9a06afdb2c4ab0f593d53b4dee48147dfaade85932" dependencies = [ "find-msvc-tools", + "jobserver", + "libc", "shlex", ] @@ -354,6 +384,15 @@ dependencies = [ "unsigned-varint", ] +[[package]] +name = "cmake" +version = "0.1.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75443c44cd6b379beb8c5b45d85d0773baf31cce901fe7bb252f4eff3008ef7d" +dependencies = [ + "cc", +] + [[package]] name = "cobs" version = "0.3.0" @@ -644,6 +683,12 @@ dependencies = [ "syn 2.0.114", ] +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "dyn-clone" version = "1.0.20" @@ -744,6 +789,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "fastrand" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" + [[package]] name = "ff" version = "0.13.1" @@ -768,6 +819,7 @@ checksum = "b375d6465b98090a5f25b1c7703f3859783755aa9a80433b36e0379a3ec2f369" dependencies = [ "crc32fast", "miniz_oxide", + "zlib-rs", ] [[package]] @@ -785,6 +837,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futures-channel" version = "0.3.31" @@ -831,7 +889,6 @@ checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" dependencies = [ "futures-core", "futures-macro", - "futures-sink", "futures-task", "pin-project-lite", "pin-utils", @@ -1087,7 +1144,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2", + "socket2 0.6.1", "system-configuration", "tokio", "tower-service", @@ -1268,6 +1325,16 @@ dependencies = [ "rustversion", ] +[[package]] +name = "io-uring" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "595a0399f411a508feb2ec1e970a4a30c249351e30208960d58298de8660b0e5" +dependencies = [ + "bitflags 1.3.2", + "libc", +] + [[package]] name = "ipld-core" version = "0.4.2" @@ -1412,6 +1479,16 @@ dependencies = [ "syn 2.0.114", ] +[[package]] +name = "jobserver" +version = "0.1.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33" +dependencies = [ + "getrandom 0.3.4", + "libc", +] + [[package]] name = "js-sys" version = "0.3.85" @@ -1664,6 +1741,16 @@ dependencies = [ "sha2", ] +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + [[package]] name = "parking_lot_core" version = "0.9.12" @@ -1844,7 +1931,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2", + "socket2 0.6.1", "thiserror 2.0.18", "tokio", "tracing", @@ -1881,7 +1968,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2", + "socket2 0.6.1", "tracing", "windows-sys 0.60.2", ] @@ -1951,7 +2038,7 @@ version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags", + "bitflags 2.10.0", ] [[package]] @@ -2095,6 +2182,8 @@ version = "0.23.36" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c665f33d38cea657d9614f766881e4d510e0eda4239891eea56b4cadcf01801b" dependencies = [ + "aws-lc-rs", + "log", "once_cell", "ring", "rustls-pki-types", @@ -2131,6 +2220,7 @@ version = "0.103.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d7df23109aa6c1567d1c575b9952556388da57401e4ace1d15f79eedad0d8f53" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -2207,7 +2297,7 @@ version = "3.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b3297343eaf830f66ede390ea39da1d462b6b0c1b000f420d0a83f898bbbe6ef" dependencies = [ - "bitflags", + "bitflags 2.10.0", "core-foundation 0.10.1", "core-foundation-sys", "libc", @@ -2416,6 +2506,12 @@ version = "0.3.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e320a6c5ad31d271ad523dcf3ad13e2767ad8b1cb8f047f75a8aeaf8da139da2" +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + [[package]] name = "slab" version = "0.4.11" @@ -2438,6 +2534,26 @@ dependencies = [ "serde_core", ] +[[package]] +name = "socket2" +version = "0.4.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7916fc008ca5542385b89a3d3ce689953c143e9304a9bf8beec1de48994c0d" +dependencies = [ + "libc", + "winapi", +] + +[[package]] +name = "socket2" +version = "0.5.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + [[package]] name = "socket2" version = "0.6.1" @@ -2448,6 +2564,32 @@ dependencies = [ "windows-sys 0.60.2", ] +[[package]] +name = "sockudo-ws" +version = "1.7.4" +source = "git+https://github.com/Sockudo/sockudo-ws#234e2d0c97c31fceac2cc62b94b0470def542dde" +dependencies = [ + "base64", + "bytes", + "dashmap", + "fastrand", + "flate2", + "futures-core", + "futures-sink", + "httparse", + "parking_lot", + "pin-project-lite", + "rustls", + "rustls-native-certs", + "sha1", + "simdutf8", + "socket2 0.5.10", + "tokio", + "tokio-rustls", + "tokio-uring", + "tokio-util", +] + [[package]] name = "spin" version = "0.9.8" @@ -2573,7 +2715,7 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b" dependencies = [ - "bitflags", + "bitflags 2.10.0", "core-foundation 0.9.4", "system-configuration-sys", ] @@ -2593,16 +2735,20 @@ name = "tapped" version = "0.2.0" dependencies = [ "base64", + "bytes", "futures-util", "libc", "reqwest", + "rustls", + "rustls-native-certs", + "rustls-pki-types", "serde", "serde_json", + "sockudo-ws", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-rustls", "tracing", - "tungstenite", "url", ] @@ -2713,7 +2859,7 @@ dependencies = [ "mio", "pin-project-lite", "signal-hook-registry", - "socket2", + "socket2 0.6.1", "tokio-macros", "windows-sys 0.61.2", ] @@ -2740,19 +2886,17 @@ dependencies = [ ] [[package]] -name = "tokio-tungstenite" -version = "0.28.0" +name = "tokio-uring" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +checksum = "748482e3e13584a34664a710168ad5068e8cb1d968aa4ffa887e83ca6dd27967" dependencies = [ "futures-util", - "log", - "rustls", - "rustls-native-certs", - "rustls-pki-types", + "io-uring", + "libc", + "slab", + "socket2 0.4.10", "tokio", - "tokio-rustls", - "tungstenite", ] [[package]] @@ -2790,7 +2934,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8" dependencies = [ "async-compression", - "bitflags", + "bitflags 2.10.0", "bytes", "futures-core", "futures-util", @@ -2866,25 +3010,6 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" -[[package]] -name = "tungstenite" -version = "0.28.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" -dependencies = [ - "bytes", - "data-encoding", - "http", - "httparse", - "log", - "rand", - "rustls", - "rustls-pki-types", - "sha1", - "thiserror 2.0.18", - "utf-8", -] - [[package]] name = "typenum" version = "1.19.0" @@ -2934,12 +3059,6 @@ dependencies = [ "serde_derive", ] -[[package]] -name = "utf-8" -version = "0.7.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" - [[package]] name = "utf8_iter" version = "1.0.4" @@ -3070,6 +3189,28 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-core" version = "0.62.2" @@ -3417,6 +3558,12 @@ dependencies = [ "syn 2.0.114", ] +[[package]] +name = "zlib-rs" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40990edd51aae2c2b6907af74ffb635029d5788228222c4bb811e9351c0caad3" + [[package]] name = "zmij" version = "1.0.16" diff --git a/tapped/Cargo.toml b/tapped/Cargo.toml index b822649..42ea112 100644 --- a/tapped/Cargo.toml +++ b/tapped/Cargo.toml @@ -12,8 +12,12 @@ repository = "https://tangled.org/octet-stream.net/tapped" [dependencies] tokio = { version = "1", features = ["net", "process", "rt", "sync", "time"] } reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] } -tokio-tungstenite = { version = "0.28", features = ["rustls-tls-native-roots"] } -tungstenite = "0.28" +tokio-rustls = "0.26" +rustls = { version = "0.23", features = ["ring"] } +rustls-native-certs = "0.8" +rustls-pki-types = "1" +sockudo-ws = { git = "https://github.com/Sockudo/sockudo-ws", features = ["rustls-native-roots"] } +bytes = "1" serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["raw_value"] } thiserror = "2" @@ -23,5 +27,8 @@ tracing = "0.1" base64 = "0.22" libc = "0.2" +[target.'cfg(target_os = "linux")'.dependencies] +sockudo-ws = { git = "https://github.com/Sockudo/sockudo-ws", features = ["io-uring"] } + [dev-dependencies] tokio = { version = "1", features = ["macros", "rt-multi-thread"] } diff --git a/tapped/src/channel.rs b/tapped/src/channel.rs index 6775d08..5de4cd0 100644 --- a/tapped/src/channel.rs +++ b/tapped/src/channel.rs @@ -1,19 +1,17 @@ //! WebSocket event channel and receiver. -use futures_util::{SinkExt, StreamExt}; use serde::Serialize; use tokio::sync::mpsc; -use tokio_tungstenite::{connect_async, tungstenite::Message}; -use tungstenite::protocol::frame::Utf8Bytes; +use sockudo_ws::{Message, Http1, Config, Stream as WsTransportStream, SplitWriter, SplitReader}; +use sockudo_ws::client::WebSocketClient; +use bytes::Bytes; use url::Url; use crate::types::RawEvent; use crate::{Error, Event, Result}; -type WsStream = - tokio_tungstenite::WebSocketStream>; -type WsSink = futures_util::stream::SplitSink; -type WsSource = futures_util::stream::SplitStream; +type WsSink = SplitWriter>; +type WsSource = SplitReader>; /// Receiver for events from a tap WebSocket channel. /// @@ -29,7 +27,7 @@ pub struct EventReceiver { } struct EventWithAck { - event: Utf8Bytes, + event: Bytes, ack_tx: mpsc::Sender, } @@ -75,24 +73,51 @@ impl EventReceiver { } ws_url.set_path("/channel"); - if let Some(password) = admin_password { - ws_url - .set_username("admin") - .map_err(|_| Error::InvalidUrl("cannot set username".into()))?; - ws_url - .set_password(Some(password)) - .map_err(|_| Error::InvalidUrl("cannot set password".into()))?; - } + // User said no admin password needed for their tap, but we keep the parameter for API compatibility. + // We'll ignore it as sockudo-ws Http1 doesn't support headers yet and user said it's okay. + let _ = admin_password; - let (ws_stream, response) = connect_async(ws_url.as_str()) - .await - .map_err(|e| Error::WebSocket(Box::new(e)))?; + let host = ws_url.host_str().ok_or(Error::InvalidUrl("missing host".into()))?; + let port = ws_url.port_or_known_default().unwrap_or(80); + let addr = format!("{}:{}", host, port); - if response.status().as_u16() == 400 { - return Err(Error::WebhookModeActive); - } + let tcp_stream = tokio::net::TcpStream::connect(addr).await + .map_err(Error::Io)?; + + let client = WebSocketClient::::new(Config::default()); - let (write, read) = ws_stream.split(); + let ws_stream = if ws_url.scheme() == "wss" { + let mut root_store = rustls::RootCertStore::empty(); + let result = rustls_native_certs::load_native_certs(); + if !result.errors.is_empty() { + // Log errors or handle them. For now we take the first one if certs is empty. + if result.certs.is_empty() { + return Err(Error::ProcessStart { message: format!("failed to load native certs: {:?}", result.errors) }); + } + } + for cert in result.certs { + root_store.add(cert).map_err(|e| Error::ProcessStart { message: e.to_string() })?; + } + let config = rustls::ClientConfig::builder() + .with_root_certificates(root_store) + .with_no_client_auth(); + let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(config)); + let domain = rustls_pki_types::ServerName::try_from(host.to_string()) + .map_err(|_| Error::InvalidUrl("invalid host".into()))? + .to_owned(); + let tls_stream = connector.connect(domain, tcp_stream).await + .map_err(Error::Io)?; + + let (ws, _) = client.connect(tls_stream, host, ws_url.path(), None).await + .map_err(|e| Error::WebSocket(Box::new(e)))?; + ws + } else { + let (ws, _) = client.connect(tcp_stream, host, ws_url.path(), None).await + .map_err(|e| Error::WebSocket(Box::new(e)))?; + ws + }; + + let (read, write) = ws_stream.split(); let (event_tx, event_rx) = mpsc::channel(100); let (ack_tx, ack_rx) = mpsc::channel(1000); @@ -125,7 +150,8 @@ impl EventReceiver { match self.event_rx.recv().await { Some(event_with_ack) => { let json = event_with_ack.event; - let raw = match serde_json::from_str::(json.as_str()) { + let json_str = std::str::from_utf8(&json).expect("must be utf8"); + let raw = match serde_json::from_str::(json_str) { Ok(raw) => raw, Err(e) => { tracing::warn!("Failed to parse event: {}", e); @@ -168,7 +194,7 @@ impl EventReceiver { } }; - if let Err(e) = write.send(Message::Text(json.into())).await { + if let Err(e) = write.send(Message::text(json)).await { tracing::warn!("Failed to send ack: {}", e); break; } diff --git a/tapped/src/error.rs b/tapped/src/error.rs index f4fdbb9..2e01387 100644 --- a/tapped/src/error.rs +++ b/tapped/src/error.rs @@ -16,7 +16,7 @@ pub enum Error { /// WebSocket error #[error("WebSocket error: {0}")] - WebSocket(#[from] Box), + WebSocket(#[from] Box), /// JSON serialisation/deserialisation error #[error("JSON error: {0}")] diff --git a/tapped/src/types.rs b/tapped/src/types.rs index d0a917d..5461c79 100644 --- a/tapped/src/types.rs +++ b/tapped/src/types.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; -use tungstenite::protocol::frame::Utf8Bytes; +use bytes::Bytes; use crate::Error; @@ -63,7 +63,7 @@ pub struct RecordEvent { /// CID of the record (None on delete). pub cid: Option, // Inner record JSON pointing into the outer JSON - json: Option, + json: Option, record_offset: usize, record_len: usize, } @@ -73,7 +73,10 @@ impl RecordEvent { pub fn record_as_str(&self) -> Option<&str> { self.json .as_ref() - .map(|j| &j.as_str()[self.record_offset..self.record_offset + self.record_len]) + .map(|j| { + let s = std::str::from_utf8(j).expect("must be utf8"); + &s[self.record_offset..self.record_offset + self.record_len] + }) } /// Parse the record's content to a compatible struct @@ -241,12 +244,12 @@ pub(crate) struct RawIdentityEvent { impl RawEvent<'_> { /// Convert to the public Event type. - pub fn into_event(self, json: Utf8Bytes) -> Option { + pub fn into_event(self, json: Bytes) -> Option { match self.type_.as_str() { "record" => { let r = self.record?; let (json, record_offset, record_len) = if let Some(rv) = r.record.as_ref() { - let json_str = json.as_str(); + let json_str = std::str::from_utf8(&json).expect("must be utf8"); let rv_str = rv.get(); let offset = rv_str.as_ptr() as usize - json_str.as_ptr() as usize; (Some(json), offset, rv_str.len()) @@ -461,8 +464,8 @@ mod tests { }) .to_string(); - let json: Utf8Bytes = json.into(); - let raw: RawEvent = serde_json::from_str(json.as_str()).unwrap(); + let json: Bytes = json.into(); + let raw: RawEvent = serde_json::from_str(std::str::from_utf8(&json).unwrap()).unwrap(); assert_eq!(raw.id, 12345); assert_eq!(raw.type_, "record"); @@ -481,7 +484,7 @@ mod tests { #[test] fn raw_identity_event_deserialize() { - let json: Utf8Bytes = json!({ + let json: Bytes = json!({ "id": 99999, "type": "identity", "identity": { @@ -494,7 +497,7 @@ mod tests { .to_string() .into(); - let raw: RawEvent = serde_json::from_str(json.as_str()).unwrap(); + let raw: RawEvent = serde_json::from_str(std::str::from_utf8(&json).unwrap()).unwrap(); let event = raw.into_event(json.clone()).unwrap(); match event { @@ -511,7 +514,7 @@ mod tests { #[test] fn raw_delete_event_no_record() { - let json: Utf8Bytes = json!({ + let json: Bytes = json!({ "id": 55555, "type": "record", "record": { @@ -528,7 +531,7 @@ mod tests { .to_string() .into(); - let raw: RawEvent = serde_json::from_str(json.as_str()).unwrap(); + let raw: RawEvent = serde_json::from_str(std::str::from_utf8(&json).unwrap()).unwrap(); let event = raw.into_event(json.clone()).unwrap(); match event { @@ -571,4 +574,4 @@ mod tests { assert_eq!(identity_event.id(), 456); assert_eq!(identity_event.did(), "did:plc:identity"); } -} +} \ No newline at end of file -- 2.51.2