update hyper to 1.0 and other libs

Reviewed-on: lavina/lavina#28

.dockerignore

@@ -1,6 +1,6 @@
 *
 !/src/
-!/migrations/
+!/crates/
 !Cargo.lock
 !Cargo.toml
 !rust-toolchain

.gitea/workflows/test-pr.yml

@@ -8,6 +8,11 @@ jobs:
       uses: actions/checkout@v4
     - name: setup rust
       uses: https://github.com/actions-rs/toolchain@v1
+    - name: check formatting
+      uses: https://github.com/actions-rs/cargo@v1
+      with:
+        command: fmt
+        args: "--check -p mgmt-api -p lavina-core -p projection-irc -p projection-xmpp -p sasl"
     - name: cargo check
       uses: https://github.com/actions-rs/cargo@v1
       with:
@@ -16,3 +21,4 @@ jobs:
       uses: https://github.com/actions-rs/cargo@v1
       with:
         command: test
+        args: "--all"

.gitignore

@@ -1,3 +1,4 @@
 /target
 /db.sqlite
 .idea/
+.DS_Store

Cargo.toml

@@ -6,10 +6,11 @@ members = [
     "crates/projection-irc",
     "crates/proto-xmpp",
     "crates/mgmt-api",
+    "crates/sasl",
 ]
 
 [workspace.package]
-version = "0.0.1-dev"
+version = "0.0.2-dev"
 
 [workspace.dependencies]
 nom = "7.1.3"
@@ -28,6 +29,8 @@ tracing = "0.1.37" # logging & tracing api
 prometheus = { version = "0.13.3", default-features = false }
 base64 = "0.21.3"
 lavina-core = { path = "crates/lavina-core" }
+tracing-subscriber = "0.3.16"
+sasl = { path = "crates/sasl" }
 
 [package]
 name = "lavina"
@@ -38,13 +41,14 @@ publish = false
 [dependencies]
 anyhow.workspace = true
 figment = { version = "0.10.8", features = ["env", "toml"] } # configuration files
-hyper = { version = "1.0.0-rc.3,<1.0.0-rc.4", features = ["server", "http1"] } # http server
+hyper = { version = "1.0.1", features = ["server", "http1"] } # http server
-http-body-util = "0.1.0-rc.3"
+http-body-util = "0.1.0"
+hyper-util = { version = "0.1", features = ["server", "http1", "tokio"] }
 serde.workspace = true
 serde_json = "1.0.93"
 tokio.workspace = true
 tracing.workspace = true
-tracing-subscriber = "0.3.16"
+tracing-subscriber.workspace = true
 futures-util.workspace = true
 prometheus.workspace = true
 nonempty.workspace = true

crates/README.md

@@ -1,30 +0,0 @@
-## Dependency diagram of the project
-
-```mermaid
-graph TD;
-    lavina-->mgmt-api;
-    lavina-->projection-irc;
-    lavina-->projection-xmpp;
-    lavina-->lavina-core;
-
-    projection-irc-->proto-irc;
-    projection-irc-->lavina-core;
-    
-    projection-xmpp-->proto-xmpp;
-    projection-xmpp-->lavina-core;
-
-    sim-irc-->proto-irc;
-    sim-irc-->mgmt-api;
-
-    sim-xmpp-->proto-xmpp;
-    sim-xmpp-->mgmt-api;
-
-    workspace-->lavina;
-    workspace-->sim-irc;
-    workspace-->sim-xmpp;
-```
-
-A few rules:
-- Only projections should be direct deps of `lavina`, there is no need to depend on `proto-*` crates.
-- On the other hand, projections should not be dependencies of `sim-*` crates.
-- `lavina-core` does not depend on protocol-specific crates.

crates/projection-irc/Cargo.toml

@@ -11,6 +11,10 @@ serde.workspace = true
 tokio.workspace = true
 prometheus.workspace = true
 futures-util.workspace = true
-
 nonempty.workspace = true
+bitflags = "2.4.1"
 proto-irc = { path = "../proto-irc" }
+sasl = { path = "../sasl" }
+
+[dev-dependencies]
+tracing-subscriber.workspace = true

crates/projection-irc/rustfmt.toml

@@ -0,0 +1,2 @@
+max_width = 120
+chain_width = 120

crates/projection-irc/src/cap.rs

@@ -0,0 +1,9 @@
+use bitflags::bitflags;
+
+bitflags! {
+    #[derive(Debug)]
+    pub struct Capabilities: u32 {
+        const None = 0;
+        const Sasl = 1 << 0;
+    }
+}

crates/projection-irc/src/lib.rs

@@ -18,10 +18,17 @@ use lavina_core::prelude::*;
 use lavina_core::repo::Storage;
 use lavina_core::room::{RoomId, RoomInfo, RoomRegistry};
 use lavina_core::terminator::Terminator;
+use proto_irc::client::CapabilitySubcommand;
 use proto_irc::client::{client_message, ClientMessage};
+use proto_irc::server::CapSubBody;
 use proto_irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
 use proto_irc::user::PrefixedNick;
 use proto_irc::{Chan, Recipient};
