Compare commits

..

4 Commits

Author SHA1 Message Date
Mikhail 317d0d45b0 Remove extra space 2024-04-01 18:33:13 +02:00
Mikhail 0499a1b209 Rename 2024-04-01 18:33:01 +02:00
Mikhail bf5460f495 Add pre-commit hook 2024-04-01 18:32:34 +02:00
Mikhail 717855fa5d Migrate 2024-04-01 18:32:15 +02:00
12 changed files with 94 additions and 469 deletions

68
Cargo.lock generated
View File

@ -45,21 +45,6 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0942ffc6dcaadf03badf6e6a2d0228460359d5e34b57ccdc720b7382dfbd5ec5" checksum = "0942ffc6dcaadf03badf6e6a2d0228460359d5e34b57ccdc720b7382dfbd5ec5"
[[package]]
name = "android-tzdata"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0"
[[package]]
name = "android_system_properties"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311"
dependencies = [
"libc",
]
[[package]] [[package]]
name = "anstream" name = "anstream"
version = "0.6.13" version = "0.6.13"
@ -231,20 +216,6 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]]
name = "chrono"
version = "0.4.37"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8a0d04d43504c61aa6c7531f1871dd0d418d91130162063b789da00fd7057a5e"
dependencies = [
"android-tzdata",
"iana-time-zone",
"js-sys",
"num-traits",
"wasm-bindgen",
"windows-targets 0.52.4",
]
[[package]] [[package]]
name = "clap" name = "clap"
version = "4.5.3" version = "4.5.3"
@ -303,12 +274,6 @@ version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e" checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e"
[[package]]
name = "core-foundation-sys"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06ea2b9bc92be3c2baa9334a323ebca2d6f074ff852cd1d7b11064035cd3868f"
[[package]] [[package]]
name = "cpufeatures" name = "cpufeatures"
version = "0.2.12" version = "0.2.12"
@ -764,29 +729,6 @@ dependencies = [
"tracing", "tracing",
] ]
[[package]]
name = "iana-time-zone"
version = "0.1.60"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e7ffbb5a1b541ea2561f8c41c087286cc091e21e556a4f09a8f6cbf17b69b141"
dependencies = [
"android_system_properties",
"core-foundation-sys",
"iana-time-zone-haiku",
"js-sys",
"wasm-bindgen",
"windows-core",
]
[[package]]
name = "iana-time-zone-haiku"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f"
dependencies = [
"cc",
]
[[package]] [[package]]
name = "idna" name = "idna"
version = "0.5.0" version = "0.5.0"
@ -876,7 +818,6 @@ name = "lavina-core"
version = "0.0.2-dev" version = "0.0.2-dev"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"chrono",
"prometheus", "prometheus",
"serde", "serde",
"sqlx", "sqlx",
@ -2442,15 +2383,6 @@ version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
[[package]]
name = "windows-core"
version = "0.52.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "33ab640c8d7e35bf8ba19b884ba838ceb4fba93a4e8c65a9059d08afcfc683d9"
dependencies = [
"windows-targets 0.52.4",
]
[[package]] [[package]]
name = "windows-sys" name = "windows-sys"
version = "0.48.0" version = "0.48.0"

View File

@ -10,4 +10,3 @@ serde.workspace = true
tokio.workspace = true tokio.workspace = true
tracing.workspace = true tracing.workspace = true
prometheus.workspace = true prometheus.workspace = true
chrono = "0.4.37"

View File

@ -87,15 +87,14 @@ impl Storage {
return Err(anyhow!("No such user")); return Err(anyhow!("No such user"));
}; };
sqlx::query( sqlx::query(
"insert into messages(room_id, id, content, author_id, created_at) "insert into messages(room_id, id, content, author_id)
values (?, ?, ?, ?, ?); values (?, ?, ?, ?);
update rooms set message_count = message_count + 1 where id = ?;", update rooms set message_count = message_count + 1 where id = ?;",
) )
.bind(room_id) .bind(room_id)
.bind(id) .bind(id)
.bind(content) .bind(content)
.bind(author_id) .bind(author_id)
.bind(chrono::Utc::now().to_string())
.bind(room_id) .bind(room_id)
.execute(&mut *executor) .execute(&mut *executor)
.await?; .await?;

