move sasl into separate crate

Reviewed-on: lavina/lavina#27

Cargo.lock

@@ -1279,6 +1279,7 @@ name = "projection-xmpp"
 version = "0.0.2-dev"
 dependencies = [
  "anyhow",
+ "assert_matches",
  "derive_more",
  "futures-util",
  "lavina-core",
@@ -1291,6 +1292,7 @@ dependencies = [
  "tokio",
  "tokio-rustls",
  "tracing",
+ "tracing-subscriber",
  "uuid",
 ]
 

Cargo.toml

@@ -29,6 +29,7 @@ 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]
@@ -46,7 +47,7 @@ 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/projection-xmpp/Cargo.toml

@@ -16,6 +16,10 @@ 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

@@ -25,18 +25,19 @@ use tokio_rustls::TlsAcceptor;
 use lavina_core::player::{PlayerConnection, PlayerId, PlayerRegistry};
 use lavina_core::prelude::*;
 use lavina_core::repo::Storage;
-use lavina_core::room::{RoomId, RoomRegistry};
+use lavina_core::room::RoomRegistry;
 use lavina_core::terminator::Terminator;
-use proto_xmpp::bind::{BindResponse, Jid, Name, Resource, Server};
-use proto_xmpp::client::{Iq, Message, MessageType, Presence};
-use proto_xmpp::disco::*;
-use proto_xmpp::roster::RosterQuery;
-use proto_xmpp::session::Session;
+use proto_xmpp::bind::{Name, Resource};
 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
-        .write_event_async(Event::Decl(BytesDecl::new("1.0", None, None)))
-        .await?;
+    xml_writer.write_event_async(Event::Decl(BytesDecl::new("1.0", None, None))).await?;
     ServerStreamStart {
         from: "localhost".into(),
         lang: "en".into(),
@@ -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
-        .write_event_async(Event::Decl(BytesDecl::new("1.0", None, None)))
-        .await?;
+    xml_writer.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 {
-                        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);
-                        }
-                        _ => {},
-                    }
+                    conn.handle_update(&mut events, update).await?;
                     for i in &events {
                         xml_writer.write_event_async(i).await?;
                     }
@@ -402,249 +395,36 @@ async fn socket_final(
     Ok(())
 }
 
-async fn handle_packet(
-    output: &mut Vec<Event<'static>>,
-    packet: ClientPacket,
-    user: &Authenticated,
-    user_handle: &mut PlayerConnection,
-    rooms: &RoomRegistry,
-) -> Result<bool> {
-    Ok(match packet {
-        proto::ClientPacket::Iq(iq) => {
-            handle_iq(output, iq, rooms).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!()
+struct XmppConnection<'a> {
+    user: &'a Authenticated,
+    user_handle: &'a mut PlayerConnection,
+    rooms: &'a RoomRegistry,
+}
+
+impl<'a> XmppConnection<'a> {
+    async fn handle_packet(&mut self, output: &mut Vec<Event<'static>>, packet: ClientPacket) -> Result<bool> {
+        let res = match packet {
+            proto::ClientPacket::Iq(iq) => {
+                self.handle_iq(output, iq).await;
+                false
             }
-        }
-        proto::ClientPacket::Presence(p) => {
-            let response = if p.to.is_none() {
-                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(user.xmpp_name.clone()),
-                        server: Server("localhost".into()),
-                        resource: Some(user.xmpp_resource.clone()),
-                    }),
-                    ..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);
-        }
+            ClientPacket::Message(m) => {
+                self.handle_message(output, m).await?;
+                false
+            }
+            proto::ClientPacket::Presence(p) => {
+                self.handle_presence(output, p).await?;
+                false
+            }
+            proto::ClientPacket::StreamEnd => {
+                ServerStreamEnd.serialize(output);
+                true
+            }
+        };
+        Ok(res)
     }
 }
 
-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,52 @@
+//! 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(())
+}