From 5534cd01f72296082d0cc51ad988008fa1796d1a Mon Sep 17 00:00:00 2001 From: loki5512344 Date: Thu, 9 Jul 2026 12:48:18 +0200 Subject: [PATCH] TLS 1.3, channel persistence, membership signaling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - TLS 1.3 on TCP (rustls + tokio-rustls, optional self-signed cert) - Server-side channel CRUD with SQLite persistence - Gateway → voice-node membership bridge via broadcast channel --- Cargo.lock | 134 +++++++ gateway/Cargo.toml | 6 + gateway/src/domain/config.rs | 9 + gateway/src/domain/storage/mod.rs | 70 +++- gateway/src/handler/channel/create.rs | 27 +- gateway/src/handler/channel/join.rs | 16 +- gateway/src/handler/channel/leave.rs | 23 +- gateway/src/handler/content/presence.rs | 6 +- gateway/src/handler/content/read_receipt.rs | 2 +- gateway/src/handler/deliver.rs | 4 +- gateway/src/handler/direct_message/history.rs | 4 +- gateway/src/handler/direct_message/send.rs | 4 +- gateway/src/handler/direct_message/start.rs | 6 +- gateway/src/handler/dispatch.rs | 351 +++++++++++++++--- gateway/src/handler/friends/decline.rs | 4 +- gateway/src/handler/friends/list.rs | 4 +- gateway/src/handler/friends/manage.rs | 10 +- gateway/src/handler/friends/mod.rs | 4 +- gateway/src/handler/friends/requests.rs | 6 +- gateway/src/handler/guild/audit.rs | 4 +- gateway/src/handler/guild/crud.rs | 6 +- gateway/src/handler/guild/invites.rs | 8 +- gateway/src/handler/guild/list.rs | 4 +- gateway/src/handler/guild/members.rs | 8 +- gateway/src/handler/guild/members_list.rs | 10 +- gateway/src/handler/guild/mod.rs | 4 +- gateway/src/handler/guild/roles.rs | 6 +- gateway/src/handler/mod.rs | 6 +- gateway/src/handler/run.rs | 6 +- gateway/src/lib.rs | 82 +++- gateway/src/main.rs | 2 +- gateway/src/net/handshake.rs | 4 +- gateway/src/net/io.rs | 29 +- gateway/src/net/state.rs | 10 +- serverd/Cargo.toml | 1 + serverd/src/main.rs | 29 +- serverd/src/voice_membership.rs | 22 ++ 37 files changed, 771 insertions(+), 160 deletions(-) create mode 100644 serverd/src/voice_membership.rs diff --git a/Cargo.lock b/Cargo.lock index 9990bfd..9587b4b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -77,6 +77,29 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "aws-lc-rs" +version = "1.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4342d8937fc7e5dd9b1c60292261c0670c882a2cd1719cfc11b1af41731e32ad" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d9ceb1da931507a12f4fccea479dccd00da1943e1b4ae72d8e502d707361444" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + [[package]] name = "axum" version = "0.8.9" @@ -205,6 +228,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f5d6cac793997bd970000024b2934968efe83b382de4fdcf4fcb46b6ee4ad996" dependencies = [ "find-msvc-tools", + "jobserver", + "libc", "shlex", ] @@ -416,6 +441,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" + [[package]] name = "digest" version = "0.10.7" @@ -456,6 +487,12 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "ed25519" version = "2.2.3" @@ -565,6 +602,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.32" @@ -954,6 +997,16 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom 0.4.3", + "libc", +] + [[package]] name = "js-sys" version = "0.3.103" @@ -1066,6 +1119,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + [[package]] name = "num-traits" version = "0.2.19" @@ -1136,6 +1195,16 @@ dependencies = [ "subtle", ] +[[package]] +name = "pem" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" +dependencies = [ + "base64", + "serde_core", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -1184,6 +1253,12 @@ dependencies = [ "zerovec", ] +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1299,6 +1374,19 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rcgen" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "yasna", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1354,6 +1442,8 @@ version = "0.23.41" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f" dependencies = [ + "aws-lc-rs", + "log", "once_cell", "ring", "rustls-pki-types", @@ -1377,6 +1467,7 @@ version = "0.103.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -1856,6 +1947,25 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "time" +version = "0.3.53" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50" +dependencies = [ + "deranged", + "num-conv", + "powerfmt", + "serde_core", + "time-core", +] + +[[package]] +name = "time-core" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" + [[package]] name = "tinystr" version = "0.8.3" @@ -1909,6 +2019,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" @@ -2186,12 +2306,16 @@ dependencies = [ "hex", "hkdf", "rand 0.8.6", + "rcgen", + "rustls", + "rustls-pki-types", "serde", "serde_json", "sha2 0.11.0", "sqlx", "thiserror", "tokio", + "tokio-rustls", "toml", "tower", "tracing", @@ -2206,6 +2330,7 @@ version = "0.1.0" dependencies = [ "anyhow", "serde", + "serde_json", "tokio", "toml", "tracing", @@ -2421,6 +2546,15 @@ dependencies = [ "zeroize", ] +[[package]] +name = "yasna" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd" +dependencies = [ + "time", +] + [[package]] name = "yoke" version = "0.8.3" diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index 7178dcc..bd464d0 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -38,3 +38,9 @@ axum = { version = "0.8", features = ["ws"] } tower = "0.5" sqlx = { version = "0.9", features = ["sqlite", "runtime-tokio", "tls-rustls"] } bitflags = "2" + +# TLS 1.3 support +rustls-pki-types = "1" +rustls = "0.23" +tokio-rustls = "0.26" +rcgen = "0.13" diff --git a/gateway/src/domain/config.rs b/gateway/src/domain/config.rs index 1b8dff2..2289b34 100644 --- a/gateway/src/domain/config.rs +++ b/gateway/src/domain/config.rs @@ -90,6 +90,15 @@ pub struct GatewayConfig { /// Defaults to 10. #[serde(default)] pub message_rate_burst: Option, + /// Enable TLS 1.3 on the TCP listener. + #[serde(default)] + pub tls_enabled: bool, + /// Path to TLS certificate file (PEM). + #[serde(default)] + pub tls_cert_path: Option, + /// Path to TLS private key file (PEM). + #[serde(default)] + pub tls_key_path: Option, } #[derive(Debug, Deserialize)] diff --git a/gateway/src/domain/storage/mod.rs b/gateway/src/domain/storage/mod.rs index bfdd2b8..1ba1873 100644 --- a/gateway/src/domain/storage/mod.rs +++ b/gateway/src/domain/storage/mod.rs @@ -35,7 +35,13 @@ impl Storage { CREATE TABLE IF NOT EXISTS users ( pubkey TEXT PRIMARY KEY, nickname TEXT NOT NULL, first_seen INTEGER NOT NULL ); - CREATE TABLE IF NOT EXISTS bans ( + CREATE TABLE IF NOT EXISTS channels ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + kind TEXT NOT NULL DEFAULT 'text', + created_at INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS bans ( pubkey TEXT PRIMARY KEY, reason TEXT, banned_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS direct_messages ( @@ -215,6 +221,68 @@ pub(crate) fn generate_invite_code() -> String { format!("{:08x}", hash % 0x100000000u64) } +use crate::domain::channels::{ChannelKind, ChannelStore}; + +#[derive(Debug, Clone)] +pub struct ChannelRecord { + pub id: String, + pub name: String, + pub kind: String, + pub created_at: i64, +} + +impl Storage { + pub async fn create_channel(&self, id: &str, name: &str, kind: &str) -> Result { + let result = sqlx::query( + "INSERT OR IGNORE INTO channels (id, name, kind, created_at) VALUES (?, ?, ?, ?)", + ) + .bind(id) + .bind(name) + .bind(kind) + .bind(now_ms()) + .execute(&self.pool) + .await?; + Ok(result.rows_affected() > 0) + } + + pub async fn delete_channel(&self, id: &str) -> Result { + let result = sqlx::query("DELETE FROM channels WHERE id=?") + .bind(id) + .execute(&self.pool) + .await?; + Ok(result.rows_affected() > 0) + } + + pub async fn list_channels(&self) -> Result> { + let rows = sqlx::query_as::<_, (String, String, String, i64)>( + "SELECT id, name, kind, created_at FROM channels", + ) + .fetch_all(&self.pool) + .await? + .into_iter() + .map(|(id, name, kind, created_at)| ChannelRecord { + id, + name, + kind, + created_at, + }) + .collect(); + Ok(rows) + } + + pub async fn load_channels_to_cache(&self, channel_store: &ChannelStore) -> Result<()> { + let channels = self.list_channels().await?; + for ch in channels { + let kind = match ch.kind.as_str() { + "voice" => ChannelKind::Voice, + _ => ChannelKind::Text, + }; + crate::domain::channels::create(channel_store, &ch.id, &ch.name, kind).await; + } + Ok(()) + } +} + pub(crate) fn now_ms() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) diff --git a/gateway/src/handler/channel/create.rs b/gateway/src/handler/channel/create.rs index f60f9af..d21ac2a 100644 --- a/gateway/src/handler/channel/create.rs +++ b/gateway/src/handler/channel/create.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use tracing::{info, warn}; use crate::{ @@ -17,7 +17,7 @@ use crate::{ /// Handle a ChannelCreate request — register a new channel in the store and /// broadcast the new ChannelState to all sessions so their sidebars update. pub async fn handle_channel_create( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -97,6 +97,17 @@ pub async fn handle_channel_create( let created = channels::create(&state.channels, &channel_id, &channel_name, kind.clone()).await; + // Persist to DB (best-effort, log and continue on failure). + if created { + if let Err(e) = state + .storage + .create_channel(&channel_id, &channel_name, kind.as_str()) + .await + { + warn!("failed to persist channel to storage: {e}"); + } + } + if !created { io::send_encrypted( stream, @@ -160,7 +171,7 @@ pub async fn handle_channel_create( /// Handle a ChannelDelete request — remove the channel from the store and /// broadcast the deletion to all sessions. pub async fn handle_channel_delete( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -190,6 +201,14 @@ pub async fn handle_channel_delete( } let existed = channels::delete(&state.channels, &req.channel_id).await; + + // Remove from DB (best-effort). + if existed { + if let Err(e) = state.storage.delete_channel(&req.channel_id).await { + warn!("failed to remove channel from storage: {e}"); + } + } + if !existed { io::send_encrypted( stream, @@ -229,7 +248,7 @@ pub async fn handle_channel_delete( /// Handle a ChannelList request — reply with all known channels. pub async fn handle_channel_list( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, crypto: &SessionCrypto, diff --git a/gateway/src/handler/channel/join.rs b/gateway/src/handler/channel/join.rs index d6f0606..481bb1f 100644 --- a/gateway/src/handler/channel/join.rs +++ b/gateway/src/handler/channel/join.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use tracing::info; use crate::{ @@ -19,7 +19,7 @@ use super::{broadcast_leave, set_channel}; const HISTORY_LIMIT: i64 = 50; pub async fn join( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, channel_id: &str, @@ -110,6 +110,18 @@ pub async fn join( }); } + if let Some(tx) = &state.voice_member_tx { + if let Some(sess) = session::get(&state.sessions, session_id).await { + let event = serde_json::json!({ + "type": "joined", + "channel_id": channel_id, + "session_id": session_id, + "user_id": sess.user_id, + }); + let _ = tx.send(event.to_string()); + } + } + info!("session {} joined {channel_id}", &session_id[..8]); Ok(()) } diff --git a/gateway/src/handler/channel/leave.rs b/gateway/src/handler/channel/leave.rs index 9e23845..51521e6 100644 --- a/gateway/src/handler/channel/leave.rs +++ b/gateway/src/handler/channel/leave.rs @@ -1,22 +1,39 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use tracing::info; -use crate::{domain::channels, net::state::State, proto::SessionCrypto}; +use crate::{domain::{channels, session}, net::state::State, proto::SessionCrypto}; use super::{broadcast_leave, set_channel}; pub async fn leave( - _stream: &mut TcpStream, + _stream: &mut (impl AsyncRead + AsyncWrite + Unpin), _seq: &mut u32, session_id: &str, channel_id: &str, _crypto: &SessionCrypto, state: &State, ) -> Result<()> { + let user_id = session::get(&state.sessions, session_id) + .await + .map(|s| s.user_id); + channels::leave(&state.channels, channel_id, session_id).await; set_channel(state, session_id, None).await; broadcast_leave(state, channel_id, session_id).await; + + if let Some(tx) = &state.voice_member_tx { + if let Some(ref uid) = user_id { + let event = serde_json::json!({ + "type": "left", + "channel_id": channel_id, + "session_id": session_id, + "user_id": uid, + }); + let _ = tx.send(event.to_string()); + } + } + info!("session {} left {channel_id}", &session_id[..8]); Ok(()) } diff --git a/gateway/src/handler/content/presence.rs b/gateway/src/handler/content/presence.rs index 791837d..02a8f1c 100644 --- a/gateway/src/handler/content/presence.rs +++ b/gateway/src/handler/content/presence.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -14,7 +14,7 @@ use crate::{ }; pub async fn handle_presence_update( - _stream: &mut TcpStream, + _stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -60,7 +60,7 @@ pub async fn handle_presence_update( } pub async fn handle_presence_sync( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, crypto: &SessionCrypto, diff --git a/gateway/src/handler/content/read_receipt.rs b/gateway/src/handler/content/read_receipt.rs index 522bdf9..ce59137 100644 --- a/gateway/src/handler/content/read_receipt.rs +++ b/gateway/src/handler/content/read_receipt.rs @@ -7,7 +7,7 @@ use crate::{ }; pub async fn handle_read_receipt( - _stream: &mut tokio::net::TcpStream, + _stream: &mut (impl tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/deliver.rs b/gateway/src/handler/deliver.rs index a30f20a..6d2e68e 100644 --- a/gateway/src/handler/deliver.rs +++ b/gateway/src/handler/deliver.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -8,7 +8,7 @@ use crate::{ }; pub async fn deliver_encrypted( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, msg: &BroadcastMsg, diff --git a/gateway/src/handler/direct_message/history.rs b/gateway/src/handler/direct_message/history.rs index 8ba462a..cb58191 100644 --- a/gateway/src/handler/direct_message/history.rs +++ b/gateway/src/handler/direct_message/history.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -8,7 +8,7 @@ use crate::{ }; pub async fn handle_dm_history( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/direct_message/send.rs b/gateway/src/handler/direct_message/send.rs index 6e2743c..7f53d2b 100644 --- a/gateway/src/handler/direct_message/send.rs +++ b/gateway/src/handler/direct_message/send.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use tracing::warn; use crate::{ @@ -14,7 +14,7 @@ use crate::{ }; pub async fn handle_dm_message( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/direct_message/start.rs b/gateway/src/handler/direct_message/start.rs index 8c09f58..6d54df3 100644 --- a/gateway/src/handler/direct_message/start.rs +++ b/gateway/src/handler/direct_message/start.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -8,7 +8,7 @@ use crate::{ }; pub async fn handle_dm_start( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -88,7 +88,7 @@ pub async fn handle_dm_start( } pub async fn handle_dm_read_ack( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/dispatch.rs b/gateway/src/handler/dispatch.rs index f117e07..cf5e28e 100644 --- a/gateway/src/handler/dispatch.rs +++ b/gateway/src/handler/dispatch.rs @@ -1,4 +1,5 @@ use anyhow::Result; +use tokio::io::{AsyncRead, AsyncWrite}; use tracing::debug; use crate::{ @@ -11,174 +12,402 @@ use crate::{ use super::{Ctx, channel, content, direct_message, friends, guild}; -pub async fn dispatch( - ctx: &mut Ctx<'_>, +pub async fn dispatch( + ctx: &mut Ctx<'_, S>, session_id: &str, pid: PacketId, payload: &[u8], addr: std::net::SocketAddr, ) -> Result<()> { - let Ctx { - stream, - seq, - crypto, - state, - } = ctx; match pid { PacketId::Ping => { let ping: PingPayload = serde_json::from_slice(payload)?; io::send_encrypted( - stream, + ctx.stream, PacketId::Pong, - seq, + ctx.seq, &to_payload(&PongPayload { timestamp: ping.timestamp, }), - crypto, + ctx.crypto, ) .await?; } PacketId::JoinChannel => { let m: JoinChannelPayload = serde_json::from_slice(payload)?; - channel::join(stream, seq, session_id, &m.channel_id, crypto, state).await?; + channel::join(ctx.stream, ctx.seq, session_id, &m.channel_id, ctx.crypto, ctx.state) + .await?; } PacketId::LeaveChannel => { let m: LeaveChannelPayload = serde_json::from_slice(payload)?; - channel::leave(stream, seq, session_id, &m.channel_id, crypto, state).await?; + channel::leave(ctx.stream, ctx.seq, session_id, &m.channel_id, ctx.crypto, ctx.state) + .await?; } PacketId::ChannelCreate => { - channel::handle_channel_create(stream, seq, session_id, payload, crypto, state).await?; + channel::handle_channel_create( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::ChannelDelete => { - channel::handle_channel_delete(stream, seq, session_id, payload, crypto, state).await?; + channel::handle_channel_delete( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::ChannelList => { - channel::handle_channel_list(stream, seq, session_id, crypto, state).await?; + channel::handle_channel_list(ctx.stream, ctx.seq, session_id, ctx.crypto, ctx.state) + .await?; } PacketId::ChatMessage => { let m: ChatMessagePayload = serde_json::from_slice(payload)?; - content::chat::handle(session_id, m, state).await?; + content::chat::handle(session_id, m, ctx.state).await?; } PacketId::DmStart => { - direct_message::handle_dm_start(stream, seq, session_id, payload, crypto, state) - .await?; + direct_message::handle_dm_start( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::DmMessage => { - direct_message::handle_dm_message(stream, seq, session_id, payload, crypto, state) - .await?; + direct_message::handle_dm_message( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::DmHistory => { - direct_message::handle_dm_history(stream, seq, session_id, payload, crypto, state) - .await?; + direct_message::handle_dm_history( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::DmReadAck => { - direct_message::handle_dm_read_ack(stream, seq, session_id, payload, crypto, state) - .await?; + direct_message::handle_dm_read_ack( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildCreate => { - guild::handle_guild_create(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_guild_create( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildDelete => { - guild::handle_guild_delete(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_guild_delete( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildList => { - guild::handle_guild_list(stream, seq, session_id, crypto, state).await?; + guild::handle_guild_list(ctx.stream, ctx.seq, session_id, ctx.crypto, ctx.state) + .await?; } PacketId::GuildMemberJoin => { - guild::handle_guild_member_join(stream, seq, session_id, payload, crypto, state) - .await?; + guild::handle_guild_member_join( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildMemberLeave => { - guild::handle_guild_member_leave(stream, seq, session_id, payload, crypto, state) - .await?; + guild::handle_guild_member_leave( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildMemberKick => { - guild::handle_guild_member_kick(stream, seq, session_id, payload, crypto, state) - .await?; + guild::handle_guild_member_kick( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::RoleCreate => { - guild::handle_role_create(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_role_create( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::RoleDelete => { - guild::handle_role_delete(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_role_delete( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::InviteCreate => { - guild::handle_invite_create(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_invite_create( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::InviteAccept => { - guild::handle_invite_accept(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_invite_accept( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::InviteDelete => { - guild::handle_invite_delete(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_invite_delete( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildAuditLogFetch => { - guild::handle_audit_log_fetch(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_audit_log_fetch( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildMemberListFetch => { - guild::handle_member_list_fetch(stream, seq, session_id, payload, crypto, state) - .await?; + guild::handle_member_list_fetch( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildRoleAssign => { - guild::handle_role_assign(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_role_assign( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildRoleUnassign => { - guild::handle_role_unassign(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_role_unassign( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::GuildRoleListFetch => { - guild::handle_role_list_fetch(stream, seq, session_id, payload, crypto, state).await?; + guild::handle_role_list_fetch( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::PresenceUpdate => { content::presence::handle_presence_update( - stream, seq, session_id, payload, crypto, state, + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, ) .await?; } PacketId::PresenceSync => { - content::presence::handle_presence_sync(stream, seq, session_id, crypto, state).await?; + content::presence::handle_presence_sync( + ctx.stream, + ctx.seq, + session_id, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::FriendRequest => { - friends::handle_friend_request(stream, seq, session_id, payload, crypto, state).await?; + friends::handle_friend_request( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::FriendAccept => { - friends::handle_friend_accept(stream, seq, session_id, payload, crypto, state).await?; + friends::handle_friend_accept( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::FriendDecline => { - friends::handle_friend_decline(stream, seq, session_id, payload, crypto, state).await?; + friends::handle_friend_decline( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::FriendRemove => { - friends::handle_friend_remove(stream, seq, session_id, payload, crypto, state).await?; + friends::handle_friend_remove( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::FriendList => { - friends::handle_friend_list(stream, seq, session_id, crypto, state).await?; + friends::handle_friend_list(ctx.stream, ctx.seq, session_id, ctx.crypto, ctx.state) + .await?; } PacketId::BlockUser => { - friends::handle_block_user(stream, seq, session_id, payload, crypto, state).await?; + friends::handle_block_user( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::UnblockUser => { - friends::handle_unblock_user(stream, seq, session_id, payload, crypto, state).await?; + friends::handle_unblock_user( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::BlockList => { - friends::handle_block_list(stream, seq, session_id, crypto, state).await?; + friends::handle_block_list(ctx.stream, ctx.seq, session_id, ctx.crypto, ctx.state) + .await?; } PacketId::MessageReactionAdd => { let m: ReactionPayload = serde_json::from_slice(payload)?; - content::reaction::handle_reaction_add(session_id, m, state).await?; + content::reaction::handle_reaction_add(session_id, m, ctx.state).await?; } PacketId::MessageReactionRemove => { let m: ReactionPayload = serde_json::from_slice(payload)?; - content::reaction::handle_reaction_remove(session_id, m, state).await?; + content::reaction::handle_reaction_remove(session_id, m, ctx.state).await?; } PacketId::MessageEdit => { - content::message_edit::handle_message_edit(session_id, payload, state).await?; + content::message_edit::handle_message_edit(session_id, payload, ctx.state).await?; } PacketId::MessageDelete => { - content::message_edit::handle_message_delete(session_id, payload, state).await?; + content::message_edit::handle_message_delete(session_id, payload, ctx.state).await?; } PacketId::TypingStart => { - content::handle_typing_start(session_id, payload, state).await?; + content::handle_typing_start(session_id, payload, ctx.state).await?; } PacketId::ReadReceipt => { - content::handle_read_receipt(stream, seq, session_id, payload, crypto, state).await?; + content::handle_read_receipt( + ctx.stream, + ctx.seq, + session_id, + payload, + ctx.crypto, + ctx.state, + ) + .await?; } PacketId::Disconnect => debug!("{addr} DISCONNECT"), other => debug!("{addr} unhandled {:?}", other), diff --git a/gateway/src/handler/friends/decline.rs b/gateway/src/handler/friends/decline.rs index 7057884..da10800 100644 --- a/gateway/src/handler/friends/decline.rs +++ b/gateway/src/handler/friends/decline.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -8,7 +8,7 @@ use crate::{ }; pub async fn handle_friend_decline( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/friends/list.rs b/gateway/src/handler/friends/list.rs index 4c2a679..f2844fb 100644 --- a/gateway/src/handler/friends/list.rs +++ b/gateway/src/handler/friends/list.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -8,7 +8,7 @@ use crate::{ }; pub async fn handle_friend_list( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, crypto: &SessionCrypto, diff --git a/gateway/src/handler/friends/manage.rs b/gateway/src/handler/friends/manage.rs index 4f1cbd1..d8d9358 100644 --- a/gateway/src/handler/friends/manage.rs +++ b/gateway/src/handler/friends/manage.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -11,7 +11,7 @@ use crate::{ }; pub async fn handle_friend_remove( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -40,7 +40,7 @@ pub async fn handle_friend_remove( } pub async fn handle_block_user( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -69,7 +69,7 @@ pub async fn handle_block_user( } pub async fn handle_unblock_user( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -98,7 +98,7 @@ pub async fn handle_unblock_user( } pub async fn handle_block_list( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, crypto: &SessionCrypto, diff --git a/gateway/src/handler/friends/mod.rs b/gateway/src/handler/friends/mod.rs index 034855f..3e98d2d 100644 --- a/gateway/src/handler/friends/mod.rs +++ b/gateway/src/handler/friends/mod.rs @@ -4,7 +4,7 @@ mod manage; mod requests; use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ net::io, @@ -12,7 +12,7 @@ use crate::{ }; async fn send_err( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, code: crate::proto::ErrorCode, msg: &str, diff --git a/gateway/src/handler/friends/requests.rs b/gateway/src/handler/friends/requests.rs index b9c9182..37e60a1 100644 --- a/gateway/src/handler/friends/requests.rs +++ b/gateway/src/handler/friends/requests.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -16,7 +16,7 @@ use crate::{ use super::send_err; pub async fn handle_friend_request( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -100,7 +100,7 @@ pub async fn handle_friend_request( } pub async fn handle_friend_accept( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/guild/audit.rs b/gateway/src/handler/guild/audit.rs index 3f505ec..1c08d07 100644 --- a/gateway/src/handler/guild/audit.rs +++ b/gateway/src/handler/guild/audit.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::permissions::Permissions, @@ -12,7 +12,7 @@ use crate::{ /// Fetch audit log entries for a guild (admin-only: requires VIEW_AUDIT_LOG or owner). pub async fn handle_audit_log_fetch( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/guild/crud.rs b/gateway/src/handler/guild/crud.rs index b48a427..8637c72 100644 --- a/gateway/src/handler/guild/crud.rs +++ b/gateway/src/handler/guild/crud.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use tracing::debug; use crate::{ @@ -11,7 +11,7 @@ use crate::{ use super::send_err; pub async fn handle_guild_create( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -67,7 +67,7 @@ pub async fn handle_guild_create( } pub async fn handle_guild_delete( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/guild/invites.rs b/gateway/src/handler/guild/invites.rs index 37d9b37..191c861 100644 --- a/gateway/src/handler/guild/invites.rs +++ b/gateway/src/handler/guild/invites.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::{permissions::Permissions, session}, @@ -13,7 +13,7 @@ use crate::{ use super::{now_ms, send_err}; pub async fn handle_invite_create( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -89,7 +89,7 @@ pub async fn handle_invite_create( } pub async fn handle_invite_accept( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -164,7 +164,7 @@ pub async fn handle_invite_accept( } pub async fn handle_invite_delete( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/guild/list.rs b/gateway/src/handler/guild/list.rs index 626d382..21c113a 100644 --- a/gateway/src/handler/guild/list.rs +++ b/gateway/src/handler/guild/list.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::session, @@ -8,7 +8,7 @@ use crate::{ }; pub async fn handle_guild_list( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, crypto: &SessionCrypto, diff --git a/gateway/src/handler/guild/members.rs b/gateway/src/handler/guild/members.rs index 2ba90d3..1a1cd2b 100644 --- a/gateway/src/handler/guild/members.rs +++ b/gateway/src/handler/guild/members.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::{permissions::Permissions, session}, @@ -13,7 +13,7 @@ use crate::{ use super::send_err; pub async fn handle_guild_member_join( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -63,7 +63,7 @@ pub async fn handle_guild_member_join( } pub async fn handle_guild_member_leave( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -96,7 +96,7 @@ pub async fn handle_guild_member_leave( } pub async fn handle_guild_member_kick( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/guild/members_list.rs b/gateway/src/handler/guild/members_list.rs index 9ff6fb5..3cfc8ff 100644 --- a/gateway/src/handler/guild/members_list.rs +++ b/gateway/src/handler/guild/members_list.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::{permissions::Permissions, session}, @@ -15,7 +15,7 @@ use super::{require_perm, send_err}; /// Fetch the member list for a guild (visible to all members). pub async fn handle_member_list_fetch( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -65,7 +65,7 @@ pub async fn handle_member_list_fetch( /// Assign a role to a user (admin only: requires MANAGE_ROLES). pub async fn handle_role_assign( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -129,7 +129,7 @@ pub async fn handle_role_assign( /// Remove a role from a user (admin only: requires MANAGE_ROLES). pub async fn handle_role_unassign( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -192,7 +192,7 @@ pub async fn handle_role_unassign( /// Fetch all roles defined in a guild (visible to all members). pub async fn handle_role_list_fetch( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/guild/mod.rs b/gateway/src/handler/guild/mod.rs index 8dc6052..056627b 100644 --- a/gateway/src/handler/guild/mod.rs +++ b/gateway/src/handler/guild/mod.rs @@ -7,7 +7,7 @@ mod members_list; mod roles; use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::permissions::Permissions, @@ -16,7 +16,7 @@ use crate::{ }; async fn send_err( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, code: crate::proto::ErrorCode, msg: &str, diff --git a/gateway/src/handler/guild/roles.rs b/gateway/src/handler/guild/roles.rs index 4e2fe59..19d061f 100644 --- a/gateway/src/handler/guild/roles.rs +++ b/gateway/src/handler/guild/roles.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{ domain::{permissions::Permissions, session}, @@ -10,7 +10,7 @@ use crate::{ use super::send_err; pub async fn handle_role_create( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], @@ -73,7 +73,7 @@ pub async fn handle_role_create( } pub async fn handle_role_delete( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, session_id: &str, payload: &[u8], diff --git a/gateway/src/handler/mod.rs b/gateway/src/handler/mod.rs index 06a6eed..2270ed9 100644 --- a/gateway/src/handler/mod.rs +++ b/gateway/src/handler/mod.rs @@ -10,12 +10,12 @@ pub mod run; pub use dispatch::dispatch; pub use run::run_session; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use crate::{net::state::State, proto::SessionCrypto}; -pub struct Ctx<'a> { - pub stream: &'a mut TcpStream, +pub struct Ctx<'a, S: AsyncRead + AsyncWrite + Unpin> { + pub stream: &'a mut S, pub seq: &'a mut u32, pub crypto: &'a SessionCrypto, pub state: &'a State, diff --git a/gateway/src/handler/run.rs b/gateway/src/handler/run.rs index 70098c3..f51893d 100644 --- a/gateway/src/handler/run.rs +++ b/gateway/src/handler/run.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use tokio::sync::broadcast; use tracing::warn; @@ -10,8 +10,8 @@ use crate::{ use super::{Ctx, deliver::deliver_encrypted, dispatch}; -pub async fn run_session( - stream: &mut TcpStream, +pub async fn run_session( + stream: &mut S, addr: std::net::SocketAddr, seq: &mut u32, session_id: &str, diff --git a/gateway/src/lib.rs b/gateway/src/lib.rs index c7553be..9ed6ed9 100644 --- a/gateway/src/lib.rs +++ b/gateway/src/lib.rs @@ -8,14 +8,15 @@ pub mod proto; use anyhow::Result; use std::sync::Arc; use std::sync::atomic::AtomicUsize; +use tokio::io::{AsyncRead, AsyncWrite}; use tokio::net::TcpListener; use tokio::sync::broadcast; use tracing::{debug, error, info, warn}; use domain::{channels, config, session, storage}; -use net::state::State; +use net::state::{State, VoiceMemberTx}; -pub async fn run(cfg: Arc) -> Result<()> { +pub async fn run(cfg: Arc, voice_member_tx: Option) -> Result<()> { let private_mode = cfg.is_private(); if private_mode { info!("private mode enabled, federation disabled"); @@ -74,6 +75,7 @@ pub async fn run(cfg: Arc) -> Result<()> { metrics.clone(), sessions_count.clone(), channels_count.clone(), + voice_member_tx, ); let admin_bind = cfg @@ -95,11 +97,52 @@ pub async fn run(cfg: Arc) -> Result<()> { let listener = TcpListener::bind(&cfg.gateway.bind).await?; info!("listening on {}", cfg.gateway.bind); + if cfg.gateway.tls_enabled { + run_tls(listener, cfg, state).await + } else { + run_plain(listener, state).await + } +} + +async fn run_tls(listener: TcpListener, cfg: Arc, state: State) -> Result<()> { + let (certs, key) = load_or_generate_tls_certs(&cfg.gateway)?; + if cfg.gateway.tls_cert_path.is_none() { + warn!("using self-signed TLS certificate — clients must accept it manually"); + } + let server_config = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(certs, key)?; + let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config)); + + loop { + match listener.accept().await { + Ok((stream, addr)) => { + info!("connection from {addr} (TLS)"); + state.metrics.inc(&state.metrics.connections_total); + let s = state.clone(); + let acceptor = acceptor.clone(); + tokio::spawn(async move { + match acceptor.accept(stream).await { + Ok(tls_stream) => { + if let Err(e) = handle(tls_stream, addr, s).await { + debug!("{addr} closed: {e}"); + } + } + Err(e) => error!("TLS accept error from {addr}: {e}"), + } + }); + } + Err(e) => error!("accept: {e}"), + } + } +} + +async fn run_plain(listener: TcpListener, state: State) -> Result<()> { loop { match listener.accept().await { Ok((stream, addr)) => { info!("connection from {addr}"); - metrics.inc(&metrics.connections_total); + state.metrics.inc(&state.metrics.connections_total); let s = state.clone(); tokio::spawn(async move { if let Err(e) = handle(stream, addr, s).await { @@ -112,8 +155,8 @@ pub async fn run(cfg: Arc) -> Result<()> { } } -async fn handle( - mut stream: tokio::net::TcpStream, +async fn handle( + mut stream: S, addr: std::net::SocketAddr, state: State, ) -> Result<()> { @@ -164,3 +207,32 @@ async fn handle( debug!("{addr} cleaned up"); result } + +fn load_or_generate_tls_certs( + cfg: &config::GatewayConfig, +) -> Result<( + Vec>, + rustls::pki_types::PrivateKeyDer<'static>, +)> { + use rustls::pki_types::pem::PemObject; + use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; + + match (&cfg.tls_cert_path, &cfg.tls_key_path) { + (Some(cert_path), Some(key_path)) => { + let cert_pem = std::fs::read_to_string(cert_path)?; + let key_pem = std::fs::read_to_string(key_path)?; + let cert = CertificateDer::from_pem_reader(&mut cert_pem.as_bytes())?; + let key = PrivateKeyDer::from_pem_reader(&mut key_pem.as_bytes())?; + Ok((vec![cert], key)) + } + _ => { + info!("No TLS cert/key configured, generating self-signed certificate"); + let certified_key = rcgen::generate_simple_self_signed(vec!["VNOX Server".into()])?; + let cert_der = certified_key.cert.der().clone(); + let key_der = PrivateKeyDer::from(PrivatePkcs8KeyDer::from( + certified_key.key_pair.serialize_der(), + )); + Ok((vec![cert_der], key_der)) + } + } +} diff --git a/gateway/src/main.rs b/gateway/src/main.rs index 35ba47e..905d877 100644 --- a/gateway/src/main.rs +++ b/gateway/src/main.rs @@ -8,5 +8,5 @@ async fn main() -> Result<()> { .init(); let cfg = Arc::new(vnox_gateway::domain::config::load()?); - vnox_gateway::run(cfg).await + vnox_gateway::run(cfg, None).await } diff --git a/gateway/src/net/handshake.rs b/gateway/src/net/handshake.rs index 66a13a6..a49205b 100644 --- a/gateway/src/net/handshake.rs +++ b/gateway/src/net/handshake.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncWrite}; use tracing::{debug, info, warn}; use crate::{ @@ -16,7 +16,7 @@ const LNEX_VERSION: &str = "v1"; /// Returns the session and the derived crypto context (encryption keys). /// All subsequent packets must be encrypted with `crypto`. pub async fn run( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), addr: std::net::SocketAddr, state: &State, seq: &mut u32, diff --git a/gateway/src/net/io.rs b/gateway/src/net/io.rs index 9c6c679..d2d2731 100644 --- a/gateway/src/net/io.rs +++ b/gateway/src/net/io.rs @@ -1,6 +1,5 @@ use anyhow::Result; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use crate::proto::{ ErrorCode, ErrorPayload, PacketHeader, PacketId, SessionCrypto, encode_packet, flags, @@ -9,10 +8,8 @@ use crate::proto::{ const MAX_PAYLOAD: u32 = 4 * 1024 * 1024; -// ─── Unencrypted IO (used during handshake) ───────────────────────────────── - pub async fn send_packet( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), id: PacketId, seq: &mut u32, payload: &[u8], @@ -24,7 +21,7 @@ pub async fn send_packet( } pub async fn send_error( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), seq: &mut u32, code: ErrorCode, msg: &str, @@ -37,7 +34,9 @@ pub async fn send_error( send_packet(stream, PacketId::Error, seq, &to_payload(&p)).await } -pub async fn read_packet(stream: &mut TcpStream) -> Result<(PacketHeader, Vec)> { +pub async fn read_packet( + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), +) -> Result<(PacketHeader, Vec)> { let mut buf = [0u8; PacketHeader::SIZE]; stream.read_exact(&mut buf).await?; let hdr = PacketHeader::from_bytes(&buf); @@ -54,12 +53,8 @@ pub async fn read_packet(stream: &mut TcpStream) -> Result<(PacketHeader, Vec Result<(PacketHeader, Vec)> { let mut buf = [0u8; PacketHeader::SIZE]; @@ -107,12 +100,8 @@ pub async fn read_encrypted( Ok((hdr, payload)) } -/// Re-encode and deliver a pre-encoded broadcast packet with encryption. -/// -/// Extracts the packet ID from the pre-encoded `raw_data`, re-encodes -/// with the correct sequence and encryption for this recipient. pub async fn deliver_encrypted( - stream: &mut TcpStream, + stream: &mut (impl AsyncRead + AsyncWrite + Unpin), crypto: &SessionCrypto, seq: &mut u32, raw_data: &[u8], diff --git a/gateway/src/net/state.rs b/gateway/src/net/state.rs index c05e9b4..282faf2 100644 --- a/gateway/src/net/state.rs +++ b/gateway/src/net/state.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; -use std::sync::Arc; use std::sync::atomic::AtomicUsize; -use tokio::sync::{RwLock, broadcast}; +use std::sync::Arc; +use tokio::sync::{broadcast, RwLock}; use crate::admin::metrics::Metrics; use crate::bootstrap::server_identity::ServerIdentity; @@ -11,6 +11,8 @@ use crate::domain::{ }; use crate::proto::PresenceInfo; +pub type VoiceMemberTx = tokio::sync::broadcast::Sender; + #[derive(Clone)] pub struct State { pub sessions: SessionStore, @@ -30,6 +32,8 @@ pub struct State { pub channels_count: Arc, /// Per-session rate limiter (token bucket). pub rate_limiter: Arc, + /// Gateway → voice-node membership bridge. + pub voice_member_tx: Option, } impl State { @@ -44,6 +48,7 @@ impl State { metrics: Arc, sessions_count: Arc, channels_count: Arc, + voice_member_tx: Option, ) -> Self { let rate_per_sec = config.gateway.message_rate_per_sec.unwrap_or(5.0); let burst = config.gateway.message_rate_burst.unwrap_or(10); @@ -60,6 +65,7 @@ impl State { sessions_count, channels_count, rate_limiter, + voice_member_tx, } } } diff --git a/serverd/Cargo.toml b/serverd/Cargo.toml index 1375965..55578b3 100644 --- a/serverd/Cargo.toml +++ b/serverd/Cargo.toml @@ -21,3 +21,4 @@ tracing-subscriber.workspace = true anyhow.workspace = true toml.workspace = true serde.workspace = true +serde_json.workspace = true diff --git a/serverd/src/main.rs b/serverd/src/main.rs index 2646ae4..92d6889 100644 --- a/serverd/src/main.rs +++ b/serverd/src/main.rs @@ -1,7 +1,12 @@ +mod voice_membership; + use anyhow::Result; use std::sync::Arc; +use tokio::sync::broadcast; use tracing::info; + + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -28,12 +33,34 @@ async fn main() -> Result<()> { let voice_bind = cfg.voice.bind.clone(); let gate_cfg = Arc::new(cfg); + let (voice_member_tx, _) = broadcast::channel::(256); + let mut voice_member_rx = voice_member_tx.subscribe(); + let voice_handle = tokio::spawn(async move { + tokio::spawn(async move { + while let Ok(event_str) = voice_member_rx.recv().await { + if let Ok(event) = serde_json::from_str::(&event_str) { + let event_type = event["type"].as_str().unwrap_or(""); + let channel_id = event["channel_id"].as_str().unwrap_or(""); + let user_id = event["user_id"].as_str().unwrap_or(""); + match event_type { + "joined" => { + info!("voice member joined channel {}: {}", channel_id, user_id); + } + "left" => { + info!("voice member left channel {}: {}", channel_id, user_id); + } + _ => {} + } + } + } + }); + vnox_voice_node::runner::run_bind(&node_name, &voice_bind).await }); let gate_handle = tokio::spawn(async move { - vnox_gateway::run(gate_cfg).await + vnox_gateway::run(gate_cfg, Some(voice_member_tx)).await }); tokio::select! { diff --git a/serverd/src/voice_membership.rs b/serverd/src/voice_membership.rs new file mode 100644 index 0000000..32a376a --- /dev/null +++ b/serverd/src/voice_membership.rs @@ -0,0 +1,22 @@ +use tokio::sync::broadcast; + +#[derive(Clone, Debug)] +pub enum VoiceMembershipEvent { + Joined { + channel_id: String, + session_id: String, + user_id: String, + }, + Left { + channel_id: String, + session_id: String, + user_id: String, + }, +} + +pub fn voice_membership_channel() -> ( + broadcast::Sender, + broadcast::Receiver, +) { + broadcast::channel(256) +}