View File

@ -48,31 +48,6 @@ impl<'a> TestScope<'a> {
Ok(()) Ok(())
} }
async fn expect_that(&mut self, validate: impl FnOnce(&str) -> bool) -> Result<()> {
let len = tokio::time::timeout(self.timeout, read_irc_message(&mut self.reader, &mut self.buffer)).await??;
let msg = std::str::from_utf8(&self.buffer[..len - 2])?;
if !validate(msg) {
return Err(anyhow!("unexpected message: {:?}", msg));
}
self.buffer.clear();
Ok(())
}
async fn expect_server_introduction(&mut self, nick: &str) -> Result<()> {
self.expect(&format!(":testserver 001 {nick} :Welcome to testserver Server")).await?;
self.expect(&format!(":testserver 002 {nick} :Welcome to testserver Server")).await?;
self.expect(&format!(":testserver 003 {nick} :Welcome to testserver Server")).await?;
self.expect(&format!(
":testserver 004 {nick} testserver {APP_VERSION} r CFILPQbcefgijklmnopqrstvz"
))
.await?;
self.expect(&format!(
":testserver 005 {nick} CHANTYPES=# :are supported by this server"
))
.await?;
Ok(())
}
async fn expect_eof(&mut self) -> Result<()> { async fn expect_eof(&mut self) -> Result<()> {
let mut buf = [0; 1]; let mut buf = [0; 1];
let len = tokio::time::timeout(self.timeout, self.reader.read(&mut buf)).await??; let len = tokio::time::timeout(self.timeout, self.reader.read(&mut buf)).await??;
@ -138,7 +113,18 @@ async fn scenario_basic() -> Result<()> {
s.send("PASS password").await?; s.send("PASS password").await?;
s.send("NICK tester").await?; s.send("NICK tester").await?;
s.send("USER UserName 0 * :Real Name").await?; s.send("USER UserName 0 * :Real Name").await?;
s.expect_server_introduction("tester").await?; s.expect(":testserver 001 tester :Welcome to testserver Server").await?;
s.expect(":testserver 002 tester :Welcome to testserver Server").await?;
s.expect(":testserver 003 tester :Welcome to testserver Server").await?;
s.expect(
format!(
":testserver 004 tester testserver {} r CFILPQbcefgijklmnopqrstvz",
&APP_VERSION
)
.as_str(),
)
.await?;
s.expect(":testserver 005 tester CHANTYPES=# :are supported by this server").await?;
s.expect_nothing().await?; s.expect_nothing().await?;
s.send("QUIT :Leaving").await?; s.send("QUIT :Leaving").await?;
s.expect(":testserver ERROR :Leaving the server").await?; s.expect(":testserver ERROR :Leaving the server").await?;
@ -152,132 +138,6 @@ async fn scenario_basic() -> Result<()> {
Ok(()) Ok(())
} }
#[tokio::test]
async fn scenario_force_join_msg() -> Result<()> {
let mut server = TestServer::start().await?;
// test scenario
server.storage.create_user("tester").await?;
server.storage.set_password("tester", "password").await?;
let mut stream1 = TcpStream::connect(server.server.addr).await?;
let mut s1 = TestScope::new(&mut stream1);
let mut stream2 = TcpStream::connect(server.server.addr).await?;
let mut s2 = TestScope::new(&mut stream2);
s1.send("PASS password").await?;
s1.send("NICK tester").await?;
s1.send("USER UserName 0 * :Real Name").await?;
s1.expect_server_introduction("tester").await?;
s1.expect_nothing().await?;
s2.send("PASS password").await?;
s2.send("NICK tester").await?;
s2.send("USER UserName 0 * :Real Name").await?;
s2.expect_server_introduction("tester").await?;
s2.expect_nothing().await?;
// We join the channel from the first connection
s1.send("JOIN #test").await?;
s1.expect(":tester JOIN #test").await?;
s1.expect(":testserver 332 tester #test :New room").await?;
s1.expect(":testserver 353 tester = #test :tester").await?;
s1.expect(":testserver 366 tester #test :End of /NAMES list").await?;
// And the second connection should receive the JOIN message (forced JOIN)
s2.expect(":tester JOIN #test").await?;
s2.expect(":testserver 332 tester #test :New room").await?;
s2.expect(":testserver 353 tester = #test :tester").await?;
s2.expect(":testserver 366 tester #test :End of /NAMES list").await?;
// We send a message to the channel from the second connection
s2.send("PRIVMSG #test :Hello").await?;
// We should not receive an acknowledgement from the server
s2.expect_nothing().await?;
// But we should receive this message from the first connection
s1.expect(":tester PRIVMSG #test :Hello").await?;
s1.send("QUIT :Leaving").await?;
s1.expect(":testserver ERROR :Leaving the server").await?;
s1.expect_eof().await?;
// Closing a connection does not kick you from the channel on a different connection
s2.expect_nothing().await?;
s2.send("QUIT :Leaving").await?;
s2.expect(":testserver ERROR :Leaving the server").await?;
s2.expect_eof().await?;
stream1.shutdown().await?;
stream2.shutdown().await?;
// wrap up
server.server.terminate().await?;
Ok(())
}
#[tokio::test]
async fn scenario_two_users() -> Result<()> {
let mut server = TestServer::start().await?;
// test scenario
server.storage.create_user("tester1").await?;
server.storage.set_password("tester1", "password").await?;
server.storage.create_user("tester2").await?;
server.storage.set_password("tester2", "password").await?;
let mut stream1 = TcpStream::connect(server.server.addr).await?;
let mut s1 = TestScope::new(&mut stream1);
let mut stream2 = TcpStream::connect(server.server.addr).await?;
let mut s2 = TestScope::new(&mut stream2);
s1.send("PASS password").await?;
s1.send("NICK tester1").await?;
s1.send("USER UserName 0 * :Real Name").await?;
s1.expect_server_introduction("tester1").await?;
s1.expect_nothing().await?;
s2.send("PASS password").await?;
s2.send("NICK tester2").await?;
s2.send("USER UserName 0 * :Real Name").await?;
s2.expect_server_introduction("tester2").await?;
s2.expect_nothing().await?;
// Join the channel from the first user
s1.send("JOIN #test").await?;
s1.expect(":tester1 JOIN #test").await?;
s1.expect(":testserver 332 tester1 #test :New room").await?;
s1.expect(":testserver 353 tester1 = #test :tester1").await?;
s1.expect(":testserver 366 tester1 #test :End of /NAMES list").await?;
// Then join the channel from the second user
s2.send("JOIN #test").await?;
s2.expect(":tester2 JOIN #test").await?;
s2.expect(":testserver 332 tester2 #test :New room").await?;
s2.expect_that(|msg| {
msg == ":testserver 353 tester2 = #test :tester1 tester2"
|| msg == ":testserver 353 tester2 = #test :tester2 tester1"
})
.await?;
s2.expect(":testserver 366 tester2 #test :End of /NAMES list").await?;
// The first user should receive the JOIN message from the second user
s1.expect(":tester2 JOIN #test").await?;
s1.expect_nothing().await?;
s2.expect_nothing().await?;
// Send a message from the second user
s2.send("PRIVMSG #test :Hello").await?;
// The first user should receive the message
s1.expect(":tester2 PRIVMSG #test :Hello").await?;
// Leave the channel from the first user
s1.send("PART #test").await?;
s1.expect(":tester1 PART #test").await?;
// The second user should receive the PART message
s2.expect(":tester1 PART #test").await?;
Ok(())
}
/* /*
IRC SASL doc: https://ircv3.net/specs/extensions/sasl-3.1.html IRC SASL doc: https://ircv3.net/specs/extensions/sasl-3.1.html
AUTHENTICATE doc: https://modern.ircdocs.horse/#authenticate-message AUTHENTICATE doc: https://modern.ircdocs.horse/#authenticate-message
@ -308,7 +168,18 @@ async fn scenario_cap_full_negotiation() -> Result<()> {
s.send("CAP END").await?; s.send("CAP END").await?;
s.expect_server_introduction("tester").await?; s.expect(":testserver 001 tester :Welcome to testserver Server").await?;
s.expect(":testserver 002 tester :Welcome to testserver Server").await?;
s.expect(":testserver 003 tester :Welcome to testserver Server").await?;
s.expect(
format!(
":testserver 004 tester testserver {} r CFILPQbcefgijklmnopqrstvz",
&APP_VERSION
)
.as_str(),
)
.await?;
s.expect(":testserver 005 tester CHANTYPES=# :are supported by this server").await?;
s.expect_nothing().await?; s.expect_nothing().await?;
s.send("QUIT :Leaving").await?; s.send("QUIT :Leaving").await?;
s.expect(":testserver ERROR :Leaving the server").await?; s.expect(":testserver ERROR :Leaving the server").await?;
@ -346,7 +217,18 @@ async fn scenario_cap_short_negotiation() -> Result<()> {
s.send("CAP END").await?; s.send("CAP END").await?;
s.expect_server_introduction("tester").await?; s.expect(":testserver 001 tester :Welcome to testserver Server").await?;
s.expect(":testserver 002 tester :Welcome to testserver Server").await?;
s.expect(":testserver 003 tester :Welcome to testserver Server").await?;
s.expect(
format!(
":testserver 004 tester testserver {} r CFILPQbcefgijklmnopqrstvz",
&APP_VERSION
)
.as_str(),
)
.await?;
s.expect(":testserver 005 tester CHANTYPES=# :are supported by this server").await?;
s.expect_nothing().await?; s.expect_nothing().await?;
s.send("QUIT :Leaving").await?; s.send("QUIT :Leaving").await?;
s.expect(":testserver ERROR :Leaving the server").await?; s.expect(":testserver ERROR :Leaving the server").await?;
@ -390,7 +272,18 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
s.send("CAP END").await?; s.send("CAP END").await?;
s.expect_server_introduction("tester").await?; s.expect(":testserver 001 tester :Welcome to testserver Server").await?;
s.expect(":testserver 002 tester :Welcome to testserver Server").await?;
s.expect(":testserver 003 tester :Welcome to testserver Server").await?;
s.expect(
format!(
":testserver 004 tester testserver {} r CFILPQbcefgijklmnopqrstvz",
&APP_VERSION
)
.as_str(),
)
.await?;
s.expect(":testserver 005 tester CHANTYPES=# :are supported by this server").await?;
s.expect_nothing().await?; s.expect_nothing().await?;
s.send("QUIT :Leaving").await?; s.send("QUIT :Leaving").await?;
s.expect(":testserver ERROR :Leaving the server").await?; s.expect(":testserver ERROR :Leaving the server").await?;

View File

@ -4,7 +4,7 @@ use quick_xml::events::Event;
use lavina_core::room::RoomRegistry; use lavina_core::room::RoomRegistry;
use proto_xmpp::bind::{BindResponse, Jid, Name, Resource, Server}; use proto_xmpp::bind::{BindResponse, Jid, Name, Resource, Server};
use proto_xmpp::client::{Iq, IqError, IqErrorType, IqType}; use proto_xmpp::client::{Iq, IqType};
use proto_xmpp::disco::{Feature, Identity, InfoQuery, Item, ItemQuery}; use proto_xmpp::disco::{Feature, Identity, InfoQuery, Item, ItemQuery};
use proto_xmpp::roster::RosterQuery; use proto_xmpp::roster::RosterQuery;
use proto_xmpp::session::Session; use proto_xmpp::session::Session;
@ -24,9 +24,9 @@ impl<'a> XmppConnection<'a> {
to: None, to: None,
r#type: IqType::Result, r#type: IqType::Result,
body: BindResponse(Jid { body: BindResponse(Jid {
name: Some(self.user.xmpp_name.clone()), name: Some(Name("darova".into())),
server: Server("localhost".into()), server: Server("localhost".into()),
resource: Some(self.user.xmpp_resource.clone()), resource: Some(Resource("kek".into())),
}), }),
}; };
req.serialize(output); req.serialize(output);
@ -79,9 +79,7 @@ impl<'a> XmppConnection<'a> {
id: iq.id, id: iq.id,
to: None, to: None,
r#type: IqType::Error, r#type: IqType::Error,
body: IqError { body: (),
r#type: IqErrorType::Cancel,
},
}; };
req.serialize(output); req.serialize(output);
} }

View File

@ -52,17 +52,9 @@ struct LoadedConfig {
} }
struct Authenticated { struct Authenticated {
/// Identifier of the authenticated player.
///
/// Used when communicating with lavina-core on behalf of the player.
player_id: PlayerId, player_id: PlayerId,
/// The user's XMPP name.
///
/// Used in `to` and `from` fields of XMPP messages.
xmpp_name: Name, xmpp_name: Name,
/// The resource given to this user by the server.
xmpp_resource: Resource, xmpp_resource: Resource,
/// The resource used by this user when joining MUCs.
xmpp_muc_name: Resource, xmpp_muc_name: Resource,
} }
@ -193,7 +185,7 @@ async fn handle_socket(
let (a, b) = tokio::io::split(new_stream); let (a, b) = tokio::io::split(new_stream);
let mut xml_reader = NsReader::from_reader(BufReader::new(a)); let mut xml_reader = NsReader::from_reader(BufReader::new(a));
let mut xml_writer = Writer::new(BufWriter::new(b)); let mut xml_writer = Writer::new(b);
pin!(termination); pin!(termination);
select! { select! {
@ -224,7 +216,7 @@ async fn handle_socket(
} }
let a = xml_reader.into_inner().into_inner(); let a = xml_reader.into_inner().into_inner();
let b = xml_writer.into_inner().into_inner(); let b = xml_writer.into_inner();
a.unsplit(b).shutdown().await?; a.unsplit(b).shutdown().await?;
Ok(()) Ok(())
} }
@ -237,6 +229,7 @@ async fn socket_force_tls(
use proto_xmpp::tls::*; use proto_xmpp::tls::*;
let xml_reader = &mut NsReader::from_reader(reader); let xml_reader = &mut NsReader::from_reader(reader);
let xml_writer = &mut Writer::new(writer); let xml_writer = &mut Writer::new(writer);
read_xml_header(xml_reader, reader_buf).await?;
let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?; let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?;
let event = Event::Decl(BytesDecl::new("1.0", None, None)); let event = Event::Decl(BytesDecl::new("1.0", None, None));
@ -268,6 +261,7 @@ async fn socket_auth(
reader_buf: &mut Vec<u8>, reader_buf: &mut Vec<u8>,
storage: &mut Storage, storage: &mut Storage,
) -> Result<Authenticated> { ) -> Result<Authenticated> {
read_xml_header(xml_reader, reader_buf).await?;
let _ = ClientStreamStart::parse(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?;
@ -290,7 +284,6 @@ async fn socket_auth(
let auth: proto_xmpp::sasl::Auth = proto_xmpp::sasl::Auth::parse(xml_reader, reader_buf).await?; let auth: proto_xmpp::sasl::Auth = proto_xmpp::sasl::Auth::parse(xml_reader, reader_buf).await?;
proto_xmpp::sasl::Success.write_xml(xml_writer).await?; proto_xmpp::sasl::Success.write_xml(xml_writer).await?;
xml_writer.get_mut().flush().await?;
match AuthBody::from_str(&auth.body) { match AuthBody::from_str(&auth.body) {
Ok(logopass) => { Ok(logopass) => {
@ -315,13 +308,11 @@ async fn socket_auth(
return Err(fail("passwords do not match")); return Err(fail("passwords do not match"));
} }
let name: Str = name.as_str().into();
Ok(Authenticated { Ok(Authenticated {
player_id: PlayerId::from(name.clone())?, player_id: PlayerId::from(name.as_str())?,
xmpp_name: Name(name.clone()), xmpp_name: Name(name.to_string().into()),
xmpp_resource: Resource(name.clone()), xmpp_resource: Resource(name.to_string().into()),
xmpp_muc_name: Resource(name.clone()), xmpp_muc_name: Resource(name.to_string().into()),
}) })
} }
Err(e) => return Err(e), Err(e) => return Err(e),
@ -336,6 +327,7 @@ async fn socket_final(
user_handle: &mut PlayerConnection, user_handle: &mut PlayerConnection,
rooms: &RoomRegistry, rooms: &RoomRegistry,
) -> Result<()> { ) -> Result<()> {
read_xml_header(xml_reader, reader_buf).await?;
let _ = ClientStreamStart::parse(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?;
@ -427,7 +419,7 @@ struct XmppConnection<'a> {
impl<'a> XmppConnection<'a> { impl<'a> XmppConnection<'a> {
async fn handle_packet(&mut self, output: &mut Vec<Event<'static>>, packet: ClientPacket) -> Result<bool> { async fn handle_packet(&mut self, output: &mut Vec<Event<'static>>, packet: ClientPacket) -> Result<bool> {
let res = match packet { let res = match packet {
ClientPacket::Iq(iq) => { proto::ClientPacket::Iq(iq) => {
self.handle_iq(output, iq).await; self.handle_iq(output, iq).await;
false false
} }
@ -435,11 +427,11 @@ impl<'a> XmppConnection<'a> {
self.handle_message(output, m).await?; self.handle_message(output, m).await?;
false false
} }
ClientPacket::Presence(p) => { proto::ClientPacket::Presence(p) => {
self.handle_presence(output, p).await?; self.handle_presence(output, p).await?;
false false
} }
ClientPacket::StreamEnd => { proto::ClientPacket::StreamEnd => {
ServerStreamEnd.serialize(output); ServerStreamEnd.serialize(output);
true true
} }
@ -447,3 +439,25 @@ impl<'a> XmppConnection<'a> {
Ok(res) Ok(res)
} }
} }
async fn read_xml_header(
xml_reader: &mut NsReader<(impl AsyncBufRead + Unpin)>,
reader_buf: &mut Vec<u8>,
) -> Result<()> {
if let Event::Decl(bytes) = xml_reader.read_event_into_async(reader_buf).await? {
// this is <?xml ...> header
if let Some(encoding) = bytes.encoding() {
let encoding = encoding?;
if &*encoding == b"UTF-8" {
Ok(())
} else {
Err(anyhow!("Unsupported encoding: {encoding:?}"))
}
} else {
// Err(fail("No XML encoding provided"))
Ok(())
}
} else {
Err(anyhow!("Expected XML header"))
}
}

View File

@ -108,7 +108,6 @@ impl<'a> TestScopeTls<'a> {
} }
struct IgnoreCertVerification; struct IgnoreCertVerification;
impl ServerCertVerifier for IgnoreCertVerification { impl ServerCertVerifier for IgnoreCertVerification {
fn verify_server_cert( fn verify_server_cert(
&self, &self,
@ -123,79 +122,6 @@ impl ServerCertVerifier for IgnoreCertVerification {
} }
} }
/// Some clients prefer to close their tags, i.e. Gajim.
#[tokio::test]
async fn scenario_basic_closed_tag() -> Result<()> {
tracing_subscriber::fmt::try_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="jabber:client" version="1.0" xmlns:stream="http://etherx.jabber.org/streams" to="sauer" xml:lang="en">
"#).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#"
<!-- Outgoing Tue 02 Apr 2024 06:39:24 PM CEST (Account Wizard) -->
</stream:stream>
<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(())
}
#[tokio::test] #[tokio::test]
async fn scenario_basic() -> Result<()> { async fn scenario_basic() -> Result<()> {
tracing_subscriber::fmt::try_init(); tracing_subscriber::fmt::try_init();
@ -261,69 +187,6 @@ async fn scenario_basic() -> Result<()> {
Ok(()) Ok(())
} }
#[tokio::test]
async fn scenario_basic_without_headers() -> Result<()> {
tracing_subscriber::fmt::try_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#"<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#"<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(())
}
#[tokio::test] #[tokio::test]
async fn terminate_socket() -> Result<()> { async fn terminate_socket() -> Result<()> {
tracing_subscriber::fmt::try_init(); tracing_subscriber::fmt::try_init();

View File

@ -49,7 +49,7 @@ pub enum ClientMessage {
}, },
Part { Part {
chan: Chan, chan: Chan,
message: Option<Str>, message: Str,
}, },
/// `PRIVMSG <target> :<msg>` /// `PRIVMSG <target> :<msg>`
PrivateMessage { PrivateMessage {
@ -194,20 +194,14 @@ fn client_message_topic(input: &str) -> IResult<&str, ClientMessage> {
fn client_message_part(input: &str) -> IResult<&str, ClientMessage> { fn client_message_part(input: &str) -> IResult<&str, ClientMessage> {
let (input, _) = tag("PART ")(input)?; let (input, _) = tag("PART ")(input)?;
let (input, chan) = chan(input)?; let (input, chan) = chan(input)?;
let (input, t) = opt(tag(" "))(input)?; let (input, _) = tag(" ")(input)?;
match t {
Some(_) => (),
None => {
return Ok((input, ClientMessage::Part { chan, message: None }));
}
}
let (input, r) = opt(tag(":"))(input)?; let (input, r) = opt(tag(":"))(input)?;
let (input, message) = match r { let (input, message) = match r {
Some(_) => token(input)?, Some(_) => token(input)?,
None => receiver(input)?, None => receiver(input)?,
}; };
let message = Some(message.into()); let message = message.into();
Ok((input, ClientMessage::Part { chan, message })) Ok((input, ClientMessage::Part { chan, message }))
} }
@ -375,18 +369,7 @@ mod test {
let input = "PART #chan :Pokasiki !!!"; let input = "PART #chan :Pokasiki !!!";
let expected = ClientMessage::Part { let expected = ClientMessage::Part {
chan: Chan::Global("chan".into()), chan: Chan::Global("chan".into()),
message: Some("Pokasiki !!!".into()), message: "Pokasiki !!!".into(),
};
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
#[test]
fn test_client_message_part_empty() {
let input = "PART #chan";
let expected = ClientMessage::Part {
chan: Chan::Global("chan".into()),
message: None,
}; };
let result = client_message(input); let result = client_message(input);

View File

@ -317,15 +317,10 @@ impl ServerMessageBody {
writer.write_all(b" = ").await?; writer.write_all(b" = ").await?;
chan.write_async(writer).await?; chan.write_async(writer).await?;
writer.write_all(b" :").await?; writer.write_all(b" :").await?;
{ for member in members {
let member = &members.head;
writer.write_all(member.prefix.to_string().as_bytes()).await?; writer.write_all(member.prefix.to_string().as_bytes()).await?;
writer.write_all(member.nick.as_bytes()).await?; writer.write_all(member.nick.as_bytes()).await?;
}
for member in &members.tail {
writer.write_all(b" ").await?; writer.write_all(b" ").await?;
writer.write_all(member.prefix.to_string().as_bytes()).await?;
writer.write_all(member.nick.as_bytes()).await?;
} }
} }
ServerMessageBody::N366NamesReplyEnd { client, chan } => { ServerMessageBody::N366NamesReplyEnd { client, chan } => {

View File

@ -11,15 +11,12 @@ pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-bind";
// TODO remove `pub` in newtypes, introduce validation // TODO remove `pub` in newtypes, introduce validation
/// Name (node identifier) of an XMPP entity. Placed before the `@` in a JID.
#[derive(PartialEq, Eq, Debug, Clone)] #[derive(PartialEq, Eq, Debug, Clone)]
pub struct Name(pub Str); pub struct Name(pub Str);
/// Server name of an XMPP entity. Placed after the `@` and before the `/` in a JID.
#[derive(PartialEq, Eq, Debug, Clone)] #[derive(PartialEq, Eq, Debug, Clone)]
pub struct Server(pub Str); pub struct Server(pub Str);
/// Resource of an XMPP entity. Placed after the `/` in a JID.
#[derive(PartialEq, Eq, Debug, Clone)] #[derive(PartialEq, Eq, Debug, Clone)]
pub struct Resource(pub Str); pub struct Resource(pub Str);

View File

@ -255,44 +255,6 @@ impl MessageType {
} }
} }
/// Error response to an IQ request.
///
/// https://xmpp.org/rfcs/rfc6120.html#stanzas-error
pub struct IqError {
pub r#type: IqErrorType,
}
pub enum IqErrorType {
/// Retry after providing credentials
Auth,
/// Do not retry (the error cannot be remedied)
Cancel,
/// Proceed (the condition was only a warning)
Continue,
/// Retry after changing the data sent
Modify,
/// Retry after waiting (the error is temporary)
Wait,
}
impl IqErrorType {
pub fn as_str(&self) -> &'static str {
match self {
IqErrorType::Auth => "auth",
IqErrorType::Cancel => "cancel",
IqErrorType::Continue => "continue",
IqErrorType::Modify => "modify",
IqErrorType::Wait => "wait",
}
}
}
impl ToXml for IqError {
fn serialize(&self, events: &mut Vec<Event<'static>>) {
let bytes = BytesStart::new(format!(r#"error xmlns="{}" type="{}""#, XMLNS, self.r#type.as_str()));
events.push(Event::Empty(bytes));
}
}
#[derive(PartialEq, Eq, Debug)] #[derive(PartialEq, Eq, Debug)]
pub struct Iq<T> { pub struct Iq<T> {
pub from: Option<String>, pub from: Option<String>,

View File

@ -24,17 +24,7 @@ impl ClientStreamStart {
reader: &mut NsReader<impl AsyncBufRead + Unpin>, reader: &mut NsReader<impl AsyncBufRead + Unpin>,
buf: &mut Vec<u8>, buf: &mut Vec<u8>,
) -> Result<ClientStreamStart> { ) -> Result<ClientStreamStart> {
let mut incoming = skip_text!(reader, buf); let incoming = skip_text!(reader, buf);
if let Event::Decl(bytes) = incoming {
// this is <?xml ...> header
if let Some(encoding) = bytes.encoding() {
let encoding = encoding?;
if &*encoding != b"UTF-8" {
return Err(anyhow!("Unsupported encoding: {encoding:?}"));
}
}
incoming = skip_text!(reader, buf);
}
if let Event::Start(e) = incoming { if let Event::Start(e) = incoming {
let (ns, local) = reader.resolve_element(e.name()); let (ns, local) = reader.resolve_element(e.name());
if ns != ResolveResult::Bound(Namespace(XMLNS.as_bytes())) { if ns != ResolveResult::Bound(Namespace(XMLNS.as_bytes())) {