lavina/crates/projection-irc/src/lib.rs

1026 lines
38 KiB
Rust
Raw Normal View History

use std::collections::HashMap;
2023-02-07 15:21:00 +00:00
use std::net::SocketAddr;
use anyhow::{anyhow, Result};
use futures_util::future::join_all;
use nonempty::nonempty;
use nonempty::NonEmpty;
2023-02-09 19:01:21 +00:00
use prometheus::{IntCounter, IntGauge, Registry as MetricsRegistry};
use serde::Deserialize;
2023-10-04 18:27:43 +00:00
use tokio::io::AsyncReadExt;
use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter};
2023-02-10 18:47:58 +00:00
use tokio::net::tcp::{ReadHalf, WriteHalf};
2023-02-07 15:21:00 +00:00
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::channel;
2023-02-07 15:21:00 +00:00
2023-09-30 23:12:11 +00:00
use lavina_core::player::*;
use lavina_core::prelude::*;
2023-09-30 23:12:11 +00:00
use lavina_core::repo::Storage;
use lavina_core::room::{RoomId, RoomInfo, RoomRegistry};
use lavina_core::terminator::Terminator;
use lavina_core::LavinaCore;
use proto_irc::client::CapabilitySubcommand;
use proto_irc::client::{client_message, ClientMessage};
use proto_irc::server::CapSubBody;
use proto_irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
use proto_irc::user::PrefixedNick;
use proto_irc::{Chan, Recipient};
use sasl::AuthBody;
mod cap;
use crate::cap::Capabilities;
2023-02-13 20:04:08 +00:00
pub const APP_VERSION: &str = concat!("lavina", "_", env!("CARGO_PKG_VERSION"));
2023-02-10 17:09:29 +00:00
#[derive(Deserialize, Debug, Clone)]
2023-02-07 15:21:00 +00:00
pub struct ServerConfig {
pub listen_on: SocketAddr,
2023-04-13 22:38:26 +00:00
pub server_name: Str,
2023-02-07 15:21:00 +00:00
}
2023-02-10 18:47:58 +00:00
#[derive(Debug)]
struct RegisteredUser {
2023-04-13 22:38:26 +00:00
nickname: Str,
2023-02-16 18:33:36 +00:00
/**
* Username is mostly unused in modern IRC.
2023-02-16 18:47:51 +00:00
*
* <https://stackoverflow.com/questions/31666247/what-is-the-difference-between-the-nick-username-and-real-name-in-irc-and-wha>
2023-02-16 18:33:36 +00:00
*/
2023-04-13 22:38:26 +00:00
username: Str,
realname: Str,
2023-02-10 18:47:58 +00:00
}
2023-02-07 15:21:00 +00:00
async fn handle_socket(
2023-02-10 17:09:29 +00:00
config: ServerConfig,
2023-02-07 15:21:00 +00:00
mut stream: TcpStream,
socket_addr: &SocketAddr,
mut core: LavinaCore,
termination: Deferred<()>, // TODO use it to stop the connection gracefully
2023-08-16 14:30:02 +00:00
mut storage: Storage,
) -> Result<()> {
2023-09-22 23:12:03 +00:00
log::info!("Received an IRC connection from {socket_addr}");
2023-02-07 15:21:00 +00:00
let (reader, writer) = stream.split();
2023-02-10 18:47:58 +00:00
let mut reader: BufReader<ReadHalf> = BufReader::new(reader);
2023-02-07 15:21:00 +00:00
let mut writer = BufWriter::new(writer);
pin!(termination);
select! {
biased;
_ = &mut termination =>{
log::info!("Socket handling was terminated");
return Ok(())
},
registered_user = handle_registration(&mut reader, &mut writer, &mut storage, &config) =>
match registered_user {
Ok(user) => {
log::debug!("User registered");
handle_registered_socket(config, core.players, core.rooms, &mut reader, &mut writer, user).await?;
}
Err(err) => {
log::debug!("Registration failed: {err}");
}
}
}
stream.shutdown().await?;
Ok(())
}
2023-02-07 15:21:00 +00:00
struct RegistrationState {
/// The last received `NICK` message.
future_nickname: Option<Str>,
/// The last received `USER` message.
future_username: Option<(Str, Str)>,
enabled_capabilities: Capabilities,
/// `CAP LS` or `CAP REQ` was received, but not `CAP END`.
cap_negotiation_in_progress: bool,
/// The last received `PASS` message.
pass: Option<Str>,
authentication_started: bool,
validated_user: Option<Str>,
}
2023-08-16 14:30:02 +00:00
impl RegistrationState {
fn new() -> RegistrationState {
RegistrationState {
future_nickname: None,
future_username: None,
enabled_capabilities: Capabilities::None,
cap_negotiation_in_progress: false,
pass: None,
authentication_started: false,
validated_user: None,
}
}
/// Handle an incoming message from the client during the registration process.
///
/// Returns `Some` if the user is fully registered, `None` if the registration is still in progress.
async fn handle_msg(
&mut self,
msg: ClientMessage,
writer: &mut BufWriter<WriteHalf<'_>>,
storage: &mut Storage,
config: &ServerConfig,
) -> Result<Option<RegisteredUser>> {
match msg {
ClientMessage::Pass { password } => {
self.pass = Some(password);
Ok(None)
}
ClientMessage::Capability { subcommand } => match subcommand {
CapabilitySubcommand::List { code: _ } => {
self.cap_negotiation_in_progress = true;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone().into()),
body: ServerMessageBody::Cap {
target: self.future_nickname.clone().unwrap_or_else(|| "*".into()),
subcmd: CapSubBody::Ls("sasl=PLAIN".into()),
},
2023-08-16 14:30:02 +00:00
}
.write_async(writer)
.await?;
writer.flush().await?;
Ok(None)
}
CapabilitySubcommand::Req(caps) => {
self.cap_negotiation_in_progress = true;
let mut acked = vec![];
let mut naked = vec![];
for cap in caps {
if &*cap.name == "sasl" {
if cap.to_disable {
self.enabled_capabilities &= !Capabilities::Sasl;
} else {
self.enabled_capabilities |= Capabilities::Sasl;
}
acked.push(cap);
2023-02-10 18:47:58 +00:00
} else {
naked.push(cap);
2023-02-10 18:47:58 +00:00
}
}
let mut ack_body = String::new();
for cap in acked {
if cap.to_disable {
ack_body.push('-');
2023-02-10 18:47:58 +00:00
}
ack_body += &*cap.name;
}
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone().into()),
body: ServerMessageBody::Cap {
target: self.future_nickname.clone().unwrap_or_else(|| "*".into()),
subcmd: CapSubBody::Ack(ack_body.into()),
},
}
.write_async(writer)
.await?;
writer.flush().await?;
Ok(None)
}
CapabilitySubcommand::End => {
let Some((ref username, ref realname)) = self.future_username else {
self.cap_negotiation_in_progress = false;
return Ok(None);
};
let Some(nickname) = self.future_nickname.clone() else {
self.cap_negotiation_in_progress = false;
return Ok(None);
};
let username = username.clone();
let realname = realname.clone();
let candidate_user = RegisteredUser {
nickname: nickname.clone(),
username,
realname,
};
self.finalize_auth(candidate_user, writer, storage, config).await
}
},
ClientMessage::Nick { nickname } => {
if self.cap_negotiation_in_progress {
self.future_nickname = Some(nickname);
Ok(None)
} else if let Some((username, realname)) = &self.future_username.clone() {
let candidate_user = RegisteredUser {
nickname: nickname.clone(),
username: username.clone(),
realname: realname.clone(),
};
self.finalize_auth(candidate_user, writer, storage, config).await
} else {
self.future_nickname = Some(nickname);
Ok(None)
2023-02-10 18:47:58 +00:00
}
}
ClientMessage::User { username, realname } => {
if self.cap_negotiation_in_progress {
self.future_username = Some((username, realname));
Ok(None)
} else if let Some(nickname) = self.future_nickname.clone() {
let candidate_user = RegisteredUser {
nickname: nickname.clone(),
username,
realname,
};
self.finalize_auth(candidate_user, writer, storage, config).await
} else {
self.future_username = Some((username, realname));
Ok(None)
}
}
ClientMessage::Authenticate(body) => {
if !self.authentication_started {
tracing::debug!("Received authentication request");
if &*body == "PLAIN" {
tracing::debug!("Authentication request with method PLAIN");
self.authentication_started = true;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone().into()),
body: ServerMessageBody::Authenticate("+".into()),
}
.write_async(writer)
.await?;
writer.flush().await?;
Ok(None)
} else {
let target = self.future_nickname.clone().unwrap_or_else(|| "*".into());
sasl_fail_message(config.server_name.clone(), target, "Unsupported mechanism".into())
.write_async(writer)
.await?;
writer.flush().await?;
Ok(None)
}
} else {
let body = AuthBody::from_str(body.as_bytes())?;
if let Err(e) = auth_user(storage, &body.login, &body.password).await {
tracing::warn!("Authentication failed: {:?}", e);
let target = self.future_nickname.clone().unwrap_or_else(|| "*".into());
sasl_fail_message(config.server_name.clone(), target, "Bad credentials".into())
.write_async(writer)
.await?;
writer.flush().await?;
Ok(None)
} else {
let login: Str = body.login.into();
self.validated_user = Some(login.clone());
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone().into()),
body: ServerMessageBody::N900LoggedIn {
nick: login.clone(),
address: login.clone(),
account: login.clone(),
message: format!("You are now logged in as {}", login).into(),
},
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone().into()),
body: ServerMessageBody::N903SaslSuccess {
nick: login.clone(),
message: "SASL authentication successful".into(),
},
}
.write_async(writer)
.await?;
writer.flush().await?;
Ok(None)
}
}
// TODO handle abortion of authentication
2023-02-10 18:47:58 +00:00
}
_ => Ok(None),
}
}
async fn finalize_auth(
&mut self,
candidate_user: RegisteredUser,
writer: &mut BufWriter<WriteHalf<'_>>,
storage: &mut Storage,
config: &ServerConfig,
) -> Result<Option<RegisteredUser>> {
if self.enabled_capabilities.contains(Capabilities::Sasl)
&& self.validated_user.as_ref() == Some(&candidate_user.nickname)
{
Ok(Some(candidate_user))
} else {
let Some(candidate_password) = &self.pass else {
sasl_fail_message(
config.server_name.clone(),
candidate_user.nickname.clone(),
"User credentials was not provided".into(),
)
.write_async(writer)
.await?;
writer.flush().await?;
return Ok(None);
};
auth_user(storage, &*candidate_user.nickname, &*candidate_password).await?;
Ok(Some(candidate_user))
}
}
}
async fn handle_registration<'a>(
reader: &mut BufReader<ReadHalf<'a>>,
writer: &mut BufWriter<WriteHalf<'a>>,
storage: &mut Storage,
config: &ServerConfig,
) -> Result<RegisteredUser> {
let mut buffer = vec![];
let mut state = RegistrationState::new();
let user = loop {
let res = read_irc_message(reader, &mut buffer).await;
tracing::trace!("Received message: {:?}", res);
let len = match res {
Ok(len) => len,
Err(err) => {
log::warn!("Failed to read from socket: {err}");
break Err(err.into());
}
};
if len == 0 {
log::info!("Terminating socket");
break Err(anyhow::Error::msg("EOF"));
}
let res = match std::str::from_utf8(&buffer[..len - 2]) {
Ok(res) => res,
Err(e) => break Err(e.into()),
};
tracing::trace!("Incoming raw IRC message: '{res}'");
let parsed = client_message(res);
let msg = match parsed {
Ok(msg) => msg,
Err(err) => {
tracing::warn!("Failed to parse IRC message: {err}");
buffer.clear();
continue;
}
};
tracing::debug!("Incoming IRC message: {msg:?}");
if let Some(user) = state.handle_msg(msg, writer, storage, config).await? {
break Ok(user);
2023-02-10 18:47:58 +00:00
}
buffer.clear();
2023-08-16 14:30:02 +00:00
}?;
// TODO properly implement session temination
Ok(user)
}
2023-08-16 14:30:02 +00:00
fn sasl_fail_message(sender: Str, nick: Str, text: Str) -> ServerMessage {
ServerMessage {
tags: vec![],
sender: Some(sender),
body: ServerMessageBody::N904SaslFail { nick, text },
}
}
async fn auth_user(storage: &mut Storage, login: &str, plain_password: &str) -> Result<()> {
let stored_user = storage.retrieve_user_by_name(login).await?;
2023-08-16 14:30:02 +00:00
let stored_user = match stored_user {
Some(u) => u,
None => {
log::info!("User '{}' not found", login);
return Err(anyhow!("no user found"));
2023-08-16 14:30:02 +00:00
}
};
let Some(expected_password) = stored_user.password else {
log::info!("Password not defined for user '{}'", login);
return Err(anyhow!("password is not defined"));
};
if expected_password != plain_password {
log::info!("Incorrect password supplied for user '{}'", login);
return Err(anyhow!("passwords do not match"));
2023-08-16 14:30:02 +00:00
}
Ok(())
2023-02-10 18:47:58 +00:00
}
async fn handle_registered_socket<'a>(
config: ServerConfig,
mut players: PlayerRegistry,
rooms: RoomRegistry,
reader: &mut BufReader<ReadHalf<'a>>,
writer: &mut BufWriter<WriteHalf<'a>>,
2023-02-10 18:47:58 +00:00
user: RegisteredUser,
) -> Result<()> {
2023-02-10 18:47:58 +00:00
let mut buffer = vec![];
log::info!("Handling registered user: {user:?}");
2023-04-13 22:38:26 +00:00
let player_id = PlayerId::from(user.nickname.clone())?;
let mut connection = players.connect_to_player(&player_id).await;
let text: Str = format!("Welcome to {} Server", &config.server_name).into();
2023-02-12 23:31:16 +00:00
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N001Welcome {
client: user.nickname.clone(),
text: text.clone(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N002YourHost {
client: user.nickname.clone(),
text: text.clone(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N003Created {
client: user.nickname.clone(),
text: text.clone(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N004MyInfo {
client: user.nickname.clone(),
2023-04-13 19:15:48 +00:00
hostname: config.server_name.clone(),
softname: APP_VERSION.into(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N005ISupport {
client: user.nickname.clone(),
2023-04-13 19:15:48 +00:00
params: "CHANTYPES=#".into(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
2023-02-14 18:28:49 +00:00
2023-02-15 20:49:52 +00:00
let rooms_list = connection.get_rooms().await?;
2023-02-14 18:28:49 +00:00
for room in &rooms_list {
produce_on_join_cmd_messages(&config, &user, &Chan::Global(room.id.as_inner().clone()), room, writer).await?;
2023-02-14 18:28:49 +00:00
}
writer.flush().await?;
2023-02-10 18:47:58 +00:00
2023-02-07 15:21:00 +00:00
loop {
select! {
biased;
2023-10-04 18:27:43 +00:00
len = read_irc_message(reader, &mut buffer) => {
let len = len?;
let len = if len == 0 {
log::info!("EOF, Terminating socket");
break;
} else {
len
};
2023-10-04 18:27:43 +00:00
let incoming = std::str::from_utf8(&buffer[0..len-2])?;
2023-04-13 19:15:48 +00:00
if let HandleResult::Leave = handle_incoming_message(incoming, &config, &user, &rooms, &mut connection, writer).await? {
2023-02-16 23:38:34 +00:00
break;
}
2023-02-07 15:21:00 +00:00
buffer.clear();
},
2023-02-13 19:16:00 +00:00
update = connection.receiver.recv() => {
if let Some(update) = update {
handle_update(&config, &user, &player_id, writer, &rooms, update).await?;
} else {
log::warn!("Player is terminated, must terminate the connection");
break;
2023-02-13 17:08:37 +00:00
}
}
2023-02-07 15:21:00 +00:00
}
}
2023-02-16 23:38:34 +00:00
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::Error {
2023-04-13 19:15:48 +00:00
reason: "Leaving the server".into(),
},
2023-02-16 23:38:34 +00:00
}
.write_async(writer)
.await?;
writer.flush().await?;
2023-02-15 16:47:48 +00:00
connection.terminate().await;
Ok(())
2023-02-07 15:21:00 +00:00
}
2023-10-09 11:35:41 +00:00
// TODO this is public only for tests, perhaps move this into proto-irc
// TODO limit buffer size in size to protect against dos attacks with large payloads
pub async fn read_irc_message(reader: &mut BufReader<ReadHalf<'_>>, buf: &mut Vec<u8>) -> Result<usize> {
2023-10-04 18:27:43 +00:00
let mut size = 0;
'outer: loop {
let res = reader.read_until(b'\r', buf).await?;
size += res;
let next = reader.read_u8().await?;
buf.push(next);
size += 1;
2023-10-09 11:35:41 +00:00
if next != b'\n' {
continue 'outer;
2023-10-04 18:27:43 +00:00
}
return Ok(size);
}
}
async fn handle_update(
config: &ServerConfig,
user: &RegisteredUser,
player_id: &PlayerId,
writer: &mut (impl AsyncWrite + Unpin),
rooms: &RoomRegistry,
update: Updates,
) -> Result<()> {
2023-07-22 14:22:49 +00:00
log::debug!("Sending irc message to player {player_id:?} on update {update:?}");
match update {
Updates::RoomJoined { new_member_id, room_id } => {
if player_id == &new_member_id {
2023-08-17 13:41:28 +00:00
if let Some(room) = rooms.get_room(&room_id).await {
let room_info = room.get_room_info().await;
2023-04-13 22:38:26 +00:00
let chan = Chan::Global(room_id.as_inner().clone());
produce_on_join_cmd_messages(&config, &user, &chan, &room_info, writer).await?;
writer.flush().await?;
} else {
log::warn!("Received join to a non-existant room");
}
} else {
ServerMessage {
tags: vec![],
2023-04-13 22:38:26 +00:00
sender: Some(new_member_id.as_inner().clone()),
body: ServerMessageBody::Join(Chan::Global(room_id.as_inner().clone())),
}
.write_async(writer)
.await?;
writer.flush().await?
}
}
Updates::RoomLeft {
room_id,
former_member_id,
} => {
ServerMessage {
tags: vec![],
2023-04-13 22:38:26 +00:00
sender: Some(former_member_id.as_inner().clone()),
body: ServerMessageBody::Part(Chan::Global(room_id.as_inner().clone())),
}
.write_async(writer)
.await?;
writer.flush().await?
}
Updates::NewMessage {
author_id,
room_id,
body,
} => {
ServerMessage {
tags: vec![],
2023-04-13 22:38:26 +00:00
sender: Some(author_id.as_inner().clone()),
body: ServerMessageBody::PrivateMessage {
2023-04-13 22:38:26 +00:00
target: Recipient::Chan(Chan::Global(room_id.as_inner().clone())),
2023-04-13 19:15:48 +00:00
body: body.clone(),
},
}
.write_async(writer)
.await?;
writer.flush().await?
}
Updates::RoomTopicChanged { room_id, new_topic } => {
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N332Topic {
client: user.nickname.clone(),
2023-04-13 22:38:26 +00:00
chat: Chan::Global(room_id.as_inner().clone()),
topic: new_topic,
},
}
.write_async(writer)
.await?;
writer.flush().await?
}
2023-02-16 21:49:17 +00:00
Updates::BannedFrom(room_id) => {
// TODO think about the case when the user was banned, but was not in the room - no need to send PART in this case
ServerMessage {
tags: vec![],
2023-04-13 22:38:26 +00:00
sender: Some(player_id.as_inner().clone()),
body: ServerMessageBody::Part(Chan::Global(room_id.as_inner().clone())),
2023-02-16 21:49:17 +00:00
}
.write_async(writer)
.await?;
writer.flush().await?
}
}
Ok(())
}
2023-02-16 23:38:34 +00:00
enum HandleResult {
Continue,
Leave,
}
async fn handle_incoming_message(
2023-04-13 19:15:48 +00:00
buffer: &str,
config: &ServerConfig,
user: &RegisteredUser,
2023-02-16 18:47:51 +00:00
rooms: &RoomRegistry,
2023-02-13 19:16:00 +00:00
user_handle: &mut PlayerConnection,
writer: &mut (impl AsyncWrite + Unpin),
2023-02-16 23:38:34 +00:00
) -> Result<HandleResult> {
2023-07-03 20:25:57 +00:00
log::debug!("Incoming raw IRC message: '{buffer}'");
let parsed = client_message(buffer);
2023-07-03 20:25:57 +00:00
log::debug!("Incoming IRC message: {parsed:?}");
match parsed {
2023-10-04 18:27:43 +00:00
Ok(msg) => match msg {
ClientMessage::Ping { token } => {
ServerMessage {
tags: vec![],
2023-07-22 14:22:49 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::Pong {
2023-04-13 19:15:48 +00:00
from: config.server_name.clone(),
token,
},
}
.write_async(writer)
.await?;
writer.flush().await?;
}
ClientMessage::Join(ref chan) => {
handle_join(&config, &user, user_handle, chan, writer).await?;
}
2023-02-15 17:54:48 +00:00
ClientMessage::Part { chan, message } => {
handle_part(config, user, user_handle, &chan, writer).await?;
}
ClientMessage::PrivateMessage { recipient, body } => match recipient {
2023-04-13 19:15:48 +00:00
Recipient::Chan(Chan::Global(chan)) => {
2023-04-13 22:38:26 +00:00
let room_id = RoomId::from(chan)?;
2023-04-13 19:15:48 +00:00
user_handle.send_message(room_id, body).await?;
}
_ => log::warn!("Unsupported target type"),
},
2023-02-14 18:46:42 +00:00
ClientMessage::Topic { chan, topic } => {
match chan {
2023-02-14 19:07:07 +00:00
Chan::Global(chan) => {
2023-04-13 22:38:26 +00:00
let room_id = RoomId::from(chan)?;
user_handle.change_topic(room_id.clone(), topic.clone()).await?;
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N332Topic {
client: user.nickname.clone(),
2023-04-13 22:38:26 +00:00
chat: Chan::Global(room_id.as_inner().clone()),
topic,
},
2023-02-14 18:46:42 +00:00
}
.write_async(writer)
.await?;
writer.flush().await?;
2023-02-14 18:46:42 +00:00
}
Chan::Local(_) => {}
};
}
ClientMessage::Who { target } => match &target {
Recipient::Nick(nick) => {
2023-02-16 18:33:36 +00:00
// TODO handle non-existing user
2023-04-13 19:15:48 +00:00
let mut username = format!("~{nick}");
let mut host = format!("user/{nick}");
2023-02-16 18:33:36 +00:00
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
2023-02-16 18:47:51 +00:00
body: user_to_who_msg(config, user, nick),
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
2023-02-16 18:47:51 +00:00
body: ServerMessageBody::N315EndOfWho {
2023-02-16 18:33:36 +00:00
client: user.nickname.clone(),
2023-02-16 18:47:51 +00:00
mask: target.clone(),
2023-04-13 19:15:48 +00:00
msg: "End of WHO list".into(),
2023-02-16 18:33:36 +00:00
},
}
2023-02-16 18:33:36 +00:00
.write_async(writer)
.await?;
2023-02-16 18:47:51 +00:00
writer.flush().await?;
}
Recipient::Chan(Chan::Global(chan)) => {
2023-08-17 13:41:28 +00:00
let room = rooms.get_room(&RoomId::from(chan.clone())?).await;
2023-02-16 18:47:51 +00:00
if let Some(room) = room {
let room_info = room.get_room_info().await;
for member in room_info.members {
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
2023-04-13 22:38:26 +00:00
body: user_to_who_msg(config, user, member.as_inner()),
2023-02-16 18:47:51 +00:00
}
.write_async(writer)
.await?;
}
}
2023-02-16 18:33:36 +00:00
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
2023-02-16 18:33:36 +00:00
body: ServerMessageBody::N315EndOfWho {
client: user.nickname.clone(),
mask: target.clone(),
2023-04-13 19:15:48 +00:00
msg: "End of WHO list".into(),
2023-02-16 18:33:36 +00:00
},
}
.write_async(writer)
.await?;
writer.flush().await?;
}
2023-02-16 18:47:51 +00:00
Recipient::Chan(Chan::Local(_)) => {
log::warn!("Local chans not supported");
}
},
2023-02-16 19:37:17 +00:00
ClientMessage::Mode { target } => {
match target {
Recipient::Nick(nickname) => {
if nickname == user.nickname {
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
2023-02-16 21:49:17 +00:00
body: ServerMessageBody::N221UserModeIs {
client: user.nickname.clone(),
2023-04-13 19:15:48 +00:00
modes: "+r".into(),
2023-02-16 21:49:17 +00:00
},
2023-02-16 19:37:17 +00:00
}
.write_async(writer)
.await?;
writer.flush().await?;
} else {
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N502UsersDontMatch {
client: user.nickname.clone(),
message: "Cant change mode for other users".into(),
},
}
.write_async(writer)
.await?;
writer.flush().await?;
2023-02-16 19:37:17 +00:00
}
2023-02-16 21:49:17 +00:00
}
2023-02-16 19:37:17 +00:00
Recipient::Chan(_) => {
// TODO handle chan mode handling
2023-02-16 21:49:17 +00:00
}
2023-02-16 19:37:17 +00:00
}
2023-02-16 21:49:17 +00:00
}
2023-02-16 23:38:34 +00:00
ClientMessage::Quit { reason } => {
log::info!("Received QUIT");
return Ok(HandleResult::Leave);
}
2023-02-14 19:56:31 +00:00
cmd => {
log::warn!("Not implemented handler for client command: {cmd:?}");
2023-02-14 18:46:42 +00:00
}
},
Err(err) => {
log::warn!("Failed to parse IRC message: {err}");
}
}
2023-02-16 23:38:34 +00:00
Ok(HandleResult::Continue)
}
fn user_to_who_msg(config: &ServerConfig, requestor: &RegisteredUser, target_user_nickname: &Str) -> ServerMessageBody {
2023-02-16 18:47:51 +00:00
// Username is equal to nickname
2023-04-13 19:15:48 +00:00
let username = format!("~{target_user_nickname}").into();
2023-02-16 18:47:51 +00:00
// User's host is not public, replace it with `user/<nickname>` pattern
2023-04-13 19:15:48 +00:00
let mut host = format!("user/{target_user_nickname}").into();
2023-02-16 18:47:51 +00:00
ServerMessageBody::N352WhoReply {
client: requestor.nickname.clone(),
username,
host,
2023-04-13 19:15:48 +00:00
server: config.server_name.clone(),
2023-02-16 18:47:51 +00:00
flags: AwayStatus::Here,
nickname: target_user_nickname.clone(),
hops: 0,
// TODO Realname is not available yet, should be matched to a future core's player field
realname: target_user_nickname.clone(),
}
}
async fn handle_join(
config: &ServerConfig,
user: &RegisteredUser,
2023-02-13 19:16:00 +00:00
user_handle: &mut PlayerConnection,
chan: &Chan,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
match chan {
Chan::Global(chan_name) => {
2023-04-13 22:38:26 +00:00
let room_id = RoomId::from(chan_name.clone())?;
2023-02-16 21:49:17 +00:00
if let JoinResult::Success(room_info) = user_handle.join_room(room_id).await? {
produce_on_join_cmd_messages(&config, &user, chan, &room_info, writer).await?;
} else {
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
2023-02-16 21:49:17 +00:00
body: ServerMessageBody::N474BannedFromChan {
client: user.nickname.clone(),
chan: chan.clone(),
2023-04-13 19:15:48 +00:00
message: "U dun goofed".into(),
2023-02-16 21:49:17 +00:00
},
}
.write_async(writer)
.await?;
}
writer.flush().await?;
}
Chan::Local(_) => {}
};
Ok(())
}
2023-02-15 17:54:48 +00:00
async fn handle_part(
config: &ServerConfig,
user: &RegisteredUser,
user_handle: &mut PlayerConnection,
chan: &Chan,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
if let Chan::Global(chan_name) = chan {
2023-04-13 22:38:26 +00:00
let room_id = RoomId::from(chan_name.clone())?;
2023-02-15 17:54:48 +00:00
user_handle.leave_room(room_id).await?;
ServerMessage {
tags: vec![],
sender: Some(user.nickname.clone()),
body: ServerMessageBody::Part(Chan::Global(chan_name.clone())),
}
.write_async(writer)
.await?;
writer.flush().await?;
} else {
log::warn!("Local chans unsupported");
}
Ok(())
}
async fn produce_on_join_cmd_messages(
config: &ServerConfig,
user: &RegisteredUser,
chan: &Chan,
room_info: &RoomInfo,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
ServerMessage {
tags: vec![],
sender: Some(user.nickname.clone()),
body: ServerMessageBody::Join(chan.clone()),
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N332Topic {
client: user.nickname.clone(),
chat: chan.clone(),
topic: room_info.topic.clone(),
},
}
.write_async(writer)
.await?;
2023-10-09 11:35:41 +00:00
let prefixed_members: Vec<PrefixedNick> =
room_info.members.iter().map(|member| PrefixedNick::from_str(member.clone().into_inner())).collect();
let non_empty_members: NonEmpty<PrefixedNick> =
NonEmpty::from_vec(prefixed_members).unwrap_or(nonempty![PrefixedNick::from_str(user.nickname.clone())]);
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N353NamesReply {
client: user.nickname.clone(),
chan: chan.clone(),
members: non_empty_members.into(),
},
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
2023-04-13 19:15:48 +00:00
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N366NamesReplyEnd {
client: user.nickname.clone(),
chan: chan.clone(),
},
}
.write_async(writer)
.await?;
Ok(())
}
2023-10-09 11:35:41 +00:00
pub struct RunningServer {
pub addr: SocketAddr,
terminator: Terminator,
}
impl RunningServer {
pub async fn terminate(self) -> Result<()> {
self.terminator.terminate().await
}
}
2023-02-09 19:01:21 +00:00
pub async fn launch(
2023-02-10 17:09:29 +00:00
config: ServerConfig,
core: LavinaCore,
2023-02-09 19:01:21 +00:00
metrics: MetricsRegistry,
2023-08-16 14:30:02 +00:00
storage: Storage,
2023-10-09 11:35:41 +00:00
) -> Result<RunningServer> {
2023-02-07 15:21:00 +00:00
log::info!("Starting IRC projection");
let (stopped_tx, mut stopped_rx) = channel(32);
let current_connections = IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
let total_connections = IntCounter::new("irc_total_connections", "Total number of opened connections")?;
2023-02-09 19:01:21 +00:00
metrics.register(Box::new(current_connections.clone()))?;
metrics.register(Box::new(total_connections.clone()))?;
2023-02-07 15:21:00 +00:00
let listener = TcpListener::bind(config.listen_on).await?;
2023-10-09 11:35:41 +00:00
let addr = listener.local_addr()?;
2023-02-07 15:21:00 +00:00
2023-02-22 15:05:28 +00:00
let terminator = Terminator::spawn(|mut rx| async move {
// TODO probably should separate logic for accepting new connection and storing them
// into two tasks so that they don't block each other
let mut actors = HashMap::new();
2023-02-07 15:21:00 +00:00
loop {
select! {
biased;
_ = &mut rx => break,
stopped = stopped_rx.recv() => match stopped {
Some(stopped) => { let _ = actors.remove(&stopped); },
None => unreachable!(),
},
2023-02-07 15:21:00 +00:00
new_conn = listener.accept() => {
match new_conn {
Ok((stream, socket_addr)) => {
2023-02-10 17:09:29 +00:00
let config = config.clone();
2023-02-09 19:01:21 +00:00
total_connections.inc();
current_connections.inc();
2023-02-07 15:21:00 +00:00
log::debug!("Incoming connection from {socket_addr}");
if actors.contains_key(&socket_addr) {
log::warn!("Already contains connection form {socket_addr}");
// TODO kill the older connection and restart it
continue;
}
2023-02-22 15:05:28 +00:00
let terminator = Terminator::spawn(|termination| {
let core = core.clone();
2023-02-22 15:05:28 +00:00
let current_connections_clone = current_connections.clone();
let stopped_tx = stopped_tx.clone();
2023-08-16 14:30:02 +00:00
let storage = storage.clone();
2023-02-22 15:05:28 +00:00
async move {
match handle_socket(config, stream, &socket_addr, core, termination, storage).await {
2023-02-22 15:05:28 +00:00
Ok(_) => log::info!("Connection terminated"),
Err(err) => log::warn!("Connection failed: {err}"),
}
current_connections_clone.dec();
stopped_tx.send(socket_addr).await?;
Ok(())
}
});
actors.insert(socket_addr, terminator);
2023-02-07 15:21:00 +00:00
},
Err(err) => log::warn!("Failed to accept new connection: {err}"),
}
},
}
}
log::info!("Stopping IRC projection");
join_all(actors.into_iter().map(|(socket_addr, terminator)| async move {
log::debug!("Stopping IRC connection at {socket_addr}");
match terminator.terminate().await {
Ok(_) => log::debug!("Stopped IRC connection at {socket_addr}"),
Err(err) => {
log::warn!("IRC connection to {socket_addr} finished with error: {err}")
}
}
}))
.await;
log::info!("Stopped IRC projection");
2023-02-07 15:21:00 +00:00
Ok(())
});
log::info!("Started IRC projection");
2023-10-09 11:35:41 +00:00
Ok(RunningServer { addr, terminator })
2023-02-07 15:21:00 +00:00
}