JustTestingV
/
lavina
forked from lavina/lavina
1
0
Fork 0
Code Pull Requests Activity

Compare commits

base: JustTestingV:77d175ccd9b976db6271ebe42f774758e9139675
Branches Tags
JustTestingV:main
JustTestingV:xmpp-sanitizing
lavina:main
lavina:0.0.0
lavina:0.0.1
lavina:0.0.2
..
compare: JustTestingV:9582160d2cfb706ed84082ae0f63b19751066b6c
Branches Tags
JustTestingV:main
JustTestingV:xmpp-sanitizing
lavina:main
lavina:0.0.0
lavina:0.0.1
lavina:0.0.2

No commits in common. "77d175ccd9b976db6271ebe42f774758e9139675" and "9582160d2cfb706ed84082ae0f63b19751066b6c" have entirely different histories.
77d175ccd9 ... 9582160d2c

35 changed files with 841 additions and 2009 deletions
Show Stats Expand all files Collapse all files

2
.dockerignore
View File

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

6
.gitea/workflows/test-pr.yml
View File

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

1
.gitignore vendored
View File

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

803
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

12
Cargo.toml
View File

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

30
crates/README.md Normal file
View File

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

8
crates/projection-irc/Cargo.toml
View File

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

2
crates/projection-irc/rustfmt.toml
View File

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

9
crates/projection-irc/src/cap.rs
View File

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

296
crates/projection-irc/src/lib.rs
View File

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

220
crates/projection-irc/tests/lib.rs
View File

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

7
crates/projection-xmpp/Cargo.toml
View File

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

2
crates/projection-xmpp/rustfmt.toml
View File

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

162
crates/projection-xmpp/src/iq.rs
View File

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

348
crates/projection-xmpp/src/lib.rs
View File

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

50
crates/projection-xmpp/src/message.rs
View File

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

55
crates/projection-xmpp/src/presence.rs
View File

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

45
crates/projection-xmpp/src/updates.rs
View File

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

52
crates/projection-xmpp/tests/certs/xmpp.key
View File

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

30
crates/projection-xmpp/tests/certs/xmpp.pem
View File

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

186
crates/projection-xmpp/tests/lib.rs
View File

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

69
crates/proto-irc/src/client.rs
View File

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

106
crates/proto-irc/src/server.rs
View File

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

1
crates/proto-xmpp/Cargo.toml
View File

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

4
crates/proto-xmpp/src/lib.rs
View File

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

96
crates/proto-xmpp/src/sasl.rs
View File

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

12
crates/proto-xmpp/src/xml/mod.rs
View File

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

46
crates/readme.md
View File

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

8
crates/sasl/Cargo.toml
View File

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

103
crates/sasl/src/lib.rs
View File

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

13
docs/cheatsheet.md
View File

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

54
docs/running.md
View File

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

4
readme.md
View File

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

2
rust-toolchain
View File

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

2
src/http.rs
View File

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