use std::collections::{HashSet, HashMap}; use async_trait::async_trait; use log::{debug, warn}; use serde::Deserialize; use tokio::{net::TcpStream, io::{BufReader, BufWriter, AsyncWrite}, select}; use crate::{supervisor::{Id, TaskResult, Sender, Task, Receiver}, linkmap::Linkmap}; use crate::supervisor::Message as BMessage; use super::{IrcSpConfig, message::{Message, Command}, network::{MessageReader, MessageWriter}}; #[derive(Debug, Deserialize)] pub struct IrcSpTask { config: IrcSpConfig, links: Linkmap, } #[async_trait] impl Task for IrcSpTask { async fn start(&self, id: Id, tx: Sender, mut rx: Receiver) -> TaskResult { let stream = TcpStream::connect((self.config.peer_host.as_ref(), self.config.peer_port)).await?; let (reader, writer) = stream.into_split(); let mut reader = MessageReader::new(BufReader::new(reader)); let mut ctx = Context { id, links: &self.links, cfg: &self.config, writer: MessageWriter::new(BufWriter::new(writer)), tx, servers: HashSet::from([self.config.server_name.clone()]), nicks: HashSet::from([self.config.user_nick.clone()]), local_users: HashSet::from([self.config.user_nick.clone()]), }; debug!("{id}: identifying"); ctx.writer.write_msg(Message::new(Command::PASS, vec![&ctx.cfg.peer_password, "0210", "abridged|"])).await?; ctx.writer.write_msg(Message::new(Command::SERVER, vec![&ctx.cfg.server_name, ":Abridged"])).await?; ctx.writer.flush().await?; debug!("{id}: checking peer identification"); let passmsg = reader.read_message().await?; if passmsg.command != Command::PASS { return Err("Invalid handshake, expected PASS".into()) } if passmsg.params.first() != Some(&ctx.cfg.password.as_str()) { return Err("Configured local password does not match password from peer".into()) } ctx.writer.write_msg(Message::with_prefix( &ctx.cfg.server_name, Command::NICK, vec![&ctx.cfg.user_nick, "0", &ctx.cfg.user_nick, "server", "1", "+", &ctx.cfg.user_nick] )).await?; for channel in self.links.iter_channels() { ctx.writer.write_msg(Message::with_prefix( &ctx.cfg.user_nick, Command::JOIN, vec![channel], )).await?; } ctx.writer.flush().await?; debug!("{id}: entering event loop"); loop { select! { msg = reader.read_message() => handle_irc_msg(msg?, &mut ctx).await?, msg = rx.recv() => handle_bridge_msg(msg?, &mut ctx).await?, }} } } struct Context<'a, W: AsyncWrite + Unpin> { id: Id, links: &'a Linkmap, cfg: &'a IrcSpConfig, writer: MessageWriter, tx: Sender, servers: HashSet, nicks: HashSet, local_users: HashMap<(String, &str), String>, } const MALFORMED: &str = "Malformed IRC message"; async fn handle_bridge_msg(msg: BMessage, ctx: &mut Context<'_, impl AsyncWrite + Unpin>) -> TaskResult { let key = &(msg.author, msg.suffix.as_ref()); let nick = match ctx.local_users.get(&key) { Some(nick) => nick, None => { let nick = msg.author + msg.suffix.as_ref(); ctx.writer.write_msg(Message::with_prefix( &ctx.cfg.server_name, Command::NICK, vec![&nick, "0", &nick, "server", "1", "+", &nick] )).await?; ctx.local_users.insert(todo!(), nick.clone()); ctx.nicks.insert(nick.clone()); &nick } }; for channel in ctx.links.get_channels(&msg.link) { if msg.origin.0 == ctx.id && &msg.origin.1 == channel { continue } ctx.writer.write_msg(Message::with_prefix( &nick, Command::PRIVMSG, vec![channel, &msg.content] )).await?; ctx.writer.flush().await?; } Ok(()) } async fn handle_irc_msg<'a>(msg: Message<'a>, ctx: &mut Context<'_, impl AsyncWrite + Unpin>) -> TaskResult { debug!("irc message: {:?}", msg); match msg.command { Command::PASS => todo!(), Command::SERVER => { let name = *msg.params.first().ok_or(MALFORMED)?; debug!("linked server {}", name); ctx.servers.insert(name.to_owned()); } Command::NICK => { let prefix = msg.prefix.ok_or(MALFORMED)?; let name = *msg.params.first().ok_or(MALFORMED)?; if ctx.nicks.contains(prefix) { ctx.nicks.remove(prefix); debug!("updated nick for {} to {}", prefix, name); } else { debug!("connecting user {}", name); } ctx.nicks.insert(name.to_owned()); }, Command::PING => { let prefix = msg.prefix.ok_or(MALFORMED)?; let params = match msg.params.first() { Some(m) => vec![prefix, m], None => vec![prefix], }; ctx.writer.write_msg(Message::with_prefix(&ctx.cfg.server_name, Command::PONG, params)).await?; ctx.writer.flush().await?; } Command::PRIVMSG => { let channel = msg.params.first().ok_or(MALFORMED)?.to_string(); if let Some(link) = ctx.links.get_link(&channel) { let author = msg.prefix.ok_or(MALFORMED)?.to_owned(); let content = msg.params.get(1).ok_or(MALFORMED)?.to_string(); ctx.tx.send(BMessage { origin: (ctx.id, channel.clone()), link: link.clone(), author, suffix: ctx.cfg.suffix.clone(), content, avatar: None, })?; } } Command::ERROR => { warn!("Error: {:?}", msg.params.first()); } _ => (), } Ok(()) } /* + PASS A_pitowoVsjRgZ 0210 proxy| + SERVER c.trimill.xyz :Server Info Text - :trimill.xyz PASS C_kTNGh3SlfpJ5 0210 proxy| - :trimill.xyz SERVER trimill.xyz 1 :Server Info Text - :trimill.xyz NICK trimill 1 trimill server 1 + :trimill - :trimill.xyz NJOIN #test2 :@trimill - :trimill.xyz NJOIN #test :@trimill - :trimill.xyz PING :trimill.xyz + :c.trimill.xyz PING :c.trimill.xyz + :c.trimill.xyz PONG trimill.xyz :trimill.xyz - :trimill.xyz PONG c.trimill.xyz :c.trimill.xyz */