+use sasl::AuthBody;
+
+mod cap;
+
+use crate::cap::Capabilities;
 
 #[derive(Deserialize, Debug, Clone)]
 pub struct ServerConfig {
@@ -55,28 +62,16 @@ async fn handle_socket(
     let mut reader: BufReader<ReadHalf> = BufReader::new(reader);
     let mut writer = BufWriter::new(writer);
 
-    ServerMessage {
+    let registered_user: Result<RegisteredUser> =
-        tags: vec![],
+        handle_registration(&mut reader, &mut writer, &mut storage, &config).await;
-        sender: Some(config.server_name.clone().into()),
-        body: ServerMessageBody::Notice {
-            first_target: "*".into(),
-            rest_targets: vec![],
-            text: "Welcome to my server!".into(),
-        },
-    }
-    .write_async(&mut writer)
-    .await?;
-    writer.flush().await?;
-
-    let registered_user: Result<RegisteredUser> = handle_registration(&mut reader, &mut writer, &mut storage).await;
 
     match registered_user {
         Ok(user) => {
             log::debug!("User registered");
             handle_registered_socket(config, players, rooms, &mut reader, &mut writer, user).await?;
         }
-        Err(_) => {
+        Err(err) => {
-            log::debug!("Registration failed");
+            log::debug!("Registration failed: {err}");
         }
     }
 
@@ -88,93 +83,239 @@ async fn handle_registration<'a>(
     reader: &mut BufReader<ReadHalf<'a>>,
     writer: &mut BufWriter<WriteHalf<'a>>,
     storage: &mut Storage,
+    config: &ServerConfig,
 ) -> Result<RegisteredUser> {
     let mut buffer = vec![];
 
     let mut future_nickname: Option<Str> = None;
     let mut future_username: Option<(Str, Str)> = None;
+    let mut enabled_capabilities = Capabilities::None;
+    let mut cap_negotiation_in_progress = false; // if true, expect `CAP END` to complete registration
 
     let mut pass: Option<Str> = None;
+    let mut authentication_started = false;
+    let mut validated_user = None;
 
     let user = loop {
         let res = read_irc_message(reader, &mut buffer).await;
-        let res = match res {
+        tracing::trace!("Received message: {:?}", res);
-            Ok(len) => {
+        let len = match res {
-                if len == 0 {
+            Ok(len) => len,
-                    log::info!("Terminating socket");
-                    break Err(anyhow::Error::msg("EOF"));
-                }
-                match std::str::from_utf8(&buffer[..len-2]) {
-                    Ok(res) => res,
-                    Err(e) => break Err(e.into()),
-                }
-            }
             Err(err) => {
                 log::warn!("Failed to read from socket: {err}");
                 break Err(err.into());
             }
         };
-        log::debug!("Incoming raw IRC message: '{res}'");
+        if len == 0 {
+            log::info!("Terminating socket");
+            break Err(anyhow::Error::msg("EOF"));
+        }
+        let res = match std::str::from_utf8(&buffer[..len - 2]) {
+            Ok(res) => res,
+            Err(e) => break Err(e.into()),
+        };
+        tracing::trace!("Incoming raw IRC message: '{res}'");
         let parsed = client_message(res);
-        match parsed {
+        let msg = match parsed {
-            Ok(msg) => {
+            Ok(msg) => msg,
-                log::debug!("Incoming IRC message: {msg:?}");
+            Err(err) => {
-                match msg {
+                tracing::warn!("Failed to parse IRC message: {err}");
-                    ClientMessage::Pass { password } => {
+                buffer.clear();
-                        pass = Some(password);
+                continue;
+            }
+        };
+        tracing::debug!("Incoming IRC message: {msg:?}");
+        match msg {
+            ClientMessage::Pass { password } => {
+                pass = Some(password);
+            }
+            ClientMessage::Capability { subcommand } => match subcommand {
+                CapabilitySubcommand::List { code: _ } => {
+                    cap_negotiation_in_progress = true;
+                    ServerMessage {
+                        tags: vec![],
+                        sender: Some(config.server_name.clone().into()),
+                        body: ServerMessageBody::Cap {
+                            target: future_nickname.clone().unwrap_or_else(|| "*".into()),
+                            subcmd: CapSubBody::Ls("sasl=PLAIN".into()),
+                        },
                     }
-                    ClientMessage::Nick { nickname } => {
+                    .write_async(writer)
-                        if let Some((username, realname)) = future_username {
+                    .await?;
-                            break Ok(RegisteredUser {
+                    writer.flush().await?;
-                                nickname,
+                }
-                                username,
+                CapabilitySubcommand::Req(caps) => {
-                                realname,
+                    cap_negotiation_in_progress = true;
-                            });
+                    let mut acked = vec![];
+                    let mut naked = vec![];
+                    for cap in caps {
+                        if &*cap.name == "sasl" {
+                            if cap.to_disable {
+                                enabled_capabilities &= !Capabilities::Sasl;
+                            } else {
+                                enabled_capabilities |= Capabilities::Sasl;
+                            }
+                            acked.push(cap);
                         } else {
-                            future_nickname = Some(nickname);
+                            naked.push(cap);
                         }
                     }
-                    ClientMessage::User { username, realname } => {
+                    let mut ack_body = String::new();
-                        if let Some(nickname) = future_nickname {
+                    for cap in acked {
-                            break Ok(RegisteredUser {
+                        if cap.to_disable {
-                                nickname,
+                            ack_body.push('-');
-                                username,
-                                realname,
-                            });
-                        } else {
-                            future_username = Some((username, realname));
                         }
+                        ack_body += &*cap.name;
                     }
-                    _ => {}
+                    ServerMessage {
+                        tags: vec![],
+                        sender: Some(config.server_name.clone().into()),
+                        body: ServerMessageBody::Cap {
+                            target: future_nickname.clone().unwrap_or_else(|| "*".into()),
+                            subcmd: CapSubBody::Ack(ack_body.into()),
+                        },
+                    }
+                    .write_async(writer)
+                    .await?;
+                    writer.flush().await?;
+                }
+                CapabilitySubcommand::End => {
+                    let Some((username, realname)) = future_username else {
+                        todo!()
+                    };
+                    let Some(nickname) = future_nickname.clone() else {
+                        todo!()
+                    };
+                    let candidate_user = RegisteredUser {
+                        nickname,
+                        username,
+                        realname,
+                    };
+                    if enabled_capabilities.contains(Capabilities::Sasl)
+                        && validated_user.as_ref() == Some(&candidate_user.nickname)
+                    {
+                        break Ok(candidate_user);
+                    } else {
+                        let Some(candidate_password) = pass else {
+                            todo!();
+                        };
+                        auth_user(storage, &*candidate_user.nickname, &*candidate_password).await?;
+                        break Ok(candidate_user);
+                    }
+                }
+            },
+            ClientMessage::Nick { nickname } => {
+                if cap_negotiation_in_progress {
+                    future_nickname = Some(nickname);
+                } else if let Some((username, realname)) = future_username.clone() {
+                    let candidate_user = RegisteredUser {
+                        nickname,
+                        username,
+                        realname,
+                    };
+                    let Some(candidate_password) = pass else {
+                        todo!();
+                    };
+                    auth_user(storage, &*candidate_user.nickname, &*candidate_password).await?;
+                    break Ok(candidate_user);
+                } else {
+                    future_nickname = Some(nickname);
                 }
             }
-            Err(err) => {
+            ClientMessage::User { username, realname } => {
-                log::warn!("Failed to parse IRC message: {err}");
+                if cap_negotiation_in_progress {
+                    future_username = Some((username, realname));
+                } else if let Some(nickname) = future_nickname.clone() {
+                    let candidate_user = RegisteredUser {
+                        nickname,
+                        username,
+                        realname,
+                    };
+                    let Some(candidate_password) = pass else {
+                        todo!();
+                    };
+                    auth_user(storage, &*candidate_user.nickname, &*candidate_password).await?;
+                    break Ok(candidate_user);
+                } else {
+                    future_username = Some((username, realname));
+                }
             }
+            ClientMessage::Authenticate(body) => {
+                if !authentication_started {
+                    tracing::debug!("Received authentication request");
+                    if &*body == "PLAIN" {
+                        tracing::debug!("Authentication request with method PLAIN");
+                        authentication_started = true;
+                        ServerMessage {
+                            tags: vec![],
+                            sender: Some(config.server_name.clone().into()),
+                            body: ServerMessageBody::Authenticate("+".into()),
+                        }
+                        .write_async(writer)
+                        .await?;
+                        writer.flush().await?;
+                    } else {
+                        // TODO respond with 904
+                        todo!();
+                    }
+                } else {
+                    let body = AuthBody::from_str(body.as_bytes())?;
+                    auth_user(storage, &body.login, &body.password).await?;
+                    let login: Str = body.login.into();
+                    validated_user = Some(login.clone());
+                    ServerMessage {
+                        tags: vec![],
+                        sender: Some(config.server_name.clone().into()),
+                        body: ServerMessageBody::N900LoggedIn {
+                            nick: login.clone(),
+                            address: login.clone(),
+                            account: login.clone(),
+                            message: format!("You are now logged in as {}", login).into(),
+                        },
+                    }
+                    .write_async(writer)
+                    .await?;
+                    ServerMessage {
+                        tags: vec![],
+                        sender: Some(config.server_name.clone().into()),
+                        body: ServerMessageBody::N903SaslSuccess {
+                            nick: login.clone(),
+                            message: "SASL authentication successful".into(),
+                        },
+                    }
+                    .write_async(writer)
+                    .await?;
+                    writer.flush().await?;
+                }
+                // TODO handle abortion of authentication
+            }
+            _ => {}
         }
         buffer.clear();
     }?;
+    // TODO properly implement session temination
+    Ok(user)
+}
 
-    let stored_user = storage.retrieve_user_by_name(&*user.nickname).await?;
+async fn auth_user(storage: &mut Storage, login: &str, plain_password: &str) -> Result<()> {
+    let stored_user = storage.retrieve_user_by_name(login).await?;
 
     let stored_user = match stored_user {
         Some(u) => u,
         None => {
-            log::info!("User '{}' not found", user.nickname);
+            log::info!("User '{}' not found", login);
             return Err(anyhow!("no user found"));
         }
     };
-    if stored_user.password.is_none() {
+    let Some(expected_password) = stored_user.password else {
-        log::info!("Password not defined for user '{}'", user.nickname);
+        log::info!("Password not defined for user '{}'", login);
         return Err(anyhow!("password is not defined"));
-    }
+    };
-    if stored_user.password.as_deref() != pass.as_deref() {
+    if expected_password != plain_password {
-        log::info!("Incorrect password supplied for user '{}'", user.nickname);
+        log::info!("Incorrect password supplied for user '{}'", login);
         return Err(anyhow!("passwords do not match"));
     }
-    // TODO properly implement session temination
+    Ok(())
-
-    Ok(user)
 }
 
 async fn handle_registered_socket<'a>(
@@ -292,7 +433,9 @@ async fn handle_registered_socket<'a>(
     Ok(())
 }
 
-async fn read_irc_message(reader: &mut BufReader<ReadHalf<'_>>, buf: &mut Vec<u8>) -> Result<usize> {
+// TODO this is public only for tests, perhaps move this into proto-irc
+// TODO limit buffer size in size to protect against dos attacks with large payloads
+pub async fn read_irc_message(reader: &mut BufReader<ReadHalf<'_>>, buf: &mut Vec<u8>) -> Result<usize> {
     let mut size = 0;
     'outer: loop {
         let res = reader.read_until(b'\r', buf).await?;
@@ -300,8 +443,8 @@ async fn read_irc_message(reader: &mut BufReader<ReadHalf<'_>>, buf: &mut Vec<u8
         let next = reader.read_u8().await?;
         buf.push(next);
         size += 1;
-        if next != b'\n' { 
+        if next != b'\n' {
-            continue 'outer; 
+            continue 'outer;
         }
         return Ok(size);
     }
@@ -668,11 +811,8 @@ async fn produce_on_join_cmd_messages(
     }
     .write_async(writer)
     .await?;
-    let prefixed_members: Vec<PrefixedNick> = room_info
+    let prefixed_members: Vec<PrefixedNick> =
-        .members
+        room_info.members.iter().map(|member| PrefixedNick::from_str(member.clone().into_inner())).collect();
-        .iter()
-        .map(|member| PrefixedNick::from_str(member.clone().into_inner()))
-        .collect();
     let non_empty_members: NonEmpty<PrefixedNick> =
         NonEmpty::from_vec(prefixed_members).unwrap_or(nonempty![PrefixedNick::from_str(user.nickname.clone())]);
 
@@ -700,13 +840,24 @@ async fn produce_on_join_cmd_messages(
     Ok(())
 }
 
+pub struct RunningServer {
+    pub addr: SocketAddr,
+    terminator: Terminator,
+}
+
+impl RunningServer {
+    pub async fn terminate(self) -> Result<()> {
+        self.terminator.terminate().await
+    }
+}
+
 pub async fn launch(
     config: ServerConfig,
     players: PlayerRegistry,
     rooms: RoomRegistry,
     metrics: MetricsRegistry,
     storage: Storage,
-) -> Result<Terminator> {
+) -> Result<RunningServer> {
     log::info!("Starting IRC projection");
     let (stopped_tx, mut stopped_rx) = channel(32);
     let current_connections = IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
@@ -715,6 +866,7 @@ pub async fn launch(
     metrics.register(Box::new(total_connections.clone()))?;
 
     let listener = TcpListener::bind(config.listen_on).await?;
+    let addr = listener.local_addr()?;
 
     let terminator = Terminator::spawn(|mut rx| async move {
         // TODO probably should separate logic for accepting new connection and storing them
@@ -782,5 +934,5 @@ pub async fn launch(
     });
 
     log::info!("Started IRC projection");
-    Ok(terminator)
+    Ok(RunningServer { addr, terminator })
 }

crates/projection-irc/tests/lib.rs

@@ -0,0 +1,220 @@
+use std::time::Duration;
+
+use anyhow::{anyhow, Result};
+use prometheus::Registry as MetricsRegistry;
+use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
+use tokio::net::tcp::{ReadHalf, WriteHalf};
+use tokio::net::TcpStream;
+
+use lavina_core::repo::{Storage, StorageConfig};
+use lavina_core::{player::PlayerRegistry, room::RoomRegistry};
+use projection_irc::{launch, read_irc_message, RunningServer, ServerConfig};
+
+struct TestScope<'a> {
+    reader: BufReader<ReadHalf<'a>>,
+    writer: WriteHalf<'a>,
+    buffer: Vec<u8>,
+    pub timeout: Duration,
+}
+
+impl<'a> TestScope<'a> {
+    fn new(stream: &mut TcpStream) -> TestScope<'_> {
+        let (reader, writer) = stream.split();
+        let reader = BufReader::new(reader);
+        let buffer = vec![];
+        let timeout = Duration::from_millis(100);
+        TestScope {
+            reader,
+            writer,
+            buffer,
+            timeout,
+        }
+    }
+
+    async fn send(&mut self, str: &(impl AsRef<str> + ?Sized)) -> Result<()> {
+        self.writer.write_all(str.as_ref().as_bytes()).await?;
+        self.writer.write_all(b"\r\n").await?;
+        self.writer.flush().await?;
+        Ok(())
+    }
+
+    async fn expect(&mut self, str: &str) -> Result<()> {
+        tracing::debug!("Expecting {}", str);
+        let len = tokio::time::timeout(self.timeout, read_irc_message(&mut self.reader, &mut self.buffer)).await??;
+        assert_eq!(std::str::from_utf8(&self.buffer[..len - 2])?, str);
+        self.buffer.clear();
+        Ok(())
+    }
+
+    async fn expect_eof(&mut self) -> Result<()> {
+        let mut buf = [0; 1];
+        let len = tokio::time::timeout(self.timeout, self.reader.read(&mut buf)).await??;
+        if len != 0 {
+            return Err(anyhow!("not a eof"));
+        }
+        Ok(())
+    }
+
+    async fn expect_nothing(&mut self) -> Result<()> {
+        let mut buf = [0; 1];
+        match tokio::time::timeout(self.timeout, self.reader.read(&mut buf)).await {
+            Ok(res) => Err(anyhow!("received something: {:?}", res)),
+            Err(_) => Ok(()),
+        }
+    }
+}
+
+struct TestServer {
+    metrics: MetricsRegistry,
+    storage: Storage,
+    rooms: RoomRegistry,
+    players: PlayerRegistry,
+    server: RunningServer,
+}
+impl TestServer {
+    async fn start() -> Result<TestServer> {
+        let _ = tracing_subscriber::fmt::try_init();
+        let config = ServerConfig {
+            listen_on: "127.0.0.1:0".parse().unwrap(),
+            server_name: "testserver".into(),
+        };
+        let mut metrics = MetricsRegistry::new();
+        let mut storage = Storage::open(StorageConfig {
+            db_path: ":memory:".into(),
+        })
+        .await?;
+        let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
+        let players = PlayerRegistry::empty(rooms.clone(), &mut metrics).unwrap();
+        let server = launch(config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await.unwrap();
+        Ok(TestServer {
+            metrics,
+            storage,
+            rooms,
+            players,
+            server,
+        })
+    }
+}
+
+#[tokio::test]
+async fn scenario_basic() -> Result<()> {
+    let mut server = TestServer::start().await?;
+
+    // test scenario
+
+    server.storage.create_user("tester").await?;
+    server.storage.set_password("tester", "password").await?;
+
+    let mut stream = TcpStream::connect(server.server.addr).await?;
+    let mut s = TestScope::new(&mut stream);
+
+    s.send("PASS password").await?;
+    s.send("NICK tester").await?;
+    s.send("USER UserName 0 * :Real Name").await?;
+    s.expect(":testserver 001 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 002 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 003 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 004 tester testserver kek-0.1.alpha.3 r CFILPQbcefgijklmnopqrstvz").await?;
+    s.expect(":testserver 005 tester CHANTYPES=# :are supported by this server").await?;
+    s.expect_nothing().await?;
+    s.send("QUIT :Leaving").await?;
+    s.expect(":testserver ERROR :Leaving the server").await?;
+    s.expect_eof().await?;
+
+    stream.shutdown().await?;
+
+    // wrap up
+
+    server.server.terminate().await?;
+    Ok(())
+}
+
+/*
+IRC SASL doc: https://ircv3.net/specs/extensions/sasl-3.1.html
+AUTHENTICATE doc: https://modern.ircdocs.horse/#authenticate-message
+*/
+#[tokio::test]
+async fn scenario_cap_full_negotiation() -> Result<()> {
+    let mut server = TestServer::start().await?;
+
+    // test scenario
+
+    server.storage.create_user("tester").await?;
+    server.storage.set_password("tester", "password").await?;
+
+    let mut stream = TcpStream::connect(server.server.addr).await?;
+    let mut s = TestScope::new(&mut stream);
+
+    s.send("CAP LS 302").await?;
+    s.send("NICK tester").await?;
+    s.send("USER UserName 0 * :Real Name").await?;
+    s.expect(":testserver CAP * LS :sasl=PLAIN").await?;
+    s.send("CAP REQ :sasl").await?;
+    s.expect(":testserver CAP tester ACK :sasl").await?;
+    s.send("AUTHENTICATE PLAIN").await?;
+    s.expect(":testserver AUTHENTICATE +").await?;
+    s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZA==").await?; // base64-encoded 'tester\x00tester\x00password'
+    s.expect(":testserver 900 tester tester tester :You are now logged in as tester").await?;
+    s.expect(":testserver 903 tester :SASL authentication successful").await?;
+
+    s.send("CAP END").await?;
+
+    s.expect(":testserver 001 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 002 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 003 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 004 tester testserver kek-0.1.alpha.3 r CFILPQbcefgijklmnopqrstvz").await?;
+    s.expect(":testserver 005 tester CHANTYPES=# :are supported by this server").await?;
+    s.expect_nothing().await?;
+    s.send("QUIT :Leaving").await?;
+    s.expect(":testserver ERROR :Leaving the server").await?;
+    s.expect_eof().await?;
+
+    stream.shutdown().await?;
+
+    // wrap up
+
+    server.server.terminate().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn scenario_cap_short_negotiation() -> Result<()> {
+    let mut server = TestServer::start().await?;
+
+    // test scenario
+
+    server.storage.create_user("tester").await?;
+    server.storage.set_password("tester", "password").await?;
+
+    let mut stream = TcpStream::connect(server.server.addr).await?;
+    let mut s = TestScope::new(&mut stream);
+
+    s.send("NICK tester").await?;
+    s.send("CAP REQ :sasl").await?;
+    s.send("USER UserName 0 * :Real Name").await?;
+    s.expect(":testserver CAP tester ACK :sasl").await?;
+    s.send("AUTHENTICATE PLAIN").await?;
+    s.expect(":testserver AUTHENTICATE +").await?;
+    s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZA==").await?; // base64-encoded 'tester\x00tester\x00password'
+    s.expect(":testserver 900 tester tester tester :You are now logged in as tester").await?;
+    s.expect(":testserver 903 tester :SASL authentication successful").await?;
+
+    s.send("CAP END").await?;
+
+    s.expect(":testserver 001 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 002 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 003 tester :Welcome to Kek Server").await?;
+    s.expect(":testserver 004 tester testserver kek-0.1.alpha.3 r CFILPQbcefgijklmnopqrstvz").await?;
+    s.expect(":testserver 005 tester CHANTYPES=# :are supported by this server").await?;
+    s.expect_nothing().await?;
+    s.send("QUIT :Leaving").await?;
+    s.expect(":testserver ERROR :Leaving the server").await?;
+    s.expect_eof().await?;
+
+    stream.shutdown().await?;
+
+    // wrap up
+
+    server.server.terminate().await?;
+    Ok(())
+}

crates/projection-xmpp/Cargo.toml

@@ -13,8 +13,13 @@ prometheus.workspace = true
 futures-util.workspace = true
 
 quick-xml.workspace = true
+sasl.workspace = true
 proto-xmpp = { path = "../proto-xmpp" }
 uuid = { version = "1.3.0", features = ["v4"] }
-tokio-rustls = "0.24.1"
+tokio-rustls = { version = "0.24.1", features = ["dangerous_configuration"] }
 rustls-pemfile = "1.0.2"
 derive_more.workspace = true
+
+[dev-dependencies]
+tracing-subscriber.workspace = true
+assert_matches = "1.5.0"

crates/projection-xmpp/rustfmt.toml

@@ -0,0 +1,2 @@
+max_width = 120
+chain_width = 120

crates/projection-xmpp/src/iq.rs

@@ -0,0 +1,162 @@
+//! Handling of all client2server iq stanzas
+
+use quick_xml::events::Event;
+
+use lavina_core::room::RoomRegistry;
+use proto_xmpp::bind::{BindResponse, Jid, Name, Resource, Server};
+use proto_xmpp::client::{Iq, IqType};
+use proto_xmpp::disco::{Feature, Identity, InfoQuery, Item, ItemQuery};
+use proto_xmpp::roster::RosterQuery;
+use proto_xmpp::session::Session;
+
+use crate::proto::IqClientBody;
+use crate::XmppConnection;
+
+use proto_xmpp::xml::ToXml;
+
+impl<'a> XmppConnection<'a> {
+    pub async fn handle_iq(&self, output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>) {
+        match iq.body {
+            IqClientBody::Bind(b) => {
+                let req = Iq {
+                    from: None,
+                    id: iq.id,
+                    to: None,
+                    r#type: IqType::Result,
+                    body: BindResponse(Jid {
+                        name: Some(Name("darova".into())),
+                        server: Server("localhost".into()),
+                        resource: Some(Resource("kek".into())),
+                    }),
+                };
+                req.serialize(output);
+            }
+            IqClientBody::Session(_) => {
+                let req = Iq {
+                    from: None,
+                    id: iq.id,
+                    to: None,
+                    r#type: IqType::Result,
+                    body: Session,
+                };
+                req.serialize(output);
+            }
+            IqClientBody::Roster(_) => {
+                let req = Iq {
+                    from: None,
+                    id: iq.id,
+                    to: None,
+                    r#type: IqType::Result,
+                    body: RosterQuery,
+                };
+                req.serialize(output);
+            }
+            IqClientBody::DiscoInfo(info) => {
+                let response = disco_info(iq.to.as_deref(), &info);
+                let req = Iq {
+                    from: iq.to,
+                    id: iq.id,
+                    to: None,
+                    r#type: IqType::Result,
+                    body: response,
+                };
+                req.serialize(output);
+            }
+            IqClientBody::DiscoItem(item) => {
+                let response = disco_items(iq.to.as_deref(), &item, self.rooms).await;
+                let req = Iq {
+                    from: iq.to,
+                    id: iq.id,
+                    to: None,
+                    r#type: IqType::Result,
+                    body: response,
+                };
+                req.serialize(output);
+            }
+            _ => {
+                let req = Iq {
+                    from: None,
+                    id: iq.id,
+                    to: None,
+                    r#type: IqType::Error,
+                    body: (),
+                };
+                req.serialize(output);
+            }
+        }
+    }
+}
+
+fn disco_info(to: Option<&str>, req: &InfoQuery) -> InfoQuery {
+    let identity;
+    let feature;
+    match to {
+        Some("localhost") => {
+            identity = vec![Identity {
+                category: "server".into(),
+                name: None,
+                r#type: "im".into(),
+            }];
+            feature = vec![
+                Feature::new("http://jabber.org/protocol/disco#info"),
+                Feature::new("http://jabber.org/protocol/disco#items"),
+                Feature::new("iq"),
+                Feature::new("presence"),
+            ]
+        }
+        Some("rooms.localhost") => {
+            identity = vec![Identity {
+                category: "conference".into(),
+                name: Some("Chat rooms".into()),
+                r#type: "text".into(),
+            }];
+            feature = vec![
+                Feature::new("http://jabber.org/protocol/disco#info"),
+                Feature::new("http://jabber.org/protocol/disco#items"),
+                Feature::new("http://jabber.org/protocol/muc"),
+            ]
+        }
+        _ => {
+            identity = vec![];
+            feature = vec![];
+        }
+    };
+    InfoQuery {
+        node: None,
+        identity,
+        feature,
+    }
+}
+
+async fn disco_items(to: Option<&str>, req: &ItemQuery, rooms: &RoomRegistry) -> ItemQuery {
+    let item = match to {
+        Some("localhost") => {
+            vec![Item {
+                jid: Jid {
+                    name: None,
+                    server: Server("rooms.localhost".into()),
+                    resource: None,
+                },
+                name: None,
+                node: None,
+            }]
+        }
+        Some("rooms.localhost") => {
+            let room_list = rooms.get_all_rooms().await;
+            room_list
+                .into_iter()
+                .map(|room_info| Item {
+                    jid: Jid {
+                        name: Some(Name(room_info.id.into_inner())),
+                        server: Server("rooms.localhost".into()),
+                        resource: None,
+                    },
+                    name: None,
+                    node: None,
+                })
+                .collect()
+        }
+        _ => vec![],
+    };
+    ItemQuery { item }
+}

crates/projection-xmpp/src/lib.rs

@@ -1,4 +1,4 @@
-#![feature(generators, generator_trait, type_alias_impl_trait, impl_trait_in_assoc_type)]
+#![feature(coroutines, coroutine_trait, type_alias_impl_trait, impl_trait_in_assoc_type)]
 
 mod proto;
 
@@ -24,19 +24,20 @@ use tokio_rustls::TlsAcceptor;
 
 use lavina_core::player::{PlayerConnection, PlayerId, PlayerRegistry};
 use lavina_core::prelude::*;
-use lavina_core::room::{RoomId, RoomRegistry};
-use lavina_core::terminator::Terminator;
 use lavina_core::repo::Storage;
-use proto_xmpp::bind::{BindResponse, Jid, Name, Resource, Server};
+use lavina_core::room::RoomRegistry;
-use proto_xmpp::client::{Iq, Message, MessageType, Presence};
+use lavina_core::terminator::Terminator;
-use proto_xmpp::disco::*;
+use proto_xmpp::bind::{Name, Resource};
-use proto_xmpp::roster::RosterQuery;
-use proto_xmpp::sasl::AuthBody;
-use proto_xmpp::session::Session;
 use proto_xmpp::stream::*;
 use proto_xmpp::xml::{Continuation, FromXml, Parser, ToXml};
+use sasl::AuthBody;
 
-use self::proto::{ClientPacket, IqClientBody};
+use self::proto::ClientPacket;
+
+mod iq;
+mod message;
+mod presence;
+mod updates;
 
 #[derive(Deserialize, Debug, Clone)]
 pub struct ServerConfig {
@@ -57,13 +58,24 @@ struct Authenticated {
     xmpp_muc_name: Resource,
 }
 
+pub struct RunningServer {
+    pub addr: SocketAddr,
+    terminator: Terminator,
+}
+
+impl RunningServer {
+    pub async fn terminate(self) -> Result<()> {
+        self.terminator.terminate().await
+    }
+}
+
 pub async fn launch(
     config: ServerConfig,
     players: PlayerRegistry,
     rooms: RoomRegistry,
     metrics: MetricsRegistry,
     storage: Storage,
-) -> Result<Terminator> {
+) -> Result<RunningServer> {
     log::info!("Starting XMPP projection");
 
     let certs = certs(&mut SyncBufReader::new(File::open(config.cert)?))?;
@@ -80,6 +92,8 @@ pub async fn launch(
     });
 
     let listener = TcpListener::bind(config.listen_on).await?;
+    let addr = listener.local_addr()?;
+
     let terminator = Terminator::spawn(|mut termination| async move {
         let (stopped_tx, mut stopped_rx) = channel(32);
         let mut actors = HashMap::new();
@@ -138,7 +152,7 @@ pub async fn launch(
         Ok(())
     });
     log::info!("Started XMPP projection");
-    Ok(terminator)
+    Ok(RunningServer { addr, terminator })
 }
 
 async fn handle_socket(
@@ -164,6 +178,7 @@ async fn handle_socket(
         .with_single_cert(vec![config.cert.clone()], config.key.clone())?;
     config.key_log = Arc::new(tokio_rustls::rustls::KeyLogFile::new());
 
+    log::debug!("Accepting TLS connection...");
     let acceptor = TlsAcceptor::from(Arc::new(config));
     let new_stream = acceptor.accept(stream).await?;
     log::debug!("TLS connection established");
@@ -234,9 +249,7 @@ async fn socket_auth(
     read_xml_header(xml_reader, reader_buf).await?;
     let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?;
 
-    xml_writer
+    xml_writer.write_event_async(Event::Decl(BytesDecl::new("1.0", None, None))).await?;
-        .write_event_async(Event::Decl(BytesDecl::new("1.0", None, None)))
-        .await?;
     ServerStreamStart {
         from: "localhost".into(),
         lang: "en".into(),
@@ -286,8 +299,8 @@ async fn socket_auth(
                 xmpp_resource: Resource(name.to_string().into()),
                 xmpp_muc_name: Resource(name.to_string().into()),
             })
-        },
+        }
-        Err(e) => return Err(e)
+        Err(e) => return Err(e),
     }
 }
 
@@ -302,9 +315,7 @@ async fn socket_final(
     read_xml_header(xml_reader, reader_buf).await?;
     let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?;
 
-    xml_writer
+    xml_writer.write_event_async(Event::Decl(BytesDecl::new("1.0", None, None))).await?;
-        .write_event_async(Event::Decl(BytesDecl::new("1.0", None, None)))
-        .await?;
     ServerStreamStart {
         from: "localhost".into(),
         lang: "en".into(),
@@ -328,6 +339,11 @@ async fn socket_final(
     let mut next_xml_event = Box::pin(xml_reader.read_resolved_event_into_async(reader_buf));
 
     'outer: loop {
+        let mut conn = XmppConnection {
+            user: authenticated,
+            user_handle,
+            rooms,
+        };
         let should_recreate_xml_future = select! {
             biased;
             res = &mut next_xml_event => 's: {
@@ -340,7 +356,7 @@ async fn socket_final(
                 match parser.consume(ns, &event) {
                     Continuation::Final(res) => {
                         let res = res?;
-                        let stop = handle_packet(&mut events, res, authenticated, user_handle, rooms).await?;
+                        let stop = conn.handle_packet(&mut events, res).await?;
                         for i in &events {
                             xml_writer.write_event_async(i).await?;
                         }
@@ -355,32 +371,9 @@ async fn socket_final(
                 }
                 true
             },
-            update = user_handle.receiver.recv() => {
+            update = conn.user_handle.receiver.recv() => {
                 if let Some(update) = update {
-                    match update {
+                    conn.handle_update(&mut events, update).await?;
-                        lavina_core::player::Updates::NewMessage { room_id, author_id, body } => {
-                            Message::<()> {
-                                to: Some(Jid {
-                                    name: Some(authenticated.xmpp_name.clone()),
-                                    server: Server("localhost".into()),
-                                    resource: Some(authenticated.xmpp_resource.clone()),
-                                }),
-                                from: Some(Jid {
-                                    name: Some(Name(room_id.into_inner().into())),
-                                    server: Server("rooms.localhost".into()),
-                                    resource: Some(Resource(author_id.into_inner().into())),
-                                }),
-                                id: None,
-                                r#type: proto_xmpp::client::MessageType::Groupchat,
-                                lang: None,
-                                subject: None,
-                                body: body.into(),
-                                custom: vec![],
-                            }
-                            .serialize(&mut events);
-                        }
-                        _ => {},
-                    }
                     for i in &events {
                         xml_writer.write_event_async(i).await?;
                     }
@@ -402,249 +395,36 @@ async fn socket_final(
     Ok(())
 }
 
-async fn handle_packet(
+struct XmppConnection<'a> {
-    output: &mut Vec<Event<'static>>,
+    user: &'a Authenticated,
-    packet: ClientPacket,
+    user_handle: &'a mut PlayerConnection,
-    user: &Authenticated,
+    rooms: &'a RoomRegistry,
-    user_handle: &mut PlayerConnection,
+}
-    rooms: &RoomRegistry,
+
-) -> Result<bool> {
+impl<'a> XmppConnection<'a> {
-    Ok(match packet {
+    async fn handle_packet(&mut self, output: &mut Vec<Event<'static>>, packet: ClientPacket) -> Result<bool> {
-        proto::ClientPacket::Iq(iq) => {
+        let res = match packet {
-            handle_iq(output, iq, rooms).await;
+            proto::ClientPacket::Iq(iq) => {
-            false
+                self.handle_iq(output, iq).await;
-        }
+                false
-        proto::ClientPacket::Message(m) => {
-            if let Some(Jid {
-                name: Some(name),
-                server,
-                resource: _,
-            }) = m.to
-            {
-                if server.0.as_ref() == "rooms.localhost" && m.r#type == MessageType::Groupchat {
-                    user_handle
-                        .send_message(RoomId::from(name.0.clone())?, m.body.clone().into())
-                        .await?;
-                    Message::<()> {
-                        to: Some(Jid {
-                            name: Some(user.xmpp_name.clone()),
-                            server: Server("localhost".into()),
-                            resource: Some(user.xmpp_resource.clone()),
-                        }),
-                        from: Some(Jid {
-                            name: Some(name),
-                            server: Server("rooms.localhost".into()),
-                            resource: Some(user.xmpp_muc_name.clone()),
-                        }),
-                        id: m.id,
-                        r#type: proto_xmpp::client::MessageType::Groupchat,
-                        lang: None,
-                        subject: None,
-                        body: m.body.clone(),
-                        custom: vec![],
-                    }
-                    .serialize(output);
-                    false
-                } else {
-                    todo!()
-                }
-            } else {
-                todo!()
             }
-        }
+            ClientPacket::Message(m) => {
-        proto::ClientPacket::Presence(p) => {
+                self.handle_message(output, m).await?;
-            let response = if p.to.is_none() {
+                false
-                Presence::<()> {
+            }
-                    to: Some(Jid {
+            proto::ClientPacket::Presence(p) => {
-                        name: Some(user.xmpp_name.clone()),
+                self.handle_presence(output, p).await?;
-                        server: Server("localhost".into()),
+                false
-                        resource: Some(user.xmpp_resource.clone()),
+            }
-                    }),
+            proto::ClientPacket::StreamEnd => {
-                    from: Some(Jid {
+                ServerStreamEnd.serialize(output);
-                        name: Some(user.xmpp_name.clone()),
+                true
-                        server: Server("localhost".into()),
+            }
-                        resource: Some(user.xmpp_resource.clone()),
+        };
-                    }),
+        Ok(res)
-                    ..Default::default()
-                }
-            } else if let Some(Jid {
-                name: Some(name),
-                server,
-                resource: Some(resource),
-            }) = p.to
-            {
-                let a = user_handle.join_room(RoomId::from(name.0.clone())?).await?;
-                Presence::<()> {
-                    to: Some(Jid {
-                        name: Some(user.xmpp_name.clone()),
-                        server: Server("localhost".into()),
-                        resource: Some(user.xmpp_resource.clone()),
-                    }),
-                    from: Some(Jid {
-                        name: Some(name.clone()),
-                        server: Server("rooms.localhost".into()),
-                        resource: Some(user.xmpp_muc_name.clone()),
-                    }),
-                    ..Default::default()
-                }
-            } else {
-                Presence::<()>::default()
-            };
-            response.serialize(output);
-            false
-        }
-        proto::ClientPacket::StreamEnd => {
-            ServerStreamEnd.serialize(output);
-            true
-        }
-    })
-}
-
-async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms: &RoomRegistry) {
-    match iq.body {
-        proto::IqClientBody::Bind(b) => {
-            let req = Iq {
-                from: None,
-                id: iq.id,
-                to: None,
-                r#type: proto_xmpp::client::IqType::Result,
-                body: BindResponse(Jid {
-                    name: Some(Name("darova".into())),
-                    server: Server("localhost".into()),
-                    resource: Some(Resource("kek".into())),
-                }),
-            };
-            req.serialize(output);
-        }
-        proto::IqClientBody::Session(_) => {
-            let req = Iq {
-                from: None,
-                id: iq.id,
-                to: None,
-                r#type: proto_xmpp::client::IqType::Result,
-                body: Session,
-            };
-            req.serialize(output);
-        }
-        proto::IqClientBody::Roster(_) => {
-            let req = Iq {
-                from: None,
-                id: iq.id,
-                to: None,
-                r#type: proto_xmpp::client::IqType::Result,
-                body: RosterQuery,
-            };
-            req.serialize(output);
-        }
-        proto::IqClientBody::DiscoInfo(info) => {
-            let response = disco_info(iq.to.as_deref(), &info);
-            let req = Iq {
-                from: iq.to,
-                id: iq.id,
-                to: None,
-                r#type: proto_xmpp::client::IqType::Result,
-                body: response,
-            };
-            req.serialize(output);
-        }
-        proto::IqClientBody::DiscoItem(item) => {
-            let response = disco_items(iq.to.as_deref(), &item, rooms).await;
-            let req = Iq {
-                from: iq.to,
-                id: iq.id,
-                to: None,
-                r#type: proto_xmpp::client::IqType::Result,
-                body: response,
-            };
-            req.serialize(output);
-        }
-        _ => {
-            let req = Iq {
-                from: None,
-                id: iq.id,
-                to: None,
-                r#type: proto_xmpp::client::IqType::Error,
-                body: (),
-            };
-            req.serialize(output);
-        }
     }
 }
 
-fn disco_info(to: Option<&str>, req: &InfoQuery) -> InfoQuery {
-    let identity;
-    let feature;
-    match to {
-        Some("localhost") => {
-            identity = vec![Identity {
-                category: "server".into(),
-                name: None,
-                r#type: "im".into(),
-            }];
-            feature = vec![
-                Feature::new("http://jabber.org/protocol/disco#info"),
-                Feature::new("http://jabber.org/protocol/disco#items"),
-                Feature::new("iq"),
-                Feature::new("presence"),
-            ]
-        }
-        Some("rooms.localhost") => {
-            identity = vec![Identity {
-                category: "conference".into(),
-                name: Some("Chat rooms".into()),
-                r#type: "text".into(),
-            }];
-            feature = vec![
-                Feature::new("http://jabber.org/protocol/disco#info"),
-                Feature::new("http://jabber.org/protocol/disco#items"),
-                Feature::new("http://jabber.org/protocol/muc"),
-            ]
-        }
-        _ => {
-            identity = vec![];
-            feature = vec![];
-        }
-    };
-    InfoQuery {
-        node: None,
-        identity,
-        feature,
-    }
-}
-
-async fn disco_items(to: Option<&str>, req: &ItemQuery, rooms: &RoomRegistry) -> ItemQuery {
-    let item = match to {
-        Some("localhost") => {
-            vec![Item {
-                jid: Jid {
-                    name: None,
-                    server: Server("rooms.localhost".into()),
-                    resource: None,
-                },
-                name: None,
-                node: None,
-            }]
-        }
-        Some("rooms.localhost") => {
-            let room_list = rooms.get_all_rooms().await;
-            room_list
-                .into_iter()
-                .map(|room_info| Item {
-                    jid: Jid {
-                        name: Some(Name(room_info.id.into_inner())),
-                        server: Server("rooms.localhost".into()),
-                        resource: None,
-                    },
-                    name: None,
-                    node: None,
-                })
-                .collect()
-        }
-        _ => vec![],
-    };
-    ItemQuery { item }
-}
-
 async fn read_xml_header(
     xml_reader: &mut NsReader<(impl AsyncBufRead + Unpin)>,
     reader_buf: &mut Vec<u8>,

crates/projection-xmpp/src/message.rs

@@ -0,0 +1,50 @@
+//! Handling of all client2server message stanzas
+
+use quick_xml::events::Event;
+
+use lavina_core::prelude::*;
+use lavina_core::room::RoomId;
+use proto_xmpp::bind::{Jid, Server};
+use proto_xmpp::client::{Message, MessageType};
+use proto_xmpp::xml::{Ignore, ToXml};
+
+use crate::XmppConnection;
+
+impl<'a> XmppConnection<'a> {
+    pub async fn handle_message(&mut self, output: &mut Vec<Event<'static>>, m: Message<Ignore>) -> Result<()> {
+        if let Some(Jid {
+            name: Some(name),
+            server,
+            resource: _,
+        }) = m.to
+        {
+            if server.0.as_ref() == "rooms.localhost" && m.r#type == MessageType::Groupchat {
+                self.user_handle.send_message(RoomId::from(name.0.clone())?, m.body.clone().into()).await?;
+                Message::<()> {
+                    to: Some(Jid {
+                        name: Some(self.user.xmpp_name.clone()),
+                        server: Server("localhost".into()),
+                        resource: Some(self.user.xmpp_resource.clone()),
+                    }),
+                    from: Some(Jid {
+                        name: Some(name),
+                        server: Server("rooms.localhost".into()),
+                        resource: Some(self.user.xmpp_muc_name.clone()),
+                    }),
+                    id: m.id,
+                    r#type: MessageType::Groupchat,
+                    lang: None,
+                    subject: None,
+                    body: m.body.clone(),
+                    custom: vec![],
+                }
+                .serialize(output);
+                Ok(())
+            } else {
+                todo!()
+            }
+        } else {
+            todo!()
+        }
+    }
+}

crates/projection-xmpp/src/presence.rs

@@ -0,0 +1,55 @@
+//! Handling of all client2server presence stanzas
+
+use quick_xml::events::Event;
+
+use lavina_core::prelude::*;
+use lavina_core::room::RoomId;
+use proto_xmpp::bind::{Jid, Server};
+use proto_xmpp::client::Presence;
+use proto_xmpp::xml::{Ignore, ToXml};
+
+use crate::XmppConnection;
+
+impl<'a> XmppConnection<'a> {
+    pub async fn handle_presence(&mut self, output: &mut Vec<Event<'static>>, p: Presence<Ignore>) -> Result<()> {
+        let response = if p.to.is_none() {
+            Presence::<()> {
+                to: Some(Jid {
+                    name: Some(self.user.xmpp_name.clone()),
+                    server: Server("localhost".into()),
+                    resource: Some(self.user.xmpp_resource.clone()),
+                }),
+                from: Some(Jid {
+                    name: Some(self.user.xmpp_name.clone()),
+                    server: Server("localhost".into()),
+                    resource: Some(self.user.xmpp_resource.clone()),
+                }),
+                ..Default::default()
+            }
+        } else if let Some(Jid {
+            name: Some(name),
+            server,
+            resource: Some(resource),
+        }) = p.to
+        {
+            let a = self.user_handle.join_room(RoomId::from(name.0.clone())?).await?;
+            Presence::<()> {
+                to: Some(Jid {
+                    name: Some(self.user.xmpp_name.clone()),
+                    server: Server("localhost".into()),
+                    resource: Some(self.user.xmpp_resource.clone()),
+                }),
+                from: Some(Jid {
+                    name: Some(name.clone()),
+                    server: Server("rooms.localhost".into()),
+                    resource: Some(self.user.xmpp_muc_name.clone()),
+                }),
+                ..Default::default()
+            }
+        } else {
+            Presence::<()>::default()
+        };
+        response.serialize(output);
+        Ok(())
+    }
+}

crates/projection-xmpp/src/updates.rs

@@ -0,0 +1,45 @@
+//! Handling of updates and converting them into server2client stanzas
+
+use anyhow::Result;
+use quick_xml::events::Event;
+
+use lavina_core::player::Updates;
+use proto_xmpp::bind::{Jid, Name, Resource, Server};
+use proto_xmpp::client::{Message, MessageType};
+use proto_xmpp::xml::ToXml;
+
+use crate::XmppConnection;
+
+impl<'a> XmppConnection<'a> {
+    pub async fn handle_update(&mut self, output: &mut Vec<Event<'static>>, update: Updates) -> Result<()> {
+        match update {
+            Updates::NewMessage {
+                room_id,
+                author_id,
+                body,
+            } => {
+                Message::<()> {
+                    to: Some(Jid {
+                        name: Some(self.user.xmpp_name.clone()),
+                        server: Server("localhost".into()),
+                        resource: Some(self.user.xmpp_resource.clone()),
+                    }),
+                    from: Some(Jid {
+                        name: Some(Name(room_id.into_inner().into())),
+                        server: Server("rooms.localhost".into()),
+                        resource: Some(Resource(author_id.into_inner().into())),
+                    }),
+                    id: None,
+                    r#type: MessageType::Groupchat,
+                    lang: None,
+                    subject: None,
+                    body: body.into(),
+                    custom: vec![],
+                }
+                .serialize(output);
+            }
+            _ => {}
+        }
+        Ok(())
+    }
+}

crates/projection-xmpp/tests/certs/xmpp.key

@@ -0,0 +1,52 @@
+-----BEGIN PRIVATE KEY-----
+MIIJQgIBADANBgkqhkiG9w0BAQEFAASCCSwwggkoAgEAAoICAQCuViBTGN8tMaQ8
+G8zixL7duElTFCuP6wQhmDsX8ut4V3eEshUpDIIFkCSX17hzfI7duBp1pe7Ket+F
+z5XjbV+ruvxpawvsCgsfGrwXE1vaDVJduy0JyRzLvRSXELWgAbcdllbvBKvGLtY1
+ogm5YJWLbtgQJjutMoLisxn7Xd04fzMQy4aqhy2ZrsxyQSMINuR1Qz/VBzDZi4EH
+Q08rb7GManQfbabbTs1I/GHuAM7PDeb/ou9AZHPASg2fzam5SJhvDYutHmX8wOS3
+b+I+amI6g3N8fJssjx0ryAEL+c+Mbv6mXQhGqh7T++kXtB8h5GoLCOg3yGtaW7o0
+PacAP1UsadDsGN13cWAAsytg1BxqgWk6IqA3Yff5uc2A+TYmX5K5DV46sonovybo
+FI9fdKmL4oCbMIz+Tq+L5vHsdUh5/5S6F2RIKQDJIDYJfE7XVCPyToabQirQsQ/B
+n27L0bCO1hD9cGR+z5td9TPGxrm7GUVGZ/fC58Q4WD/TrVhC3pUq8n7hMzGakg+w
+Ri7FSJTPIZQXiTL/HtPleW+1y6d8Q84UI6Qm39vVueS5YSCjFCiQW2feod+L4sTU
+sE0Rumbvb+saS6cmX7ZBzdgJhP9J5FiAOPqswgCS5w54F5hvfbg+yS4SgeKrvZF3
+dAf+3wW8r039sFN+R/gQowxZcYZwOwIDAQABAoICAEB/p8TmmkcXqxn79RDe1nik
+QiiE+VrtCaG+Nvq0ym5C+fpzgkWmFYKmYgt1aY38gsS/5LYrFk3+KK1ScDNsly0r
+aFA+JPKGgrfWxcjJxj1FmXgJFHAe4lL0WOZM7c1NZSiCoxYaBc00Ldc45F0bwSgN
+cc2Dv6dj3S2vMokfoIVS9hscGW4ExheqJoSM2b+jw2Eo6LhRST7rEGkV+3foAmmf
+RugLwuQ3YtbCXR7XWKwdCh4A84BAydxV6XV6evUMSS0o90isyvG4kcXWFH+gD0hz
+sqnXVfel2RaGD/EU0rczp234VGQEc5RdCk9VOgFphtwfRv7AXQtYjWrfdmYuiD1i
+csiRZxcDgaDbWssBu7oppnbQo0XQ/pynLRJy3Lf9ETIjQ2Xkn6GdoXZY3Phj/XtW
+N7WSQTSEqRfVt7XwWx0wGXlPimIaYJDXmYFU1A/XESrvDgjjwtVrSojipwr5kdYl
+XgCjJVGZyNqAavYzuu7qiH7nXGmEtDQRQlbEQS135ukDkKkv2fIOFbX10iIPgIVM
+Y5dk7Q3gqCWcgcQ/rcR2LLo2H3PdJl0yQnml8rKz2CNpAHp5p4VF66jfsPC0q0qE
+1ad3JAweX042/k0/XPbwIjPCvDBUIeZei3gu8AotgmaRQWgLb7ICgvavjLnJKn/J
+VK2PJHKv/eFvEw6l1T85AoIBAQDVeoTOYe9AdAtsj6oqqjci4WZHJ/keO3pbeh4q
+JKgxzLlyag+8fSGzNkR0RE4f7EF0lnlcRTUpscaes19tmYMzbL84/wmD1SDAjRnG
+OVFkBHIgmXRnyRGQvP0QbBk3mcJKz+8qi7Mvb3YkSdh9lyYs/JtsPDI0XeQrq9J7
+ABIHElu9lCMTtUauljoYZH/pHYjpvk22Ijj6f+0dT5MHHYfsEoxIPB7Ow4JS7buT
+m+O7vlfYZLxSn3OvbdbzGo2IL+AYVDcszSdx6qmHbeu9uMWOvO/YbIX4VyLtzUuc
+MQad7nBOkiKVrTFXb0b+g1dxvUV3+FreZ1K9oFYy82bomcDZAoIBAQDRD7XxRkGO
+OSaWkx2FFMLvntAtO06RpshxtU/rv3vDYwYDulrHc3AaJ+P9rWjb7v15P5OpwRYW
+x1ve8lm8ycKnR6UgD3EYTQkEQkuZ68+ndhVarYCJaelWM9HdhFiWKdPyyrKwcXLr
+blcTZjnq4WC76YtTdSdZfn0KRoSAxuAmVwfWBI4LxaUeMk3TMxbJ4aPUCjkKEw0m
+Jie6S6419d+2aVNXjw4KqaPoSREWlUT8U+iVo1xQp5eg3cOIe2lVukqZTqe7j9Ze
+zP+Gq/RyTk+dvBWcsK/RzhyC73+KD8qAEmdzPAdEwhkxRioTk2PGtU4/K0nDZk3c
+TNlLdeOcqg0zAoIBAEwV9MuSADHaqk+xDJdUP36BE3D9AD8UN9Huvl2K3x+Qte/f
+eWhWuPIkv1UpGycpj1K8ZtjKGd6YbBAYIkTv1+E2OxlXXM7N4XR/VdZei3G4W+ze
+hKyQ71/E2/VEceBtPuBnJ/jj/aNEeLkKUMzCWGrkRYjYE5SyeiZOgSAxsDsxAd2Z
+tL7Ldzu2c1JKT4SIcEnO9+eYXvJ5McumluKMVet/2NvOAbTz3bks3hQIFazOdIS9
+splIF3VJErlml1cYqShCq7+eBxcE6hNIzCK8fj0XfeyHEWCnvd0/tFkg6BjV6NU4
+JHdwWQuur4D60unI6b+OluR5svW+9boHIoB4fFECggEAHeCW6fJWcBLu1toTf+9l
+pIUXzz8IjXw+bTGySEjHUTcXpvS9AIAY50QIKzrbH4NaKjfRzJLRq1O2Z3hPJtHW
+xb1RdfF/AjAQN9GZqFexB4eyqZDeK8U9GZqyRWwilONJbQtW2ix8dfUA8L7NTCoF
+fxVzWewGQZ34FL3bNeQ2KISLlCR2gGwwms4pnSNSAGwE08raN/xdBrSxPMiQDxoi
+bJlE1eCV6yQvToUSsh2HDGCZfrkn+kbZPp4y0ZCBj0TeYGaDRiTaSBYX9pEgkC1s
+52f31rrRhbRlErlTitGS6Ra4Phm4GDV9EDOs07teqQlEM3bmRcybF/7LlyMz8jHD
+TQKCAQEAvjdo48rvwjlj4IKkhkEWP2Qn19bxeQT6lWb9Wtgqar1CIIk9ei74V6xF
+5cyhw9vCXqxvlDM37n0JRHc0EM70aCE7IFL7WxCmmwGXBwT/PeyIJMaaZippd+Ff
+QjBx4z6OSsCnzWnM8YPkRwSSCGVk9EDvxQtwS7BmpqSgilHL1/n1yDKAiiGsvAPI
+uR5WhXzeN/OUHji7Gp5tCxc6Lo4dKqMhQrMFTomUbupw/o4X8TZ0O3hZdzLv9d4Q
+9LM5nI4JOwB4qEUOY9kFQNjvxwPvrTPj8QuIyPIjPQYWZ4Jw8lNCOK3COPV8H+Rb
+kO/SCOEdhp17yWQspky/Uo1RC4lKVA==
+-----END PRIVATE KEY-----

crates/projection-xmpp/tests/certs/xmpp.pem

@@ -0,0 +1,30 @@
+-----BEGIN CERTIFICATE-----
+MIIFDTCCAvWgAwIBAgIUGd9jmau898T/kGdIeiRCcdXueaowDQYJKoZIhvcNAQEL
+BQAwFjEUMBIGA1UEAwwLZXhhbXBsZS5jb20wHhcNMjMxMDA5MTIyMTU2WhcNMjMx
+MTA4MTIyMTU2WjAWMRQwEgYDVQQDDAtleGFtcGxlLmNvbTCCAiIwDQYJKoZIhvcN
+AQEBBQADggIPADCCAgoCggIBAK5WIFMY3y0xpDwbzOLEvt24SVMUK4/rBCGYOxfy
+63hXd4SyFSkMggWQJJfXuHN8jt24GnWl7sp634XPleNtX6u6/GlrC+wKCx8avBcT
+W9oNUl27LQnJHMu9FJcQtaABtx2WVu8Eq8Yu1jWiCblglYtu2BAmO60yguKzGftd
+3Th/MxDLhqqHLZmuzHJBIwg25HVDP9UHMNmLgQdDTytvsYxqdB9tpttOzUj8Ye4A
+zs8N5v+i70Bkc8BKDZ/NqblImG8Ni60eZfzA5Ldv4j5qYjqDc3x8myyPHSvIAQv5
+z4xu/qZdCEaqHtP76Re0HyHkagsI6DfIa1pbujQ9pwA/VSxp0OwY3XdxYACzK2DU
+HGqBaToioDdh9/m5zYD5NiZfkrkNXjqyiei/JugUj190qYvigJswjP5Or4vm8ex1
+SHn/lLoXZEgpAMkgNgl8TtdUI/JOhptCKtCxD8GfbsvRsI7WEP1wZH7Pm131M8bG
+ubsZRUZn98LnxDhYP9OtWELelSryfuEzMZqSD7BGLsVIlM8hlBeJMv8e0+V5b7XL
+p3xDzhQjpCbf29W55LlhIKMUKJBbZ96h34vixNSwTRG6Zu9v6xpLpyZftkHN2AmE
+/0nkWIA4+qzCAJLnDngXmG99uD7JLhKB4qu9kXd0B/7fBbyvTf2wU35H+BCjDFlx
+hnA7AgMBAAGjUzBRMB0GA1UdDgQWBBSFuG4k9lOlZweMeqvBejQic3Me1zAfBgNV
+HSMEGDAWgBSFuG4k9lOlZweMeqvBejQic3Me1zAPBgNVHRMBAf8EBTADAQH/MA0G
+CSqGSIb3DQEBCwUAA4ICAQAjLko9eLBofC9QzF/rm59kOtS6P8fRD0OJKawRdxvP
+5ClQFz9Q/2I7E1jbXL9y8r8iB31hpfnMT/XYxHAmJXyV+X2jHlnKmhzWpfrHx84V
+ZYeFIEFlwJHacPU6mVUUnXLwZIt0VWB3RUT9bgbkXFfkg7Qrccl7Blc0458l8wd2
+mvNgdEGG9Z8VhyaExAHerpOD299llaDTcV+jKLLAboXzDGJsPmMfRu2DWwosfYQJ
+MIKajyylmGOrAk+8fVTVWOPJBE6AmxDpKtZQO04nQq78PohP/CsibalimOf00Wqb
+2x05ssWFP71lmdGilaXCp/mkhaVz/G7ZuDSY9AxCPH4+3pTUbMgsfEmLXRTrUo9c
+B4zz0eDoUiqU9I4TWAHGhnn7b2e6o8ko0baq+PaCSCDK8haL/17CUyMYQBBdhlG6
+sSR2IBSXkxGQZnhfmQdYwc7Y3IgsJJoN0ZfqnyBY+u/ZOBegcZFuhMiDGILJBgFi
+QmgQFTozXKkeEzFdWoRn57lgcBABcMHSucnjekk80uLIWIlL36IyYkPbeI6/I0/l
+ajKftsVEY96hYUXtDBykBpg6gsJXj2gP2FXHW7ngtuI4/mOqI3ltcWh1C83MvM5N
+zMYfTNtxVM1vyvN1M7b7iMeosvmAIvQE3bFk/pJCHAZPfsu0zmeizPHBgCySygLn
+JQ==
+-----END CERTIFICATE-----

crates/projection-xmpp/tests/lib.rs

@@ -0,0 +1,186 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::Result;
+use assert_matches::*;
+use prometheus::Registry as MetricsRegistry;
+use quick_xml::events::Event;
+use quick_xml::NsReader;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{ReadHalf as GenericReadHalf, WriteHalf as GenericWriteHalf};
+use tokio::net::tcp::{ReadHalf, WriteHalf};
+use tokio::net::TcpStream;
+use tokio_rustls::client::TlsStream;
+use tokio_rustls::rustls::client::ServerCertVerifier;
+use tokio_rustls::rustls::{ClientConfig, ServerName};
+use tokio_rustls::TlsConnector;
+
+use lavina_core::player::PlayerRegistry;
+use lavina_core::repo::{Storage, StorageConfig};
+use lavina_core::room::RoomRegistry;
+use projection_xmpp::{launch, ServerConfig};
+use proto_xmpp::xml::{Continuation, FromXml, Parser};
+
+pub async fn read_irc_message(reader: &mut BufReader<ReadHalf<'_>>, buf: &mut Vec<u8>) -> Result<usize> {
+    let mut size = 0;
+    let res = reader.read_until(b'\n', buf).await?;
+    size += res;
+    return Ok(size);
+}
+
+struct TestScope<'a> {
+    reader: NsReader<BufReader<ReadHalf<'a>>>,
+    writer: WriteHalf<'a>,
+    buffer: Vec<u8>,
+}
+
+impl<'a> TestScope<'a> {
+    fn new(stream: &mut TcpStream) -> TestScope<'_> {
+        let (reader, writer) = stream.split();
+        let reader = NsReader::from_reader(BufReader::new(reader));
+        let buffer = vec![];
+        TestScope { reader, writer, buffer }
+    }
+
+    async fn send(&mut self, str: &str) -> Result<()> {
+        self.writer.write_all(str.as_bytes()).await?;
+        self.writer.flush().await?;
+        Ok(())
+    }
+
+    async fn next_xml_event(&mut self) -> Result<Event<'_>> {
+        self.buffer.clear();
+        let event = self.reader.read_event_into_async(&mut self.buffer).await?;
+        Ok(event)
+    }
+
+    async fn read<T: FromXml>(&mut self) -> Result<T> {
+        self.buffer.clear();
+        let (ns, event) = self.reader.read_resolved_event_into_async(&mut self.buffer).await?;
+        let mut parser: Continuation<_, std::result::Result<T, anyhow::Error>> = T::parse().consume(ns, &event);
+        loop {
+            match parser {
+                Continuation::Final(res) => return Ok(res?),
+                Continuation::Continue(next) => {
+                    let (ns, event) = self.reader.read_resolved_event_into_async(&mut self.buffer).await?;
+                    parser = next.consume(ns, &event);
+                }
+            }
+        }
+    }
+}
+
+struct TestScopeTls<'a> {
+    reader: NsReader<BufReader<GenericReadHalf<&'a mut TlsStream<TcpStream>>>>,
+    writer: GenericWriteHalf<&'a mut TlsStream<TcpStream>>,
+    buffer: Vec<u8>,
+    pub timeout: Duration,
+}
+
+impl<'a> TestScopeTls<'a> {
+    fn new(stream: &'a mut TlsStream<TcpStream>, buffer: Vec<u8>) -> TestScopeTls<'a> {
+        let (reader, writer) = tokio::io::split(stream);
+        let reader = NsReader::from_reader(BufReader::new(reader));
+        let timeout = Duration::from_millis(100);
+
+        TestScopeTls {
+            reader,
+            writer,
+            buffer,
+            timeout,
+        }
+    }
+
+    async fn send(&mut self, str: &str) -> Result<()> {
+        self.writer.write_all(str.as_bytes()).await?;
+        self.writer.flush().await?;
+        Ok(())
+    }
+
+    async fn next_xml_event(&mut self) -> Result<Event<'_>> {
+        self.buffer.clear();
+        let event = self.reader.read_event_into_async(&mut self.buffer);
+        let event = tokio::time::timeout(self.timeout, event).await??;
+        Ok(event)
+    }
+}
+
+struct IgnoreCertVerification;
+impl ServerCertVerifier for IgnoreCertVerification {
+    fn verify_server_cert(
+        &self,
+        _end_entity: &tokio_rustls::rustls::Certificate,
+        _intermediates: &[tokio_rustls::rustls::Certificate],
+        _server_name: &ServerName,
+        _scts: &mut dyn Iterator<Item = &[u8]>,
+        _ocsp_response: &[u8],
+        _now: std::time::SystemTime,
+    ) -> std::result::Result<tokio_rustls::rustls::client::ServerCertVerified, tokio_rustls::rustls::Error> {
+        Ok(tokio_rustls::rustls::client::ServerCertVerified::assertion())
+    }
+}
+
+#[tokio::test]
+async fn scenario_basic() -> Result<()> {
+    tracing_subscriber::fmt::init();
+    let config = ServerConfig {
+        listen_on: "127.0.0.1:0".parse().unwrap(),
+        cert: "tests/certs/xmpp.pem".parse().unwrap(),
+        key: "tests/certs/xmpp.key".parse().unwrap(),
+    };
+    let mut metrics = MetricsRegistry::new();
+    let mut storage = Storage::open(StorageConfig {
+        db_path: ":memory:".into(),
+    })
+    .await?;
+    let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
+    let players = PlayerRegistry::empty(rooms.clone(), &mut metrics).unwrap();
+    let server = launch(config, players, rooms, metrics, storage.clone()).await.unwrap();
+
+    // test scenario
+
+    storage.create_user("tester").await?;
+    storage.set_password("tester", "password").await?;
+
+    let mut stream = TcpStream::connect(server.addr).await?;
+    let mut s = TestScope::new(&mut stream);
+    tracing::info!("TCP connection established");
+
+    s.send(r#"<?xml version="1.0"?>"#).await?;
+    s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"features"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"starttls"));
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"required"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"starttls"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"features"));
+    s.send(r#"<starttls/>"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed"));
+    let buffer = s.buffer;
+    tracing::info!("TLS feature negotiation complete");
+
+    let connector = TlsConnector::from(Arc::new(
+        ClientConfig::builder()
+            .with_safe_defaults()
+            .with_custom_certificate_verifier(Arc::new(IgnoreCertVerification))
+            .with_no_client_auth(),
+    ));
+    tracing::info!("Initiating TLS connection...");
+    let mut stream = connector.connect(ServerName::IpAddress(server.addr.ip()), stream).await?;
+    tracing::info!("TLS connection established");
+
+    let mut s = TestScopeTls::new(&mut stream, buffer);
+
+    s.send(r#"<?xml version="1.0"?>"#).await?;
+    s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+
+    stream.shutdown().await?;
+
+    // wrap up
+
+    server.terminate().await?;
+    Ok(())
+}

crates/proto-irc/src/client.rs

@@ -2,6 +2,7 @@ use super::*;
 
 use anyhow::{anyhow, Result};
 use nom::combinator::{all_consuming, opt};
+use nonempty::NonEmpty;
 
 /// Client-to-server command.
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -59,6 +60,7 @@ pub enum ClientMessage {
     Quit {
         reason: Str,
     },
+    Authenticate(Str),
 }
 
 pub fn client_message(input: &str) -> Result<ClientMessage> {
@@ -76,6 +78,7 @@ pub fn client_message(input: &str) -> Result<ClientMessage> {
         client_message_part,
         client_message_privmsg,
         client_message_quit,
+        client_message_authenticate,
     )))(input);
     match res {
         Ok((_, e)) => Ok(e),
@@ -223,16 +226,29 @@ fn client_message_quit(input: &str) -> IResult<&str, ClientMessage> {
     Ok((input, ClientMessage::Quit { reason: reason.into() }))
 }
 
+fn client_message_authenticate(input: &str) -> IResult<&str, ClientMessage> {
+    let (input, _) = tag("AUTHENTICATE ")(input)?;
+    let (input, body) = token(input)?;
+
+    Ok((input, ClientMessage::Authenticate(body.into())))
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum CapabilitySubcommand {
     /// CAP LS {code}
     List { code: [u8; 3] },
+    /// CAP REQ :...
+    Req(NonEmpty<CapReq>),
     /// CAP END
     End,
 }
 
 fn capability_subcommand(input: &str) -> IResult<&str, CapabilitySubcommand> {
-    alt((capability_subcommand_ls, capability_subcommand_end))(input)
+    alt((
+        capability_subcommand_ls,
+        capability_subcommand_end,
+        capability_subcommand_req,
+    ))(input)
 }
 
 fn capability_subcommand_ls(input: &str) -> IResult<&str, CapabilitySubcommand> {
@@ -247,14 +263,46 @@ fn capability_subcommand_ls(input: &str) -> IResult<&str, CapabilitySubcommand>
     ))
 }
 
+fn capability_subcommand_req(input: &str) -> IResult<&str, CapabilitySubcommand> {
+    let (input, _) = tag("REQ ")(input)?;
+    let (input, r) = opt(tag(":"))(input)?;
+    let (input, body) = match r {
+        Some(_) => token(input)?,
+        None => receiver(input)?,
+    };
+
+    let caps = body
+        .split(' ')
+        .map(|cap| {
+            let to_disable = cap.starts_with('-');
+            let name = if to_disable { &cap[1..] } else { &cap[..] };
+            CapReq {
+                to_disable,
+                name: name.into(),
+            }
+        })
+        .collect::<Vec<_>>();
+
+    let caps = NonEmpty::from_vec(caps).ok_or_else(|| todo!())?;
+
+    Ok((input, CapabilitySubcommand::Req(caps)))
+}
+
 fn capability_subcommand_end(input: &str) -> IResult<&str, CapabilitySubcommand> {
     let (input, _) = tag("END")(input)?;
     Ok((input, CapabilitySubcommand::End))
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CapReq {
+    pub to_disable: bool,
+    pub name: Str,
+}
+
 #[cfg(test)]
 mod test {
     use assert_matches::*;
+    use nonempty::nonempty;
 
     use super::*;
     #[test]
@@ -324,6 +372,25 @@ mod test {
             message: "Pokasiki !!!".into(),
         };
 
+        let result = client_message(input);
+        assert_matches!(result, Ok(result) => assert_eq!(expected, result));
+    }
+    #[test]
+    fn test_client_cap_req() {
+        let input = "CAP REQ :multi-prefix -sasl";
+        let expected = ClientMessage::Capability {
+            subcommand: CapabilitySubcommand::Req(nonempty![
+                CapReq {
+                    to_disable: false,
+                    name: "multi-prefix".into()
+                },
+                CapReq {
+                    to_disable: true,
+                    name: "sasl".into()
+                }
+            ]),
+        };
+
         let result = client_message(input);
         assert_matches!(result, Ok(result) => assert_eq!(expected, result));
     }

crates/proto-irc/src/server.rs

@@ -66,6 +66,11 @@ pub enum ServerMessageBody {
     Error {
         reason: Str,
     },
+    Cap {
+        target: Str,
+        subcmd: CapSubBody,
+    },
+    Authenticate(Str),
     N001Welcome {
         client: Str,
         text: Str,
@@ -138,6 +143,16 @@ pub enum ServerMessageBody {
         client: Str,
         message: Str,
     },
+    N900LoggedIn {
+        nick: Str,
+        address: Str,
+        account: Str,
+        message: Str,
+    },
+    N903SaslSuccess {
+        nick: Str,
+        message: Str,
+    }
 }
 
 impl ServerMessageBody {
@@ -181,6 +196,24 @@ impl ServerMessageBody {
                 writer.write_all(b"ERROR :").await?;
                 writer.write_all(reason.as_bytes()).await?;
             }
+            ServerMessageBody::Cap { target, subcmd } => {
+                writer.write_all(b"CAP ").await?;
+                writer.write_all(target.as_bytes()).await?;
+                match subcmd {
+                    CapSubBody::Ls(caps) => {
+                        writer.write_all(b" LS :").await?;
+                        writer.write_all(caps.as_bytes()).await?;
+                    }
+                    CapSubBody::Ack(caps) => {
+                        writer.write_all(b" ACK :").await?;
+                        writer.write_all(caps.as_bytes()).await?;
+                    }
+                }
+            }
+            ServerMessageBody::Authenticate(body) => {
+                writer.write_all(b"AUTHENTICATE ").await?;
+                writer.write_all(body.as_bytes()).await?;
+            }
             ServerMessageBody::N001Welcome { client, text } => {
                 writer.write_all(b"001 ").await?;
                 writer.write_all(client.as_bytes()).await?;
@@ -320,11 +353,33 @@ impl ServerMessageBody {
                 writer.write_all(b" :").await?;
                 writer.write_all(message.as_bytes()).await?;
             }
+            ServerMessageBody::N900LoggedIn { nick, address, account, message } => {
+                writer.write_all(b"900 ").await?;
+                writer.write_all(nick.as_bytes()).await?;
+                writer.write_all(b" ").await?;
+                writer.write_all(address.as_bytes()).await?;
+                writer.write_all(b" ").await?;
+                writer.write_all(account.as_bytes()).await?;
+                writer.write_all(b" :").await?;
+                writer.write_all(message.as_bytes()).await?;
+            }
+            ServerMessageBody::N903SaslSuccess { nick, message } => {
+                writer.write_all(b"903 ").await?;
+                writer.write_all(nick.as_bytes()).await?;
+                writer.write_all(b" :").await?;
+                writer.write_all(message.as_bytes()).await?;
+            }
         }
         Ok(())
     }
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum CapSubBody {
+    Ls(Str),
+    Ack(Str),
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum AwayStatus {
     Here,
@@ -336,6 +391,7 @@ fn server_message_body(input: &str) -> IResult<&str, ServerMessageBody> {
         server_message_body_notice,
         server_message_body_ping,
         server_message_body_pong,
+        server_message_body_cap,
     ))(input)
 }
 
@@ -361,12 +417,7 @@ fn server_message_body_ping(input: &str) -> IResult<&str, ServerMessageBody> {
     let (input, _) = tag("PING ")(input)?;
     let (input, token) = token(input)?;
 
-    Ok((
+    Ok((input, ServerMessageBody::Ping { token: token.into() }))
-        input,
-        ServerMessageBody::Ping {
-            token: token.into(),
-        },
-    ))
 }
 
 fn server_message_body_pong(input: &str) -> IResult<&str, ServerMessageBody> {
@@ -384,6 +435,21 @@ fn server_message_body_pong(input: &str) -> IResult<&str, ServerMessageBody> {
     ))
 }
 
+fn server_message_body_cap(input: &str) -> IResult<&str, ServerMessageBody> {
+    let (input, _) = tag("CAP ")(input)?;
+    let (input, from) = receiver(input)?;
+    let (input, _) = tag(" LS :")(input)?;
+    let (input, token) = token(input)?;
+
+    Ok((
+        input,
+        ServerMessageBody::Cap {
+            target: from.into(),
+            subcmd: CapSubBody::Ls(token.into()),
+        },
+    ))
+}
+
 #[cfg(test)]
 mod test {
     use assert_matches::*;
@@ -408,9 +474,7 @@ mod test {
         assert_matches!(result, Ok((_, result)) => assert_eq!(expected, result));
 
         let mut bytes = vec![];
-        sync_future(expected.write_async(&mut bytes))
+        sync_future(expected.write_async(&mut bytes)).unwrap().unwrap();
-            .unwrap()
-            .unwrap();
         assert_eq!(bytes, input.as_bytes());
     }
 
@@ -430,9 +494,27 @@ mod test {
         assert_matches!(result, Ok((_, result)) => assert_eq!(expected, result));
 
         let mut bytes = vec![];
-        sync_future(expected.write_async(&mut bytes))
+        sync_future(expected.write_async(&mut bytes)).unwrap().unwrap();
-            .unwrap()
+        assert_eq!(bytes, input.as_bytes());
-            .unwrap();
+    }
+
+    #[test]
+    fn test_server_message_cap_ls() {
+        let input = "CAP * LS :sasl\r\n";
+        let expected = ServerMessage {
+            tags: vec![],
+            sender: None,
+            body: ServerMessageBody::Cap {
+                target: "*".into(),
+                subcmd: CapSubBody::Ls("sasl".into()),
+            },
+        };
+
+        let result = server_message(input);
+        assert_matches!(result, Ok((_, result)) => assert_eq!(expected, result));
+
+        let mut bytes = vec![];
+        sync_future(expected.write_async(&mut bytes)).unwrap().unwrap();
         assert_eq!(bytes, input.as_bytes());
     }
 }

crates/proto-xmpp/Cargo.toml

@@ -10,7 +10,6 @@ regex.workspace = true
 anyhow.workspace = true
 tokio.workspace = true
 derive_more.workspace = true
-base64.workspace = true
 
 [dev-dependencies]
 assert_matches.workspace = true

crates/proto-xmpp/src/lib.rs

@@ -1,6 +1,6 @@
 #![feature(
-    generators,
+    coroutines,
-    generator_trait,
+    coroutine_trait,
     type_alias_impl_trait,
     impl_trait_in_assoc_type
 )]

crates/proto-xmpp/src/sasl.rs

@@ -1,14 +1,11 @@
 use std::borrow::Borrow;
 
-use quick_xml::{
+use anyhow::{anyhow, Result};
-    events::{BytesStart, Event},
+use quick_xml::events::{BytesStart, Event};
-    NsReader, Writer,
+use quick_xml::{NsReader, Writer};
-};
 use tokio::io::{AsyncBufRead, AsyncWrite};
-use base64::{Engine as _, engine::general_purpose};
 
 use super::skip_text;
-use anyhow::{anyhow, Result};
 
 pub enum Mechanism {
     Plain,
@@ -29,98 +26,13 @@ impl Mechanism {
     }
 }
 
-#[derive(PartialEq, Debug)]
-pub struct AuthBody {
-    pub login: String,
-    pub password: String,
-}
-
-impl AuthBody {
-    pub fn from_str(input: &[u8]) -> Result<AuthBody> {
-        match general_purpose::STANDARD.decode(input){
-            Ok(decoded_body) => {
-                match String::from_utf8(decoded_body) {
-                    Ok(parsed_to_string) => {
-                        let separated_words: Vec<&str> = parsed_to_string.split("\x00").collect::<Vec<_>>().clone();
-                        if separated_words.len() == 3 {
-                            // first segment ignored (might be needed in the future)
-                            Ok(AuthBody { login: separated_words[1].to_string(), password: separated_words[2].to_string() })
-                        } else { return Err(anyhow!("Incorrect auth format")) }
-                    },
-                    Err(e) => return Err(anyhow!(e))
-                }
-            },
-            Err(e) => return Err(anyhow!(e))
-        }
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::*;
-    #[test]
-    fn test_returning_auth_body() {
-        let orig = b"\x00login\x00pass";
-        let encoded = general_purpose::STANDARD.encode(orig);
-        let expected = AuthBody {login: "login".to_string(), password: "pass".to_string()};
-        let result = AuthBody::from_str(encoded.as_bytes()).unwrap();
-
-        assert_eq!(expected, result);
-    }
-
-    #[test]
-    fn test_ignoring_first_segment() {
-        let orig = b"ignored\x00login\x00pass";
-        let encoded = general_purpose::STANDARD.encode(orig);
-        let expected = AuthBody {login: "login".to_string(), password: "pass".to_string()};
-        let result = AuthBody::from_str(encoded.as_bytes()).unwrap();
-
-        assert_eq!(expected, result);
-    }
-
-    #[test]
-    fn test_returning_auth_body_with_empty_strings() {
-        let orig = b"\x00\x00";
-        let encoded = general_purpose::STANDARD.encode(orig);
-        let expected = AuthBody {login: "".to_string(), password: "".to_string()};
-        let result = AuthBody::from_str(encoded.as_bytes()).unwrap();
-
-        assert_eq!(expected, result);
-    }
-
-
-    #[test]
-    fn test_fail_if_size_less_then_3() {
-        let orig = b"login\x00pass";
-        let encoded = general_purpose::STANDARD.encode(orig);
-        let expected = AuthBody {login: "login".to_string(), password: "pass".to_string()};
-        let result = AuthBody::from_str(encoded.as_bytes());
-
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn test_fail_if_size_greater_then_3() {
-        let orig = b"first\x00login\x00pass\x00other";
-        let encoded = general_purpose::STANDARD.encode(orig);
-        let expected = AuthBody {login: "login".to_string(), password: "pass".to_string()};
-        let result = AuthBody::from_str(encoded.as_bytes());
-
-        assert!(result.is_err());
-    }
-
-}
-
 pub struct Auth {
     pub mechanism: Mechanism,
     pub body: Vec<u8>,
 }
 
 impl Auth {
-    pub async fn parse(
+    pub async fn parse(reader: &mut NsReader<impl AsyncBufRead + Unpin>, buf: &mut Vec<u8>) -> Result<Auth> {
-        reader: &mut NsReader<impl AsyncBufRead + Unpin>,
-        buf: &mut Vec<u8>,
-    ) -> Result<Auth> {
         let event = skip_text!(reader, buf);
         let mechanism = if let Event::Start(bytes) = event {
             let mut mechanism = None;

crates/proto-xmpp/src/xml/mod.rs

@@ -1,4 +1,4 @@
-use std::ops::Generator;
+use std::ops::Coroutine;
 use std::pin::Pin;
 
 use quick_xml::NsReader;
@@ -37,7 +37,7 @@ pub trait Parser: Sized {
 
 impl<T, Out> Parser for T
 where
-    T: Generator<(ResolveResult<'static>, &'static Event<'static>), Yield = (), Return = Out>
+    T: Coroutine<(ResolveResult<'static>, &'static Event<'static>), Yield = (), Return = Out>
         + Unpin,
 {
     type Output = Out;
@@ -48,13 +48,13 @@ where
         event: &Event<'a>,
     ) -> Continuation<Self, Self::Output> {
         let s = Pin::new(&mut self);
-        // this is a very rude workaround fixing the fact that rust generators
+        // this is a very rude workaround fixing the fact that rust coroutines
-        // 1. don't support higher-kinded lifetimes (i.e. no `impl for <'a> Generator<Event<'a>>)
+        // 1. don't support higher-kinded lifetimes (i.e. no `impl for <'a> Coroutine<Event<'a>>)
         // 2. don't track borrows across yield points and lack thereof
         // implementors of Parser should manually check that inputs are not used across yields
         match s.resume(unsafe { std::mem::transmute((namespace, event)) }) {
-            std::ops::GeneratorState::Yielded(()) => Continuation::Continue(self),
+            std::ops::CoroutineState::Yielded(()) => Continuation::Continue(self),
-            std::ops::GeneratorState::Complete(res) => Continuation::Final(res),
+            std::ops::CoroutineState::Complete(res) => Continuation::Final(res),
         }
     }
 }

crates/readme.md

@@ -0,0 +1,46 @@
+## Dependency diagram of the project
+
+```mermaid
+graph TD;
+    lavina-->mgmt-api;
+    lavina-->projection-irc;
+    lavina-->projection-xmpp;
+    lavina-->lavina-core;
+
+    projection-irc-->proto-irc;
+    projection-irc-->lavina-core;
+    
+    projection-xmpp-->proto-xmpp;
+    projection-xmpp-->lavina-core;
+
+    sim-irc-->proto-irc;
+    sim-irc-->mgmt-api;
+
+    sim-xmpp-->proto-xmpp;
+    sim-xmpp-->mgmt-api;
+
+    workspace-->lavina;
+    workspace-->sim-irc;
+    workspace-->sim-xmpp;
+```
+
+A few rules:
+- Only projections should be direct deps of `lavina`, there is no need to depend on `proto-*` crates.
+- On the other hand, projections should not be dependencies of `sim-*` crates.
+- `lavina-core` does not depend on protocol-specific crates.
+
+## Lavina
+
+**lavina-core**. This crate implements the core functionality of the server not specific to public interfaces. It includes handling general persistence, message casts, in future room authorization and cross-node communication.
+
+**lavina**. The application crate. It's used to link other crates together and build the final binary with support for all protocols.
+
+**mgmt-api**. Model definitions for management API to be used both in the server and client implementations.
+
+## IRC
+
+**proto-irc**. Protocol definition for IRC, includes both server-side and client-side messages and both serialization and deserialization where needed.
+
+**projection-irc**. Projection of lavina-core onto the IRC client-to-server protocol implementation.
+
+**sim-irc**. Future implementation of IRC client simulator to be used in integration and load testing.

crates/sasl/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "sasl"
+edition = "2021"
+version.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+base64.workspace = true

crates/sasl/src/lib.rs

@@ -0,0 +1,103 @@
+use anyhow::{anyhow, Result};
+use base64::engine::general_purpose;
+use base64::Engine;
+
+#[derive(PartialEq, Debug)]
+pub struct AuthBody {
+    pub login: String,
+    pub password: String,
+}
+
+impl AuthBody {
+    pub fn from_str(input: &[u8]) -> Result<AuthBody> {
+        match general_purpose::STANDARD.decode(input) {
+            Ok(decoded_body) => {
+                match String::from_utf8(decoded_body) {
+                    Ok(parsed_to_string) => {
+                        let separated_words: Vec<&str> = parsed_to_string.split("\x00").collect::<Vec<_>>().clone();
+                        if separated_words.len() == 3 {
+                            // first segment ignored (might be needed in the future)
+                            Ok(AuthBody {
+                                login: separated_words[1].to_string(),
+                                password: separated_words[2].to_string(),
+                            })
+                        } else {
+                            return Err(anyhow!("Incorrect auth format"));
+                        }
+                    }
+                    Err(e) => return Err(e.into()),
+                }
+            }
+            Err(e) => return Err(e.into()),
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    #[test]
+    fn test_returning_auth_body() {
+        let orig = b"\x00login\x00pass";
+        let encoded = general_purpose::STANDARD.encode(orig);
+        let expected = AuthBody {
+            login: "login".to_string(),
+            password: "pass".to_string(),
+        };
+        let result = AuthBody::from_str(encoded.as_bytes()).unwrap();
+
+        assert_eq!(expected, result);
+    }
+
+    #[test]
+    fn test_ignoring_first_segment() {
+        let orig = b"ignored\x00login\x00pass";
+        let encoded = general_purpose::STANDARD.encode(orig);
+        let expected = AuthBody {
+            login: "login".to_string(),
+            password: "pass".to_string(),
+        };
+        let result = AuthBody::from_str(encoded.as_bytes()).unwrap();
+
+        assert_eq!(expected, result);
+    }
+
+    #[test]
+    fn test_returning_auth_body_with_empty_strings() {
+        let orig = b"\x00\x00";
+        let encoded = general_purpose::STANDARD.encode(orig);
+        let expected = AuthBody {
+            login: "".to_string(),
+            password: "".to_string(),
+        };
+        let result = AuthBody::from_str(encoded.as_bytes()).unwrap();
+
+        assert_eq!(expected, result);
+    }
+
+    #[test]
+    fn test_fail_if_size_less_then_3() {
+        let orig = b"login\x00pass";
+        let encoded = general_purpose::STANDARD.encode(orig);
+        let expected = AuthBody {
+            login: "login".to_string(),
+            password: "pass".to_string(),
+        };
+        let result = AuthBody::from_str(encoded.as_bytes());
+
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_fail_if_size_greater_then_3() {
+        let orig = b"first\x00login\x00pass\x00other";
+        let encoded = general_purpose::STANDARD.encode(orig);
+        let expected = AuthBody {
+            login: "login".to_string(),
+            password: "pass".to_string(),
+        };
+        let result = AuthBody::from_str(encoded.as_bytes());
+
+        assert!(result.is_err());
+    }
+}

docs/cheatsheet.md

@@ -18,25 +18,12 @@ Print content of a TLS certificate:
 
     openssl x509 -in certs/xmpp.pem -text
 
-Make sure `xmpp.key` starts and ends with:
-```
------BEGIN RSA PRIVATE KEY-----
------END RSA PRIVATE KEY-----
-```
-
-
 ## Protocol Specs
 
 XMPP XSDs - [https://xmpp.org/schemas/index.shtml]
 
 IRC modern spec - [https://modern.ircdocs.horse/]
 
-## Initializing DB with some users
-
-    sqlite3 db.sqlite < test/init_state.sql
-
-Same test migration could be used for integration tests in the future.
-
 ## Using irssi
 
 irssi in a TUI-based IRC client.

docs/running.md

@@ -0,0 +1,54 @@
+# Building & Running Lavina
+
+## Configuration
+
+A server instance must be supplied with a configuration file to start up. The path to configuration is specified with a required CLI argument `--config <path>`.
+
+Example configuration:
+```toml
+[telemetry]
+# address for management and telemetry API
+# should be kept private
+listen_on = "127.0.0.1:8080"
+
+[irc]
+listen_on = "127.0.0.1:6667"
+server_name = "irc.localhost"
+
+[xmpp]
+listen_on = "127.0.0.1:5222"
+cert = "./certs/xmpp.pem"
+key = "./certs/xmpp.key"
+
+[storage]
+db_path = "db.sqlite"
+```
+
+## With Docker Compose
+
+Example `docker-compose.yml` file:
+```yaml
+version: '3.0'
+
+services:
+  lavina:
+    image: git.vilunov.me/lavina/lavina:0.0.1
+    volumes:
+      - './config/:/etc/lavina/'
+      - './data/:/var/lib/lavina/'
+    ports:
+      - '5222:5222'  # xmpp
+      - '6667:6667'  # irc non-tls
+      - '127.0.0.1:1380:8080' # management http (private)
+```
+
+## With Cargo
+
+You can run it via cargo:
+
+    cargo run -- --config config.toml
+
+Or you can build it and run manually:
+
+    cargo build --release
+    ./target/release/lavina --config config.toml

readme.md

@@ -2,6 +2,10 @@
 
 Multiprotocol chat server based on open protocols.
 
+- [How to run Lavina locally](docs/running.md)
+- [Architectural diagrams](docs/flow.md)
+- [Project structure](crates/readme.md)
+
 ## Goals
 
 #### Support for multiple open protocols

rust-toolchain

@@ -1 +1 @@
-nightly-2023-10-06
+nightly-2023-12-07

src/http.rs

@@ -7,6 +7,7 @@ use hyper::body::Bytes;
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::{Method, Request, Response, StatusCode};
+use hyper_util::rt::TokioIo;
 use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
 use serde::{Deserialize, Serialize};
 use tokio::net::TcpListener;
@@ -52,6 +53,7 @@ async fn main_loop(
             _ = &mut termination => break,
             result = listener.accept() => {
                 let (stream, _) = result?;
+                let stream = TokioIo::new(stream);
                 let metrics = metrics.clone();
                 let rooms = rooms.clone();
                 let storage = storage.clone();