From adf1d8c14c39b0bdcdbab12d30f9bd4317e5f262 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 5 May 2024 15:12:58 +0000 Subject: [PATCH 1/5] xmpp: Implement Message Archive Management stub for XEP-0313 (#60) https://xmpp.org/extensions/xep-0313.html Reviewed-on: https://git.vilunov.me/lavina/lavina/pulls/60 Co-authored-by: Mikhail Co-committed-by: Mikhail --- crates/projection-xmpp/src/iq.rs | 18 ++- crates/projection-xmpp/src/proto.rs | 3 + crates/projection-xmpp/tests/lib.rs | 170 ++++++++++++++++----- crates/proto-xmpp/src/bind.rs | 2 +- crates/proto-xmpp/src/client.rs | 12 +- crates/proto-xmpp/src/lib.rs | 1 + crates/proto-xmpp/src/mam.rs | 225 ++++++++++++++++++++++++++++ crates/proto-xmpp/src/roster.rs | 41 ++++- crates/proto-xmpp/src/stream.rs | 8 +- 9 files changed, 427 insertions(+), 53 deletions(-) create mode 100644 crates/proto-xmpp/src/mam.rs diff --git a/crates/projection-xmpp/src/iq.rs b/crates/projection-xmpp/src/iq.rs index 0031c36..fdff68d 100644 --- a/crates/projection-xmpp/src/iq.rs +++ b/crates/projection-xmpp/src/iq.rs @@ -4,16 +4,16 @@ use quick_xml::events::Event; use lavina_core::room::{RoomId, RoomRegistry}; use proto_xmpp::bind::{BindResponse, Jid, Name, Server}; -use proto_xmpp::client::{Iq, IqError, IqErrorType, IqType}; +use proto_xmpp::client::{Iq, IqError, IqErrorType, IqType, Message, MessageType}; use proto_xmpp::disco::{Feature, Identity, InfoQuery, Item, ItemQuery}; +use proto_xmpp::mam::{Fin, Set}; use proto_xmpp::roster::RosterQuery; use proto_xmpp::session::Session; +use proto_xmpp::xml::ToXml; 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>, iq: Iq) { match iq.body { @@ -87,6 +87,18 @@ impl<'a> XmppConnection<'a> { }; req.serialize(output); } + IqClientBody::MessageArchiveRequest(_) => { + let response = Iq { + from: iq.to, + id: iq.id, + to: None, + r#type: IqType::Result, + body: Fin { + set: Set { count: Some(0) }, + }, + }; + response.serialize(output); + } _ => { let req = Iq { from: None, diff --git a/crates/projection-xmpp/src/proto.rs b/crates/projection-xmpp/src/proto.rs index 0b157a6..2943509 100644 --- a/crates/projection-xmpp/src/proto.rs +++ b/crates/projection-xmpp/src/proto.rs @@ -7,6 +7,7 @@ use lavina_core::prelude::*; use proto_xmpp::bind::BindRequest; use proto_xmpp::client::{Iq, Message, Presence}; use proto_xmpp::disco::{InfoQuery, ItemQuery}; +use proto_xmpp::mam::MessageArchiveRequest; use proto_xmpp::roster::RosterQuery; use proto_xmpp::session::Session; use proto_xmpp::xml::*; @@ -18,6 +19,7 @@ pub enum IqClientBody { Roster(RosterQuery), DiscoInfo(InfoQuery), DiscoItem(ItemQuery), + MessageArchiveRequest(MessageArchiveRequest), Unknown(Ignore), } @@ -38,6 +40,7 @@ impl FromXml for IqClientBody { RosterQuery, InfoQuery, ItemQuery, + MessageArchiveRequest, { delegate_parsing!(Ignore, namespace, event).into() } diff --git a/crates/projection-xmpp/tests/lib.rs b/crates/projection-xmpp/tests/lib.rs index 3c82979..c1ea13a 100644 --- a/crates/projection-xmpp/tests/lib.rs +++ b/crates/projection-xmpp/tests/lib.rs @@ -1,4 +1,5 @@ use std::io::ErrorKind; +use std::str::from_utf8; use std::sync::Arc; use std::time::Duration; @@ -6,6 +7,7 @@ use anyhow::Result; use assert_matches::*; use prometheus::Registry as MetricsRegistry; use quick_xml::events::Event; +use quick_xml::name::LocalName; use quick_xml::NsReader; use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::io::{ReadHalf as GenericReadHalf, WriteHalf as GenericWriteHalf}; @@ -22,6 +24,10 @@ use lavina_core::LavinaCore; use projection_xmpp::{launch, RunningServer, ServerConfig}; use proto_xmpp::xml::{Continuation, FromXml, Parser}; +fn element_name<'a>(local_name: &LocalName<'a>) -> &'a str { + from_utf8(local_name.into_inner()).unwrap() +} + pub async fn read_irc_message(reader: &mut BufReader>, buf: &mut Vec) -> Result { let mut size = 0; let res = reader.read_until(b'\n', buf).await?; @@ -56,11 +62,11 @@ impl<'a> TestScope<'a> { } async fn expect_starttls_required(&mut self) -> Result<()> { - assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"features")); - assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"starttls")); - assert_matches!(self.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"required")); - assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"starttls")); - assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"features")); + assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "features")); + assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "starttls")); + assert_matches!(self.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "required")); + assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "starttls")); + assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "features")); Ok(()) } } @@ -93,20 +99,20 @@ impl<'a> TestScopeTls<'a> { } async fn expect_auth_mechanisms(&mut self) -> Result<()> { - assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"features")); - assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"mechanisms")); - assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"mechanism")); + assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "features")); + assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "mechanisms")); + assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "mechanism")); assert_matches!(self.next_xml_event().await?, Event::Text(b) => assert_eq!(&*b, b"PLAIN")); - assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"mechanism")); - assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"mechanisms")); - assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"features")); + assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "mechanism")); + assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "mechanisms")); + assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "features")); Ok(()) } async fn expect_bind_feature(&mut self) -> Result<()> { - assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"features")); - assert_matches!(self.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"bind")); - assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"features")); + assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "features")); + assert_matches!(self.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "bind")); + assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "features")); Ok(()) } @@ -119,6 +125,7 @@ impl<'a> TestScopeTls<'a> { } struct IgnoreCertVerification; + impl ServerCertVerifier for IgnoreCertVerification { fn verify_server_cert( &self, @@ -139,6 +146,7 @@ struct TestServer { core: LavinaCore, server: RunningServer, } + impl TestServer { async fn start() -> Result { let _ = tracing_subscriber::fmt::try_init(); @@ -187,10 +195,10 @@ async fn scenario_basic() -> Result<()> { s.send(r#""#).await?; s.send(r#""#).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!(element_name(&b.local_name()), "stream")); s.expect_starttls_required().await?; s.send(r#""#).await?; - assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed")); + assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed")); let buffer = s.buffer; tracing::info!("TLS feature negotiation complete"); @@ -209,26 +217,26 @@ async fn scenario_basic() -> Result<()> { s.send(r#""#).await?; s.send(r#""#).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!(element_name(&b.local_name()), "stream")); s.expect_auth_mechanisms().await?; - // base64-encoded b"\x00tester\x00password" + // base64-encoded "\x00tester\x00password" s.send(r#"AHRlc3RlcgBwYXNzd29yZA=="#) .await?; - assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"success")); + assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "success")); s.send(r#""#).await?; s.send(r#""#).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!(element_name(&b.local_name()), "stream")); s.expect_bind_feature().await?; s.send(r#"kek"#).await?; - assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"iq")); - assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"bind")); - assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"jid")); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "iq")); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "bind")); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "jid")); assert_matches!(s.next_xml_event().await?, Event::Text(b) => assert_eq!(&*b, b"tester@localhost/tester")); - assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"jid")); - assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"bind")); - assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"iq")); + assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "jid")); + assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "bind")); + assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "iq")); s.send(r#"Logged out"#).await?; stream.shutdown().await?; @@ -255,10 +263,10 @@ async fn scenario_wrong_password() -> Result<()> { s.send(r#""#).await?; s.send(r#""#).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!(element_name(&b.local_name()), "stream")); s.expect_starttls_required().await?; s.send(r#""#).await?; - assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed")); + assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed")); let buffer = s.buffer; tracing::info!("TLS feature negotiation complete"); @@ -277,14 +285,14 @@ async fn scenario_wrong_password() -> Result<()> { s.send(r#""#).await?; s.send(r#""#).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!(element_name(&b.local_name()), "stream")); s.expect_auth_mechanisms().await?; - // base64-encoded b"\x00tester\x00password2" + // base64-encoded "\x00tester\x00password2" s.send(r#"AHRlc3RlcgBwYXNzd29yZDI="#) .await?; - assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"failure")); - assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"not-authorized")); - assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"failure")); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "failure")); + assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "not-authorized")); + assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "failure")); let _ = stream.shutdown().await; @@ -309,10 +317,10 @@ async fn scenario_basic_without_headers() -> Result<()> { s.send(r#""#).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!(element_name(&b.local_name()), "stream")); s.expect_starttls_required().await?; s.send(r#""#).await?; - assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed")); + assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed")); let buffer = s.buffer; tracing::info!("TLS feature negotiation complete"); @@ -330,7 +338,7 @@ async fn scenario_basic_without_headers() -> Result<()> { s.send(r#""#).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!(element_name(&b.local_name()), "stream")); stream.shutdown().await?; @@ -357,10 +365,10 @@ async fn terminate_socket() -> Result<()> { s.send(r#""#).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!(element_name(&b.local_name()), "stream")); s.expect_starttls_required().await?; s.send(r#""#).await?; - assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed")); + assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed")); let connector = TlsConnector::from(Arc::new( ClientConfig::builder() @@ -379,3 +387,89 @@ async fn terminate_socket() -> Result<()> { Ok(()) } + +#[tokio::test] +async fn test_message_archive_request() -> Result<()> { + let mut server = TestServer::start().await?; + + // test scenario + + server.storage.create_user("tester").await?; + Authenticator::new(&server.storage).set_password("tester", "password").await?; + + let mut stream = TcpStream::connect(server.server.addr).await?; + let mut s = TestScope::new(&mut stream); + tracing::info!("TCP connection established"); + + s.send(r#""#).await?; + s.send(r#""#).await?; + assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {}); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream")); + s.expect_starttls_required().await?; + s.send(r#""#).await?; + assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "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.server.addr.ip()), stream).await?; + tracing::info!("TLS connection established"); + + let mut s = TestScopeTls::new(&mut stream, buffer); + + s.send(r#""#).await?; + s.send(r#""#).await?; + assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {}); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream")); + s.expect_auth_mechanisms().await?; + + // base64-encoded "\x00tester\x00password" + s.send(r#"AHRlc3RlcgBwYXNzd29yZA=="#) + .await?; + assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "success")); + s.send(r#""#).await?; + s.send(r#""#).await?; + assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {}); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream")); + s.expect_bind_feature().await?; + s.send(r#"kek"#).await?; + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "iq")); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "bind")); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "jid")); + assert_matches!(s.next_xml_event().await?, Event::Text(b) => assert_eq!(&*b, b"tester@localhost/tester")); + assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "jid")); + assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "bind")); + assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "iq")); + + s.send(r#""#).await?; + assert_matches!(s.next_xml_event().await?, Event::Start(b) => { + assert_eq!(element_name(&b.local_name()), "iq") + }); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => { + assert_eq!(element_name(&b.local_name()), "fin") + }); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => { + assert_eq!(element_name(&b.local_name()), "set") + }); + assert_matches!(s.next_xml_event().await?, Event::Start(b) => { + assert_eq!(element_name(&b.local_name()), "count") + }); + assert_matches!(s.next_xml_event().await?, Event::Text(b) => { + assert_eq!(&*b, b"0") + }); + + s.send(r#"Logged out"#).await?; + + stream.shutdown().await?; + + // wrap up + + server.shutdown().await?; + Ok(()) +} diff --git a/crates/proto-xmpp/src/bind.rs b/crates/proto-xmpp/src/bind.rs index d27a00e..d546141 100644 --- a/crates/proto-xmpp/src/bind.rs +++ b/crates/proto-xmpp/src/bind.rs @@ -74,8 +74,8 @@ impl Jid { pub struct BindRequest(pub Resource); impl FromXmlTag for BindRequest { - const NS: &'static str = XMLNS; const NAME: &'static str = "bind"; + const NS: &'static str = XMLNS; } impl FromXml for BindRequest { diff --git a/crates/proto-xmpp/src/client.rs b/crates/proto-xmpp/src/client.rs index 05807bd..08d203a 100644 --- a/crates/proto-xmpp/src/client.rs +++ b/crates/proto-xmpp/src/client.rs @@ -658,7 +658,7 @@ mod tests { #[tokio::test] async fn parse_message() { - let input = r#"daabbb"#; + let input = r#"daabbb"#; let result: Message = crate::xml::parse(input).unwrap(); assert_eq!( result, @@ -666,8 +666,8 @@ mod tests { from: None, id: Some("aacea".to_string()), to: Some(Jid { - name: Some(Name("nikita".into())), - server: Server("vlnv.dev".into()), + name: Some(Name("chelik".into())), + server: Server("xmpp.ru".into()), resource: None }), r#type: MessageType::Chat, @@ -681,7 +681,7 @@ mod tests { #[tokio::test] async fn parse_message_empty_custom() { - let input = r#"daabbb"#; + let input = r#"daabbb"#; let result: Message = crate::xml::parse(input).unwrap(); assert_eq!( result, @@ -689,8 +689,8 @@ mod tests { from: None, id: Some("aacea".to_string()), to: Some(Jid { - name: Some(Name("nikita".into())), - server: Server("vlnv.dev".into()), + name: Some(Name("chelik".into())), + server: Server("xmpp.ru".into()), resource: None }), r#type: MessageType::Chat, diff --git a/crates/proto-xmpp/src/lib.rs b/crates/proto-xmpp/src/lib.rs index d3e25ba..71e8a94 100644 --- a/crates/proto-xmpp/src/lib.rs +++ b/crates/proto-xmpp/src/lib.rs @@ -3,6 +3,7 @@ pub mod bind; pub mod client; pub mod disco; +pub mod mam; pub mod muc; mod prelude; pub mod roster; diff --git a/crates/proto-xmpp/src/mam.rs b/crates/proto-xmpp/src/mam.rs new file mode 100644 index 0000000..c8151f2 --- /dev/null +++ b/crates/proto-xmpp/src/mam.rs @@ -0,0 +1,225 @@ +use anyhow::{anyhow, Result}; +use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event}; +use quick_xml::name::{Namespace, ResolveResult}; +use std::io::Read; + +use crate::xml::*; + +pub const MAM_XMLNS: &'static str = "urn:xmpp:mam:2"; +pub const DATA_XMLNS: &'static str = "jabber:x:data"; +pub const RESULT_SET_XMLNS: &'static str = "http://jabber.org/protocol/rsm"; + +#[derive(PartialEq, Eq, Debug, Clone)] +pub struct MessageArchiveRequest { + pub x: Option, +} + +#[derive(PartialEq, Eq, Debug, Clone)] +pub struct X { + pub fields: Vec, +} + +#[derive(PartialEq, Eq, Debug, Clone)] +pub struct Field { + pub values: Vec, +} + +// Message archive response styled as a result set. +#[derive(PartialEq, Eq, Debug, Clone)] +pub struct Fin { + pub set: Set, +} + +#[derive(PartialEq, Eq, Debug, Clone)] +pub struct Set { + pub count: Option, +} + +impl ToXml for Fin { + fn serialize(&self, events: &mut Vec>) { + let fin_bytes = BytesStart::new(format!(r#"fin xmlns="{}" complete=True"#, MAM_XMLNS)); + let set_bytes = BytesStart::new(format!(r#"set xmlns="{}""#, RESULT_SET_XMLNS)); + events.push(Event::Start(fin_bytes)); + events.push(Event::Start(set_bytes)); + + if let &Some(count) = &self.set.count { + events.push(Event::Start(BytesStart::new("count"))); + events.push(Event::Text(BytesText::new(count.to_string().as_str()).into_owned())); + events.push(Event::End(BytesEnd::new("count"))); + } + events.push(Event::End(BytesEnd::new("set"))); + events.push(Event::End(BytesEnd::new("fin"))); + } +} + +impl FromXmlTag for X { + const NAME: &'static str = "x"; + const NS: &'static str = DATA_XMLNS; +} + +impl FromXmlTag for MessageArchiveRequest { + const NAME: &'static str = "query"; + const NS: &'static str = MAM_XMLNS; +} + +impl FromXml for X { + type P = impl Parser>; + + fn parse() -> Self::P { + |(mut namespace, mut event): (ResolveResult<'static>, &'static Event<'static>)| -> Result { + println!("X::parse {:?}", event); + + let bytes = match event { + Event::Start(bytes) if bytes.name().0 == X::NAME.as_bytes() => bytes, + Event::Empty(bytes) if bytes.name().0 == X::NAME.as_bytes() => return Ok(X { fields: vec![] }), + _ => return Err(anyhow!("Unexpected XML event: {event:?}")), + }; + let mut fields = vec![]; + loop { + (namespace, event) = yield; + match event { + Event::Start(_) => { + // start of + let mut values = vec![]; + loop { + (namespace, event) = yield; + match event { + Event::Start(bytes) if bytes.name().0 == b"value" => { + // start of + } + Event::End(bytes) if bytes.name().0 == b"field" => { + // end of + break; + } + _ => return Err(anyhow!("Unexpected XML event: {event:?}")), + } + (namespace, event) = yield; + let text: String = match event { + Event::Text(bytes) => { + // text inside + String::from_utf8(bytes.to_vec())? + } + _ => return Err(anyhow!("Unexpected XML event: {event:?}")), + }; + (namespace, event) = yield; + match event { + Event::End(bytes) if bytes.name().0 == b"value" => { + // end of + } + _ => return Err(anyhow!("Unexpected XML event: {event:?}")), + } + values.push(text); + } + fields.push(Field { values }) + } + Event::End(bytes) if bytes.name().0 == X::NAME.as_bytes() => { + // end of + return Ok(X { fields }); + } + _ => return Err(anyhow!("Unexpected XML event: {event:?}")), + } + } + } + } +} + +impl FromXml for MessageArchiveRequest { + type P = impl Parser>; + + fn parse() -> Self::P { + |(mut namespace, mut event): (ResolveResult<'static>, &'static Event<'static>)| -> Result { + println!("MessageArchiveRequest::parse {:?}", event); + + let bytes = match event { + Event::Empty(_) => return Ok(MessageArchiveRequest { x: None }), + Event::Start(bytes) => bytes, + _ => return Err(anyhow!("Unexpected XML event: {event:?}")), + }; + if bytes.name().0 != MessageArchiveRequest::NAME.as_bytes() { + return Err(anyhow!("Unexpected XML tag: {:?}", bytes.name())); + } + let ResolveResult::Bound(Namespace(ns)) = namespace else { + return Err(anyhow!("No namespace provided")); + }; + if ns != MAM_XMLNS.as_bytes() { + return Err(anyhow!("Incorrect namespace")); + } + (namespace, event) = yield; + match event { + Event::End(bytes) if bytes.name().0 == MessageArchiveRequest::NAME.as_bytes() => { + Ok(MessageArchiveRequest { x: None }) + } + Event::Start(bytes) | Event::Empty(bytes) if bytes.name().0 == X::NAME.as_bytes() => { + let x = delegate_parsing!(X, namespace, event)?; + Ok(MessageArchiveRequest { x: Some(x) }) + } + _ => Err(anyhow!("Unexpected XML event: {event:?}")), + } + } + } +} + +impl MessageArchiveRequest {} + +#[cfg(test)] +mod tests { + use super::*; + use crate::bind::{Jid, Name, Server}; + use crate::client::{Iq, IqType}; + + #[test] + fn test_parse_archive_query() { + let input = r#""#; + + let result: Iq = parse(input).unwrap(); + assert_eq!( + result, + Iq { + from: None, + id: "juliet1".to_string(), + to: Option::from(Jid { + name: None, + server: Server("pubsub.shakespeare.lit".into()), + resource: None, + }), + r#type: IqType::Set, + body: MessageArchiveRequest { x: None }, + } + ); + } + + #[test] + fn test_parse_query_messages_from_jid() { + let input = r#"value1juliet@capulet.lit"#; + + let result: Iq = parse(input).unwrap(); + assert_eq!( + result, + Iq { + from: None, + id: "juliet1".to_string(), + to: None, + r#type: IqType::Set, + body: MessageArchiveRequest { + x: Some(X { + fields: vec![ + Field { + values: vec!["value1".to_string()], + }, + Field { + values: vec!["juliet@capulet.lit".to_string()], + }, + ] + }) + }, + } + ); + } + + #[test] + fn test_parse_query_messages_from_jid_with_unclosed_tag() { + let input = r#"value1juliet@capulet.lit"#; + + assert!(parse::>(input).is_err()) + } +} diff --git a/crates/proto-xmpp/src/roster.rs b/crates/proto-xmpp/src/roster.rs index f7d0305..7eb97d1 100644 --- a/crates/proto-xmpp/src/roster.rs +++ b/crates/proto-xmpp/src/roster.rs @@ -2,7 +2,7 @@ use quick_xml::events::{BytesStart, Event}; use crate::xml::*; use anyhow::{anyhow, Result}; -use quick_xml::name::ResolveResult; +use quick_xml::name::{Namespace, ResolveResult}; pub const XMLNS: &'static str = "jabber:iq:roster"; @@ -14,6 +14,9 @@ impl FromXml for RosterQuery { fn parse() -> Self::P { |(mut namespace, mut event): (ResolveResult<'static>, &'static Event<'static>)| -> Result { + let ResolveResult::Bound(Namespace(ns)) = namespace else { + return Err(anyhow!("No namespace provided")); + }; match event { Event::Start(_) => (), Event::Empty(_) => return Ok(RosterQuery), @@ -38,3 +41,39 @@ impl ToXml for RosterQuery { events.push(Event::Empty(BytesStart::new(format!(r#"query xmlns="{}""#, XMLNS)))); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::bind::{Jid, Name, Resource, Server}; + use crate::client::{Iq, IqType}; + + #[test] + fn test_parse() { + let input = + r#""#; + + let result: Iq = parse(input).unwrap(); + assert_eq!( + result, + Iq { + from: Option::from(Jid { + name: Option::from(Name("juliet".into())), + server: Server("example.com".into()), + resource: Option::from(Resource("balcony".into())), + }), + id: "bv1bs71f".to_string(), + to: None, + r#type: IqType::Get, + body: RosterQuery, + } + ) + } + + #[test] + fn test_missing_namespace() { + let input = r#""#; + + assert!(parse::>(input).is_err()); + } +} diff --git a/crates/proto-xmpp/src/stream.rs b/crates/proto-xmpp/src/stream.rs index 8f46f31..b12891e 100644 --- a/crates/proto-xmpp/src/stream.rs +++ b/crates/proto-xmpp/src/stream.rs @@ -170,14 +170,14 @@ mod test { #[tokio::test] async fn client_stream_start_correct_parse() { - let input = r###""###; + let input = r###""###; let mut reader = NsReader::from_reader(input.as_bytes()); let mut buf = vec![]; let res = ClientStreamStart::parse(&mut reader, &mut buf).await.unwrap(); assert_eq!( res, ClientStreamStart { - to: "vlnv.dev".to_owned(), + to: "xmpp.ru".to_owned(), lang: Some("en".to_owned()), version: "1.0".to_owned() } @@ -187,12 +187,12 @@ mod test { #[tokio::test] async fn server_stream_start_write() { let input = ServerStreamStart { - from: "vlnv.dev".to_owned(), + from: "xmpp.ru".to_owned(), lang: "en".to_owned(), id: "stream_id".to_owned(), version: "1.0".to_owned(), }; - let expected = r###""###; + let expected = r###""###; let mut output: Vec = vec![]; let mut writer = Writer::new(&mut output); input.write_xml(&mut writer).await.unwrap(); From abe9a26925d25cc3cf0082f848338f9c9516d6b2 Mon Sep 17 00:00:00 2001 From: homycdev Date: Sun, 5 May 2024 17:21:40 +0000 Subject: [PATCH 2/5] irc: implement WHOIS command (#43) Reviewed-on: https://git.vilunov.me/lavina/lavina/pulls/43 Co-authored-by: homycdev Co-committed-by: homycdev --- crates/lavina-core/src/player.rs | 32 ++++++ crates/lavina-core/src/repo/mod.rs | 13 ++- crates/projection-irc/src/handler.rs | 18 +++ crates/projection-irc/src/lib.rs | 16 ++- crates/projection-irc/src/whois.rs | 67 +++++++++++ crates/projection-irc/tests/lib.rs | 7 ++ crates/projection-xmpp/src/lib.rs | 2 +- crates/proto-irc/src/client.rs | 106 +++++++++++++++++- crates/proto-irc/src/commands/mod.rs | 1 + crates/proto-irc/src/commands/whois/error.rs | 67 +++++++++++ crates/proto-irc/src/commands/whois/mod.rs | 2 + .../proto-irc/src/commands/whois/response.rs | 24 ++++ crates/proto-irc/src/lib.rs | 2 + crates/proto-irc/src/response.rs | 47 ++++++++ crates/proto-irc/src/server.rs | 32 +++++- 15 files changed, 427 insertions(+), 9 deletions(-) create mode 100644 crates/projection-irc/src/handler.rs create mode 100644 crates/projection-irc/src/whois.rs create mode 100644 crates/proto-irc/src/commands/mod.rs create mode 100644 crates/proto-irc/src/commands/whois/error.rs create mode 100644 crates/proto-irc/src/commands/whois/mod.rs create mode 100644 crates/proto-irc/src/commands/whois/response.rs create mode 100644 crates/proto-irc/src/response.rs diff --git a/crates/lavina-core/src/player.rs b/crates/lavina-core/src/player.rs index 08ce37a..21914c4 100644 --- a/crates/lavina-core/src/player.rs +++ b/crates/lavina-core/src/player.rs @@ -8,6 +8,7 @@ //! A player actor is a serial handler of commands from a single player. It is preferable to run all per-player validations in the player actor, //! so that they don't overload the room actor. use std::collections::{HashMap, HashSet}; +use std::ops::Deref; use std::sync::Arc; use chrono::{DateTime, Utc}; @@ -124,6 +125,15 @@ impl PlayerConnection { self.player_handle.send(ActorCommand::ClientCommand(cmd, self.connection_id.clone())).await; Ok(deferred.await?) } + + /// Handler in [Player::check_user_existence]. + #[tracing::instrument(skip(self), name = "PlayerConnection::check_user_existence")] + pub async fn check_user_existence(&self, recipient: PlayerId) -> Result { + let (promise, deferred) = oneshot(); + let cmd = ClientCommand::GetInfo { recipient, promise }; + self.player_handle.send(ActorCommand::ClientCommand(cmd, self.connection_id.clone())).await; + Ok(deferred.await?) + } } /// Handle to a player actor. @@ -200,6 +210,15 @@ pub enum ClientCommand { body: Str, promise: Promise<()>, }, + GetInfo { + recipient: PlayerId, + promise: Promise, + }, +} + +pub enum GetInfoResult { + UserExists, + UserDoesntExist, } pub enum JoinResult { @@ -500,6 +519,10 @@ impl Player { self.send_dialog_message(connection_id, recipient, body).await; let _ = promise.send(()); } + ClientCommand::GetInfo { recipient, promise } => { + let result = self.check_user_existence(recipient).await; + let _ = promise.send(result); + } } } @@ -596,6 +619,15 @@ impl Player { self.broadcast_update(update, connection_id).await; } + #[tracing::instrument(skip(self), name = "Player::check_user_existence")] + async fn check_user_existence(&self, recipient: PlayerId) -> GetInfoResult { + if self.storage.check_user_existence(recipient.as_inner().as_ref()).await.unwrap() { + GetInfoResult::UserExists + } else { + GetInfoResult::UserDoesntExist + } + } + /// Broadcasts an update to all connections except the one with the given id. /// /// This is called after handling a client command. diff --git a/crates/lavina-core/src/repo/mod.rs b/crates/lavina-core/src/repo/mod.rs index f0d210c..4c29651 100644 --- a/crates/lavina-core/src/repo/mod.rs +++ b/crates/lavina-core/src/repo/mod.rs @@ -7,7 +7,7 @@ use anyhow::anyhow; use chrono::{DateTime, Utc}; use serde::Deserialize; use sqlx::sqlite::SqliteConnectOptions; -use sqlx::{ConnectOptions, Connection, FromRow, Sqlite, SqliteConnection, Transaction}; +use sqlx::{ConnectOptions, Connection, Execute, FromRow, Sqlite, SqliteConnection, Transaction}; use tokio::sync::Mutex; use crate::prelude::*; @@ -56,6 +56,17 @@ impl Storage { Ok(res) } + #[tracing::instrument(skip(self), name = "Storage::check_user_existence")] + pub async fn check_user_existence(&self, username: &str) -> Result { + let mut executor = self.conn.lock().await; + let result: Option<(String,)> = sqlx::query_as("select name from users where name = ?;") + .bind(username) + .fetch_optional(&mut *executor) + .await?; + + Ok(result.is_some()) + } + #[tracing::instrument(skip(self), name = "Storage::retrieve_room_by_name")] pub async fn retrieve_room_by_name(&self, name: &str) -> Result> { let mut executor = self.conn.lock().await; diff --git a/crates/projection-irc/src/handler.rs b/crates/projection-irc/src/handler.rs new file mode 100644 index 0000000..2e7e0da --- /dev/null +++ b/crates/projection-irc/src/handler.rs @@ -0,0 +1,18 @@ +use lavina_core::{player::PlayerConnection, prelude::Str, LavinaCore}; +use std::future::Future; +use tokio::io::AsyncWrite; + +pub struct IrcConnection<'a, T: AsyncWrite + Unpin> { + pub server_name: Str, + /// client is nick of requester + pub client: Str, + pub writer: &'a mut T, + pub player_connection: &'a mut PlayerConnection, +} + +pub trait Handler +where + T: AsyncWrite + Unpin, +{ + fn handle(&self, arg: IrcConnection) -> impl Future>; +} diff --git a/crates/projection-irc/src/lib.rs b/crates/projection-irc/src/lib.rs index 2a310a1..a18e5fc 100644 --- a/crates/projection-irc/src/lib.rs +++ b/crates/projection-irc/src/lib.rs @@ -28,11 +28,14 @@ use proto_irc::server::{AwayStatus, ServerMessage, ServerMessageBody}; use proto_irc::user::PrefixedNick; use proto_irc::{Chan, Recipient, Tag}; use sasl::AuthBody; - mod cap; +use handler::Handler; +mod whois; use crate::cap::Capabilities; +mod handler; + pub const APP_VERSION: &str = concat!("lavina", "_", env!("CARGO_PKG_VERSION")); #[derive(Deserialize, Debug, Clone)] @@ -806,6 +809,17 @@ async fn handle_incoming_message( log::warn!("Local chans not supported"); } }, + ClientMessage::Whois { arg } => { + arg.handle(handler::IrcConnection { + server_name: config.server_name.clone(), + client: user.nickname.clone(), + writer, + player_connection: user_handle, + }) + .await?; + + writer.flush().await?; + } ClientMessage::Mode { target } => { match target { Recipient::Nick(nickname) => { diff --git a/crates/projection-irc/src/whois.rs b/crates/projection-irc/src/whois.rs new file mode 100644 index 0000000..f48b90f --- /dev/null +++ b/crates/projection-irc/src/whois.rs @@ -0,0 +1,67 @@ +use lavina_core::{ + player::{GetInfoResult, PlayerId}, + prelude::Str, +}; +use proto_irc::{ + client::command_args::Whois, + commands::whois::{ + error::{ErrNoNicknameGiven431, ErrNoSuchNick401}, + response::RplEndOfWhois318, + }, + response::{IrcResponseMessage, WriteResponse}, +}; +use tokio::io::AsyncWrite; + +use crate::handler::{Handler, IrcConnection}; + +impl Handler for Whois { + async fn handle(&self, body: IrcConnection<'_, T>) -> anyhow::Result<()> { + match self { + Whois::Nick(nick) => handle_nick_target(nick.clone(), body).await?, + Whois::TargetNick(_, nick) => handle_nick_target(nick.clone(), body).await?, + Whois::EmptyArgs => { + let IrcConnection { + server_name, + mut writer, + .. + } = body; + IrcResponseMessage::empty_tags( + Some(server_name.clone()), + ErrNoNicknameGiven431::new(server_name.clone()), + ) + .write_response(&mut writer) + .await? + } + } + Ok(()) + } +} + +async fn handle_nick_target(nick: Str, body: IrcConnection<'_, impl AsyncWrite + Unpin>) -> anyhow::Result<()> { + let IrcConnection { + server_name, + mut writer, + client, + player_connection, + } = body; + + if let GetInfoResult::UserDoesntExist = + player_connection.check_user_existence(PlayerId::from(nick.clone())?).await? + { + IrcResponseMessage::empty_tags( + Some(server_name.clone()), + ErrNoSuchNick401::new(client.clone(), nick.clone()), + ) + .write_response(&mut writer) + .await? + } + + IrcResponseMessage::empty_tags( + Some(server_name.clone()), + RplEndOfWhois318::new(client.clone(), nick.clone()), + ) + .write_response(&mut writer) + .await?; + + Ok(()) +} diff --git a/crates/projection-irc/tests/lib.rs b/crates/projection-irc/tests/lib.rs index 2b069ef..9b72947 100644 --- a/crates/projection-irc/tests/lib.rs +++ b/crates/projection-irc/tests/lib.rs @@ -383,7 +383,14 @@ async fn scenario_two_users() -> Result<()> { // The second user should receive the PART message s2.expect(":tester1 PART #test").await?; + s1.send("WHOIS tester2").await?; + s1.expect(":testserver 318 tester1 tester2 :End of /WHOIS list").await?; + stream1.shutdown().await?; + s2.send("WHOIS tester3").await?; + s2.expect(":testserver 401 tester2 tester3 :No such nick/channel").await?; + s2.expect(":testserver 318 tester2 tester3 :End of /WHOIS list").await?; + stream2.shutdown().await?; server.shutdown().await?; diff --git a/crates/projection-xmpp/src/lib.rs b/crates/projection-xmpp/src/lib.rs index eec6fc3..9ae294f 100644 --- a/crates/projection-xmpp/src/lib.rs +++ b/crates/projection-xmpp/src/lib.rs @@ -201,7 +201,7 @@ async fn handle_socket( pin!(termination); select! { biased; - _ = &mut termination =>{ + _ = &mut termination => { log::info!("Socket handling was terminated"); return Ok(()) }, diff --git a/crates/proto-irc/src/client.rs b/crates/proto-irc/src/client.rs index a692e92..5a62427 100644 --- a/crates/proto-irc/src/client.rs +++ b/crates/proto-irc/src/client.rs @@ -1,9 +1,9 @@ -use super::*; - use anyhow::{anyhow, Result}; use nom::combinator::{all_consuming, opt}; use nonempty::NonEmpty; +use super::*; + /// Client-to-server command. #[derive(Clone, Debug, PartialEq, Eq)] pub enum ClientMessage { @@ -42,6 +42,10 @@ pub enum ClientMessage { Who { target: Recipient, // aka mask }, + /// WHOIS [] + Whois { + arg: command_args::Whois, + }, /// `TOPIC :` Topic { chan: Chan, @@ -63,6 +67,17 @@ pub enum ClientMessage { Authenticate(Str), } +pub mod command_args { + use crate::prelude::Str; + + #[derive(Clone, Debug, PartialEq, Eq)] + pub enum Whois { + Nick(Str), + TargetNick(Str, Str), + EmptyArgs, + } +} + pub fn client_message(input: &str) -> Result { let res = all_consuming(alt(( client_message_capability, @@ -74,6 +89,7 @@ pub fn client_message(input: &str) -> Result { client_message_join, client_message_mode, client_message_who, + client_message_whois, client_message_topic, client_message_part, client_message_privmsg, @@ -177,6 +193,31 @@ fn client_message_who(input: &str) -> IResult<&str, ClientMessage> { Ok((input, ClientMessage::Who { target })) } +fn client_message_whois(input: &str) -> IResult<&str, ClientMessage> { + let (input, _) = tag("WHOIS ")(input)?; + let args: Vec<_> = input.split_whitespace().collect(); + match args.as_slice()[..] { + [nick] => Ok(( + "", + ClientMessage::Whois { + arg: command_args::Whois::Nick(nick.into()), + }, + )), + [target, nick, ..] => Ok(( + "", + ClientMessage::Whois { + arg: command_args::Whois::TargetNick(target.into(), nick.into()), + }, + )), + [] => Ok(( + "", + ClientMessage::Whois { + arg: command_args::Whois::EmptyArgs, + }, + )), + } +} + fn client_message_topic(input: &str) -> IResult<&str, ClientMessage> { let (input, _) = tag("TOPIC ")(input)?; let (input, chan) = chan(input)?; @@ -311,6 +352,7 @@ mod test { use nonempty::nonempty; use super::*; + #[test] fn test_client_message_cap_ls() { let input = "CAP LS 302"; @@ -360,6 +402,66 @@ mod test { assert_matches!(result, Ok(result) => assert_eq!(expected, result)); } #[test] + fn test_client_message_whois() { + let test_user = "WHOIS val"; + let test_user_user = "WHOIS val val"; + let test_server_user = "WHOIS com.test.server user"; + let test_user_server = "WHOIS user com.test.server"; + let test_users_list = "WHOIS user_1,user_2,user_3"; + let test_server_users_list = "WHOIS com.test.server user_1,user_2,user_3"; + let test_more_than_two_params = "WHOIS test.server user_1,user_2,user_3 whatever spam"; + let test_none_none_params = "WHOIS "; + + let res_one_arg = client_message(test_user); + let res_user_user = client_message(test_user_user); + + let res_server_user = client_message(test_server_user); + let res_user_server = client_message(test_user_server); + let res_users_list = client_message(test_users_list); + let res_server_users_list = client_message(test_server_users_list); + let res_more_than_two_params = client_message(test_more_than_two_params); + let res_none_none_params = client_message(test_none_none_params); + + let expected_arg = ClientMessage::Whois { + arg: command_args::Whois::Nick("val".into()), + }; + let expected_user_user = ClientMessage::Whois { + arg: command_args::Whois::TargetNick("val".into(), "val".into()), + }; + + let expected_server_user = ClientMessage::Whois { + arg: command_args::Whois::TargetNick("com.test.server".into(), "user".into()), + }; + + let expected_user_server = ClientMessage::Whois { + arg: command_args::Whois::TargetNick("user".into(), "com.test.server".into()), + }; + + let expected_user_list = ClientMessage::Whois { + arg: command_args::Whois::Nick("user_1,user_2,user_3".into()), + }; + let expected_server_user_list = ClientMessage::Whois { + arg: command_args::Whois::TargetNick("com.test.server".into(), "user_1,user_2,user_3".into()), + }; + + let expected_more_than_two_params = ClientMessage::Whois { + arg: command_args::Whois::TargetNick("test.server".into(), "user_1,user_2,user_3".into()), + }; + + let expected_none_none_params = ClientMessage::Whois { + arg: command_args::Whois::EmptyArgs, + }; + + assert_matches!(res_one_arg, Ok(result) => assert_eq!(expected_arg, result)); + assert_matches!(res_user_user, Ok(result) => assert_eq!(expected_user_user, result)); + assert_matches!(res_server_user, Ok(result) => assert_eq!(expected_server_user, result)); + assert_matches!(res_user_server, Ok(result) => assert_eq!(expected_user_server, result)); + assert_matches!(res_users_list, Ok(result) => assert_eq!(expected_user_list, result)); + assert_matches!(res_server_users_list, Ok(result) => assert_eq!(expected_server_user_list, result)); + assert_matches!(res_more_than_two_params, Ok(result) => assert_eq!(expected_more_than_two_params, result)); + assert_matches!(res_none_none_params, Ok(result) => assert_eq!(expected_none_none_params, result)) + } + #[test] fn test_client_message_user() { let input = "USER SomeNick 8 * :Real Name"; let expected = ClientMessage::User { diff --git a/crates/proto-irc/src/commands/mod.rs b/crates/proto-irc/src/commands/mod.rs new file mode 100644 index 0000000..80525f8 --- /dev/null +++ b/crates/proto-irc/src/commands/mod.rs @@ -0,0 +1 @@ +pub mod whois; diff --git a/crates/proto-irc/src/commands/whois/error.rs b/crates/proto-irc/src/commands/whois/error.rs new file mode 100644 index 0000000..426095b --- /dev/null +++ b/crates/proto-irc/src/commands/whois/error.rs @@ -0,0 +1,67 @@ +use tokio::io::{AsyncWrite, AsyncWriteExt}; + +use crate::{prelude::Str, response::WriteResponse}; + +/// ErrNoSuchNick401 +pub struct ErrNoSuchNick401 { + client: Str, + nick: Str, +} + +impl ErrNoSuchNick401 { + pub fn new(client: Str, nick: Str) -> Self { + ErrNoSuchNick401 { client, nick } + } +} + +/// ErrNoSuchServer402 +struct ErrNoSuchServer402 { + client: Str, + /// target parameter in WHOIS + /// example: `/whois ` + server_name: Str, +} + +/// ErrNoNicknameGiven431 +pub struct ErrNoNicknameGiven431 { + client: Str, +} +impl ErrNoNicknameGiven431 { + pub fn new(client: Str) -> Self { + ErrNoNicknameGiven431 { client } + } +} + +impl WriteResponse for ErrNoSuchNick401 { + async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> { + writer.write_all(b"401 ").await?; + writer.write_all(self.client.as_bytes()).await?; + writer.write_all(b" ").await?; + writer.write_all(self.nick.as_bytes()).await?; + writer.write_all(b" :").await?; + writer.write_all("No such nick/channel".as_bytes()).await?; + Ok(()) + } +} + +impl WriteResponse for ErrNoNicknameGiven431 { + async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> { + writer.write_all(b"431").await?; + writer.write_all(self.client.as_bytes()).await?; + writer.write_all(b" :").await?; + writer.write_all("No nickname given".as_bytes()).await?; + Ok(()) + } +} + +impl WriteResponse for ErrNoSuchServer402 { + async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> { + writer.write_all(b"402 ").await?; + writer.write_all(self.client.as_bytes()).await?; + writer.write_all(b" ").await?; + writer.write_all(self.server_name.as_bytes()).await?; + writer.write_all(b" :").await?; + writer.write_all("No such server".as_bytes()).await?; + Ok(()) + } +} diff --git a/crates/proto-irc/src/commands/whois/mod.rs b/crates/proto-irc/src/commands/whois/mod.rs new file mode 100644 index 0000000..52897d3 --- /dev/null +++ b/crates/proto-irc/src/commands/whois/mod.rs @@ -0,0 +1,2 @@ +pub mod error; +pub mod response; diff --git a/crates/proto-irc/src/commands/whois/response.rs b/crates/proto-irc/src/commands/whois/response.rs new file mode 100644 index 0000000..506ebf4 --- /dev/null +++ b/crates/proto-irc/src/commands/whois/response.rs @@ -0,0 +1,24 @@ +use tokio::io::{AsyncWrite, AsyncWriteExt}; + +use crate::{prelude::Str, response::WriteResponse}; + +pub struct RplEndOfWhois318 { + client: Str, + nick: Str, +} +impl RplEndOfWhois318 { + pub fn new(client: Str, nick: Str) -> Self { + RplEndOfWhois318 { client, nick } + } +} +impl WriteResponse for RplEndOfWhois318 { + async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> { + writer.write_all(b"318 ").await?; + writer.write_all(self.client.as_bytes()).await?; + writer.write_all(b" ").await?; + writer.write_all(self.nick.as_bytes()).await?; + writer.write_all(b" :").await?; + writer.write_all("End of /WHOIS list".as_bytes()).await?; + Ok(()) + } +} diff --git a/crates/proto-irc/src/lib.rs b/crates/proto-irc/src/lib.rs index 3c7ade2..b2c8a37 100644 --- a/crates/proto-irc/src/lib.rs +++ b/crates/proto-irc/src/lib.rs @@ -1,6 +1,8 @@ //! Client-to-Server IRC protocol. pub mod client; +pub mod commands; mod prelude; +pub mod response; pub mod server; #[cfg(test)] mod testkit; diff --git a/crates/proto-irc/src/response.rs b/crates/proto-irc/src/response.rs new file mode 100644 index 0000000..8c2ac4f --- /dev/null +++ b/crates/proto-irc/src/response.rs @@ -0,0 +1,47 @@ +use std::future::Future; + +use tokio::io::{AsyncWrite, AsyncWriteExt}; + +use crate::prelude::Str; +use crate::Tag; + +pub trait WriteResponse { + fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> impl Future>; +} + +/// Server-to-client enum agnostic message +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct IrcResponseMessage { + /// Optional tags section, prefixed with `@` + pub tags: Vec, + /// Optional server name, prefixed with `:`. + pub sender: Option, + pub body: T, +} + +impl IrcResponseMessage { + pub fn empty_tags(sender: Option, body: T) -> Self { + IrcResponseMessage { + tags: vec![], + sender, + body, + } + } + + pub fn new(tags: Vec, sender: Option, body: T) -> Self { + IrcResponseMessage { tags, sender, body } + } +} + +impl WriteResponse for IrcResponseMessage { + async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> { + if let Some(sender) = &self.sender { + writer.write_all(b":").await?; + writer.write_all(sender.as_bytes()).await?; + writer.write_all(b" ").await?; + } + self.body.write_response(writer).await?; + writer.write_all(b"\r\n").await?; + Ok(()) + } +} diff --git a/crates/proto-irc/src/server.rs b/crates/proto-irc/src/server.rs index 53cc05d..ac50f72 100644 --- a/crates/proto-irc/src/server.rs +++ b/crates/proto-irc/src/server.rs @@ -1,12 +1,11 @@ -use std::sync::Arc; - use nonempty::NonEmpty; use tokio::io::AsyncWrite; use tokio::io::AsyncWriteExt; -use super::*; use crate::user::PrefixedNick; +use super::*; + /// Server-to-client message. #[derive(Clone, Debug, PartialEq, Eq)] pub struct ServerMessage { @@ -114,6 +113,12 @@ pub enum ServerMessageBody { /// Usually `b"End of WHO list"` msg: Str, }, + N318EndOfWhois { + client: Str, + nick: Str, + /// Usually `b"End of /WHOIS list"` + msg: Str, + }, N332Topic { client: Str, chat: Chan, @@ -143,6 +148,10 @@ pub enum ServerMessageBody { client: Str, chan: Chan, }, + N431ErrNoNicknameGiven { + client: Str, + message: Str, + }, N474BannedFromChan { client: Str, chan: Chan, @@ -280,6 +289,14 @@ impl ServerMessageBody { writer.write_all(b" :").await?; writer.write_all(msg.as_bytes()).await?; } + ServerMessageBody::N318EndOfWhois { client, nick, msg } => { + writer.write_all(b"318 ").await?; + writer.write_all(client.as_bytes()).await?; + writer.write_all(b" ").await?; + writer.write_all(nick.as_bytes()).await?; + writer.write_all(b" :").await?; + writer.write_all(msg.as_bytes()).await?; + } ServerMessageBody::N332Topic { client, chat, topic } => { writer.write_all(b"332 ").await?; writer.write_all(client.as_bytes()).await?; @@ -342,6 +359,12 @@ impl ServerMessageBody { chan.write_async(writer).await?; writer.write_all(b" :End of /NAMES list").await?; } + ServerMessageBody::N431ErrNoNicknameGiven { client, message } => { + writer.write_all(b"431").await?; + writer.write_all(client.as_bytes()).await?; + writer.write_all(b" :").await?; + writer.write_all(message.as_bytes()).await?; + } ServerMessageBody::N474BannedFromChan { client, chan, message } => { writer.write_all(b"474 ").await?; writer.write_all(client.as_bytes()).await?; @@ -470,9 +493,10 @@ fn server_message_body_cap(input: &str) -> IResult<&str, ServerMessageBody> { mod test { use assert_matches::*; - use super::*; use crate::testkit::*; + use super::*; + #[test] fn test_server_message_notice() { let input = "NOTICE * :*** Looking up your hostname...\r\n"; From 8ac64ba8f5c07bc67052c9bb9ad11627b6b0f26d Mon Sep 17 00:00:00 2001 From: Nikita Vilunov Date: Sat, 4 May 2024 00:37:49 +0200 Subject: [PATCH 3/5] get rid of storage usages in projections --- crates/lavina-core/src/auth.rs | 9 +++--- crates/lavina-core/src/lib.rs | 4 +++ crates/projection-irc/src/lib.rs | 44 +++++++++++------------------ crates/projection-irc/tests/lib.rs | 26 ++++++++--------- crates/projection-xmpp/src/lib.rs | 24 ++++++---------- crates/projection-xmpp/tests/lib.rs | 12 ++++---- src/http.rs | 10 +++---- src/main.rs | 4 +-- 8 files changed, 60 insertions(+), 73 deletions(-) diff --git a/crates/lavina-core/src/auth.rs b/crates/lavina-core/src/auth.rs index 6e6f4ba..7d923de 100644 --- a/crates/lavina-core/src/auth.rs +++ b/crates/lavina-core/src/auth.rs @@ -17,11 +17,12 @@ pub enum UpdatePasswordResult { UserNotFound, } -pub struct Authenticator<'a> { - storage: &'a Storage, +#[derive(Clone)] +pub struct Authenticator { + storage: Storage, } -impl<'a> Authenticator<'a> { - pub fn new(storage: &'a Storage) -> Self { +impl Authenticator { + pub fn new(storage: Storage) -> Self { Self { storage } } diff --git a/crates/lavina-core/src/lib.rs b/crates/lavina-core/src/lib.rs index b251ed9..3ea0b33 100644 --- a/crates/lavina-core/src/lib.rs +++ b/crates/lavina-core/src/lib.rs @@ -2,6 +2,7 @@ use anyhow::Result; use prometheus::Registry as MetricsRegistry; +use crate::auth::Authenticator; use crate::dialog::DialogRegistry; use crate::player::PlayerRegistry; use crate::repo::Storage; @@ -22,6 +23,7 @@ pub struct LavinaCore { pub players: PlayerRegistry, pub rooms: RoomRegistry, pub dialogs: DialogRegistry, + pub authenticator: Authenticator, } impl LavinaCore { @@ -31,10 +33,12 @@ impl LavinaCore { let dialogs = DialogRegistry::new(storage.clone()); let players = PlayerRegistry::empty(rooms.clone(), dialogs.clone(), storage.clone(), &mut metrics)?; dialogs.set_players(players.clone()).await; + let authenticator = Authenticator::new(storage.clone()); Ok(LavinaCore { players, rooms, dialogs, + authenticator, }) } diff --git a/crates/projection-irc/src/lib.rs b/crates/projection-irc/src/lib.rs index a18e5fc..bb64824 100644 --- a/crates/projection-irc/src/lib.rs +++ b/crates/projection-irc/src/lib.rs @@ -17,7 +17,6 @@ use tokio::sync::mpsc::channel; use lavina_core::auth::{Authenticator, Verdict}; use lavina_core::player::*; use lavina_core::prelude::*; -use lavina_core::repo::Storage; use lavina_core::room::{RoomId, RoomInfo, RoomRegistry}; use lavina_core::terminator::Terminator; use lavina_core::LavinaCore; @@ -61,9 +60,8 @@ async fn handle_socket( config: ServerConfig, mut stream: TcpStream, socket_addr: &SocketAddr, - mut core: LavinaCore, + core: LavinaCore, termination: Deferred<()>, // TODO use it to stop the connection gracefully - mut storage: Storage, ) -> Result<()> { log::info!("Received an IRC connection from {socket_addr}"); let (reader, writer) = stream.split(); @@ -77,7 +75,7 @@ async fn handle_socket( log::info!("Socket handling was terminated"); return Ok(()) }, - registered_user = handle_registration(&mut reader, &mut writer, &mut storage, &config) => + registered_user = handle_registration(&mut reader, &mut writer, &core, &config) => match registered_user { Ok(user) => { log::debug!("User registered"); @@ -126,7 +124,7 @@ impl RegistrationState { &mut self, msg: ClientMessage, writer: &mut BufWriter>, - storage: &mut Storage, + core: &LavinaCore, config: &ServerConfig, ) -> Result> { match msg { @@ -217,7 +215,7 @@ impl RegistrationState { realname, enabled_capabilities: self.enabled_capabilities, }; - self.finalize_auth(candidate_user, writer, storage, config).await + self.finalize_auth(candidate_user, writer, &core.authenticator, config).await } }, ClientMessage::Nick { nickname } => { @@ -231,7 +229,7 @@ impl RegistrationState { realname: realname.clone(), enabled_capabilities: self.enabled_capabilities, }; - self.finalize_auth(candidate_user, writer, storage, config).await + self.finalize_auth(candidate_user, writer, &core.authenticator, config).await } else { self.future_nickname = Some(nickname); Ok(None) @@ -248,7 +246,7 @@ impl RegistrationState { realname, enabled_capabilities: self.enabled_capabilities, }; - self.finalize_auth(candidate_user, writer, storage, config).await + self.finalize_auth(candidate_user, writer, &core.authenticator, config).await } else { self.future_username = Some((username, realname)); Ok(None) @@ -279,7 +277,7 @@ impl RegistrationState { } } else { let body = AuthBody::from_str(body.as_bytes())?; - if let Err(e) = auth_user(storage, &body.login, &body.password).await { + if let Err(e) = auth_user(&core.authenticator, &body.login, &body.password).await { tracing::warn!("Authentication failed: {:?}", e); let target = self.future_nickname.clone().unwrap_or_else(|| "*".into()); sasl_fail_message(config.server_name.clone(), target, "Bad credentials".into()) @@ -327,7 +325,7 @@ impl RegistrationState { &mut self, candidate_user: RegisteredUser, writer: &mut BufWriter>, - storage: &mut Storage, + authenticator: &Authenticator, config: &ServerConfig, ) -> Result> { if self.enabled_capabilities.contains(Capabilities::Sasl) @@ -346,7 +344,7 @@ impl RegistrationState { writer.flush().await?; return Ok(None); }; - auth_user(storage, &*candidate_user.nickname, &*candidate_password).await?; + auth_user(authenticator, &*candidate_user.nickname, &*candidate_password).await?; Ok(Some(candidate_user)) } } @@ -355,7 +353,7 @@ impl RegistrationState { async fn handle_registration<'a>( reader: &mut BufReader>, writer: &mut BufWriter>, - storage: &mut Storage, + core: &LavinaCore, config: &ServerConfig, ) -> Result { let mut buffer = vec![]; @@ -391,7 +389,7 @@ async fn handle_registration<'a>( } }; tracing::debug!("Incoming IRC message: {msg:?}"); - if let Some(user) = state.handle_msg(msg, writer, storage, config).await? { + if let Some(user) = state.handle_msg(msg, writer, core, config).await? { break Ok(user); } buffer.clear(); @@ -408,8 +406,8 @@ fn sasl_fail_message(sender: Str, nick: Str, text: Str) -> ServerMessage { } } -async fn auth_user(storage: &mut Storage, login: &str, plain_password: &str) -> Result<()> { - let verdict = Authenticator::new(storage).authenticate(login, plain_password).await?; +async fn auth_user(authenticator: &Authenticator, login: &str, plain_password: &str) -> Result<()> { + let verdict = authenticator.authenticate(login, plain_password).await?; // TODO properly map these onto protocol messages match verdict { Verdict::Authenticated => Ok(()), @@ -420,7 +418,7 @@ async fn auth_user(storage: &mut Storage, login: &str, plain_password: &str) -> async fn handle_registered_socket<'a>( config: ServerConfig, - mut players: PlayerRegistry, + players: PlayerRegistry, rooms: RoomRegistry, reader: &mut BufReader>, writer: &mut BufWriter>, @@ -756,8 +754,6 @@ async fn handle_incoming_message( ClientMessage::Who { target } => match &target { Recipient::Nick(nick) => { // TODO handle non-existing user - let mut username = format!("~{nick}"); - let mut host = format!("user/{nick}"); ServerMessage { tags: vec![], sender: Some(config.server_name.clone()), @@ -874,7 +870,7 @@ fn user_to_who_msg(config: &ServerConfig, requestor: &RegisteredUser, target_use let username = format!("~{target_user_nickname}").into(); // User's host is not public, replace it with `user/` pattern - let mut host = format!("user/{target_user_nickname}").into(); + let host = format!("user/{target_user_nickname}").into(); ServerMessageBody::N352WhoReply { client: requestor.nickname.clone(), @@ -1010,12 +1006,7 @@ impl RunningServer { } } -pub async fn launch( - config: ServerConfig, - core: LavinaCore, - metrics: MetricsRegistry, - storage: Storage, -) -> Result { +pub async fn launch(config: ServerConfig, core: LavinaCore, metrics: MetricsRegistry) -> Result { 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")?; @@ -1056,9 +1047,8 @@ pub async fn launch( let core = core.clone(); let current_connections_clone = current_connections.clone(); let stopped_tx = stopped_tx.clone(); - let storage = storage.clone(); async move { - match handle_socket(config, stream, &socket_addr, core, termination, storage).await { + match handle_socket(config, stream, &socket_addr, core, termination).await { Ok(_) => log::info!("Connection terminated"), Err(err) => log::warn!("Connection failed: {err}"), } diff --git a/crates/projection-irc/tests/lib.rs b/crates/projection-irc/tests/lib.rs index 9b72947..583b18f 100644 --- a/crates/projection-irc/tests/lib.rs +++ b/crates/projection-irc/tests/lib.rs @@ -119,7 +119,7 @@ impl TestServer { }) .await?; let core = LavinaCore::new(metrics.clone(), storage.clone()).await?; - let server = launch(config, core.clone(), metrics.clone(), storage.clone()).await.unwrap(); + let server = launch(config, core.clone(), metrics.clone()).await.unwrap(); Ok(TestServer { metrics, storage, @@ -143,7 +143,7 @@ impl TestServer { core.shutdown().await?; let metrics = MetricsRegistry::new(); let core = LavinaCore::new(metrics.clone(), storage.clone()).await?; - let server = launch(config, core.clone(), metrics.clone(), storage.clone()).await.unwrap(); + let server = launch(config, core.clone(), metrics.clone()).await.unwrap(); Ok(TestServer { metrics, storage, @@ -167,7 +167,7 @@ async fn scenario_basic() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -196,7 +196,7 @@ async fn scenario_join_and_reboot() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -266,7 +266,7 @@ async fn scenario_force_join_msg() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream1 = TcpStream::connect(server.server.addr).await?; let mut s1 = TestScope::new(&mut stream1); @@ -332,9 +332,9 @@ async fn scenario_two_users() -> Result<()> { // test scenario server.storage.create_user("tester1").await?; - Authenticator::new(&server.storage).set_password("tester1", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester1", "password").await?; server.storage.create_user("tester2").await?; - Authenticator::new(&server.storage).set_password("tester2", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester2", "password").await?; let mut stream1 = TcpStream::connect(server.server.addr).await?; let mut s1 = TestScope::new(&mut stream1); @@ -408,7 +408,7 @@ async fn scenario_cap_full_negotiation() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -448,7 +448,7 @@ async fn scenario_cap_full_negotiation_nick_last() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -487,7 +487,7 @@ async fn scenario_cap_short_negotiation() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -525,7 +525,7 @@ async fn scenario_cap_sasl_fail() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -569,7 +569,7 @@ async fn terminate_socket_scenario() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -594,7 +594,7 @@ async fn server_time_capability() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); diff --git a/crates/projection-xmpp/src/lib.rs b/crates/projection-xmpp/src/lib.rs index 9ae294f..a391b72 100644 --- a/crates/projection-xmpp/src/lib.rs +++ b/crates/projection-xmpp/src/lib.rs @@ -22,10 +22,9 @@ use tokio::sync::mpsc::channel; use tokio_rustls::rustls::{Certificate, PrivateKey}; use tokio_rustls::TlsAcceptor; -use lavina_core::auth::{Authenticator, Verdict}; -use lavina_core::player::{ConnectionMessage, PlayerConnection, PlayerId, PlayerRegistry, StopReason}; +use lavina_core::auth::Verdict; +use lavina_core::player::{ConnectionMessage, PlayerConnection, PlayerId, StopReason}; use lavina_core::prelude::*; -use lavina_core::repo::Storage; use lavina_core::room::RoomRegistry; use lavina_core::terminator::Terminator; use lavina_core::LavinaCore; @@ -81,12 +80,7 @@ impl RunningServer { } } -pub async fn launch( - config: ServerConfig, - core: LavinaCore, - metrics: MetricsRegistry, - storage: Storage, -) -> Result { +pub async fn launch(config: ServerConfig, core: LavinaCore, metrics: MetricsRegistry) -> Result { log::info!("Starting XMPP projection"); let certs = certs(&mut SyncBufReader::new(File::open(config.cert)?))?; @@ -126,13 +120,12 @@ pub async fn launch( continue; } let core = core.clone(); - let storage = storage.clone(); let hostname = config.hostname.clone(); let terminator = Terminator::spawn(|termination| { let stopped_tx = stopped_tx.clone(); let loaded_config = loaded_config.clone(); async move { - match handle_socket(loaded_config, stream, &socket_addr, core, storage, hostname, termination).await { + match handle_socket(loaded_config, stream, &socket_addr, core, hostname, termination).await { Ok(_) => log::info!("Connection terminated"), Err(err) => log::warn!("Connection failed: {err}"), } @@ -170,8 +163,7 @@ async fn handle_socket( cert_config: Arc, mut stream: TcpStream, socket_addr: &SocketAddr, - mut core: LavinaCore, - mut storage: Storage, + core: LavinaCore, hostname: Str, termination: Deferred<()>, // TODO use it to stop the connection gracefully ) -> Result<()> { @@ -205,7 +197,7 @@ async fn handle_socket( log::info!("Socket handling was terminated"); return Ok(()) }, - authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf, &mut storage, &hostname) => { + authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf, &core, &hostname) => { match authenticated { Ok(authenticated) => { let mut connection = core.players.connect_to_player(&authenticated.player_id).await; @@ -272,7 +264,7 @@ async fn socket_auth( xml_reader: &mut NsReader<(impl AsyncBufRead + Unpin)>, xml_writer: &mut Writer<(impl AsyncWrite + Unpin)>, reader_buf: &mut Vec, - storage: &mut Storage, + core: &LavinaCore, hostname: &Str, ) -> Result { // TODO validate the server hostname received in the stream start @@ -301,7 +293,7 @@ async fn socket_auth( match AuthBody::from_str(&auth.body) { Ok(logopass) => { let name = &logopass.login; - let verdict = Authenticator::new(storage).authenticate(name, &logopass.password).await?; + let verdict = core.authenticator.authenticate(name, &logopass.password).await?; match verdict { Verdict::Authenticated => { proto_xmpp::sasl::Success.write_xml(xml_writer).await?; diff --git a/crates/projection-xmpp/tests/lib.rs b/crates/projection-xmpp/tests/lib.rs index c1ea13a..29e0a69 100644 --- a/crates/projection-xmpp/tests/lib.rs +++ b/crates/projection-xmpp/tests/lib.rs @@ -162,7 +162,7 @@ impl TestServer { }) .await?; let core = LavinaCore::new(metrics.clone(), storage.clone()).await?; - let server = launch(config, core.clone(), metrics.clone(), storage.clone()).await.unwrap(); + let server = launch(config, core.clone(), metrics.clone()).await.unwrap(); Ok(TestServer { metrics, storage, @@ -186,7 +186,7 @@ async fn scenario_basic() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -254,7 +254,7 @@ async fn scenario_wrong_password() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -309,7 +309,7 @@ async fn scenario_basic_without_headers() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -355,7 +355,7 @@ async fn terminate_socket() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); @@ -395,7 +395,7 @@ async fn test_message_archive_request() -> Result<()> { // test scenario server.storage.create_user("tester").await?; - Authenticator::new(&server.storage).set_password("tester", "password").await?; + Authenticator::new(server.storage.clone()).set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.server.addr).await?; let mut s = TestScope::new(&mut stream); diff --git a/src/http.rs b/src/http.rs index f13ee7a..a879a85 100644 --- a/src/http.rs +++ b/src/http.rs @@ -12,7 +12,7 @@ use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder}; use serde::{Deserialize, Serialize}; use tokio::net::TcpListener; -use lavina_core::auth::{Authenticator, UpdatePasswordResult}; +use lavina_core::auth::UpdatePasswordResult; use lavina_core::player::{PlayerId, PlayerRegistry, SendMessageResult}; use lavina_core::prelude::*; use lavina_core::repo::Storage; @@ -87,7 +87,7 @@ async fn route( (&Method::GET, "/rooms") => endpoint_rooms(core.rooms).await, (&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(), (&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, core.players).await.or5xx(), - (&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, storage).await.or5xx(), + (&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, core).await.or5xx(), (&Method::POST, rooms::paths::SEND_MESSAGE) => endpoint_send_room_message(request, core).await.or5xx(), (&Method::POST, rooms::paths::SET_TOPIC) => endpoint_set_room_topic(request, core).await.or5xx(), _ => endpoint_not_found(), @@ -147,13 +147,13 @@ async fn endpoint_stop_player( #[tracing::instrument(skip_all)] async fn endpoint_set_password( request: Request, - storage: Storage, + core: LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(res) = serde_json::from_slice::(&str[..]) else { return Ok(malformed_request()); }; - let verdict = Authenticator::new(&storage).set_password(&res.player_name, &res.password).await?; + let verdict = core.authenticator.set_password(&res.player_name, &res.password).await?; match verdict { UpdatePasswordResult::PasswordUpdated => {} UpdatePasswordResult::UserNotFound => { @@ -165,7 +165,7 @@ async fn endpoint_set_password( async fn endpoint_send_room_message( request: Request, - mut core: LavinaCore, + core: LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(req) = serde_json::from_slice::(&str[..]) else { diff --git a/src/main.rs b/src/main.rs index 9b73b1f..173b683 100644 --- a/src/main.rs +++ b/src/main.rs @@ -69,8 +69,8 @@ async fn main() -> Result<()> { let storage = Storage::open(storage_config).await?; let core = LavinaCore::new(metrics.clone(), storage.clone()).await?; let telemetry_terminator = http::launch(telemetry_config, metrics.clone(), core.clone(), storage.clone()).await?; - let irc = projection_irc::launch(irc_config, core.clone(), metrics.clone(), storage.clone()).await?; - let xmpp = projection_xmpp::launch(xmpp_config, core.clone(), metrics.clone(), storage.clone()).await?; + let irc = projection_irc::launch(irc_config, core.clone(), metrics.clone()).await?; + let xmpp = projection_xmpp::launch(xmpp_config, core.clone(), metrics.clone()).await?; tracing::info!("Started"); sleep.await; From bb0fe3bf0bf09ba6123d839a1f700ec3dfa58bf4 Mon Sep 17 00:00:00 2001 From: Nikita Vilunov Date: Sat, 4 May 2024 01:07:23 +0200 Subject: [PATCH 4/5] use borrows in http endpoint handlers --- crates/lavina-core/src/repo/mod.rs | 2 +- src/http.rs | 30 ++++++++++++++---------------- 2 files changed, 15 insertions(+), 17 deletions(-) diff --git a/crates/lavina-core/src/repo/mod.rs b/crates/lavina-core/src/repo/mod.rs index 4c29651..0089c22 100644 --- a/crates/lavina-core/src/repo/mod.rs +++ b/crates/lavina-core/src/repo/mod.rs @@ -143,7 +143,7 @@ impl Storage { } #[tracing::instrument(skip(self), name = "Storage::create_user")] - pub async fn create_user(&mut self, name: &str) -> Result<()> { + pub async fn create_user(&self, name: &str) -> Result<()> { let query = sqlx::query( "insert into users(name) values (?);", diff --git a/src/http.rs b/src/http.rs index a879a85..dfe23a2 100644 --- a/src/http.rs +++ b/src/http.rs @@ -61,10 +61,8 @@ async fn main_loop( let core = core.clone(); let storage = storage.clone(); tokio::task::spawn(async move { - let registry = metrics.clone(); - let core = core.clone(); - let storage = storage.clone(); - let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), core.clone(), storage.clone(), r))); + let svc_fn = service_fn(|r| route(&metrics, &core, &storage, r)); + let server = http1::Builder::new().serve_connection(stream, svc_fn); if let Err(err) = server.await { tracing::error!("Error serving connection: {:?}", err); } @@ -77,16 +75,16 @@ async fn main_loop( } async fn route( - registry: MetricsRegistry, - core: LavinaCore, - storage: Storage, + registry: &MetricsRegistry, + core: &LavinaCore, + storage: &Storage, request: Request, ) -> HttpResult>> { let res = match (request.method(), request.uri().path()) { (&Method::GET, "/metrics") => endpoint_metrics(registry), - (&Method::GET, "/rooms") => endpoint_rooms(core.rooms).await, + (&Method::GET, "/rooms") => endpoint_rooms(&core.rooms).await, (&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(), - (&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, core.players).await.or5xx(), + (&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, &core.players).await.or5xx(), (&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, core).await.or5xx(), (&Method::POST, rooms::paths::SEND_MESSAGE) => endpoint_send_room_message(request, core).await.or5xx(), (&Method::POST, rooms::paths::SET_TOPIC) => endpoint_set_room_topic(request, core).await.or5xx(), @@ -95,7 +93,7 @@ async fn route( Ok(res) } -fn endpoint_metrics(registry: MetricsRegistry) -> Response> { +fn endpoint_metrics(registry: &MetricsRegistry) -> Response> { let mf = registry.gather(); let mut buffer = vec![]; TextEncoder.encode(&mf, &mut buffer).expect("write to vec cannot fail"); @@ -103,7 +101,7 @@ fn endpoint_metrics(registry: MetricsRegistry) -> Response> { } #[tracing::instrument(skip_all)] -async fn endpoint_rooms(rooms: RoomRegistry) -> Response> { +async fn endpoint_rooms(rooms: &RoomRegistry) -> Response> { // TODO introduce management API types independent from core-domain types // TODO remove `Serialize` implementations from all core-domain types let room_list = rooms.get_all_rooms().await.to_body(); @@ -113,7 +111,7 @@ async fn endpoint_rooms(rooms: RoomRegistry) -> Response> { #[tracing::instrument(skip_all)] async fn endpoint_create_player( request: Request, - mut storage: Storage, + storage: &Storage, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(res) = serde_json::from_slice::(&str[..]) else { @@ -129,7 +127,7 @@ async fn endpoint_create_player( #[tracing::instrument(skip_all)] async fn endpoint_stop_player( request: Request, - players: PlayerRegistry, + players: &PlayerRegistry, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(res) = serde_json::from_slice::(&str[..]) else { @@ -147,7 +145,7 @@ async fn endpoint_stop_player( #[tracing::instrument(skip_all)] async fn endpoint_set_password( request: Request, - core: LavinaCore, + core: &LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(res) = serde_json::from_slice::(&str[..]) else { @@ -165,7 +163,7 @@ async fn endpoint_set_password( async fn endpoint_send_room_message( request: Request, - core: LavinaCore, + core: &LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(req) = serde_json::from_slice::(&str[..]) else { @@ -187,7 +185,7 @@ async fn endpoint_send_room_message( async fn endpoint_set_room_topic( request: Request, - core: LavinaCore, + core: &LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(req) = serde_json::from_slice::(&str[..]) else { From 7f2c6a10132a38cfc005e12f67a247a070baf17d Mon Sep 17 00:00:00 2001 From: Nikita Vilunov Date: Sat, 4 May 2024 01:59:43 +0200 Subject: [PATCH 5/5] continue propagated traces in http request handlers --- src/http.rs | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/src/http.rs b/src/http.rs index dfe23a2..1d24c14 100644 --- a/src/http.rs +++ b/src/http.rs @@ -74,12 +74,14 @@ async fn main_loop( Ok(()) } +#[tracing::instrument(skip_all)] async fn route( registry: &MetricsRegistry, core: &LavinaCore, storage: &Storage, request: Request, ) -> HttpResult>> { + propagade_span_from_headers(&request); let res = match (request.method(), request.uri().path()) { (&Method::GET, "/metrics") => endpoint_metrics(registry), (&Method::GET, "/rooms") => endpoint_rooms(&core.rooms).await, @@ -161,6 +163,7 @@ async fn endpoint_set_password( Ok(empty_204_request()) } +#[tracing::instrument(skip_all)] async fn endpoint_send_room_message( request: Request, core: &LavinaCore, @@ -183,6 +186,7 @@ async fn endpoint_send_room_message( } } +#[tracing::instrument(skip_all)] async fn endpoint_set_room_topic( request: Request, core: &LavinaCore, @@ -282,3 +286,24 @@ where Full::new(Bytes::from(buffer)) } } + +fn propagade_span_from_headers(req: &Request) { + use opentelemetry::propagation::Extractor; + use tracing::Span; + use tracing_opentelemetry::OpenTelemetrySpanExt; + struct HttpReqExtractor<'a, T> { + req: &'a Request, + } + impl<'a, T> Extractor for HttpReqExtractor<'a, T> { + fn get(&self, key: &str) -> Option<&str> { + self.req.headers().get(key).and_then(|v| v.to_str().ok()) + } + + fn keys(&self) -> Vec<&str> { + self.req.headers().keys().map(|k| k.as_str()).collect() + } + } + + let ctx = opentelemetry::global::get_text_map_propagator(|pp| pp.extract(&HttpReqExtractor { req })); + Span::current().set_parent(ctx); +}