Compare commits

...

19 Commits

Author SHA1 Message Date
JustTestingV 82f8fc90b3 merge main 2023-10-01 22:37:16 +03:00
JustTestingV da520f5a25 [xmpp] logopass auth 2023-10-01 22:15:44 +03:00
Nikita Vilunov 8047a97baa remove /test/ 2023-10-01 18:16:11 +02:00
Nikita Vilunov 47195f5eee add crates/README.md 2023-10-01 18:15:37 +02:00
Nikita Vilunov 2f034284cf rename telemetry mod to http 2023-10-01 01:50:04 +02:00
Nikita Vilunov dc0a101fe6 split xmpp projection into a separate crate 2023-10-01 01:47:18 +02:00
Nikita Vilunov 854a244dbc split irc proejction into a separate crate (#18) 2023-09-30 23:34:35 +00:00
Nikita Vilunov a1db17c779 split core into a separate crate (#17) 2023-09-30 23:12:11 +00:00
Nikita Vilunov 1b3551f108 xmpp: rewrite bind request parser as a coroutine (#12)
Reviewed-on: lavina/lavina#12
2023-09-30 16:02:18 +00:00
Nikita Vilunov 444b608e96 split proto xmpp defs into its own crate (#14)
Reviewed-on: lavina/lavina#14
2023-09-30 15:43:46 +00:00
Nikita Vilunov 563811cbca pull-request workflow (#15) 2023-09-30 15:38:51 +00:00
Nikita Vilunov 58f6a5d90a add management API endpoints 2023-09-24 22:59:34 +02:00
Nikita Vilunov df6cdd4861 add a bit more logging 2023-09-23 01:12:03 +02:00
Nikita Vilunov 3d59f6aae5 read config path from cli args 2023-09-22 17:24:36 +02:00
Nikita Vilunov ad49703714 split proto irc defs into its own crate (#13)
Reviewed-on: lavina/lavina#13
2023-09-22 13:20:47 +00:00
JustTestingV 87d73af811 [irc] used nonempty prefixed nicks in 353 reply (#9)
Reviewed-on: lavina/lavina#9
Co-authored-by: JustTestingV <JustTestingV@gmail.com>
Co-committed-by: JustTestingV <JustTestingV@gmail.com>
2023-09-15 16:33:25 +00:00
Nikita Vilunov 298245f3f5 bump version 2023-09-15 17:12:12 +02:00
Nikita Vilunov 3de7a131f0 add dockerfile 2023-09-15 17:11:29 +02:00
Nikita Vilunov c662b64f11 disable tls support in sqlx, remove webpki-roots 2023-09-14 18:55:03 +02:00
57 changed files with 1038 additions and 703 deletions

6
.dockerignore Normal file
View File

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

View File

@ -0,0 +1,18 @@
name: check-and-test
on: [push, pull_request]
jobs:
check-and-test:
runs-on: ubuntu-latest
steps:
- name: git checkout repository
uses: actions/checkout@v4
- name: setup rust
uses: https://github.com/actions-rs/toolchain@v1
- name: cargo check
uses: https://github.com/actions-rs/cargo@v1
with:
command: check
- name: test
uses: https://github.com/actions-rs/cargo@v1
with:
command: test

1
.gitignore vendored
View File

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

231
Cargo.lock generated
View File

@ -44,6 +44,54 @@ 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 = "anstream"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1f58811cfac344940f1a400b6e6231ce35171f614f26439e80f8c1465c5cc0c"
dependencies = [
"anstyle",
"anstyle-parse",
"anstyle-query",
"anstyle-wincon",
"colorchoice",
"utf8parse",
]
[[package]]
name = "anstyle"
version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b84bf0a05bbb2a83e5eb6fa36bb6e87baa08193c35ff52bbf6b38d8af2890e46"
[[package]]
name = "anstyle-parse"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "938874ff5980b03a87c5524b3ae5b59cf99b1d6bc836848df7bc5ada9643c333"
dependencies = [
"utf8parse",
]
[[package]]
name = "anstyle-query"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5ca11d4be1bab0c8bc8734a9aa7bf4ee8316d462a08c6ac5052f888fef5b494b"
dependencies = [
"windows-sys",
]
[[package]]
name = "anstyle-wincon"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "58f54d10c6dfa51283a066ceab3ec1ab78d13fae00aa49243a45e4571fb79dfd"
dependencies = [
"anstyle",
"windows-sys",
]
[[package]] [[package]]
name = "anyhow" name = "anyhow"
version = "1.0.75" version = "1.0.75"
@ -161,6 +209,52 @@ 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 = "clap"
version = "4.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1d7b8d5ec32af0fadc644bf1fd509a688c2103b185644bb1e29d164e0703136"
dependencies = [
"clap_builder",
"clap_derive",
]
[[package]]
name = "clap_builder"
version = "4.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5179bb514e4d7c2051749d8fcefa2ed6d06a9f4e6d69faf3805f5d80b8cf8d56"
dependencies = [
"anstream",
"anstyle",
"clap_lex",
"strsim",
]
[[package]]
name = "clap_derive"
version = "4.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0862016ff20d69b84ef8247369fabf5c008a7417002411897d40ee1f4532b873"
dependencies = [
"heck",
"proc-macro2",
"quote",
"syn 2.0.29",
]
[[package]]
name = "clap_lex"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cd7cc57abe963c6d3b9d8be5b06ba7c8957a930305ca90304f24ef040aa6f961"
[[package]]
name = "colorchoice"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "acbf1af155f9b9ef647e42cdc158db4b64a1b61f743629225fde6f3e0be2a7c7"
[[package]] [[package]]
name = "const-oid" name = "const-oid"
version = "0.9.5" version = "0.9.5"
@ -750,30 +844,41 @@ dependencies = [
[[package]] [[package]]
name = "lavina" name = "lavina"
version = "0.1.0" version = "0.0.1-dev"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"assert_matches", "assert_matches",
"clap",
"derive_more", "derive_more",
"figment", "figment",
"futures-util", "futures-util",
"http-body-util", "http-body-util",
"hyper 1.0.0-rc.3", "hyper 1.0.0-rc.3",
"lazy_static", "lavina-core",
"nom", "mgmt-api",
"nonempty",
"projection-irc",
"projection-xmpp",
"prometheus", "prometheus",
"quick-xml",
"regex", "regex",
"reqwest", "reqwest",
"rustls-pemfile",
"serde", "serde",
"serde_json", "serde_json",
"sqlx",
"tokio", "tokio",
"tokio-rustls",
"tracing", "tracing",
"tracing-subscriber", "tracing-subscriber",
"uuid", ]
[[package]]
name = "lavina-core"
version = "0.0.1-dev"
dependencies = [
"anyhow",
"prometheus",
"serde",
"sqlx",
"tokio",
"tracing",
] ]
[[package]] [[package]]
@ -845,6 +950,13 @@ version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d" checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
[[package]]
name = "mgmt-api"
version = "0.0.1-dev"
dependencies = [
"serde",
]
[[package]] [[package]]
name = "mime" name = "mime"
version = "0.3.17" version = "0.3.17"
@ -887,6 +999,12 @@ dependencies = [
"minimal-lexical", "minimal-lexical",
] ]
[[package]]
name = "nonempty"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aeaf4ad7403de93e699c191202f017118df734d3850b01e13a3a8b2e6953d3c9"
[[package]] [[package]]
name = "nu-ansi-term" name = "nu-ansi-term"
version = "0.46.0" version = "0.46.0"
@ -1130,6 +1248,40 @@ dependencies = [
"yansi", "yansi",
] ]
[[package]]
name = "projection-irc"
version = "0.0.1-dev"
dependencies = [
"anyhow",
"futures-util",
"lavina-core",
"nonempty",
"prometheus",
"proto-irc",
"serde",
"tokio",
"tracing",
]
[[package]]
name = "projection-xmpp"
version = "0.0.1-dev"
dependencies = [
"anyhow",
"derive_more",
"futures-util",
"lavina-core",
"prometheus",
"proto-xmpp",
"quick-xml",
"rustls-pemfile",
"serde",
"tokio",
"tokio-rustls",
"tracing",
"uuid",
]
[[package]] [[package]]
name = "prometheus" name = "prometheus"
version = "0.13.3" version = "0.13.3"
@ -1144,6 +1296,31 @@ dependencies = [
"thiserror", "thiserror",
] ]
[[package]]
name = "proto-irc"
version = "0.0.1-dev"
dependencies = [
"anyhow",
"assert_matches",
"futures-util",
"nom",
"nonempty",
"tokio",
]
[[package]]
name = "proto-xmpp"
version = "0.0.1-dev"
dependencies = [
"anyhow",
"base64",
"derive_more",
"lazy_static",
"quick-xml",
"regex",
"tokio",
]
[[package]] [[package]]
name = "quick-xml" name = "quick-xml"
version = "0.30.0" version = "0.30.0"
@ -1603,19 +1780,14 @@ dependencies = [
"once_cell", "once_cell",
"paste", "paste",
"percent-encoding", "percent-encoding",
"rustls",
"rustls-pemfile",
"serde", "serde",
"serde_json", "serde_json",
"sha2", "sha2",
"smallvec", "smallvec",
"sqlformat", "sqlformat",
"thiserror", "thiserror",
"tokio",
"tokio-stream",
"tracing", "tracing",
"url", "url",
"webpki-roots",
] ]
[[package]] [[package]]
@ -1652,7 +1824,6 @@ dependencies = [
"sqlx-sqlite", "sqlx-sqlite",
"syn 1.0.109", "syn 1.0.109",
"tempfile", "tempfile",
"tokio",
"url", "url",
] ]
@ -1769,6 +1940,12 @@ dependencies = [
"unicode-normalization", "unicode-normalization",
] ]
[[package]]
name = "strsim"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
[[package]] [[package]]
name = "subtle" name = "subtle"
version = "2.5.0" version = "2.5.0"
@ -1895,17 +2072,6 @@ dependencies = [
"tokio", "tokio",
] ]
[[package]]
name = "tokio-stream"
version = "0.1.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "397c988d37662c7dda6d2208364a706264bf3d6138b11d436cbac0ad38832842"
dependencies = [
"futures-core",
"pin-project-lite",
"tokio",
]
[[package]] [[package]]
name = "tokio-util" name = "tokio-util"
version = "0.7.8" version = "0.7.8"
@ -2090,6 +2256,12 @@ dependencies = [
"percent-encoding", "percent-encoding",
] ]
[[package]]
name = "utf8parse"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a"
[[package]] [[package]]
name = "uuid" name = "uuid"
version = "1.4.1" version = "1.4.1"
@ -2208,15 +2380,6 @@ dependencies = [
"wasm-bindgen", "wasm-bindgen",
] ]
[[package]]
name = "webpki-roots"
version = "0.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b291546d5d9d1eab74f069c77749f2cb8504a12caa20f0f2de93ddbf6f411888"
dependencies = [
"rustls-webpki",
]
[[package]] [[package]]
name = "whoami" name = "whoami"
version = "1.4.1" version = "1.4.1"

View File

@ -1,32 +1,61 @@
[workspace]
members = [
".",
"crates/lavina-core",
"crates/proto-irc",
"crates/projection-irc",
"crates/proto-xmpp",
"crates/mgmt-api",
]
[workspace.package]
version = "0.0.1-dev"
[workspace.dependencies]
nom = "7.1.3"
assert_matches = "1.5.0"
tokio = { version = "1.24.1", features = ["full"] } # async runtime
futures-util = "0.3.25"
anyhow = "1.0.68" # error utils
nonempty = "0.8.1"
quick-xml = { version = "0.30.0", features = ["async-tokio"] }
lazy_static = "1.4.0"
regex = "1.7.1"
derive_more = "0.99.17"
clap = { version = "4.4.4", features = ["derive"] }
serde = { version = "1.0.152", features = ["rc", "serde_derive"] }
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" }
[package] [package]
name = "lavina" name = "lavina"
version = "0.1.0" version.workspace = true
edition = "2021" edition = "2021"
publish = false publish = false
[dependencies] [dependencies]
anyhow = "1.0.68" # error utils anyhow.workspace = true
figment = { version = "0.10.8", features = ["env", "toml"] } # configuration files figment = { version = "0.10.8", features = ["env", "toml"] } # configuration files
hyper = { version = "1.0.0-rc.3,<1.0.0-rc.4", features = ["server", "http1"] } # http server hyper = { version = "1.0.0-rc.3,<1.0.0-rc.4", features = ["server", "http1"] } # http server
http-body-util = "0.1.0-rc.3" http-body-util = "0.1.0-rc.3"
serde = { version = "1.0.152", features = ["rc", "serde_derive"] } serde.workspace = true
serde_json = "1.0.93" serde_json = "1.0.93"
tokio = { version = "1.24.1", features = ["full"] } # async runtime tokio.workspace = true
tracing = "0.1.37" # logging & tracing api tracing.workspace = true
tracing-subscriber = "0.3.16" tracing-subscriber = "0.3.16"
futures-util = "0.3.25" futures-util.workspace = true
prometheus = { version = "0.13.3", default-features = false } prometheus.workspace = true
regex = "1.7.1" nonempty.workspace = true
lazy_static = "1.4.0" derive_more.workspace = true
nom = "7.1.3" lavina-core.workspace = true
tokio-rustls = "0.24.1" projection-irc = { path = "crates/projection-irc" }
rustls-pemfile = "1.0.2" projection-xmpp = { path = "crates/projection-xmpp" }
quick-xml = { version = "0.30.0", features = ["async-tokio"] } mgmt-api = { path = "crates/mgmt-api" }
derive_more = "0.99.17" clap.workspace = true
uuid = { version = "1.3.0", features = ["v4"] }
sqlx = { version = "0.7.0-alpha.2", features = ["sqlite", "runtime-tokio-rustls", "migrate"] }
[dev-dependencies] [dev-dependencies]
assert_matches = "1.5.0" assert_matches.workspace = true
regex = "1.7.1" regex = "1.7.1"
reqwest = { version = "0.11", default-features = false } reqwest = { version = "0.11", default-features = false }

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.

View File

@ -0,0 +1,12 @@
[package]
name = "lavina-core"
edition = "2021"
version.workspace = true
[dependencies]
anyhow.workspace = true
sqlx = { version = "0.7.0-alpha.2", features = ["sqlite", "migrate"] }
serde.workspace = true
tokio.workspace = true
tracing.workspace = true
prometheus.workspace = true

View File

@ -1,4 +1,8 @@
//! Domain definitions and implementation of common chat logic. //! Domain definitions and implementation of common chat logic.
pub mod player; pub mod player;
pub mod prelude;
pub mod repo; pub mod repo;
pub mod room; pub mod room;
pub mod terminator;
mod table;

View File

@ -19,11 +19,9 @@ use tokio::{
task::JoinHandle, task::JoinHandle,
}; };
use crate::{ use crate::prelude::*;
core::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry}, use crate::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry};
prelude::*, use crate::table::{AnonTable, Key as AnonKey};
util::table::{AnonTable, Key as AnonKey},
};
/// Opaque player identifier. Cannot contain spaces, must be shorter than 32. /// Opaque player identifier. Cannot contain spaces, must be shorter than 32.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)] #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
@ -63,9 +61,7 @@ impl PlayerConnection {
} }
pub async fn join_room(&mut self, room_id: RoomId) -> Result<JoinResult> { pub async fn join_room(&mut self, room_id: RoomId) -> Result<JoinResult> {
self.player_handle self.player_handle.join_room(room_id, self.connection_id.clone()).await
.join_room(room_id, self.connection_id.clone())
.await
} }
pub async fn change_topic(&mut self, room_id: RoomId, new_topic: Str) -> Result<()> { pub async fn change_topic(&mut self, room_id: RoomId, new_topic: Str) -> Result<()> {
@ -100,9 +96,7 @@ impl PlayerConnection {
pub async fn get_rooms(&self) -> Result<Vec<RoomInfo>> { pub async fn get_rooms(&self) -> Result<Vec<RoomInfo>> {
let (promise, deferred) = oneshot(); let (promise, deferred) = oneshot();
self.player_handle self.player_handle.send(PlayerCommand::GetRooms(promise)).await;
.send(PlayerCommand::GetRooms(promise))
.await;
Ok(deferred.await?) Ok(deferred.await?)
} }
} }
@ -126,27 +120,14 @@ impl PlayerHandle {
} }
} }
pub async fn send_message( pub async fn send_message(&self, room_id: RoomId, connection_id: ConnectionId, body: Str) -> Result<()> {
&self,
room_id: RoomId,
connection_id: ConnectionId,
body: Str,
) -> Result<()> {
let (promise, deferred) = oneshot(); let (promise, deferred) = oneshot();
let cmd = Cmd::SendMessage { let cmd = Cmd::SendMessage { room_id, body, promise };
room_id,
body,
promise,
};
let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await; let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await;
Ok(deferred.await?) Ok(deferred.await?)
} }
pub async fn join_room( pub async fn join_room(&self, room_id: RoomId, connection_id: ConnectionId) -> Result<JoinResult> {
&self,
room_id: RoomId,
connection_id: ConnectionId,
) -> Result<JoinResult> {
let (promise, deferred) = oneshot(); let (promise, deferred) = oneshot();
let cmd = Cmd::JoinRoom { room_id, promise }; let cmd = Cmd::JoinRoom { room_id, promise };
let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await; let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await;
@ -230,12 +211,8 @@ pub enum Updates {
#[derive(Clone)] #[derive(Clone)]
pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>); pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
impl PlayerRegistry { impl PlayerRegistry {
pub fn empty( pub fn empty(room_registry: RoomRegistry, metrics: &mut MetricsRegistry) -> Result<PlayerRegistry> {
room_registry: RoomRegistry, let metric_active_players = IntGauge::new("chat_players_active", "Number of alive player actors")?;
metrics: &mut MetricsRegistry,
) -> Result<PlayerRegistry> {
let metric_active_players =
IntGauge::new("chat_players_active", "Number of alive player actors")?;
metrics.register(Box::new(metric_active_players.clone()))?; metrics.register(Box::new(metric_active_players.clone()))?;
let inner = PlayerRegistryInner { let inner = PlayerRegistryInner {
room_registry, room_registry,
@ -302,7 +279,7 @@ impl Player {
player_id, player_id,
connections: AnonTable::new(), connections: AnonTable::new(),
my_rooms: HashMap::new(), my_rooms: HashMap::new(),
banned_from: HashSet::from([RoomId::from("empty").unwrap()]), banned_from: HashSet::from([RoomId::from("Empty").unwrap()]),
rx, rx,
handle, handle,
rooms, rooms,
@ -375,8 +352,7 @@ impl Player {
return; return;
} }
}; };
room.subscribe(self.player_id.clone(), self.handle.clone()) room.subscribe(self.player_id.clone(), self.handle.clone()).await;
.await;
self.my_rooms.insert(room_id.clone(), room.clone()); self.my_rooms.insert(room_id.clone(), room.clone());
let room_info = room.get_room_info().await; let room_info = room.get_room_info().await;
let _ = promise.send(JoinResult::Success(room_info)); let _ = promise.send(JoinResult::Success(room_info));
@ -399,15 +375,10 @@ impl Player {
}; };
self.broadcast_update(update, connection_id).await; self.broadcast_update(update, connection_id).await;
} }
Cmd::SendMessage { Cmd::SendMessage { room_id, body, promise } => {
room_id,
body,
promise,
} => {
let room = self.rooms.get_room(&room_id).await; let room = self.rooms.get_room(&room_id).await;
if let Some(room) = room { if let Some(room) = room {
room.send_message(self.player_id.clone(), body.clone()) room.send_message(self.player_id.clone(), body.clone()).await;
.await;
} else { } else {
tracing::info!("no room found"); tracing::info!("no room found");
} }
@ -426,8 +397,7 @@ impl Player {
} => { } => {
let room = self.rooms.get_room(&room_id).await; let room = self.rooms.get_room(&room_id).await;
if let Some(mut room) = room { if let Some(mut room) = room {
room.set_topic(self.player_id.clone(), new_topic.clone()) room.set_topic(self.player_id.clone(), new_topic.clone()).await;
.await;
} else { } else {
tracing::info!("no room found"); tracing::info!("no room found");
} }

View File

@ -15,11 +15,3 @@ pub type Str = std::sync::Arc<str>;
pub fn fail(msg: &str) -> anyhow::Error { pub fn fail(msg: &str) -> anyhow::Error {
anyhow::Error::msg(msg.to_owned()) anyhow::Error::msg(msg.to_owned())
} }
macro_rules! ffail {
($($arg:tt)*) => {
fail(&format!($($arg)*))
};
}
pub(crate) use ffail;

View File

@ -5,7 +5,7 @@ use std::sync::Arc;
use serde::Deserialize; use serde::Deserialize;
use sqlx::sqlite::SqliteConnectOptions; use sqlx::sqlite::SqliteConnectOptions;
use sqlx::{ConnectOptions, Connection, FromRow, SqliteConnection}; use sqlx::{ConnectOptions, Connection, FromRow, Sqlite, SqliteConnection, Transaction};
use tokio::sync::Mutex; use tokio::sync::Mutex;
use crate::prelude::*; use crate::prelude::*;
@ -93,15 +93,59 @@ impl Storage {
Ok(()) Ok(())
} }
pub async fn close(mut self) -> Result<()> { pub async fn close(self) -> Result<()> {
let res = match Arc::try_unwrap(self.conn) { let res = match Arc::try_unwrap(self.conn) {
Ok(e) => e, Ok(e) => e,
Err(e) => return Err(fail("failed to acquire DB ownership on shutdown")), Err(_) => return Err(fail("failed to acquire DB ownership on shutdown")),
}; };
let res = res.into_inner(); let res = res.into_inner();
res.close().await?; res.close().await?;
Ok(()) Ok(())
} }
pub async fn create_user(&mut self, name: &str) -> Result<()> {
let query = sqlx::query(
"insert into users(name)
values (?);",
)
.bind(name);
let mut executor = self.conn.lock().await;
query.execute(&mut *executor).await?;
Ok(())
}
pub async fn set_password<'a>(&'a mut self, name: &'a str, pwd: &'a str) -> Result<Option<()>> {
async fn inner(txn: &mut Transaction<'_, Sqlite>, name: &str, pwd: &str) -> Result<Option<()>> {
let id: Option<(u32,)> = sqlx::query_as("select * from users where name = ? limit 1;")
.bind(name)
.fetch_optional(&mut **txn)
.await?;
let Some((id,)) = id else {
return Ok(None);
};
sqlx::query("insert or replace into challenges_plain_password(user_id, password) values (?, ?);")
.bind(id)
.bind(pwd)
.execute(&mut **txn)
.await?;
Ok(Some(()))
}
let mut executor = self.conn.lock().await;
let mut tx = executor.begin().await?;
let res = inner(&mut tx, name, pwd).await;
match res {
Ok(e) => {
tx.commit().await?;
Ok(e)
}
Err(e) => {
tx.rollback().await?;
Err(e)
}
}
}
} }
#[derive(FromRow)] #[derive(FromRow)]

View File

@ -1,17 +1,13 @@
//! Domain of rooms — chats with multiple participants. //! Domain of rooms — chats with multiple participants.
use std::{ use std::{collections::HashMap, hash::Hash, sync::Arc};
collections::HashMap,
hash::Hash,
sync::Arc,
};
use prometheus::{IntGauge, Registry as MetricRegistry}; use prometheus::{IntGauge, Registry as MetricRegistry};
use serde::Serialize; use serde::Serialize;
use tokio::sync::RwLock as AsyncRwLock; use tokio::sync::RwLock as AsyncRwLock;
use crate::core::player::{PlayerHandle, PlayerId, Updates}; use crate::player::{PlayerHandle, PlayerId, Updates};
use crate::core::repo::Storage;
use crate::prelude::*; use crate::prelude::*;
use crate::repo::Storage;
/// Opaque room id /// Opaque room id
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)] #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
@ -20,9 +16,7 @@ impl RoomId {
pub fn from(str: impl Into<Str>) -> Result<RoomId> { pub fn from(str: impl Into<Str>) -> Result<RoomId> {
let bytes = str.into(); let bytes = str.into();
if bytes.len() > 32 { if bytes.len() > 32 {
return Err(anyhow::Error::msg( return Err(anyhow::Error::msg("Room name cannot be longer than 32 symbols"));
"Room name cannot be longer than 32 symbols",
));
} }
if bytes.contains(' ') { if bytes.contains(' ') {
return Err(anyhow::Error::msg("Room name cannot contain spaces")); return Err(anyhow::Error::msg("Room name cannot contain spaces"));
@ -42,8 +36,7 @@ impl RoomId {
pub struct RoomRegistry(Arc<AsyncRwLock<RoomRegistryInner>>); pub struct RoomRegistry(Arc<AsyncRwLock<RoomRegistryInner>>);
impl RoomRegistry { impl RoomRegistry {
pub fn new(metrics: &mut MetricRegistry, storage: Storage) -> Result<RoomRegistry> { pub fn new(metrics: &mut MetricRegistry, storage: Storage) -> Result<RoomRegistry> {
let metric_active_rooms = let metric_active_rooms = IntGauge::new("chat_rooms_active", "Number of alive room actors")?;
IntGauge::new("chat_rooms_active", "Number of alive room actors")?;
metrics.register(Box::new(metric_active_rooms.clone()))?; metrics.register(Box::new(metric_active_rooms.clone()))?;
let inner = RoomRegistryInner { let inner = RoomRegistryInner {
rooms: HashMap::new(), rooms: HashMap::new(),
@ -140,7 +133,7 @@ impl RoomHandle {
pub async fn send_message(&self, player_id: PlayerId, body: Str) { pub async fn send_message(&self, player_id: PlayerId, body: Str) {
let mut lock = self.0.write().await; let mut lock = self.0.write().await;
let res = lock.send_message(player_id, body).await; let res = lock.send_message(player_id, body).await;
if let Err(err) = res { if let Err(err) = res {
log::warn!("Failed to send message: {err:?}"); log::warn!("Failed to send message: {err:?}");
} }
@ -150,11 +143,7 @@ impl RoomHandle {
let lock = self.0.read().await; let lock = self.0.read().await;
RoomInfo { RoomInfo {
id: lock.room_id.clone(), id: lock.room_id.clone(),
members: lock members: lock.subscriptions.keys().map(|x| x.clone()).collect::<Vec<_>>(),
.subscriptions
.keys()
.map(|x| x.clone())
.collect::<Vec<_>>(),
topic: lock.topic.clone(), topic: lock.topic.clone(),
} }
} }
@ -191,7 +180,9 @@ impl Room {
async fn send_message(&mut self, author_id: PlayerId, body: Str) -> Result<()> { async fn send_message(&mut self, author_id: PlayerId, body: Str) -> Result<()> {
tracing::info!("Adding a message to room"); tracing::info!("Adding a message to room");
self.storage.insert_message(self.storage_id, self.message_count, &body).await?; self.storage
.insert_message(self.storage_id, self.message_count, &body)
.await?;
self.message_count += 1; self.message_count += 1;
let update = Updates::NewMessage { let update = Updates::NewMessage {
room_id: self.room_id.clone(), room_id: self.room_id.clone(),

View File

@ -21,7 +21,7 @@ impl<V> AnonTable<V> {
pub fn insert(&mut self, value: V) -> Key { pub fn insert(&mut self, value: V) -> Key {
let id = self.next; let id = self.next;
self.next += 1; self.next += 1;
self.inner.insert(id, value); // should be always empty self.inner.insert(id, value); // should be always Empty
Key(id) Key(id)
} }

View File

@ -1,12 +1,5 @@
use crate::prelude::*; use crate::prelude::*;
pub mod http;
pub mod table;
pub mod telemetry;
#[cfg(test)]
pub mod testkit;
pub mod xml;
pub struct Terminator { pub struct Terminator {
signal: Promise<()>, signal: Promise<()>,
completion: JoinHandle<Result<()>>, completion: JoinHandle<Result<()>>,

View File

@ -0,0 +1,8 @@
[package]
name = "mgmt-api"
edition = "2021"
version.workspace = true
publish = false
[dependencies]
serde.workspace = true

View File

@ -0,0 +1,29 @@
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize)]
pub struct ErrorResponse<'a> {
pub code: &'a str,
pub message: &'a str,
}
#[derive(Serialize, Deserialize)]
pub struct CreatePlayerRequest<'a> {
pub name: &'a str,
}
#[derive(Serialize, Deserialize)]
pub struct ChangePasswordRequest<'a> {
pub player_name: &'a str,
pub password: &'a str,
}
pub mod paths {
pub const CREATE_PLAYER: &'static str = "/mgmt/create_player";
pub const SET_PASSWORD: &'static str = "/mgmt/set_password";
}
pub mod errors {
pub const INVALID_PATH: &'static str = "invalid_path";
pub const MALFORMED_REQUEST: &'static str = "malformed_request";
pub const PLAYER_NOT_FOUND: &'static str = "player_not_found";
}

View File

@ -0,0 +1,16 @@
[package]
name = "projection-irc"
edition = "2021"
version.workspace = true
[dependencies]
lavina-core.workspace = true
tracing.workspace = true
anyhow.workspace = true
serde.workspace = true
tokio.workspace = true
prometheus.workspace = true
futures-util.workspace = true
nonempty.workspace = true
proto-irc = { path = "../proto-irc" }

View File

@ -1,7 +1,10 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::net::SocketAddr; use std::net::SocketAddr;
use anyhow::{anyhow, Result};
use futures_util::future::join_all; use futures_util::future::join_all;
use nonempty::nonempty;
use nonempty::NonEmpty;
use prometheus::{IntCounter, IntGauge, Registry as MetricsRegistry}; use prometheus::{IntCounter, IntGauge, Registry as MetricsRegistry};
use serde::Deserialize; use serde::Deserialize;
use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter}; use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter};
@ -9,17 +12,15 @@ use tokio::net::tcp::{ReadHalf, WriteHalf};
use tokio::net::{TcpListener, TcpStream}; use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::channel; use tokio::sync::mpsc::channel;
use crate::core::player::*; use lavina_core::player::*;
use crate::core::repo::Storage; use lavina_core::prelude::*;
use crate::core::room::{RoomId, RoomInfo, RoomRegistry}; use lavina_core::repo::Storage;
use crate::prelude::*; use lavina_core::room::{RoomId, RoomInfo, RoomRegistry};
use crate::protos::irc::client::{client_message, ClientMessage}; use lavina_core::terminator::Terminator;
use crate::protos::irc::server::{AwayStatus, ServerMessage, ServerMessageBody}; use proto_irc::client::{client_message, ClientMessage};
use crate::protos::irc::{Chan, Recipient}; use proto_irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
use crate::util::Terminator; use proto_irc::user::PrefixedNick;
use proto_irc::{Chan, Recipient};
#[cfg(test)]
mod test;
#[derive(Deserialize, Debug, Clone)] #[derive(Deserialize, Debug, Clone)]
pub struct ServerConfig { pub struct ServerConfig {
@ -48,6 +49,7 @@ async fn handle_socket(
termination: Deferred<()>, // TODO use it to stop the connection gracefully termination: Deferred<()>, // TODO use it to stop the connection gracefully
mut storage: Storage, mut storage: Storage,
) -> Result<()> { ) -> Result<()> {
log::info!("Received an IRC connection from {socket_addr}");
let (reader, writer) = stream.split(); let (reader, writer) = stream.split();
let mut reader: BufReader<ReadHalf> = BufReader::new(reader); let mut reader: BufReader<ReadHalf> = BufReader::new(reader);
let mut writer = BufWriter::new(writer); let mut writer = BufWriter::new(writer);
@ -65,15 +67,16 @@ async fn handle_socket(
.await?; .await?;
writer.flush().await?; writer.flush().await?;
let registered_user: Result<RegisteredUser> = let registered_user: Result<RegisteredUser> = handle_registration(&mut reader, &mut writer, &mut storage).await;
handle_registration(&mut reader, &mut writer, &mut storage).await;
match registered_user { match registered_user {
Ok(user) => { Ok(user) => {
handle_registered_socket(config, players, rooms, &mut reader, &mut writer, user) log::debug!("User registered");
.await?; handle_registered_socket(config, players, rooms, &mut reader, &mut writer, user).await?;
}
Err(_) => {
log::debug!("Registration failed");
} }
Err(_) => {}
} }
stream.shutdown().await?; stream.shutdown().await?;
@ -151,23 +154,22 @@ async fn handle_registration<'a>(
buffer.clear(); buffer.clear();
}?; }?;
let stored_user = storage.retrieve_user_by_name(&*user.nickname).await?; let stored_user = storage.retrieve_user_by_name(&*user.nickname).await?;
let stored_user = match stored_user { let stored_user = match stored_user {
Some(u) => u, Some(u) => u,
None => { None => {
log::info!("User '{}' not found", user.nickname); log::info!("User '{}' not found", user.nickname);
return Err(fail("no user found")); return Err(anyhow!("no user found"));
} }
}; };
if stored_user.password.is_none() { if stored_user.password.is_none() {
log::info!("Password not defined for user '{}'", user.nickname); log::info!("Password not defined for user '{}'", user.nickname);
return Err(fail("password is not defined")); return Err(anyhow!("password is not defined"));
} }
if stored_user.password.as_deref() != pass.as_deref() { if stored_user.password.as_deref() != pass.as_deref() {
log::info!("Incorrect password supplied for user '{}'", user.nickname); log::info!("Incorrect password supplied for user '{}'", user.nickname);
return Err(fail("passwords do not match")); return Err(anyhow!("passwords do not match"));
} }
// TODO properly implement session temination // TODO properly implement session temination
@ -242,14 +244,7 @@ async fn handle_registered_socket<'a>(
let rooms_list = connection.get_rooms().await?; let rooms_list = connection.get_rooms().await?;
for room in &rooms_list { for room in &rooms_list {
produce_on_join_cmd_messages( produce_on_join_cmd_messages(&config, &user, &Chan::Global(room.id.as_inner().clone()), room, writer).await?;
&config,
&user,
&Chan::Global(room.id.as_inner().clone()),
room,
writer,
)
.await?;
} }
writer.flush().await?; writer.flush().await?;
@ -306,10 +301,7 @@ async fn handle_update(
) -> Result<()> { ) -> Result<()> {
log::debug!("Sending irc message to player {player_id:?} on update {update:?}"); log::debug!("Sending irc message to player {player_id:?} on update {update:?}");
match update { match update {
Updates::RoomJoined { Updates::RoomJoined { new_member_id, room_id } => {
new_member_id,
room_id,
} => {
if player_id == &new_member_id { if player_id == &new_member_id {
if let Some(room) = rooms.get_room(&room_id).await { if let Some(room) = rooms.get_room(&room_id).await {
let room_info = room.get_room_info().await; let room_info = room.get_room_info().await;
@ -430,16 +422,14 @@ async fn handle_incoming_message(
Recipient::Chan(Chan::Global(chan)) => { Recipient::Chan(Chan::Global(chan)) => {
let room_id = RoomId::from(chan)?; let room_id = RoomId::from(chan)?;
user_handle.send_message(room_id, body).await?; user_handle.send_message(room_id, body).await?;
}, }
_ => log::warn!("Unsupported target type"), _ => log::warn!("Unsupported target type"),
}, },
ClientMessage::Topic { chan, topic } => { ClientMessage::Topic { chan, topic } => {
match chan { match chan {
Chan::Global(chan) => { Chan::Global(chan) => {
let room_id = RoomId::from(chan)?; let room_id = RoomId::from(chan)?;
user_handle user_handle.change_topic(room_id.clone(), topic.clone()).await?;
.change_topic(room_id.clone(), topic.clone())
.await?;
ServerMessage { ServerMessage {
tags: vec![], tags: vec![],
sender: Some(config.server_name.clone()), sender: Some(config.server_name.clone()),
@ -561,11 +551,7 @@ async fn handle_incoming_message(
Ok(HandleResult::Continue) Ok(HandleResult::Continue)
} }
fn user_to_who_msg( fn user_to_who_msg(config: &ServerConfig, requestor: &RegisteredUser, target_user_nickname: &Str) -> ServerMessageBody {
config: &ServerConfig,
requestor: &RegisteredUser,
target_user_nickname: &Str,
) -> ServerMessageBody {
// Username is equal to nickname // Username is equal to nickname
let username = format!("~{target_user_nickname}").into(); let username = format!("~{target_user_nickname}").into();
@ -666,22 +652,21 @@ async fn produce_on_join_cmd_messages(
} }
.write_async(writer) .write_async(writer)
.await?; .await?;
let mut members: String = if let Some(head) = room_info.members.first() { let prefixed_members: Vec<PrefixedNick> = room_info
head.as_inner().clone() .members
} else { .iter()
user.nickname.clone() .map(|member| PrefixedNick::from_str(member.clone().into_inner()))
}.as_ref().into(); .collect();
for i in &room_info.members[1..] { let non_empty_members: NonEmpty<PrefixedNick> =
members.push(' '); NonEmpty::from_vec(prefixed_members).unwrap_or(nonempty![PrefixedNick::from_str(user.nickname.clone())]);
members.push_str(i.as_inner());
}
ServerMessage { ServerMessage {
tags: vec![], tags: vec![],
sender: Some(config.server_name.clone()), sender: Some(config.server_name.clone()),
body: ServerMessageBody::N353NamesReply { body: ServerMessageBody::N353NamesReply {
client: user.nickname.clone(), client: user.nickname.clone(),
chan: chan.clone(), chan: chan.clone(),
members: members.into(), members: non_empty_members.into(),
}, },
} }
.write_async(writer) .write_async(writer)
@ -708,12 +693,8 @@ pub async fn launch(
) -> Result<Terminator> { ) -> Result<Terminator> {
log::info!("Starting IRC projection"); log::info!("Starting IRC projection");
let (stopped_tx, mut stopped_rx) = channel(32); let (stopped_tx, mut stopped_rx) = channel(32);
let current_connections = let current_connections = IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
IntGauge::new("irc_current_connections", "Open and alive TCP connections")?; let total_connections = IntCounter::new("irc_total_connections", "Total number of opened connections")?;
let total_connections = IntCounter::new(
"irc_total_connections",
"Total number of opened connections",
)?;
metrics.register(Box::new(current_connections.clone()))?; metrics.register(Box::new(current_connections.clone()))?;
metrics.register(Box::new(total_connections.clone()))?; metrics.register(Box::new(total_connections.clone()))?;
@ -778,7 +759,8 @@ pub async fn launch(
log::warn!("IRC connection to {socket_addr} finished with error: {err}") log::warn!("IRC connection to {socket_addr} finished with error: {err}")
} }
} }
})).await; }))
.await;
log::info!("Stopped IRC projection"); log::info!("Stopped IRC projection");
Ok(()) Ok(())
}); });

View File

@ -0,0 +1,20 @@
[package]
name = "projection-xmpp"
edition = "2021"
version.workspace = true
[dependencies]
lavina-core.workspace = true
tracing.workspace = true
anyhow.workspace = true
serde.workspace = true
tokio.workspace = true
prometheus.workspace = true
futures-util.workspace = true
quick-xml.workspace = true
proto-xmpp = { path = "../proto-xmpp" }
uuid = { version = "1.3.0", features = ["v4"] }
tokio-rustls = "0.24.1"
rustls-pemfile = "1.0.2"
derive_more.workspace = true

View File

@ -1,3 +1,5 @@
#![feature(generators, generator_trait, type_alias_impl_trait, impl_trait_in_assoc_type)]
mod proto; mod proto;
use std::collections::HashMap; use std::collections::HashMap;
@ -7,6 +9,7 @@ use std::net::SocketAddr;
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::Arc; use std::sync::Arc;
use anyhow::anyhow;
use futures_util::future::join_all; use futures_util::future::join_all;
use prometheus::Registry as MetricsRegistry; use prometheus::Registry as MetricsRegistry;
use quick_xml::events::{BytesDecl, Event}; use quick_xml::events::{BytesDecl, Event};
@ -19,18 +22,19 @@ use tokio::sync::mpsc::channel;
use tokio_rustls::rustls::{Certificate, PrivateKey}; use tokio_rustls::rustls::{Certificate, PrivateKey};
use tokio_rustls::TlsAcceptor; use tokio_rustls::TlsAcceptor;
use crate::core::player::{PlayerConnection, PlayerId, PlayerRegistry}; use lavina_core::player::{PlayerConnection, PlayerId, PlayerRegistry};
use crate::core::room::{RoomId, RoomRegistry}; use lavina_core::prelude::*;
use crate::prelude::*; use lavina_core::room::{RoomId, RoomRegistry};
use crate::protos::xmpp; use lavina_core::terminator::Terminator;
use crate::protos::xmpp::bind::{BindResponse, Jid, Name, Resource, Server}; use lavina_core::repo::Storage;
use crate::protos::xmpp::client::{Iq, Message, MessageType, Presence}; use proto_xmpp::bind::{BindResponse, Jid, Name, Resource, Server};
use crate::protos::xmpp::disco::*; use proto_xmpp::client::{Iq, Message, MessageType, Presence};
use crate::protos::xmpp::roster::RosterQuery; use proto_xmpp::disco::*;
use crate::protos::xmpp::session::Session; use proto_xmpp::roster::RosterQuery;
use crate::protos::xmpp::stream::*; use proto_xmpp::sasl::AuthBody;
use crate::util::xml::{Continuation, FromXml, Parser, ToXml}; use proto_xmpp::session::Session;
use crate::util::Terminator; use proto_xmpp::stream::*;
use proto_xmpp::xml::{Continuation, FromXml, Parser, ToXml};
use self::proto::{ClientPacket, IqClientBody}; use self::proto::{ClientPacket, IqClientBody};
@ -58,6 +62,7 @@ pub async fn launch(
players: PlayerRegistry, players: PlayerRegistry,
rooms: RoomRegistry, rooms: RoomRegistry,
metrics: MetricsRegistry, metrics: MetricsRegistry,
storage: Storage,
) -> Result<Terminator> { ) -> Result<Terminator> {
log::info!("Starting XMPP projection"); log::info!("Starting XMPP projection");
@ -97,11 +102,12 @@ pub async fn launch(
} }
let players = players.clone(); let players = players.clone();
let rooms = rooms.clone(); let rooms = rooms.clone();
let storage = storage.clone();
let terminator = Terminator::spawn(|termination| { let terminator = Terminator::spawn(|termination| {
let stopped_tx = stopped_tx.clone(); let stopped_tx = stopped_tx.clone();
let loaded_config = loaded_config.clone(); let loaded_config = loaded_config.clone();
async move { async move {
match handle_socket(loaded_config, stream, &socket_addr, players, rooms, termination).await { match handle_socket(loaded_config, stream, &socket_addr, players, rooms, storage, termination).await {
Ok(_) => log::info!("Connection terminated"), Ok(_) => log::info!("Connection terminated"),
Err(err) => log::warn!("Connection failed: {err}"), Err(err) => log::warn!("Connection failed: {err}"),
} }
@ -118,21 +124,15 @@ pub async fn launch(
} }
} }
log::info!("Stopping XMPP projection"); log::info!("Stopping XMPP projection");
join_all( join_all(actors.into_iter().map(|(socket_addr, terminator)| async move {
actors log::debug!("Stopping XMPP connection at {socket_addr}");
.into_iter() match terminator.terminate().await {
.map(|(socket_addr, terminator)| async move { Ok(_) => log::debug!("Stopped XMPP connection at {socket_addr}"),
log::debug!("Stopping XMPP connection at {socket_addr}"); Err(err) => {
match terminator.terminate().await { log::warn!("XMPP connection to {socket_addr} finished with error: {err}")
Ok(_) => log::debug!("Stopped XMPP connection at {socket_addr}"), }
Err(err) => { }
log::warn!( }))
"XMPP connection to {socket_addr} finished with error: {err}"
)
}
}
}),
)
.await; .await;
log::info!("Stopped XMPP projection"); log::info!("Stopped XMPP projection");
Ok(()) Ok(())
@ -147,9 +147,10 @@ async fn handle_socket(
socket_addr: &SocketAddr, socket_addr: &SocketAddr,
mut players: PlayerRegistry, mut players: PlayerRegistry,
rooms: RoomRegistry, rooms: RoomRegistry,
storage: Storage,
termination: Deferred<()>, // TODO use it to stop the connection gracefully termination: Deferred<()>, // TODO use it to stop the connection gracefully
) -> Result<()> { ) -> Result<()> {
log::debug!("Received an XMPP connection from {socket_addr}"); log::info!("Received an XMPP connection from {socket_addr}");
let mut reader_buf = vec![]; let mut reader_buf = vec![];
let (reader, writer) = stream.split(); let (reader, writer) = stream.split();
let mut buf_reader = BufReader::new(reader); let mut buf_reader = BufReader::new(reader);
@ -171,10 +172,10 @@ async fn handle_socket(
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(b); let mut xml_writer = Writer::new(b);
let authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf).await?; let mut storage = storage.clone();
let mut connection = players let authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf, &mut storage).await?;
.connect_to_player(authenticated.player_id.clone()) log::debug!("User authenticated");
.await; let mut connection = players.connect_to_player(authenticated.player_id.clone()).await;
socket_final( socket_final(
&mut xml_reader, &mut xml_reader,
&mut xml_writer, &mut xml_writer,
@ -196,7 +197,7 @@ async fn socket_force_tls(
writer: &mut (impl AsyncWrite + Unpin), writer: &mut (impl AsyncWrite + Unpin),
reader_buf: &mut Vec<u8>, reader_buf: &mut Vec<u8>,
) -> Result<()> { ) -> Result<()> {
use crate::protos::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?; read_xml_header(xml_reader, reader_buf).await?;
@ -229,6 +230,7 @@ async fn socket_auth(
xml_reader: &mut NsReader<(impl AsyncBufRead + Unpin)>, xml_reader: &mut NsReader<(impl AsyncBufRead + Unpin)>,
xml_writer: &mut Writer<(impl AsyncWrite + Unpin)>, xml_writer: &mut Writer<(impl AsyncWrite + Unpin)>,
reader_buf: &mut Vec<u8>, reader_buf: &mut Vec<u8>,
storage: &mut Storage,
) -> Result<Authenticated> { ) -> Result<Authenticated> {
read_xml_header(xml_reader, reader_buf).await?; read_xml_header(xml_reader, reader_buf).await?;
let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?; let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?;
@ -253,16 +255,41 @@ async fn socket_auth(
.await?; .await?;
xml_writer.get_mut().flush().await?; xml_writer.get_mut().flush().await?;
let _ = 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?;
xmpp::sasl::Success.write_xml(xml_writer).await?; proto_xmpp::sasl::Success.write_xml(xml_writer).await?;
let name: Str = "darova".into(); let auth_body = AuthBody::from_str(auth.body);
Ok(Authenticated {
player_id: PlayerId::from("darova")?, match auth_body {
xmpp_name: Name(name.clone()), Ok(logopass) => {
xmpp_resource: Resource(name.clone()), let name = &logopass.login;
xmpp_muc_name: Resource(name), let stored_user = storage.retrieve_user_by_name(name).await?;
})
let stored_user = match stored_user {
Some(u) => u,
None => {
log::info!("User '{}' not found", name);
return Err(fail("no user found"));
}
};
if stored_user.password.is_none() {
log::info!("Password not defined for user '{}'", name);
return Err(fail("password is not defined"));
}
if stored_user.password.as_deref() != Some(&logopass.password) {
log::info!("Incorrect password supplied for user '{}'", name);
return Err(fail("passwords do not match"));
}
Ok(Authenticated {
player_id: PlayerId::from("darova")?,
xmpp_name: Name(name.to_string().into()),
xmpp_resource: Resource(name.to_string().into()),
xmpp_muc_name: Resource(name.to_string().into()),
})
}
Err(e) => Err(e)
}
} }
async fn socket_final( async fn socket_final(
@ -314,7 +341,6 @@ async fn socket_final(
match parser.consume(ns, &event) { match parser.consume(ns, &event) {
Continuation::Final(res) => { Continuation::Final(res) => {
let res = res?; let res = res?;
dbg!(&res);
let stop = handle_packet(&mut events, res, authenticated, user_handle, rooms).await?; let stop = handle_packet(&mut events, res, authenticated, user_handle, rooms).await?;
for i in &events { for i in &events {
xml_writer.write_event_async(i).await?; xml_writer.write_event_async(i).await?;
@ -333,7 +359,7 @@ async fn socket_final(
update = user_handle.receiver.recv() => { update = user_handle.receiver.recv() => {
if let Some(update) = update { if let Some(update) = update {
match update { match update {
crate::core::player::Updates::NewMessage { room_id, author_id, body } => { lavina_core::player::Updates::NewMessage { room_id, author_id, body } => {
Message { Message {
to: Some(Jid { to: Some(Jid {
name: Some(authenticated.xmpp_name.clone()), name: Some(authenticated.xmpp_name.clone()),
@ -346,7 +372,7 @@ async fn socket_final(
resource: Some(Resource(author_id.into_inner().into())), resource: Some(Resource(author_id.into_inner().into())),
}), }),
id: None, id: None,
r#type: xmpp::client::MessageType::Groupchat, r#type: proto_xmpp::client::MessageType::Groupchat,
lang: None, lang: None,
subject: None, subject: None,
body: body.into(), body: body.into(),
@ -397,10 +423,7 @@ async fn handle_packet(
{ {
if server.0.as_ref() == "rooms.localhost" && m.r#type == MessageType::Groupchat { if server.0.as_ref() == "rooms.localhost" && m.r#type == MessageType::Groupchat {
user_handle user_handle
.send_message( .send_message(RoomId::from(name.0.clone())?, m.body.clone().into())
RoomId::from(name.0.clone())?,
m.body.clone().into(),
)
.await?; .await?;
Message { Message {
to: Some(Jid { to: Some(Jid {
@ -414,7 +437,7 @@ async fn handle_packet(
resource: Some(user.xmpp_muc_name.clone()), resource: Some(user.xmpp_muc_name.clone()),
}), }),
id: m.id, id: m.id,
r#type: xmpp::client::MessageType::Groupchat, r#type: proto_xmpp::client::MessageType::Groupchat,
lang: None, lang: None,
subject: None, subject: None,
body: m.body.clone(), body: m.body.clone(),
@ -449,9 +472,7 @@ async fn handle_packet(
resource: Some(resource), resource: Some(resource),
}) = p.to }) = p.to
{ {
let a = user_handle let a = user_handle.join_room(RoomId::from(name.0.clone())?).await?;
.join_room(RoomId::from(name.0.clone())?)
.await?;
Presence::<()> { Presence::<()> {
to: Some(Jid { to: Some(Jid {
name: Some(user.xmpp_name.clone()), name: Some(user.xmpp_name.clone()),
@ -485,7 +506,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
from: None, from: None,
id: iq.id, id: iq.id,
to: None, to: None,
r#type: xmpp::client::IqType::Result, r#type: proto_xmpp::client::IqType::Result,
body: BindResponse(Jid { body: BindResponse(Jid {
name: Some(Name("darova".into())), name: Some(Name("darova".into())),
server: Server("localhost".into()), server: Server("localhost".into()),
@ -499,7 +520,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
from: None, from: None,
id: iq.id, id: iq.id,
to: None, to: None,
r#type: xmpp::client::IqType::Result, r#type: proto_xmpp::client::IqType::Result,
body: Session, body: Session,
}; };
req.serialize(output); req.serialize(output);
@ -509,7 +530,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
from: None, from: None,
id: iq.id, id: iq.id,
to: None, to: None,
r#type: xmpp::client::IqType::Result, r#type: proto_xmpp::client::IqType::Result,
body: RosterQuery, body: RosterQuery,
}; };
req.serialize(output); req.serialize(output);
@ -520,7 +541,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
from: iq.to, from: iq.to,
id: iq.id, id: iq.id,
to: None, to: None,
r#type: xmpp::client::IqType::Result, r#type: proto_xmpp::client::IqType::Result,
body: response, body: response,
}; };
req.serialize(output); req.serialize(output);
@ -531,7 +552,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
from: iq.to, from: iq.to,
id: iq.id, id: iq.id,
to: None, to: None,
r#type: xmpp::client::IqType::Result, r#type: proto_xmpp::client::IqType::Result,
body: response, body: response,
}; };
req.serialize(output); req.serialize(output);
@ -541,7 +562,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
from: None, from: None,
id: iq.id, id: iq.id,
to: None, to: None,
r#type: xmpp::client::IqType::Error, r#type: proto_xmpp::client::IqType::Error,
body: (), body: (),
}; };
req.serialize(output); req.serialize(output);
@ -634,13 +655,13 @@ async fn read_xml_header(
if &*encoding == b"UTF-8" { if &*encoding == b"UTF-8" {
Ok(()) Ok(())
} else { } else {
Err(fail(format!("Unsupported encoding: {encoding:?}").as_str())) Err(anyhow!("Unsupported encoding: {encoding:?}"))
} }
} else { } else {
// Err(fail("No XML encoding provided")) // Err(fail("No XML encoding provided"))
Ok(()) Ok(())
} }
} else { } else {
Err(fail("Expected XML header")) Err(anyhow!("Expected XML header"))
} }
} }

View File

@ -1,15 +1,15 @@
use anyhow::anyhow;
use derive_more::From; use derive_more::From;
use quick_xml::events::Event; use quick_xml::events::Event;
use quick_xml::name::{Namespace, ResolveResult}; use quick_xml::name::{Namespace, ResolveResult};
use crate::protos::xmpp::bind::BindRequest; use lavina_core::prelude::*;
use crate::protos::xmpp::client::{Iq, Message, Presence}; use proto_xmpp::bind::BindRequest;
use crate::protos::xmpp::disco::{InfoQuery, ItemQuery}; use proto_xmpp::client::{Iq, Message, Presence};
use crate::protos::xmpp::roster::RosterQuery; use proto_xmpp::disco::{InfoQuery, ItemQuery};
use crate::protos::xmpp::session::Session; use proto_xmpp::roster::RosterQuery;
use crate::util::xml::*; use proto_xmpp::session::Session;
use proto_xmpp::xml::*;
use crate::prelude::*;
#[derive(PartialEq, Eq, Debug, From)] #[derive(PartialEq, Eq, Debug, From)]
pub enum IqClientBody { pub enum IqClientBody {
@ -29,7 +29,7 @@ impl FromXml for IqClientBody {
let bytes = match event { let bytes = match event {
Event::Start(bytes) => bytes, Event::Start(bytes) => bytes,
Event::Empty(bytes) => bytes, Event::Empty(bytes) => bytes,
_ => return Err(ffail!("Unexpected XML event: {event:?}")), _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
}; };
let name = bytes.name(); let name = bytes.name();
match_parser!(name, namespace, event; match_parser!(name, namespace, event;
@ -67,7 +67,7 @@ impl FromXml for ClientPacket {
Presence::<Ignore>, Presence::<Ignore>,
Message, Message,
{ {
Err(ffail!( Err(anyhow!(
"Unexpected XML event of name {:?} in namespace {:?}", "Unexpected XML event of name {:?} in namespace {:?}",
name, name,
namespace namespace
@ -80,11 +80,11 @@ impl FromXml for ClientPacket {
if name.local_name().as_ref() == b"stream" { if name.local_name().as_ref() == b"stream" {
return Ok(ClientPacket::StreamEnd); return Ok(ClientPacket::StreamEnd);
} else { } else {
return Err(ffail!("Unexpected XML event: {event:?}")); return Err(anyhow!("Unexpected XML event: {event:?}"));
} }
} }
_ => { _ => {
return Err(ffail!("Unexpected XML event: {event:?}")); return Err(anyhow!("Unexpected XML event: {event:?}"));
} }
} }
} }

View File

@ -0,0 +1,15 @@
[package]
name = "proto-irc"
edition = "2021"
version.workspace = true
publish = false
[dependencies]
nom.workspace = true
nonempty.workspace = true
tokio.workspace = true
futures-util.workspace = true
anyhow.workspace = true
[dev-dependencies]
assert_matches.workspace = true

View File

@ -1,9 +1,13 @@
//! Client-to-Server IRC protocol. //! Client-to-Server IRC protocol.
pub mod client; pub mod client;
mod prelude;
pub mod server; pub mod server;
#[cfg(test)]
mod testkit;
pub mod user;
use std::io::Result;
use crate::prelude::Str; use crate::prelude::Str;
use std::io::Result;
use nom::{ use nom::{
branch::alt, branch::alt,
@ -100,7 +104,7 @@ mod test {
use assert_matches::*; use assert_matches::*;
use super::*; use super::*;
use crate::util::testkit::*; use crate::testkit::*;
#[test] #[test]
fn test_chan_global() { fn test_chan_global() {

View File

@ -0,0 +1,3 @@
use std::sync::Arc;
pub type Str = Arc<str>;

View File

@ -1,7 +1,9 @@
use nonempty::NonEmpty;
use tokio::io::AsyncWrite; use tokio::io::AsyncWrite;
use tokio::io::AsyncWriteExt; use tokio::io::AsyncWriteExt;
use super::*; use super::*;
use crate::user::PrefixedNick;
/// Server-to-client message. /// Server-to-client message.
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
@ -121,7 +123,7 @@ pub enum ServerMessageBody {
N353NamesReply { N353NamesReply {
client: Str, client: Str,
chan: Chan, chan: Chan,
members: Str, // TODO make this a non-empty list with prefixes members: NonEmpty<PrefixedNick>,
}, },
N366NamesReplyEnd { N366NamesReplyEnd {
client: Str, client: Str,
@ -284,7 +286,13 @@ 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?;
writer.write_all(members.as_bytes()).await?; for member in members {
writer
.write_all(member.prefix.to_string().as_bytes())
.await?;
writer.write_all(member.nick.as_bytes()).await?;
writer.write_all(b" ").await?;
}
} }
ServerMessageBody::N366NamesReplyEnd { client, chan } => { ServerMessageBody::N366NamesReplyEnd { client, chan } => {
writer.write_all(b"366 ").await?; writer.write_all(b"366 ").await?;
@ -381,7 +389,7 @@ mod test {
use assert_matches::*; use assert_matches::*;
use super::*; use super::*;
use crate::util::testkit::*; use crate::testkit::*;
#[test] #[test]
fn test_server_message_notice() { fn test_server_message_notice() {

View File

@ -1,10 +1,10 @@
use std::future::Future;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use futures_util::task::noop_waker_ref; use futures_util::task::noop_waker_ref;
use tokio::pin;
use crate::prelude::*; pub fn sync_future<T>(future: impl Future<Output = T>) -> anyhow::Result<T> {
pub fn sync_future<T>(future: impl Future<Output = T>) -> Result<T> {
let waker = noop_waker_ref(); let waker = noop_waker_ref();
let mut context = Context::from_waker(waker); let mut context = Context::from_waker(waker);
pin!(future); pin!(future);

View File

@ -0,0 +1,30 @@
use super::*;
use std::fmt;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Prefix {
Empty,
}
impl fmt::Display for Prefix {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
Prefix::Empty => write!(f, ""),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PrefixedNick {
pub prefix: Prefix,
pub nick: Str,
}
impl PrefixedNick {
pub fn from_str(nick: Str) -> PrefixedNick {
PrefixedNick {
prefix: Prefix::Empty,
nick,
}
}
}

View File

@ -0,0 +1,13 @@
[package]
name = "proto-xmpp"
edition = "2021"
version.workspace = true
[dependencies]
quick-xml.workspace = true
lazy_static.workspace = true
regex.workspace = true
anyhow.workspace = true
tokio.workspace = true
derive_more.workspace = true
base64.workspace = true

View File

@ -1,11 +1,11 @@
use std::fmt::Display; use std::fmt::Display;
use nom::AsBytes; use anyhow::{anyhow, Result};
use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event}; use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event};
use quick_xml::name::{Namespace, ResolveResult}; use quick_xml::name::{Namespace, ResolveResult};
use crate::prelude::*; use crate::prelude::*;
use crate::util::xml::*; use crate::xml::*;
pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-bind"; pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-bind";
@ -49,7 +49,7 @@ impl Jid {
} }
let m = RE let m = RE
.captures(i) .captures(i)
.ok_or(ffail!("Incorrectly format jid: {i}"))?; .ok_or(anyhow!("Incorrectly format jid: {i}"))?;
let name = m.get(2).map(|name| Name(name.as_str().into())); let name = m.get(2).map(|name| Name(name.as_str().into()));
let server = m.get(3).unwrap(); let server = m.get(3).unwrap();
@ -78,89 +78,56 @@ impl Jid {
#[derive(PartialEq, Eq, Debug)] #[derive(PartialEq, Eq, Debug)]
pub struct BindRequest(pub Resource); pub struct BindRequest(pub Resource);
pub struct BindRequestParser(BindRequestParserInner);
enum BindRequestParserInner {
Initial,
/// Consumed <bind> start and expects <resource>
InBind(Option<String>),
/// Consumed <resource> start
InBindResourceInitial,
/// Consumer <resource> start and inner text
InBindResourceEnd(String),
}
impl FromXmlTag for BindRequest { impl FromXmlTag for BindRequest {
const NS: &'static str = XMLNS; const NS: &'static str = XMLNS;
const NAME: &'static str = "bind"; const NAME: &'static str = "bind";
} }
impl FromXml for BindRequest { impl FromXml for BindRequest {
type P = BindRequestParser; type P = impl Parser<Output = Result<Self>>;
fn parse() -> Self::P { fn parse() -> Self::P {
BindRequestParser(BindRequestParserInner::Initial) |(namespace, event): (ResolveResult<'static>, &'static Event<'static>)| -> Result<Self> {
} let mut resource: Option<Str> = None;
} let Event::Start(bytes) = event else {
return Err(anyhow!("Unexpected XML event: {event:?}"));
// TODO rewrite as a generator };
impl Parser for BindRequestParser { if bytes.name().0 != BindRequest::NAME.as_bytes() {
type Output = Result<BindRequest>; return Err(anyhow!("Unexpected XML tag: {:?}", bytes.name()));
fn consume<'a>(
self: Self,
namespace: ResolveResult,
event: &Event<'a>,
) -> Continuation<Self, Self::Output> {
// TODO validate tag names and namespaces
use BindRequestParserInner::*;
match self.0 {
Initial => {
let Event::Start(bytes) = event else {
return Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}")));
};
if bytes.name().0 != BindRequest::NAME.as_bytes() {
return Continuation::Final(Err(ffail!(
"Unexpected XML tag: {:?}",
bytes.name()
)));
}
let ResolveResult::Bound(Namespace(ns)) = namespace else {
return Continuation::Final(Err(ffail!("No namespace provided")));
};
if ns != XMLNS.as_bytes() {
return Continuation::Final(Err(ffail!("Incorrect namespace")));
}
Continuation::Continue(BindRequestParser(InBind(None)))
} }
InBind(resource) => match event { let ResolveResult::Bound(Namespace(ns)) = namespace else {
Event::Start(bytes) => { return Err(anyhow!("No namespace provided"));
Continuation::Continue(BindRequestParser(InBindResourceInitial)) };
} if ns != XMLNS.as_bytes() {
Event::End(bytes) => { return Err(anyhow!("Incorrect namespace"));
let Some(resource) = resource else { }
return Continuation::Final(Err(ffail!("No resource was provided"))); loop {
let (namespace, event) = yield;
match event {
Event::Start(bytes) if bytes.name().0 == b"resource" => {
let (namespace, event) = yield;
let Event::Text(text) = event else {
return Err(anyhow!("Unexpected XML event: {event:?}"));
}; };
Continuation::Final(Ok(BindRequest(Resource(resource.into())))) resource = Some(std::str::from_utf8(&*text)?.into());
let (namespace, event) = yield;
let Event::End(bytes) = event else {
return Err(anyhow!("Unexpected XML event: {event:?}"));
};
if bytes.name().0 != b"resource" {
return Err(anyhow!("Unexpected XML tag: {:?}", bytes.name()));
}
}
Event::End(bytes) if bytes.name().0 == BindRequest::NAME.as_bytes() => {
break;
}
_ => return Err(anyhow!("Unexpected XML event: {event:?}")),
} }
_ => Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}"))),
},
InBindResourceInitial => {
let Event::Text(text) = event else {
return Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}")));
};
let resource = match std::str::from_utf8(text.as_bytes()) {
Ok(e) => e.to_string(),
Err(err) => return Continuation::Final(Err(err.into())),
};
Continuation::Continue(BindRequestParser(InBindResourceEnd(resource)))
}
InBindResourceEnd(resource) => {
let Event::End(bytes) = event else {
return Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}")));
};
Continuation::Continue(BindRequestParser(InBind(Some(resource))))
} }
let Some(resource) = resource else {
return Err(anyhow!("No resource was provided"));
};
Ok(BindRequest(Resource(resource)))
} }
} }
} }

View File

@ -3,8 +3,11 @@ use quick_xml::events::attributes::Attribute;
use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event}; use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event};
use quick_xml::name::{QName, ResolveResult}; use quick_xml::name::{QName, ResolveResult};
use anyhow::{Result, anyhow as ffail};
use crate::prelude::*; use crate::prelude::*;
use crate::util::xml::*; use crate::xml::*;
use super::bind::Jid; use super::bind::Jid;
@ -308,9 +311,9 @@ impl<T: FromXml> Parser for IqParser<T> {
} }
IqParserInner::Final(state) => { IqParserInner::Final(state) => {
if let Event::End(ref bytes) = event { if let Event::End(ref bytes) = event {
let id = fail_fast!(state.id.ok_or_else(|| fail("No id provided"))); let id = fail_fast!(state.id.ok_or_else(|| ffail!("No id provided")));
let r#type = fail_fast!(state.r#type.ok_or_else(|| fail("No type provided"))); let r#type = fail_fast!(state.r#type.ok_or_else(|| ffail!("No type provided")));
let body = fail_fast!(state.body.ok_or_else(|| fail("No body provided"))); let body = fail_fast!(state.body.ok_or_else(|| ffail!("No body provided")));
Continuation::Final(Ok(Iq { Continuation::Final(Ok(Iq {
from: state.from, from: state.from,
id, id,
@ -579,7 +582,7 @@ impl<T: ToXml> ToXml for Presence<T> {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use crate::protos::xmpp::bind::{BindRequest, Name, Resource, Server}; use crate::bind::{BindRequest, Name, Resource, Server};
use super::*; use super::*;
use quick_xml::NsReader; use quick_xml::NsReader;

View File

@ -2,8 +2,8 @@ use quick_xml::events::attributes::Attribute;
use quick_xml::events::{BytesEnd, BytesStart, Event}; use quick_xml::events::{BytesEnd, BytesStart, Event};
use quick_xml::name::{QName, ResolveResult}; use quick_xml::name::{QName, ResolveResult};
use crate::prelude::*; use anyhow::{Result, anyhow as ffail};
use crate::util::xml::*; use crate::xml::*;
use super::bind::Jid; use super::bind::Jid;

View File

@ -1,3 +1,10 @@
#![feature(
generators,
generator_trait,
type_alias_impl_trait,
impl_trait_in_assoc_type
)]
pub mod bind; pub mod bind;
pub mod client; pub mod client;
pub mod disco; pub mod disco;
@ -8,6 +15,8 @@ pub mod session;
pub mod stanzaerror; pub mod stanzaerror;
pub mod stream; pub mod stream;
pub mod tls; pub mod tls;
mod prelude;
pub mod xml;
// Implemented as a macro instead of a fn due to borrowck limitations // Implemented as a macro instead of a fn due to borrowck limitations
macro_rules! skip_text { macro_rules! skip_text {
@ -25,4 +34,4 @@ macro_rules! skip_text {
}; };
} }
pub(super) use skip_text; pub(crate) use skip_text;

View File

@ -1,8 +1,8 @@
use quick_xml::events::Event; use quick_xml::events::Event;
use quick_xml::name::ResolveResult; use quick_xml::name::ResolveResult;
use crate::prelude::*; use anyhow::{anyhow, Result};
use crate::util::xml::*; use crate::xml::*;
pub const XMLNS: &'static str = "http://jabber.org/protocol/muc"; pub const XMLNS: &'static str = "http://jabber.org/protocol/muc";
@ -22,7 +22,7 @@ impl FromXml for History {
let (bytes, end) = match event { let (bytes, end) = match event {
Event::Start(bytes) => (bytes, false), Event::Start(bytes) => (bytes, false),
Event::Empty(bytes) => (bytes, true), Event::Empty(bytes) => (bytes, true),
_ => return Err(ffail!("Unexpected XML event: {event:?}")), _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
}; };
for attr in bytes.attributes() { for attr in bytes.attributes() {
let attr = attr?; let attr = attr?;
@ -51,7 +51,7 @@ impl FromXml for History {
let (namespace, event) = yield; let (namespace, event) = yield;
let Event::End(bytes) = event else { let Event::End(bytes) = event else {
return Err(ffail!("Unexpected XML event: {event:?}")); return Err(anyhow!("Unexpected XML event: {event:?}"));
}; };
Ok(history) Ok(history)
} }
@ -73,15 +73,15 @@ impl FromXml for Password {
fn parse() -> Self::P { fn parse() -> Self::P {
|(namespace, event): (ResolveResult<'static>, &'static Event<'static>)| -> Result<Self> { |(namespace, event): (ResolveResult<'static>, &'static Event<'static>)| -> Result<Self> {
let Event::Start(bytes) = event else { let Event::Start(bytes) = event else {
return Err(ffail!("Unexpected XML event: {event:?}")); return Err(anyhow!("Unexpected XML event: {event:?}"));
}; };
let (namespace, event) = yield; let (namespace, event) = yield;
let Event::Text(bytes) = event else { let Event::Text(bytes) = event else {
return Err(ffail!("Unexpected XML event: {event:?}")); return Err(anyhow!("Unexpected XML event: {event:?}"));
}; };
let s = std::str::from_utf8(bytes)?.to_string(); let s = std::str::from_utf8(bytes)?.to_string();
let Event::End(bytes) = event else { let Event::End(bytes) = event else {
return Err(ffail!("Unexpected XML event: {event:?}")); return Err(anyhow!("Unexpected XML event: {event:?}"));
}; };
Ok(Password(s)) Ok(Password(s))
} }
@ -109,7 +109,7 @@ impl FromXml for X {
let (bytes, end) = match event { let (bytes, end) = match event {
Event::Start(bytes) => (bytes, false), Event::Start(bytes) => (bytes, false),
Event::Empty(bytes) => (bytes, true), Event::Empty(bytes) => (bytes, true),
_ => return Err(ffail!("Unexpected XML event: {event:?}")), _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
}; };
if end { if end {
return Ok(res); return Ok(res);
@ -121,7 +121,7 @@ impl FromXml for X {
Event::Start(bytes) => bytes, Event::Start(bytes) => bytes,
Event::Empty(bytes) => bytes, Event::Empty(bytes) => bytes,
Event::End(_) => break, Event::End(_) => break,
_ => return Err(ffail!("Unexpected XML event: {event:?}")), _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
}; };
if bytes.name().0 == Password::NAME.as_bytes() { if bytes.name().0 == Password::NAME.as_bytes() {
let password = delegate_parsing!(Password, namespace, event)?; let password = delegate_parsing!(Password, namespace, event)?;
@ -130,7 +130,7 @@ impl FromXml for X {
let history = delegate_parsing!(History, namespace, event)?; let history = delegate_parsing!(History, namespace, event)?;
res.history = Some(history); res.history = Some(history);
} else { } else {
return Err(ffail!("Unexpected XML event: {event:?}")); return Err(anyhow!("Unexpected XML event: {event:?}"));
} }
} }

View File

@ -0,0 +1,3 @@
use std::sync::Arc;
pub type Str = Arc<str>;

View File

@ -1,7 +1,7 @@
use quick_xml::events::{BytesStart, Event}; use quick_xml::events::{BytesStart, Event};
use crate::prelude::*; use crate::xml::*;
use crate::util::xml::*; use anyhow::{anyhow as ffail, Result};
pub const XMLNS: &'static str = "jabber:iq:roster"; pub const XMLNS: &'static str = "jabber:iq:roster";

View File

@ -5,13 +5,15 @@ use quick_xml::{
NsReader, Writer, NsReader, Writer,
}; };
use tokio::io::{AsyncBufRead, AsyncWrite}; use tokio::io::{AsyncBufRead, AsyncWrite};
use base64::{Engine as _, engine::general_purpose};
use super::skip_text; use super::skip_text;
use crate::prelude::*; use anyhow::{anyhow, Result};
pub enum Mechanism { pub enum Mechanism {
Plain, Plain,
} }
impl Mechanism { impl Mechanism {
pub fn to_str(&self) -> &'static str { pub fn to_str(&self) -> &'static str {
match self { match self {
@ -22,15 +24,33 @@ impl Mechanism {
pub fn from_str(input: &[u8]) -> Result<Mechanism> { pub fn from_str(input: &[u8]) -> Result<Mechanism> {
match input { match input {
b"PLAIN" => Ok(Mechanism::Plain), b"PLAIN" => Ok(Mechanism::Plain),
_ => Err(fail(format!("unknown auth mechanism: {input:?}").as_str())), _ => Err(anyhow!("unknown auth mechanism: {input:?}")),
} }
} }
} }
pub struct AuthBody {
pub login: String,
pub password: String,
}
impl AuthBody {
pub fn from_str(input: Vec<u8>) -> Result<AuthBody> {
let decoded_body = general_purpose::STANDARD.decode(input).unwrap();
let parsed_to_string = String::from_utf8(decoded_body).unwrap();
let separated_words: Vec<&str> = parsed_to_string.split("\x00").filter(|&part| !part.is_empty()).collect::<Vec<_>>().clone();
if &separated_words.len() == &2_usize {
Ok(AuthBody { login: separated_words[0].to_string(), password: separated_words[1].to_string() })
} else { Err(anyhow!("Incorrect auth format")) }
}
}
pub struct Auth { pub struct Auth {
pub mechanism: Mechanism, pub mechanism: Mechanism,
pub body: Vec<u8>, pub body: Vec<u8>,
} }
impl Auth { impl Auth {
pub async fn parse( pub async fn parse(
reader: &mut NsReader<impl AsyncBufRead + Unpin>, reader: &mut NsReader<impl AsyncBufRead + Unpin>,
@ -48,20 +68,20 @@ impl Auth {
if let Some(mechanism) = mechanism { if let Some(mechanism) = mechanism {
Mechanism::from_str(mechanism.borrow())? Mechanism::from_str(mechanism.borrow())?
} else { } else {
return Err(fail("expected mechanism attribute in <auth>")); return Err(anyhow!("expected mechanism attribute in <auth>"));
} }
} else { } else {
return Err(fail("expected start of <auth>")); return Err(anyhow!("expected start of <auth>"));
}; };
let body = if let Event::Text(text) = reader.read_event_into_async(buf).await? { let body = if let Event::Text(text) = reader.read_event_into_async(buf).await? {
text.into_inner().into_owned() text.into_inner().into_owned()
} else { } else {
return Err(fail("expected text body in <auth>")); return Err(anyhow!("expected text body in <auth>"));
}; };
if let Event::End(_) = reader.read_event_into_async(buf).await? { if let Event::End(_) = reader.read_event_into_async(buf).await? {
//TODO //TODO
} else { } else {
return Err(fail("expected end of <auth>")); return Err(anyhow!("expected end of <auth>"));
}; };
Ok(Auth { mechanism, body }) Ok(Auth { mechanism, body })
@ -69,6 +89,7 @@ impl Auth {
} }
pub struct Success; pub struct Success;
impl Success { impl Success {
pub async fn write_xml(&self, writer: &mut Writer<impl AsyncWrite + Unpin>) -> Result<()> { pub async fn write_xml(&self, writer: &mut Writer<impl AsyncWrite + Unpin>) -> Result<()> {
let event = BytesStart::new(r#"success xmlns="urn:ietf:params:xml:ns:xmpp-sasl""#); let event = BytesStart::new(r#"success xmlns="urn:ietf:params:xml:ns:xmpp-sasl""#);

View File

@ -1,7 +1,7 @@
use quick_xml::events::{BytesStart, Event}; use quick_xml::events::{BytesStart, Event};
use crate::prelude::*; use crate::xml::*;
use crate::util::xml::*; use anyhow::{anyhow, Result};
pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-session"; pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-session";
@ -29,11 +29,11 @@ impl Parser for SessionParser {
Continuation::Continue(SessionParser(SessionParserInner::InSession)) Continuation::Continue(SessionParser(SessionParserInner::InSession))
} }
Event::Empty(_) => Continuation::Final(Ok(Session)), Event::Empty(_) => Continuation::Final(Ok(Session)),
_ => Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}"))), _ => Continuation::Final(Err(anyhow!("Unexpected XML event: {event:?}"))),
}, },
SessionParserInner::InSession => match event { SessionParserInner::InSession => match event {
Event::End(_) => Continuation::Final(Ok(Session)), Event::End(_) => Continuation::Final(Ok(Session)),
_ => Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}"))), _ => Continuation::Final(Err(anyhow!("Unexpected XML event: {event:?}"))),
}, },
} }
} }

View File

@ -5,8 +5,9 @@ use quick_xml::{NsReader, Writer};
use tokio::io::{AsyncBufRead, AsyncWrite}; use tokio::io::{AsyncBufRead, AsyncWrite};
use super::skip_text; use super::skip_text;
use crate::prelude::*;
use crate::util::xml::ToXml; use anyhow::{anyhow, Result};
use crate::xml::ToXml;
pub static XMLNS: &'static str = "http://etherx.jabber.org/streams"; pub static XMLNS: &'static str = "http://etherx.jabber.org/streams";
pub static PREFIX: &'static str = "stream"; pub static PREFIX: &'static str = "stream";
@ -63,7 +64,6 @@ impl ClientStreamStart {
version: version.unwrap(), version: version.unwrap(),
}) })
} else { } else {
log::error!("WAT: {incoming:?}");
Err(panic!()) Err(panic!())
} }
} }

View File

@ -5,7 +5,8 @@ use quick_xml::{NsReader, Writer};
use tokio::io::{AsyncBufRead, AsyncWrite}; use tokio::io::{AsyncBufRead, AsyncWrite};
use super::skip_text; use super::skip_text;
use crate::prelude::*;
use anyhow::{anyhow, Result};
pub static XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-tls"; pub static XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-tls";
@ -21,7 +22,7 @@ impl StartTLS {
return Ok(StartTLS); return Ok(StartTLS);
} }
} }
Err(ffail!("XML tag starttls expected, received: {incoming:?}")) Err(anyhow!("XML tag starttls expected, received: {incoming:?}"))
} }
} }

View File

@ -4,7 +4,7 @@ use std::pin::Pin;
use quick_xml::events::Event; use quick_xml::events::Event;
use quick_xml::name::ResolveResult; use quick_xml::name::ResolveResult;
use crate::prelude::Result; use anyhow::{anyhow, Result};
mod ignore; mod ignore;
pub use ignore::Ignore; pub use ignore::Ignore;
@ -72,6 +72,7 @@ macro_rules! fail_fast {
}; };
} }
#[macro_export]
macro_rules! delegate_parsing { macro_rules! delegate_parsing {
($parser: ty, $namespace: expr, $event: expr) => {{ ($parser: ty, $namespace: expr, $event: expr) => {{
let mut parser = <$parser as FromXml>::parse().consume($namespace, $event); let mut parser = <$parser as FromXml>::parse().consume($namespace, $event);
@ -88,6 +89,7 @@ macro_rules! delegate_parsing {
}}; }};
} }
#[macro_export]
macro_rules! match_parser { macro_rules! match_parser {
($name: expr, $ns: expr, $event: expr; $subtype: ty, $fin: block) => { ($name: expr, $ns: expr, $event: expr; $subtype: ty, $fin: block) => {
if $name.0 == <$subtype as FromXmlTag>::NAME.as_bytes() && $ns == ResolveResult::Bound(Namespace(<$subtype as FromXmlTag>::NS.as_bytes())) { if $name.0 == <$subtype as FromXmlTag>::NAME.as_bytes() && $ns == ResolveResult::Bound(Namespace(<$subtype as FromXmlTag>::NS.as_bytes())) {
@ -105,6 +107,6 @@ macro_rules! match_parser {
}; };
} }
pub(crate) use delegate_parsing; pub use delegate_parsing;
pub(crate) use fail_fast; pub(crate) use fail_fast;
pub(crate) use match_parser; pub use match_parser;

View File

@ -17,14 +17,18 @@ allow = [
"MIT", "MIT",
"Apache-2.0", "Apache-2.0",
"ISC", "ISC",
"MPL-2.0",
"BSD-3-Clause", "BSD-3-Clause",
] ]
exceptions = [ exceptions = [
{ allow = ["Unicode-DFS-2016"], name = "unicode-ident" }, { allow = ["Unicode-DFS-2016"], name = "unicode-ident" },
{ allow = ["OpenSSL"], name = "ring" }, { allow = ["OpenSSL"], name = "ring" },
] ]
deny = ["GPL-2.0", "GPL-3.0", "AGPL-3.0"] deny = [
"GPL-2.0",
"GPL-3.0",
"AGPL-3.0",
"MPL-2.0" # it is used by webpki-roots; we do not hardcode root certs in the binary
]
copyleft = "deny" copyleft = "deny"
confidence-threshold = 0.93 confidence-threshold = 0.93
private = { ignore = true } private = { ignore = true }

11
dist/alpine3.18.Dockerfile vendored Normal file
View File

@ -0,0 +1,11 @@
FROM rust:1.72.0-alpine3.18@sha256:2f5592c561cef195c9fa4462633a674458dc375fc0ba4b80e7efe4c3c8e68403 as bld
RUN apk add --no-cache musl-dev
COPY . .
RUN cargo build --release
FROM alpine:3.18@sha256:7144f7bab3d4c2648d7e59409f15ec52a18006a128c733fcff20d3a4a54ba44a
COPY --from=bld target/release/lavina /usr/bin/lavina
VOLUME ["/etc/lavina/", "/var/lib/lavina/"]
ENTRYPOINT ["lavina", "--config", "/etc/lavina/config.toml"]

1
rustfmt.toml Normal file
View File

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

197
src/http.rs Normal file
View File

@ -0,0 +1,197 @@
use std::convert::Infallible;
use std::net::SocketAddr;
use futures_util::FutureExt;
use http_body_util::{BodyExt, Full};
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Method, Request, Response, StatusCode};
use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use lavina_core::prelude::*;
use lavina_core::repo::Storage;
use lavina_core::room::RoomRegistry;
use lavina_core::terminator::Terminator;
use mgmt_api::*;
type HttpResult<T> = std::result::Result<T, Infallible>;
#[derive(Deserialize, Debug)]
pub struct ServerConfig {
pub listen_on: SocketAddr,
}
pub async fn launch(
config: ServerConfig,
metrics: MetricsRegistry,
rooms: RoomRegistry,
storage: Storage,
) -> Result<Terminator> {
log::info!("Starting the http service");
let listener = TcpListener::bind(config.listen_on).await?;
log::debug!("Listener started");
let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, rooms, storage, rx.map(|_| ())));
Ok(terminator)
}
async fn main_loop(
listener: TcpListener,
metrics: MetricsRegistry,
rooms: RoomRegistry,
storage: Storage,
termination: impl Future<Output = ()>,
) -> Result<()> {
pin!(termination);
loop {
select! {
biased;
_ = &mut termination => break,
result = listener.accept() => {
let (stream, _) = result?;
let metrics = metrics.clone();
let rooms = rooms.clone();
let storage = storage.clone();
tokio::task::spawn(async move {
let registry = metrics.clone();
let rooms = rooms.clone();
let storage = storage.clone();
let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), rooms.clone(), storage.clone(), r)));
if let Err(err) = server.await {
tracing::error!("Error serving connection: {:?}", err);
}
});
},
}
}
log::info!("Terminating the http service");
Ok(())
}
async fn route(
registry: MetricsRegistry,
rooms: RoomRegistry,
storage: Storage,
request: Request<hyper::body::Incoming>,
) -> HttpResult<Response<Full<Bytes>>> {
let res = match (request.method(), request.uri().path()) {
(&Method::GET, "/metrics") => endpoint_metrics(registry),
(&Method::GET, "/rooms") => endpoint_rooms(rooms).await,
(&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(),
(&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, storage).await.or5xx(),
_ => not_found(),
};
Ok(res)
}
fn endpoint_metrics(registry: MetricsRegistry) -> Response<Full<Bytes>> {
let mf = registry.gather();
let mut buffer = vec![];
TextEncoder.encode(&mf, &mut buffer).expect("write to vec cannot fail");
Response::new(Full::new(Bytes::from(buffer)))
}
async fn endpoint_rooms(rooms: RoomRegistry) -> Response<Full<Bytes>> {
// TODO introduce management API types independent from core-domain types
// TODO remove `Serialize` implementations from all core-domain types
let room_list = rooms.get_all_rooms().await.to_body();
Response::new(room_list)
}
async fn endpoint_create_player(
request: Request<hyper::body::Incoming>,
mut storage: Storage,
) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else {
let payload = ErrorResponse {
code: errors::MALFORMED_REQUEST,
message: "The request payload contains incorrect JSON value",
}
.to_body();
let mut response = Response::new(payload);
*response.status_mut() = StatusCode::BAD_REQUEST;
return Ok(response);
};
storage.create_user(&res.name).await?;
log::info!("Player {} created", res.name);
let mut response = Response::new(Full::<Bytes>::default());
*response.status_mut() = StatusCode::CREATED;
Ok(response)
}
async fn endpoint_set_password(
request: Request<hyper::body::Incoming>,
mut storage: Storage,
) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else {
let payload = ErrorResponse {
code: errors::MALFORMED_REQUEST,
message: "The request payload contains incorrect JSON value",
}
.to_body();
let mut response = Response::new(payload);
*response.status_mut() = StatusCode::BAD_REQUEST;
return Ok(response);
};
let Some(_) = storage.set_password(&res.player_name, &res.password).await? else {
let payload = ErrorResponse {
code: errors::PLAYER_NOT_FOUND,
message: "No such player exists",
}
.to_body();
let mut response = Response::new(payload);
*response.status_mut() = StatusCode::UNPROCESSABLE_ENTITY;
return Ok(response);
};
log::info!("Password changed for player {}", res.player_name);
let mut response = Response::new(Full::<Bytes>::default());
*response.status_mut() = StatusCode::NO_CONTENT;
Ok(response)
}
pub fn not_found() -> Response<Full<Bytes>> {
let payload = ErrorResponse {
code: errors::INVALID_PATH,
message: "The path does not exist",
}
.to_body();
let mut response = Response::new(payload);
*response.status_mut() = StatusCode::NOT_FOUND;
response
}
trait Or5xx {
fn or5xx(self) -> Response<Full<Bytes>>;
}
impl Or5xx for Result<Response<Full<Bytes>>> {
fn or5xx(self) -> Response<Full<Bytes>> {
match self {
Ok(e) => e,
Err(e) => {
let mut response = Response::new(Full::new(e.to_string().into()));
*response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
response
}
}
}
}
trait ToBody {
fn to_body(&self) -> Full<Bytes>;
}
impl<T> ToBody for T
where
T: Serialize,
{
fn to_body(&self) -> Full<Bytes> {
let mut buffer = vec![];
serde_json::to_writer(&mut buffer, self).expect("unexpected fail when writing to vec");
Full::new(Bytes::from(buffer))
}
}

View File

@ -1,38 +1,36 @@
#![feature( mod http;
generators,
generator_trait,
type_alias_impl_trait,
impl_trait_in_assoc_type
)]
mod core;
mod prelude;
mod projections;
mod protos;
mod util;
use std::future::Future; use std::future::Future;
use std::path::Path;
use clap::Parser;
use figment::providers::Format; use figment::providers::Format;
use figment::{providers::Toml, Figment}; use figment::{providers::Toml, Figment};
use prometheus::Registry as MetricsRegistry; use prometheus::Registry as MetricsRegistry;
use serde::Deserialize; use serde::Deserialize;
use crate::core::player::PlayerRegistry; use lavina_core::player::PlayerRegistry;
use crate::core::repo::Storage; use lavina_core::prelude::*;
use crate::core::room::RoomRegistry; use lavina_core::repo::Storage;
use crate::prelude::*; use lavina_core::room::RoomRegistry;
#[derive(Deserialize, Debug)] #[derive(Deserialize, Debug)]
struct ServerConfig { struct ServerConfig {
telemetry: util::telemetry::ServerConfig, telemetry: http::ServerConfig,
irc: projections::irc::ServerConfig, irc: projection_irc::ServerConfig,
xmpp: projections::xmpp::ServerConfig, xmpp: projection_xmpp::ServerConfig,
storage: core::repo::StorageConfig, storage: lavina_core::repo::StorageConfig,
}
#[derive(Parser)]
struct CliArgs {
#[arg(long)]
config: Box<Path>,
} }
fn load_config() -> Result<ServerConfig> { fn load_config() -> Result<ServerConfig> {
let raw_config = Figment::new().merge(Toml::file("config.toml")); let args = CliArgs::parse();
let raw_config = Figment::from(Toml::file(args.config));
let config: ServerConfig = raw_config.extract()?; let config: ServerConfig = raw_config.extract()?;
Ok(config) Ok(config)
} }
@ -55,10 +53,16 @@ async fn main() -> Result<()> {
let storage = Storage::open(storage_config).await?; let storage = Storage::open(storage_config).await?;
let rooms = RoomRegistry::new(&mut metrics, storage.clone())?; let rooms = RoomRegistry::new(&mut metrics, storage.clone())?;
let mut players = PlayerRegistry::empty(rooms.clone(), &mut metrics)?; let mut players = PlayerRegistry::empty(rooms.clone(), &mut metrics)?;
let telemetry_terminator = let telemetry_terminator = http::launch(telemetry_config, metrics.clone(), rooms.clone(), storage.clone()).await?;
util::telemetry::launch(telemetry_config, metrics.clone(), rooms.clone()).await?; let irc = projection_irc::launch(
let irc = projections::irc::launch(irc_config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await?; irc_config,
let xmpp = projections::xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone()).await?; players.clone(),
rooms.clone(),
metrics.clone(),
storage.clone(),
)
.await?;
let xmpp = projection_xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await?;
tracing::info!("Started"); tracing::info!("Started");
sleep.await; sleep.await;

View File

@ -1,166 +0,0 @@
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::tcp::{ReadHalf, WriteHalf};
use tokio::net::TcpStream;
use crate::prelude::*;
struct TestScope<'a> {
reader: BufReader<ReadHalf<'a>>,
writer: WriteHalf<'a>,
buffer: Vec<u8>,
}
impl<'a> TestScope<'a> {
async fn send(&mut self, str: &(impl AsRef<str> + ?Sized)) -> Result<()> {
self.writer.write_all(str.as_ref().as_bytes()).await?;
self.writer.flush().await?;
Ok(())
}
async fn expect(&mut self, str: &(impl AsRef<str> + ?Sized)) -> Result<()> {
let len = self.reader.read_until(b'\n', &mut self.buffer).await?;
assert_eq!(std::str::from_utf8(&self.buffer[0..len])?, str.as_ref());
self.buffer.clear();
Ok(())
}
async fn assert(&mut self, f: impl FnOnce(&str) -> Result<()>) -> Result<()> {
let len = self.reader.read_until(b'\n', &mut self.buffer).await?;
let res = f(std::str::from_utf8(&self.buffer[0..len])?);
self.buffer.clear();
res
}
}
async fn init_client(stream: &mut TcpStream) -> Result<TestScope> {
let (reader, writer) = stream.split();
let reader = BufReader::new(reader);
let buffer = vec![];
Ok(TestScope {
reader,
writer,
buffer,
})
}
macro_rules! send {
($scope: expr, $($arg:tt),*) => {{
$scope.send(&format!($($arg,)*)).await?;
}};
}
macro_rules! expect {
($scope: expr, $($arg:tt),*) => {{
$scope.expect(&format!($($arg,)*)).await?;
}};
}
const SERVER_HOSTNAME: &'static str = "irc.localhost";
#[rustfmt::skip]
async fn registration(scope: &mut TestScope<'_>, nickname: &str) -> Result<()> {
expect!(scope, ":irc.localhost NOTICE * :Welcome to my server!\r\n");
send!(scope, "NICK {nickname}\r\n");
send!(scope, "USER UserName 0 * :Real Name\r\n");
expect!(scope, ":irc.localhost 001 {nickname} :Welcome to Kek Server\r\n");
expect!(scope, ":irc.localhost 002 {nickname} :Welcome to Kek Server\r\n");
expect!(scope, ":irc.localhost 003 {nickname} :Welcome to Kek Server\r\n");
expect!(scope, ":irc.localhost 004 {nickname} irc.localhost kek-0.1.alpha.3 r CFILPQbcefgijklmnopqrstvz\r\n");
expect!(scope, ":irc.localhost 005 {nickname} CHANTYPES=# :are supported by this server\r\n");
Ok(())
}
#[rustfmt::skip]
async fn join(scope: &mut TestScope<'_>, nickname: &str, chan: &str) -> Result<()> {
send!(scope, "JOIN #{chan}\r\n");
expect!(scope, ":{nickname} JOIN #{chan}\r\n");
expect!(scope, ":irc.localhost 332 {nickname} #{chan} :New room\r\n");
expect!(scope, ":irc.localhost 353 {nickname} = #{chan} :{nickname}\r\n");
expect!(scope, ":irc.localhost 366 {nickname} #{chan} :End of /NAMES list\r\n");
Ok(())
}
#[tokio::test]
async fn test_registration() -> Result<()> {
let mut stream = TcpStream::connect("127.0.0.1:6667").await?;
let mut scope = init_client(&mut stream).await?;
registration(&mut scope, "NickName1").await?;
join(&mut scope, "NickName1", "chan1").await?;
Ok(())
}
#[rustfmt::skip]
#[tokio::test]
async fn test_two_connections_one_player() -> Result<()> {
let mut stream1 = TcpStream::connect("127.0.0.1:6667").await?;
let mut scope1 = init_client(&mut stream1).await?;
let mut stream2 = TcpStream::connect("127.0.0.1:6667").await?;
let mut scope2 = init_client(&mut stream2).await?;
let nickname = "NickName2";
let chan = "chan2";
registration(&mut scope1, nickname).await?;
registration(&mut scope2, nickname).await?;
join(&mut scope1, nickname, chan).await?;
// force join on second connection when the other one joins a room
expect!(scope2, ":{nickname} JOIN #{chan}\r\n");
expect!(scope2, ":{SERVER_HOSTNAME} 332 {nickname} #{chan} :New room\r\n");
expect!(scope2, ":{SERVER_HOSTNAME} 353 {nickname} = #{chan} :{nickname}\r\n");
expect!(scope2, ":{SERVER_HOSTNAME} 366 {nickname} #{chan} :End of /NAMES list\r\n");
// force send PRIVMSG to other connections
send!(scope1, "PRIVMSG #{chan} :Chmoki vsem v etam chati!\r\n");
expect!(scope2, ":{nickname} PRIVMSG #{chan} :Chmoki vsem v etam chati!\r\n");
send!(scope2, "PRIVMSG #{chan} :I tebe privetiki\r\n");
expect!(scope1, ":{nickname} PRIVMSG #{chan} :I tebe privetiki\r\n");
let mut stream3 = TcpStream::connect("127.0.0.1:6667").await?;
let mut scope3 = init_client(&mut stream3).await?;
registration(&mut scope3, nickname).await?;
// force join on registration
expect!(scope3, ":{nickname} JOIN #{chan}\r\n");
expect!(scope3, ":{SERVER_HOSTNAME} 332 {nickname} #{chan} :New room\r\n");
expect!(scope3, ":{SERVER_HOSTNAME} 353 {nickname} = #{chan} :{nickname}\r\n");
expect!(scope3, ":{SERVER_HOSTNAME} 366 {nickname} #{chan} :End of /NAMES list\r\n");
Ok(())
}
#[rustfmt::skip]
#[tokio::test]
async fn test_two_players() -> Result<()> {
let mut stream1 = TcpStream::connect("127.0.0.1:6667").await?;
let mut scope1 = init_client(&mut stream1).await?;
let mut stream2 = TcpStream::connect("127.0.0.1:6667").await?;
let mut scope2 = init_client(&mut stream2).await?;
let nickname1 = "NickName3";
let nickname2 = "NickName4";
let chan = "chan3";
registration(&mut scope1, "NickName3").await?;
registration(&mut scope2, "NickName4").await?;
join(&mut scope1, nickname1, "chan3").await?;
send!(scope2, "JOIN #{chan}\r\n");
expect!(scope2, ":{nickname2} JOIN #{chan}\r\n");
expect!(scope2, ":irc.localhost 332 {nickname2} #{chan} :New room\r\n");
scope2.assert(|line| {
if line == format!(":irc.localhost 353 {nickname2} = #{chan} :{nickname1} {nickname2}\r\n") { return Ok(()) }
if line == format!(":irc.localhost 353 {nickname2} = #{chan} :{nickname2} {nickname1}\r\n") { return Ok(()) }
panic!("incorrect chan member list received: {line}");
}).await?;
expect!(scope2, ":irc.localhost 366 {nickname2} #{chan} :End of /NAMES list\r\n");
expect!(scope1, ":{nickname2} JOIN #{chan}\r\n");
send!(scope1, "PRIVMSG #chan3 :Chmoki vsem v etam chati!\r\n");
expect!(scope2, ":{nickname1} PRIVMSG #chan3 :Chmoki vsem v etam chati!\r\n");
send!(scope2, "PRIVMSG #chan3 :I tebe privetiki\r\n");
expect!(scope1, ":{nickname2} PRIVMSG #chan3 :I tebe privetiki\r\n");
Ok(())
}

View File

@ -1,3 +0,0 @@
//! Protocol projections — implementations of public APIs.
pub mod irc;
pub mod xmpp;

View File

@ -1,3 +0,0 @@
//! Definitions of wire protocols to be used in implementations of projections.
pub mod irc;
pub mod xmpp;

View File

@ -1,10 +0,0 @@
use std::convert::Infallible;
use http_body_util::Full;
use hyper::{body::Bytes, Response, StatusCode};
pub fn not_found() -> std::result::Result<Response<Full<Bytes>>, Infallible> {
let mut response = Response::new(Full::new(Bytes::from("404")));
*response.status_mut() = StatusCode::NOT_FOUND;
Ok(response)
}

View File

@ -1,96 +0,0 @@
use std::convert::Infallible;
use std::net::SocketAddr;
use futures_util::FutureExt;
use http_body_util::{BodyExt, Full};
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Method, Request, Response};
use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
use serde::Deserialize;
use tokio::net::TcpListener;
use crate::core::room::RoomRegistry;
use crate::prelude::*;
use crate::util::http::*;
use crate::util::Terminator;
type BoxBody = http_body_util::combinators::BoxBody<Bytes, Infallible>;
type HttpResult<T> = std::result::Result<T, Infallible>;
#[derive(Deserialize, Debug)]
pub struct ServerConfig {
pub listen_on: SocketAddr,
}
pub async fn launch(
config: ServerConfig,
metrics: MetricsRegistry,
rooms: RoomRegistry,
) -> Result<Terminator> {
log::info!("Starting the telemetry service");
let listener = TcpListener::bind(config.listen_on).await?;
log::debug!("Listener started");
let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, rooms, rx.map(|_| ())));
Ok(terminator)
}
async fn main_loop(
listener: TcpListener,
metrics: MetricsRegistry,
rooms: RoomRegistry,
termination: impl Future<Output = ()>,
) -> Result<()> {
pin!(termination);
loop {
select! {
biased;
_ = &mut termination => break,
result = listener.accept() => {
let (stream, _) = result?;
let metrics = metrics.clone();
let rooms = rooms.clone();
tokio::task::spawn(async move {
let registry = metrics.clone();
let rooms = rooms.clone();
let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), rooms.clone(), r)));
if let Err(err) = server.await {
tracing::error!("Error serving connection: {:?}", err);
}
});
},
}
}
log::info!("Terminating the telemetry service");
Ok(())
}
async fn route(
registry: MetricsRegistry,
rooms: RoomRegistry,
request: Request<hyper::body::Incoming>,
) -> std::result::Result<Response<BoxBody>, Infallible> {
match (request.method(), request.uri().path()) {
(&Method::GET, "/metrics") => Ok(endpoint_metrics(registry)?.map(BodyExt::boxed)),
(&Method::GET, "/rooms") => Ok(endpoint_rooms(rooms).await?.map(BodyExt::boxed)),
_ => Ok(not_found()?.map(BodyExt::boxed)),
}
}
fn endpoint_metrics(registry: MetricsRegistry) -> HttpResult<Response<Full<Bytes>>> {
let mf = registry.gather();
let mut buffer = vec![];
TextEncoder
.encode(&mf, &mut buffer)
.expect("write to vec cannot fail");
Ok(Response::new(Full::new(Bytes::from(buffer))))
}
async fn endpoint_rooms(rooms: RoomRegistry) -> HttpResult<Response<Full<Bytes>>> {
let room_list = rooms.get_all_rooms().await;
let mut buffer = vec![];
serde_json::to_writer(&mut buffer, &room_list).expect("unexpected fail when writing to vec");
Ok(Response::new(Full::new(Bytes::from(buffer))))
}

1
test/.gitignore vendored
View File

@ -1 +0,0 @@
*.log

View File

@ -1,6 +0,0 @@
insert into users(name)
values ('kek'), ('shrek')
returning id;
insert into challenges_plain_password(user_id, password)
values (1, 'parolchik1'), (2, 'qwerty123');

View File

@ -1,5 +0,0 @@
#!/bin/sh
cargo build
./target/debug/lavina 1>test/lavina.stdout.log 2>test/lavina.stderr.log &
cargo test
kill %1