diff --git a/programs/Cargo.toml b/programs/Cargo.toml index 07cd3e2..b72c495 100644 --- a/programs/Cargo.toml +++ b/programs/Cargo.toml @@ -9,7 +9,7 @@ publish = false [dependencies] clap = { version = "4.5", features = ["derive"] } libremetaverse = { path = "../crates/libremetaverse", default-features = false } -tokio = { version = "1.47", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] } +tokio = { version = "1.47", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } [lints] workspace = true diff --git a/programs/README.md b/programs/README.md index 04d8e17..fbe2324 100644 --- a/programs/README.md +++ b/programs/README.md @@ -12,7 +12,7 @@ here so a source entry is never mistaken for a completed port. | `packet-dump` | PacketDump | Implemented with live and deterministic fake-grid capture | | `prim-inspector` | PrimInspector | Implemented with live and deterministic fake-grid discovery | | `inventory-explorer` | InventoryExplorer | Implemented with live inventory and deterministic AIS fixtures | -| `irc-gateway` | IRCGateway | Pending milestone 11 issue #90 | +| `irc-gateway` | IRCGateway | Implemented with live IRC/grid transports and deterministic offline scripts | | `test-client` | TestClient | Pending milestone 11 issues #91–#94 | | `vivox-test` | VivoxTest | Pending milestone 11 issue #95 | | `webrtc-test` | WebRtcTest | Pending milestone 11 issue #96 | @@ -101,6 +101,56 @@ cargo test -p libremetaverse-programs --test simple_bot_cli --locked cargo test --manifest-path tests/compat/Cargo.toml --test core_runtime_shims --locked ``` +## IRCGateway + +`irc-gateway` preserves the upstream seven-position live interface and bridges +fully audible normal local chat to one IRC channel in both directions: + +```text +irc-gateway FIRSTNAME LASTNAME PASSWORD MASTER_UUID IRC_HOST IRC_PORT '#channel' +``` + +The positional values can instead come from `GRID_FIRST_NAME`, +`GRID_LAST_NAME`, `GRID_PASSWORD`, `GRID_MASTER_UUID`, `IRC_HOST`, `IRC_PORT`, +and `IRC_CHANNEL`. `GRID_LOGIN_URL` or `--login-uri` selects a grid endpoint. +`--nickname` defaults to the upstream `SLGateway` name. Login is bounded by +`--login-timeout-seconds`; `--messages-per-second` and `--burst` control the +per-direction token buckets. + +The native IRC transport registers with `NICK`/`USER`, responds to `PING`, +joins after numeric `001`, handles fragmented UTF-8 input, caps inbound and +outbound lines, and reconnects with exponential backoff. The bridge uses +bounded queues, sanitizes names and line breaks, suppresses self messages, +short-window duplicates, and reflected cross-transport messages, and accepts +teleport lures only from `MASTER_UUID`. Login messages and console output +redact credentials, tokens, capability URLs, and other URLs. Ctrl-C cancels +login and reconnect waits, closes the IRC socket, joins workers, unsubscribes +grid callbacks, logs out, and disposes the native client. + +Offline tests use the same routing and rate-limit engine without credentials or +network access: + +```text +irc-connect +irc-disconnect +irctargetnicknamemessage +grid-chatsource-uuidnamenormal|otherfull|othermessage +grid-loginmessage +teleportsource-uuidsession-uuid +advancemilliseconds +shutdown +``` + +Scripts are limited to 1 MiB and 4,096 events; names, messages, transport +queues, and loop histories are also bounded. Run the issue-focused suite and +the related native social-message compatibility cases with: + +```sh +cargo test -p libremetaverse-programs --test irc_gateway_cli --locked +cargo test -p libremetaverse-programs irc_gateway::tests --locked +cargo test --manifest-path tests/compat/Cargo.toml --test social_message_semantics --locked +``` + ## PacketDump `packet-dump` preserves the upstream live arguments and its 20-second login diff --git a/programs/src/bin/irc_gateway.rs b/programs/src/bin/irc_gateway.rs index 6204d5f..758fac5 100644 --- a/programs/src/bin/irc_gateway.rs +++ b/programs/src/bin/irc_gateway.rs @@ -1,3 +1,3 @@ fn main() -> std::process::ExitCode { - libremetaverse_programs::pending_program("IRCGateway") + libremetaverse_programs::irc_gateway::main_entry() } diff --git a/programs/src/irc_gateway.rs b/programs/src/irc_gateway.rs new file mode 100644 index 0000000..c224bf0 --- /dev/null +++ b/programs/src/irc_gateway.rs @@ -0,0 +1,1573 @@ +//! Native, cancellation-safe IRC/local-grid chat bridge. + +use clap::Parser; +use libremetaverse::types::UUID; +use libremetaverse::types::compat::{CancellationToken, CancellationTokenSource, Subscription}; +use libremetaverse::{ + AgentManager, ChatAudibleLevel, ChatEventArgs, ChatType, GridClient, InstantMessageDialog, + InstantMessageEventArgs, LoginParams, LoginProgressEventArgs, NetworkManager, +}; +use std::collections::VecDeque; +use std::fmt; +use std::fs::File; +use std::io::{self, BufRead, BufReader, Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::ExitCode; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::{Duration, Instant}; +use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader as AsyncBufReader}; +use tokio::net::TcpStream; +use tokio::sync::mpsc; + +pub const EXIT_SUCCESS: u8 = 0; +pub const EXIT_USAGE: u8 = 2; +pub const EXIT_INPUT: u8 = 3; +pub const EXIT_CLIENT: u8 = 4; + +const MAX_SCRIPT_BYTES: u64 = 1024 * 1024; +const MAX_SCRIPT_EVENTS: usize = 4096; +const MAX_NAME_BYTES: usize = 256; +const MAX_MESSAGE_BYTES: usize = 4096; +const EVENT_QUEUE_CAPACITY: usize = 256; +const OUTBOUND_QUEUE_CAPACITY: usize = 128; +const BRIDGE_QUEUE_CAPACITY: usize = 64; +const IRC_LINE_BYTES: usize = 512; +const MAX_IRC_INPUT_BYTES: usize = 8192; +const DUPLICATE_WINDOW_MS: u64 = 2_000; +const LOOP_WINDOW_MS: u64 = 30_000; +const MAX_RECENT_MESSAGES: usize = 256; +const TICK_MS: u64 = 100; + +#[derive(Parser)] +#[command( + name = "irc-gateway", + version, + about = "Bridge one IRC channel and LibreMetaverse local chat", + long_about = None, + arg_required_else_help = true, + after_help = "Live mode preserves the upstream positional interface. Offline scripts use tab-separated events: irc-connect, irc-disconnect, irc, grid-chat, grid-login, teleport, advance, and shutdown." +)] +struct Cli { + /// Avatar first name. May also be supplied as `GRID_FIRST_NAME`. + #[arg(value_name = "FIRSTNAME")] + first_name: Option, + /// Avatar last name. May also be supplied as `GRID_LAST_NAME`. + #[arg(value_name = "LASTNAME")] + last_name: Option, + /// Avatar password. May also be supplied as `GRID_PASSWORD`. + #[arg(value_name = "PASSWORD")] + password: Option, + /// UUID whose teleport lures are accepted. May also be `GRID_MASTER_UUID`. + #[arg(value_name = "MASTER_UUID")] + master_uuid: Option, + /// IRC server host. May also be `IRC_HOST`. + #[arg(value_name = "IRC_HOST")] + irc_host: Option, + /// IRC server port. May also be `IRC_PORT`. + #[arg(value_name = "IRC_PORT")] + irc_port: Option, + /// IRC channel, including its leading '#'. May also be `IRC_CHANNEL`. + #[arg(value_name = "#CHANNEL")] + channel: Option, + + /// Run a deterministic, bounded offline bridge script. + #[arg(long, value_name = "FILE")] + fake_script: Option, + /// Channel used in fake mode when the positional channel is absent. + #[arg(long, default_value = "#metacrate", value_name = "#CHANNEL")] + fake_channel: String, + /// Master UUID used in fake mode when the positional UUID is absent. + #[arg( + long, + default_value = "11111111-2222-3333-4444-555555555555", + value_name = "UUID" + )] + fake_master_uuid: String, + /// IRC nickname. + #[arg(long, default_value = "SLGateway", value_name = "NICK")] + nickname: String, + /// IRC real name sent during registration. + #[arg(long, default_value = "Second Life Gateway", value_name = "TEXT")] + real_name: String, + /// Override the grid login endpoint; `GRID_LOGIN_URL` is the fallback. + #[arg(long, value_name = "URL")] + login_uri: Option, + /// Maximum time allowed for grid login. + #[arg(long, default_value_t = 30, value_name = "SECONDS", value_parser = clap::value_parser!(u64).range(1..=300))] + login_timeout_seconds: u64, + /// Per-direction bridge rate, in messages per second. + #[arg(long, default_value_t = 4, value_name = "COUNT", value_parser = clap::value_parser!(u32).range(1..=100))] + messages_per_second: u32, + /// Per-direction burst capacity. + #[arg(long, default_value_t = 4, value_name = "COUNT", value_parser = clap::value_parser!(u32).range(1..=100))] + burst: u32, +} + +struct LiveConfig { + first_name: String, + last_name: String, + password: String, + master: UUID, + irc_host: String, + irc_port: u16, + bridge: BridgeConfig, + real_name: String, + login_uri: Option, + login_timeout: Duration, +} + +#[derive(Clone)] +struct BridgeConfig { + channel: String, + nickname: String, + messages_per_second: u32, + burst: u32, +} + +#[derive(Debug)] +enum ProgramError { + Usage(&'static str), + Io { action: String, source: io::Error }, + InvalidScript { line: usize, reason: &'static str }, + Client(&'static str), + LoginFailed, + LoginTimedOut, + Signal, +} + +impl ProgramError { + const fn exit_code(&self) -> u8 { + match self { + Self::Usage(_) => EXIT_USAGE, + Self::Io { .. } | Self::InvalidScript { .. } => EXIT_INPUT, + Self::Client(_) | Self::LoginFailed | Self::LoginTimedOut | Self::Signal => EXIT_CLIENT, + } + } +} + +impl fmt::Display for ProgramError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Usage(message) => f.write_str(message), + Self::Io { action, source } => write!(f, "{action}: {source}"), + Self::InvalidScript { line, reason } => { + write!(f, "invalid bridge script at line {line}: {reason}") + } + Self::Client(operation) => write!(f, "native client operation failed: {operation}"), + Self::LoginFailed => f.write_str("grid login failed"), + Self::LoginTimedOut => f.write_str("grid login timed out"), + Self::Signal => f.write_str("could not install the Ctrl-C handler"), + } + } +} + +#[derive(Clone, Debug)] +enum BridgeEvent { + IrcConnected, + IrcDisconnected, + IrcMessage { + target: String, + nick: String, + message: String, + }, + GridChat { + source_id: UUID, + name: String, + message: String, + normal: bool, + fully_audible: bool, + }, + GridLogin(String), + OwnGridId(UUID), + Teleport { + source_id: UUID, + session_id: UUID, + }, +} + +trait BridgeTransport { + fn join_irc(&mut self, channel: &str) -> Result<(), &'static str>; + fn send_irc(&mut self, channel: &str, message: &str) -> Result<(), &'static str>; + fn send_grid(&mut self, message: &str) -> Result<(), &'static str>; + fn accept_teleport(&mut self, source: UUID, session: UUID) -> Result<(), &'static str>; +} + +#[derive(Clone, Copy, Eq, PartialEq)] +enum Direction { + ToIrc, + ToGrid, +} + +struct PendingMessage { + direction: Direction, + message: String, +} + +struct RecentMessage { + direction: Direction, + message: String, + at_ms: u64, +} + +struct TokenBucket { + milli_tokens: u64, + last_ms: u64, + rate: u64, + burst_milli: u64, +} + +impl TokenBucket { + fn new(rate: u32, burst: u32) -> Self { + Self { + milli_tokens: u64::from(burst) * 1_000, + last_ms: 0, + rate: u64::from(rate), + burst_milli: u64::from(burst) * 1_000, + } + } + + fn take(&mut self, now_ms: u64) -> bool { + let elapsed = now_ms.saturating_sub(self.last_ms); + self.last_ms = now_ms; + self.milli_tokens = self + .milli_tokens + .saturating_add(elapsed.saturating_mul(self.rate)) + .min(self.burst_milli); + if self.milli_tokens >= 1_000 { + self.milli_tokens -= 1_000; + true + } else { + false + } + } +} + +struct Bridge { + config: BridgeConfig, + master: UUID, + own_grid_id: UUID, + irc_connected: bool, + pending: VecDeque, + recent: VecDeque, + inbound_recent: VecDeque, + irc_rate: TokenBucket, + grid_rate: TokenBucket, + dropped: usize, +} + +impl Bridge { + fn new(config: BridgeConfig, master: UUID, own_grid_id: UUID) -> Self { + Self { + irc_rate: TokenBucket::new(config.messages_per_second, config.burst), + grid_rate: TokenBucket::new(config.messages_per_second, config.burst), + config, + master, + own_grid_id, + irc_connected: false, + pending: VecDeque::new(), + recent: VecDeque::new(), + inbound_recent: VecDeque::new(), + dropped: 0, + } + } + + fn handle( + &mut self, + event: BridgeEvent, + now_ms: u64, + transport: &mut dyn BridgeTransport, + log: &dyn BridgeLogger, + ) { + self.expire_recent(now_ms); + match event { + BridgeEvent::IrcConnected => { + self.irc_connected = true; + if let Err(error) = transport.join_irc(&self.config.channel) { + log.line(format!("IRC join failed: {error}")); + } else { + log.line(format!("IRC connected; joined {}", self.config.channel)); + } + } + BridgeEvent::IrcDisconnected => { + self.irc_connected = false; + log.line("IRC disconnected; reconnect scheduled".into()); + } + BridgeEvent::IrcMessage { + target, + nick, + message, + } => { + if !irc_eq(&target, &self.config.channel) + || irc_eq(&nick, &self.config.nickname) + || !valid_message(&message) + { + return; + } + let message = clean_message(&message); + if self.is_loop(Direction::ToGrid, &message, now_ms) { + log.line("Suppressed duplicate/loop IRC message".into()); + return; + } + let formatted = format!("<{}> {}", map_name(&nick), message); + if self.is_duplicate(Direction::ToGrid, &formatted, now_ms) { + log.line("Suppressed duplicate/loop IRC message".into()); + return; + } + self.remember_inbound(Direction::ToGrid, &formatted, now_ms); + self.enqueue(Direction::ToGrid, formatted); + } + BridgeEvent::GridChat { + source_id, + name, + message, + normal, + fully_audible, + } => { + if source_id == self.own_grid_id + || !normal + || !fully_audible + || !valid_message(&message) + { + return; + } + let message = clean_message(&message); + if self.is_loop(Direction::ToIrc, &message, now_ms) { + log.line("Suppressed duplicate/loop grid message".into()); + return; + } + let formatted = format!("<{}> {}", map_name(&name), message); + if self.is_duplicate(Direction::ToIrc, &formatted, now_ms) { + log.line("Suppressed duplicate/loop grid message".into()); + return; + } + self.remember_inbound(Direction::ToIrc, &formatted, now_ms); + self.enqueue(Direction::ToIrc, formatted); + } + BridgeEvent::GridLogin(message) => { + if valid_message(&message) { + self.enqueue(Direction::ToIrc, redact_text(&clean_message(&message))); + } + } + BridgeEvent::OwnGridId(agent_id) => self.own_grid_id = agent_id, + BridgeEvent::Teleport { + source_id, + session_id, + } => { + if source_id == self.master { + if let Err(error) = transport.accept_teleport(source_id, session_id) { + log.line(format!("Teleport response failed: {error}")); + } else { + log.line("Accepted teleport lure from configured master".into()); + } + } + } + } + self.flush(now_ms, transport, log); + } + + fn tick(&mut self, now_ms: u64, transport: &mut dyn BridgeTransport, log: &dyn BridgeLogger) { + self.expire_recent(now_ms); + self.flush(now_ms, transport, log); + } + + fn enqueue(&mut self, direction: Direction, message: String) { + if self.pending.len() == BRIDGE_QUEUE_CAPACITY { + self.pending.pop_front(); + self.dropped += 1; + } + self.pending + .push_back(PendingMessage { direction, message }); + } + + fn flush(&mut self, now_ms: u64, transport: &mut dyn BridgeTransport, log: &dyn BridgeLogger) { + let mut retained = VecDeque::with_capacity(self.pending.len()); + while let Some(item) = self.pending.pop_front() { + let permitted = match item.direction { + Direction::ToIrc => self.irc_connected && self.irc_rate.take(now_ms), + Direction::ToGrid => self.grid_rate.take(now_ms), + }; + if !permitted { + retained.push_back(item); + continue; + } + let result = match item.direction { + Direction::ToIrc => transport.send_irc(&self.config.channel, &item.message), + Direction::ToGrid => transport.send_grid(&item.message), + }; + match result { + Ok(()) => { + self.recent.push_back(RecentMessage { + direction: item.direction, + message: item.message.clone(), + at_ms: now_ms, + }); + log.line(format!( + "[{}] {}", + if item.direction == Direction::ToIrc { + "Grid->IRC" + } else { + "IRC->Grid" + }, + item.message + )); + } + Err(error) => { + log.line(format!("Bridge send failed: {error}")); + retained.push_front(item); + break; + } + } + } + retained.append(&mut self.pending); + self.pending = retained; + while self.recent.len() > MAX_RECENT_MESSAGES { + self.recent.pop_front(); + } + } + + fn is_loop(&self, direction: Direction, message: &str, now_ms: u64) -> bool { + let opposite = if direction == Direction::ToIrc { + Direction::ToGrid + } else { + Direction::ToIrc + }; + self.recent.iter().any(|recent| { + recent.direction == opposite + && recent.message == message + && now_ms.saturating_sub(recent.at_ms) <= LOOP_WINDOW_MS + }) + } + + fn is_duplicate(&self, direction: Direction, message: &str, now_ms: u64) -> bool { + self.inbound_recent.iter().any(|recent| { + recent.direction == direction + && recent.message == message + && now_ms.saturating_sub(recent.at_ms) <= DUPLICATE_WINDOW_MS + }) + } + + fn remember_inbound(&mut self, direction: Direction, message: &str, now_ms: u64) { + self.inbound_recent.push_back(RecentMessage { + direction, + message: message.into(), + at_ms: now_ms, + }); + while self.inbound_recent.len() > MAX_RECENT_MESSAGES { + self.inbound_recent.pop_front(); + } + } + + fn expire_recent(&mut self, now_ms: u64) { + self.recent + .retain(|entry| now_ms.saturating_sub(entry.at_ms) <= LOOP_WINDOW_MS); + self.inbound_recent + .retain(|entry| now_ms.saturating_sub(entry.at_ms) <= DUPLICATE_WINDOW_MS); + } +} + +trait BridgeLogger: Send + Sync { + fn line(&self, line: String); +} + +struct ConsoleLogger; + +impl BridgeLogger for ConsoleLogger { + fn line(&self, line: String) { + println!("{}", redact_text(&line)); + } +} + +#[derive(Default)] +struct ScriptTransport { + calls: Vec, +} + +impl BridgeTransport for ScriptTransport { + fn join_irc(&mut self, channel: &str) -> Result<(), &'static str> { + self.calls.push(format!("CALL irc-join {channel}")); + Ok(()) + } + fn send_irc(&mut self, channel: &str, message: &str) -> Result<(), &'static str> { + self.calls + .push(format!("CALL irc-privmsg {channel} {message}")); + Ok(()) + } + fn send_grid(&mut self, message: &str) -> Result<(), &'static str> { + self.calls + .push(format!("CALL grid-chat channel=0 type=Normal {message}")); + Ok(()) + } + fn accept_teleport(&mut self, source: UUID, session: UUID) -> Result<(), &'static str> { + self.calls.push(format!( + "CALL teleport-lure-respond {source} {session} accept=true" + )); + Ok(()) + } +} + +struct LiveGrid { + client: Mutex, + network: NetworkManager, + subscriptions: Mutex>, + shutdown: AtomicBool, +} + +impl LiveGrid { + fn new(client: GridClient, network: NetworkManager) -> Self { + Self { + client: Mutex::new(client), + network, + subscriptions: Mutex::new(Vec::new()), + shutdown: AtomicBool::new(false), + } + } + + fn with_agent(&self, action: impl FnOnce(&mut AgentManager) -> T) -> T { + action(lock(&self.client).self_()) + } + + fn install(self: &Arc, sender: &mpsc::Sender, dropped: &Arc) { + let chat_sender = sender.clone(); + let chat_dropped = Arc::clone(dropped); + let chat = self.with_agent(|agent| { + agent.subscribe_chat_from_simulator(Arc::new(move |event: ChatEventArgs| { + try_event( + &chat_sender, + BridgeEvent::GridChat { + source_id: event.source_id(), + name: event.from_name(), + message: event.message(), + normal: event.type_() == ChatType::Normal, + fully_audible: event.audible_level() == ChatAudibleLevel::Fully, + }, + &chat_dropped, + ); + })) + }); + let im_sender = sender.clone(); + let im_dropped = Arc::clone(dropped); + let im = self.with_agent(|agent| { + agent.subscribe_im(Arc::new(move |event: InstantMessageEventArgs| { + let im = event.im(); + if im.dialog == InstantMessageDialog::RequestTeleport { + try_event( + &im_sender, + BridgeEvent::Teleport { + source_id: im.from_agent_id, + session_id: im.im_session_id, + }, + &im_dropped, + ); + } + })) + }); + let progress_sender = sender.clone(); + let progress_dropped = Arc::clone(dropped); + let progress = self.network.subscribe_login_progress(Arc::new( + move |event: LoginProgressEventArgs| { + try_event( + &progress_sender, + BridgeEvent::GridLogin(event.message()), + &progress_dropped, + ); + }, + )); + *lock(&self.subscriptions) = vec![chat, im, progress]; + } + + fn shutdown(&self) { + if self.shutdown.swap(true, Ordering::AcqRel) { + return; + } + lock(&self.subscriptions).clear(); + let _ = self.network.logout_with_method(); + let mut client = lock(&self.client); + let _ = client.self_().dispose(); + let _ = client.dispose_with_method(); + } +} + +struct LiveTransport { + grid: Arc, + irc: mpsc::Sender, +} + +impl BridgeTransport for LiveTransport { + fn join_irc(&mut self, channel: &str) -> Result<(), &'static str> { + self.irc + .try_send(IrcCommand::Join(channel.into())) + .map_err(|_| "IRC command queue full or closed") + } + fn send_irc(&mut self, channel: &str, message: &str) -> Result<(), &'static str> { + self.irc + .try_send(IrcCommand::Message(channel.into(), message.into())) + .map_err(|_| "IRC command queue full or closed") + } + fn send_grid(&mut self, message: &str) -> Result<(), &'static str> { + self.grid + .with_agent(|agent| agent.chat(message.into(), 0, ChatType::Normal, Some(false))) + .map_err(|_| "grid chat") + } + fn accept_teleport(&mut self, source: UUID, session: UUID) -> Result<(), &'static str> { + self.grid + .with_agent(|agent| agent.teleport_lure_respond(source, session, true)) + .map_err(|_| "teleport lure response") + } +} + +enum IrcCommand { + Join(String), + Message(String, String), +} + +struct IrcRuntime { + host: String, + port: u16, + nick: String, + real_name: String, + commands: mpsc::Receiver, + events: mpsc::Sender, + dropped: Arc, + cancellation: CancellationToken, +} + +#[must_use] +pub fn main_entry() -> ExitCode { + let cli = Cli::parse(); + let Ok(runtime) = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + else { + eprintln!("irc-gateway: could not initialize the async runtime"); + return ExitCode::from(EXIT_CLIENT); + }; + match runtime.block_on(run(cli)) { + Ok(()) => ExitCode::from(EXIT_SUCCESS), + Err(error) => { + eprintln!("irc-gateway: {error}"); + ExitCode::from(error.exit_code()) + } + } +} + +async fn run(cli: Cli) -> Result<(), ProgramError> { + if let Some(path) = cli.fake_script.clone() { + return run_fake(&cli, &path); + } + run_live(resolve_live(cli)?).await +} + +fn resolve_live(cli: Cli) -> Result { + let first_name = required( + cli.first_name, + "GRID_FIRST_NAME", + "FIRSTNAME is required (or set GRID_FIRST_NAME)", + )?; + let last_name = required( + cli.last_name, + "GRID_LAST_NAME", + "LASTNAME is required (or set GRID_LAST_NAME)", + )?; + let password = required( + cli.password, + "GRID_PASSWORD", + "PASSWORD is required (or set GRID_PASSWORD)", + )?; + let master_text = required( + cli.master_uuid, + "GRID_MASTER_UUID", + "MASTER_UUID is required (or set GRID_MASTER_UUID)", + )?; + let master = UUID::new_with_string(master_text) + .map_err(|_| ProgramError::Usage("MASTER_UUID is invalid"))?; + let irc_host = required(cli.irc_host, "IRC_HOST", "IRC_HOST is required")?; + let irc_port = cli + .irc_port + .or_else(|| std::env::var("IRC_PORT").ok()?.parse().ok()) + .ok_or(ProgramError::Usage( + "IRC_PORT is required and must be 1..65535", + ))?; + if irc_port == 0 { + return Err(ProgramError::Usage( + "IRC_PORT is required and must be 1..65535", + )); + } + let channel = required(cli.channel, "IRC_CHANNEL", "#CHANNEL is required")?; + validate_channel(&channel)?; + validate_nick(&cli.nickname)?; + Ok(LiveConfig { + first_name, + last_name, + password, + master, + irc_host, + irc_port, + bridge: BridgeConfig { + channel, + nickname: cli.nickname, + messages_per_second: cli.messages_per_second, + burst: cli.burst, + }, + real_name: clean_message(&cli.real_name), + login_uri: cli + .login_uri + .or_else(|| std::env::var("GRID_LOGIN_URL").ok()) + .filter(|value| !value.is_empty()), + login_timeout: Duration::from_secs(cli.login_timeout_seconds), + }) +} + +fn required( + value: Option, + env: &str, + message: &'static str, +) -> Result { + value + .or_else(|| std::env::var(env).ok()) + .filter(|value| !value.is_empty()) + .ok_or(ProgramError::Usage(message)) +} + +async fn run_live(mut config: LiveConfig) -> Result<(), ProgramError> { + let client = GridClient::new().map_err(|_| ProgramError::Client("construct GridClient"))?; + let network = client.network(); + let mut login = network + .default_login_params( + std::mem::take(&mut config.first_name), + std::mem::take(&mut config.last_name), + std::mem::take(&mut config.password), + "IRCGateway".into(), + env!("CARGO_PKG_VERSION").into(), + ) + .map_err(|_| ProgramError::Client("build login parameters"))?; + if let Some(uri) = config.login_uri.take() { + login.uri = uri; + } + + let grid = Arc::new(LiveGrid::new(client, network.clone())); + let own_grid_id = grid.with_agent(|agent| agent.agent_id()); + let dropped = Arc::new(AtomicUsize::new(0)); + let (event_tx, event_rx) = mpsc::channel(EVENT_QUEUE_CAPACITY); + let (irc_tx, irc_rx) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY); + grid.install(&event_tx, &dropped); + let cancellation = CancellationTokenSource::new(); + let irc_worker = tokio::spawn(irc_loop(IrcRuntime { + host: config.irc_host.clone(), + port: config.irc_port, + nick: config.bridge.nickname.clone(), + real_name: config.real_name.clone(), + commands: irc_rx, + events: event_tx.clone(), + dropped: Arc::clone(&dropped), + cancellation: cancellation.token(), + })); + let bridge_worker = tokio::spawn(bridge_loop( + Bridge::new(config.bridge.clone(), config.master, own_grid_id), + LiveTransport { + grid: Arc::clone(&grid), + irc: irc_tx.clone(), + }, + event_rx, + Arc::new(ConsoleLogger), + cancellation.token(), + )); + + let result = match await_grid_login(&network, login, config.login_timeout).await { + Ok(Some(true)) => { + try_event( + &event_tx, + BridgeEvent::OwnGridId(grid.with_agent(|agent| agent.agent_id())), + &dropped, + ); + tokio::signal::ctrl_c() + .await + .map_err(|_| ProgramError::Signal) + } + Ok(Some(false)) => Err(ProgramError::LoginFailed), + Ok(None) => Ok(()), + Err(error) => Err(error), + }; + finish_live( + grid, + event_tx, + irc_tx, + cancellation, + bridge_worker, + irc_worker, + dropped, + ) + .await; + result +} + +async fn await_grid_login( + network: &NetworkManager, + login: LoginParams, + timeout: Duration, +) -> Result, ProgramError> { + let cancellation = CancellationTokenSource::new(); + tokio::select! { + result = network.login_with_login_params_cancellation_token(login, Some(cancellation.token())) => { + result.map(Some).map_err(|_| ProgramError::LoginFailed) + }, + () = tokio::time::sleep(timeout) => { + cancellation.cancel(); + let _ = network.abort_login(); + Err(ProgramError::LoginTimedOut) + }, + signal = tokio::signal::ctrl_c() => { + cancellation.cancel(); + let _ = network.abort_login(); + signal.map_err(|_| ProgramError::Signal)?; + Ok(None) + } + } +} + +async fn finish_live( + grid: Arc, + event_tx: mpsc::Sender, + irc_tx: mpsc::Sender, + cancellation: CancellationTokenSource, + bridge: tokio::task::JoinHandle, + irc: tokio::task::JoinHandle<()>, + dropped: Arc, +) { + cancellation.cancel(); + drop(event_tx); + drop(irc_tx); + let finished = bridge.await.ok(); + let _ = irc.await; + grid.shutdown(); + let total_dropped = + dropped.load(Ordering::Acquire) + finished.map_or(0, |bridge| bridge.dropped); + if total_dropped > 0 { + eprintln!("irc-gateway: dropped {total_dropped} messages because bounded queues were full"); + } +} + +async fn bridge_loop( + mut bridge: Bridge, + mut transport: T, + mut events: mpsc::Receiver, + logger: Arc, + cancellation: CancellationToken, +) -> Bridge { + let start = Instant::now(); + let mut interval = tokio::time::interval(Duration::from_millis(TICK_MS)); + loop { + tokio::select! { + () = cancellation.cancelled() => break, + _ = interval.tick() => bridge.tick(elapsed_ms(start), &mut transport, logger.as_ref()), + event = events.recv() => match event { + Some(event) => bridge.handle(event, elapsed_ms(start), &mut transport, logger.as_ref()), + None => break, + } + } + } + events.close(); + bridge +} + +async fn irc_loop(mut runtime: IrcRuntime) { + let mut backoff = Duration::from_secs(1); + loop { + let connection = tokio::select! { + () = runtime.cancellation.cancelled() => break, + result = TcpStream::connect((runtime.host.as_str(), runtime.port)) => result, + }; + match connection { + Ok(stream) => { + backoff = Duration::from_secs(1); + if run_irc_connection( + stream, + &runtime.nick, + &runtime.real_name, + &mut runtime.commands, + &runtime.events, + &runtime.dropped, + runtime.cancellation.clone(), + ) + .await + .is_err() + { + try_event( + &runtime.events, + BridgeEvent::IrcDisconnected, + &runtime.dropped, + ); + } + } + Err(_) => try_event( + &runtime.events, + BridgeEvent::IrcDisconnected, + &runtime.dropped, + ), + } + tokio::select! { + () = runtime.cancellation.cancelled() => break, + () = tokio::time::sleep(backoff) => {} + } + backoff = (backoff * 2).min(Duration::from_secs(30)); + } +} + +async fn run_irc_connection( + stream: TcpStream, + nick: &str, + real_name: &str, + commands: &mut mpsc::Receiver, + events: &mpsc::Sender, + dropped: &AtomicUsize, + cancellation: CancellationToken, +) -> io::Result<()> { + let (reader, mut writer) = stream.into_split(); + write_irc(&mut writer, &format!("USER {nick} 0 * :{real_name}")).await?; + write_irc(&mut writer, &format!("NICK {nick}")).await?; + let mut reader = AsyncBufReader::new(reader); + loop { + tokio::select! { + () = cancellation.cancelled() => return Ok(()), + command = commands.recv() => match command { + Some(IrcCommand::Join(channel)) => write_irc(&mut writer, &format!("JOIN {}", sanitize_irc_atom(&channel))).await?, + Some(IrcCommand::Message(target, message)) => { + let prefix = format!("PRIVMSG {} :", sanitize_irc_atom(&target)); + let message = truncate_utf8(&clean_message(&message), IRC_LINE_BYTES - 2 - prefix.len()); + write_irc(&mut writer, &format!("{prefix}{message}")).await?; + } + None => return Ok(()), + }, + line = read_irc_line(&mut reader) => { + let Some(line) = line? else { return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "IRC disconnected")); }; + if let Some(message) = parse_irc_line(&line) { + match message.command.as_str() { + "PING" => { + let payload = message.trailing.or_else(|| message.params.first().cloned()).unwrap_or_default(); + write_irc(&mut writer, &format!("PONG :{}", clean_message(&payload))).await?; + } + "001" => try_event(events, BridgeEvent::IrcConnected, dropped), + "PRIVMSG" => { + if let (Some(prefix), Some(target), Some(text)) = (message.prefix, message.params.first(), message.trailing) { + let nick = prefix.split('!').next().unwrap_or(&prefix).to_owned(); + try_event(events, BridgeEvent::IrcMessage { target: target.clone(), nick, message: text }, dropped); + } + } + _ => {} + } + } + } + } + } +} + +async fn read_irc_line(reader: &mut R) -> io::Result> { + let mut bytes = Vec::with_capacity(IRC_LINE_BYTES); + loop { + let available = reader.fill_buf().await?; + if available.is_empty() { + return if bytes.is_empty() { + Ok(None) + } else { + Ok(Some(String::from_utf8_lossy(&bytes).into_owned())) + }; + } + let newline = available.iter().position(|byte| *byte == b'\n'); + let consumed = newline.map_or(available.len(), |index| index + 1); + if bytes.len().saturating_add(consumed) > MAX_IRC_INPUT_BYTES { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "oversized IRC line", + )); + } + bytes.extend_from_slice(&available[..consumed]); + reader.consume(consumed); + if newline.is_some() { + return Ok(Some(String::from_utf8_lossy(&bytes).into_owned())); + } + } +} + +async fn write_irc(writer: &mut tokio::net::tcp::OwnedWriteHalf, line: &str) -> io::Result<()> { + let line = truncate_utf8(&clean_message(line), IRC_LINE_BYTES - 2); + writer.write_all(line.as_bytes()).await?; + writer.write_all(b"\r\n").await +} + +struct IrcMessage { + prefix: Option, + command: String, + params: Vec, + trailing: Option, +} + +fn parse_irc_line(line: &str) -> Option { + let line = line.trim_matches(['\r', '\n']); + if line.is_empty() || line.as_bytes().contains(&0) { + return None; + } + let line = if let Some(tagged) = line.strip_prefix('@') { + tagged.split_once(' ')?.1.trim_start() + } else { + line + }; + let (prefix, rest) = if let Some(rest) = line.strip_prefix(':') { + let (prefix, rest) = rest.split_once(' ')?; + (Some(prefix.to_owned()), rest.trim_start()) + } else { + (None, line) + }; + let (middle, trailing) = if let Some((middle, trailing)) = rest.split_once(" :") { + (middle, Some(trailing.to_owned())) + } else { + (rest, None) + }; + let mut words = middle.split_whitespace(); + let command = words.next()?.to_ascii_uppercase(); + Some(IrcMessage { + prefix, + command, + params: words.map(str::to_owned).collect(), + trailing, + }) +} + +fn run_fake(cli: &Cli, path: &Path) -> Result<(), ProgramError> { + validate_channel(&cli.fake_channel)?; + validate_nick(&cli.nickname)?; + let master_text = cli.master_uuid.as_deref().unwrap_or(&cli.fake_master_uuid); + let master = parse_uuid(master_text, 0, "master UUID is invalid")?; + let own = UUID::new_with_string("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee".into()) + .map_err(|_| ProgramError::Client("construct fake agent UUID"))?; + let config = BridgeConfig { + channel: cli + .channel + .clone() + .unwrap_or_else(|| cli.fake_channel.clone()), + nickname: cli.nickname.clone(), + messages_per_second: cli.messages_per_second, + burst: cli.burst, + }; + validate_channel(&config.channel)?; + let events = read_script(path)?; + let mut bridge = Bridge::new(config, master, own); + let mut transport = ScriptTransport::default(); + let logger = ScriptLogger::default(); + let mut now_ms = 0; + let mut shutdown = false; + for event in events { + match event { + ScriptEvent::Bridge(event) => bridge.handle(event, now_ms, &mut transport, &logger), + ScriptEvent::Advance(milliseconds) => { + now_ms = now_ms.saturating_add(milliseconds); + bridge.tick(now_ms, &mut transport, &logger); + } + ScriptEvent::Shutdown => { + shutdown = true; + break; + } + } + } + bridge.tick(now_ms.saturating_add(60_000), &mut transport, &logger); + let stdout = io::stdout(); + let mut output = stdout.lock(); + for line in logger.lines.into_inner().unwrap_or_default() { + writeln!(output, "{line}").map_err(|source| io_error("writing standard output", source))?; + } + for call in transport.calls { + writeln!(output, "{}", redact_text(&call)) + .map_err(|source| io_error("writing standard output", source))?; + } + writeln!(output, "STATE irc-connected={} queued={} dropped={} active-transports=0 active-tasks=0 shutdown={shutdown}", bridge.irc_connected, bridge.pending.len(), bridge.dropped) + .map_err(|source| io_error("writing standard output", source))?; + Ok(()) +} + +#[derive(Default)] +struct ScriptLogger { + lines: Mutex>, +} +impl BridgeLogger for ScriptLogger { + fn line(&self, line: String) { + lock(&self.lines).push(redact_text(&line)); + } +} + +enum ScriptEvent { + Bridge(BridgeEvent), + Advance(u64), + Shutdown, +} + +fn read_script(path: &Path) -> Result, ProgramError> { + let file = File::open(path) + .map_err(|source| io_error(format!("opening {}", path.display()), source))?; + if file + .metadata() + .map_err(|source| io_error("reading script metadata", source))? + .len() + > MAX_SCRIPT_BYTES + { + return Err(ProgramError::InvalidScript { + line: 0, + reason: "script exceeds 1 MiB", + }); + } + let mut bounded = file.take(MAX_SCRIPT_BYTES + 1); + let mut bytes = Vec::new(); + bounded + .read_to_end(&mut bytes) + .map_err(|source| io_error("reading bridge script", source))?; + if bytes.len() as u64 > MAX_SCRIPT_BYTES { + return Err(ProgramError::InvalidScript { + line: 0, + reason: "script exceeds 1 MiB", + }); + } + let reader = BufReader::new(bytes.as_slice()); + let mut events = Vec::new(); + for (index, line) in reader.lines().enumerate() { + let line_number = index + 1; + let line = line.map_err(|source| io_error("decoding bridge script as UTF-8", source))?; + if line.trim().is_empty() || line.trim_start().starts_with('#') { + continue; + } + if events.len() == MAX_SCRIPT_EVENTS { + return Err(ProgramError::InvalidScript { + line: line_number, + reason: "script has more than 4,096 events", + }); + } + events.push(parse_script_line(&line, line_number)?); + } + Ok(events) +} + +fn parse_script_line(line: &str, number: usize) -> Result { + let fields: Vec<_> = line.split('\t').collect(); + let invalid = || ProgramError::InvalidScript { + line: number, + reason: "expected irc-connect, irc-disconnect, irctargetnickmessage, grid-chatuuidnamenormal|otherfull|othermessage, grid-loginmessage, teleportsource-uuidsession-uuid, advancemilliseconds, or shutdown", + }; + match fields.as_slice() { + ["irc-connect"] => Ok(ScriptEvent::Bridge(BridgeEvent::IrcConnected)), + ["irc-disconnect"] => Ok(ScriptEvent::Bridge(BridgeEvent::IrcDisconnected)), + ["irc", target, nick, message] + if target.len() <= MAX_NAME_BYTES + && nick.len() <= MAX_NAME_BYTES + && message.len() <= MAX_MESSAGE_BYTES => + { + Ok(ScriptEvent::Bridge(BridgeEvent::IrcMessage { + target: (*target).into(), + nick: (*nick).into(), + message: (*message).into(), + })) + } + ["grid-chat", uuid, name, kind, audible, message] + if name.len() <= MAX_NAME_BYTES && message.len() <= MAX_MESSAGE_BYTES => + { + Ok(ScriptEvent::Bridge(BridgeEvent::GridChat { + source_id: parse_uuid(uuid, number, "grid source UUID is invalid")?, + name: (*name).into(), + message: (*message).into(), + normal: *kind == "normal", + fully_audible: *audible == "full", + })) + } + ["grid-login", message] if message.len() <= MAX_MESSAGE_BYTES => Ok(ScriptEvent::Bridge( + BridgeEvent::GridLogin((*message).into()), + )), + ["teleport", source, session] => Ok(ScriptEvent::Bridge(BridgeEvent::Teleport { + source_id: parse_uuid(source, number, "teleport source UUID is invalid")?, + session_id: parse_uuid(session, number, "teleport session UUID is invalid")?, + })), + ["advance", milliseconds] => Ok(ScriptEvent::Advance( + milliseconds.parse().map_err(|_| invalid())?, + )), + ["shutdown"] => Ok(ScriptEvent::Shutdown), + _ => Err(invalid()), + } +} + +fn validate_channel(channel: &str) -> Result<(), ProgramError> { + if channel.starts_with('#') + && channel.len() > 1 + && channel.len() <= 200 + && !channel + .chars() + .any(|ch| ch.is_control() || ch == ' ' || ch == ',') + { + Ok(()) + } else { + Err(ProgramError::Usage( + "IRC channel must begin with '#' and contain no spaces, commas, or controls", + )) + } +} + +fn validate_nick(nick: &str) -> Result<(), ProgramError> { + if !nick.is_empty() && nick.len() <= 32 && nick.chars().all(is_nick_char) { + Ok(()) + } else { + Err(ProgramError::Usage( + "IRC nickname must be 1..32 portable IRC nickname characters", + )) + } +} + +fn is_nick_char(ch: char) -> bool { + ch.is_ascii_alphanumeric() || "-_[]{}\\`^|".contains(ch) +} +fn irc_eq(left: &str, right: &str) -> bool { + left.chars().map(irc_fold).eq(right.chars().map(irc_fold)) +} +fn irc_fold(ch: char) -> char { + match ch { + 'A'..='Z' => ch.to_ascii_lowercase(), + '[' | '{' => '{', + ']' | '}' => '}', + '\\' | '|' => '|', + '^' | '~' => '~', + _ => ch, + } +} + +fn map_name(name: &str) -> String { + let mapped: String = name + .chars() + .filter(|ch| !ch.is_control()) + .map(|ch| if ch == '<' || ch == '>' { '_' } else { ch }) + .take(64) + .collect(); + let mapped = mapped.split_whitespace().collect::>().join(" "); + if mapped.is_empty() { + "unknown".into() + } else { + mapped + } +} + +fn valid_message(message: &str) -> bool { + !message.trim().is_empty() + && message.len() <= MAX_MESSAGE_BYTES + && !message.as_bytes().contains(&0) +} +fn clean_message(message: &str) -> String { + message + .chars() + .map(|ch| { + if ch == '\r' || ch == '\n' || ch == '\0' { + ' ' + } else { + ch + } + }) + .collect::() + .trim() + .to_owned() +} +fn sanitize_irc_atom(value: &str) -> String { + value + .chars() + .filter(|ch| !ch.is_control() && *ch != ' ' && *ch != ',' && *ch != ':') + .collect() +} + +fn truncate_utf8(value: &str, max: usize) -> String { + if value.len() <= max { + return value.into(); + } + let mut end = max; + while !value.is_char_boundary(end) { + end -= 1; + } + value[..end].into() +} + +fn redact_text(value: &str) -> String { + let lower = value.to_ascii_lowercase(); + if [ + "password", + "passwd", + "authorization", + "capability", + "token=", + "sasl", + "oauth", + ] + .iter() + .any(|marker| lower.contains(marker)) + { + return "".into(); + } + value + .split_whitespace() + .map(|word| { + if word.contains("://") { + "" + } else { + word + } + }) + .collect::>() + .join(" ") +} + +fn elapsed_ms(start: Instant) -> u64 { + u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX) +} +fn try_event(sender: &mpsc::Sender, event: BridgeEvent, dropped: &AtomicUsize) { + if sender.try_send(event).is_err() { + dropped.fetch_add(1, Ordering::Relaxed); + } +} +fn parse_uuid(value: &str, line: usize, reason: &'static str) -> Result { + UUID::new_with_string(value.into()).map_err(|_| ProgramError::InvalidScript { line, reason }) +} +fn io_error(action: impl Into, source: io::Error) -> ProgramError { + ProgramError::Io { + action: action.into(), + source, + } +} +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +#[cfg(test)] +mod tests { + use super::*; + + struct DropProbe(Arc); + + impl Drop for DropProbe { + fn drop(&mut self) { + self.0.store(false, Ordering::Release); + } + } + + impl BridgeTransport for DropProbe { + fn join_irc(&mut self, _channel: &str) -> Result<(), &'static str> { + Ok(()) + } + fn send_irc(&mut self, _channel: &str, _message: &str) -> Result<(), &'static str> { + Ok(()) + } + fn send_grid(&mut self, _message: &str) -> Result<(), &'static str> { + Ok(()) + } + fn accept_teleport(&mut self, _source: UUID, _session: UUID) -> Result<(), &'static str> { + Ok(()) + } + } + + #[test] + fn parses_partial_protocol_boundaries() { + let ping = parse_irc_line("PING :server.example").unwrap(); + assert_eq!(ping.command, "PING"); + assert_eq!(ping.trailing.as_deref(), Some("server.example")); + let message = parse_irc_line(":Alice!user@host PRIVMSG #room :hello UTF-8 ☃").unwrap(); + assert_eq!(message.prefix.as_deref(), Some("Alice!user@host")); + assert_eq!(message.params, ["#room"]); + assert_eq!(message.trailing.as_deref(), Some("hello UTF-8 ☃")); + let tagged = parse_irc_line("@time=now :Bob!u@h PRIVMSG #room :tagged").unwrap(); + assert_eq!(tagged.command, "PRIVMSG"); + assert_eq!(tagged.trailing.as_deref(), Some("tagged")); + assert!(parse_irc_line(":broken").is_none()); + } + + #[tokio::test] + async fn line_reader_handles_fragments_and_rejects_unbounded_input() { + let (mut writer, reader) = tokio::io::duplex(MAX_IRC_INPUT_BYTES * 2); + let sender = tokio::spawn(async move { + writer.write_all(b":nick PRIV").await.unwrap(); + writer.write_all(b"MSG #room :hello\r\n").await.unwrap(); + }); + let mut reader = AsyncBufReader::new(reader); + assert_eq!( + read_irc_line(&mut reader).await.unwrap().as_deref(), + Some(":nick PRIVMSG #room :hello\r\n") + ); + sender.await.unwrap(); + + let bytes = vec![b'x'; MAX_IRC_INPUT_BYTES + 1]; + let mut reader = AsyncBufReader::new(bytes.as_slice()); + assert_eq!( + read_irc_line(&mut reader).await.unwrap_err().kind(), + io::ErrorKind::InvalidData + ); + } + + #[test] + fn utf8_truncation_never_splits_a_scalar() { + assert_eq!(truncate_utf8("ab☃cd", 4), "ab"); + assert_eq!(truncate_utf8("ab☃cd", 5), "ab☃"); + } + + #[test] + fn duplicate_and_loop_detection_are_directional() { + let master = UUID::new_with_string("11111111-2222-3333-4444-555555555555".into()).unwrap(); + let own = UUID::new_with_string("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee".into()).unwrap(); + let other = UUID::new_with_string("bbbbbbbb-cccc-dddd-eeee-ffffffffffff".into()).unwrap(); + let mut bridge = Bridge::new( + BridgeConfig { + channel: "#room".into(), + nickname: "SLGateway".into(), + messages_per_second: 10, + burst: 10, + }, + master, + own, + ); + let mut transport = ScriptTransport::default(); + let logger = ScriptLogger::default(); + bridge.handle(BridgeEvent::IrcConnected, 0, &mut transport, &logger); + let event = BridgeEvent::GridChat { + source_id: other, + name: "Alice".into(), + message: "hello".into(), + normal: true, + fully_audible: true, + }; + bridge.handle(event.clone(), 1, &mut transport, &logger); + bridge.handle(event, 2, &mut transport, &logger); + bridge.handle( + BridgeEvent::IrcMessage { + target: "#room".into(), + nick: "Relay".into(), + message: " hello".into(), + }, + 3, + &mut transport, + &logger, + ); + assert_eq!( + transport + .calls + .iter() + .filter(|call| call.contains("irc-privmsg")) + .count(), + 1 + ); + assert!( + !transport + .calls + .iter() + .any(|call| call.contains("grid-chat")) + ); + } + + #[test] + fn rate_limit_queues_and_drains() { + let master = UUID::new_with_string("11111111-2222-3333-4444-555555555555".into()).unwrap(); + let own = UUID::new_with_string("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee".into()).unwrap(); + let mut bridge = Bridge::new( + BridgeConfig { + channel: "#room".into(), + nickname: "SLGateway".into(), + messages_per_second: 1, + burst: 1, + }, + master, + own, + ); + let mut transport = ScriptTransport::default(); + let logger = ScriptLogger::default(); + bridge.handle(BridgeEvent::IrcConnected, 0, &mut transport, &logger); + bridge.handle( + BridgeEvent::GridLogin("one".into()), + 0, + &mut transport, + &logger, + ); + bridge.handle( + BridgeEvent::GridLogin("two".into()), + 0, + &mut transport, + &logger, + ); + assert_eq!(bridge.pending.len(), 1); + bridge.tick(1_000, &mut transport, &logger); + assert!(bridge.pending.is_empty()); + } + + #[test] + fn bounded_bridge_queue_drops_oldest_messages() { + let master = UUID::new_with_string("11111111-2222-3333-4444-555555555555".into()).unwrap(); + let own = UUID::new_with_string("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee".into()).unwrap(); + let mut bridge = Bridge::new( + BridgeConfig { + channel: "#room".into(), + nickname: "SLGateway".into(), + messages_per_second: 1, + burst: 1, + }, + master, + own, + ); + for index in 0..=BRIDGE_QUEUE_CAPACITY { + bridge.enqueue(Direction::ToIrc, format!("message-{index}")); + } + assert_eq!(bridge.pending.len(), BRIDGE_QUEUE_CAPACITY); + assert_eq!(bridge.dropped, 1); + assert_eq!(bridge.pending.front().unwrap().message, "message-1"); + } + + #[test] + fn irc_channel_comparison_uses_rfc1459_case_mapping() { + assert!(irc_eq("#Rust[Bridge]", "#rUST{bRIDGE}")); + assert!(irc_eq("Nick\\Name", "nICK|nAME")); + assert!(!irc_eq("#one", "#two")); + } + + #[tokio::test] + async fn cancellation_joins_workers_and_drops_transport() { + let master = UUID::new_with_string("11111111-2222-3333-4444-555555555555".into()).unwrap(); + let own = UUID::new_with_string("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee".into()).unwrap(); + let config = BridgeConfig { + channel: "#room".into(), + nickname: "SLGateway".into(), + messages_per_second: 1, + burst: 1, + }; + let alive = Arc::new(AtomicBool::new(true)); + let bridge_cancel = CancellationTokenSource::new(); + let (event_tx, event_rx) = mpsc::channel(1); + let bridge = tokio::spawn(bridge_loop( + Bridge::new(config, master, own), + DropProbe(Arc::clone(&alive)), + event_rx, + Arc::new(ScriptLogger::default()), + bridge_cancel.token(), + )); + bridge_cancel.cancel(); + drop(event_tx); + tokio::time::timeout(Duration::from_secs(1), bridge) + .await + .expect("bridge stopped before timeout") + .expect("bridge worker did not panic"); + assert!(!alive.load(Ordering::Acquire)); + + let irc_cancel = CancellationTokenSource::new(); + let (command_tx, command_rx) = mpsc::channel(1); + let (events, _event_rx) = mpsc::channel(1); + let irc = tokio::spawn(irc_loop(IrcRuntime { + host: "127.0.0.1".into(), + port: 0, + nick: "SLGateway".into(), + real_name: "test".into(), + commands: command_rx, + events, + dropped: Arc::new(AtomicUsize::new(0)), + cancellation: irc_cancel.token(), + })); + irc_cancel.cancel(); + drop(command_tx); + tokio::time::timeout(Duration::from_secs(1), irc) + .await + .expect("IRC worker stopped before timeout") + .expect("IRC worker did not panic"); + } +} diff --git a/programs/src/lib.rs b/programs/src/lib.rs index 1f504c4..a8867da 100644 --- a/programs/src/lib.rs +++ b/programs/src/lib.rs @@ -2,6 +2,7 @@ pub mod commands; pub mod inventory_explorer; +pub mod irc_gateway; pub mod osd_inspector; pub mod packet_dump; pub mod prim_inspector; diff --git a/programs/tests/irc_gateway_cli.rs b/programs/tests/irc_gateway_cli.rs new file mode 100644 index 0000000..e870120 --- /dev/null +++ b/programs/tests/irc_gateway_cli.rs @@ -0,0 +1,222 @@ +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; + +const MASTER: &str = "11111111-2222-3333-4444-555555555555"; +const SESSION: &str = "22222222-3333-4444-5555-666666666666"; +const OTHER: &str = "bbbbbbbb-cccc-dddd-eeee-ffffffffffff"; +const SELF_ID: &str = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"; +const EXIT_USAGE: i32 = 2; +const EXIT_INPUT: i32 = 3; + +static TEMP_ID: AtomicU64 = AtomicU64::new(0); + +struct TestDir(PathBuf); + +impl TestDir { + fn new(name: &str) -> Self { + let id = TEMP_ID.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir().join(format!( + "metacrate-irc-gateway-{name}-{}-{id}", + std::process::id() + )); + fs::create_dir(&path).expect("create test directory"); + Self(path) + } + + fn write(&self, contents: &str) -> PathBuf { + let path = self.0.join("bridge.tsv"); + fs::write(&path, contents).expect("write bridge script"); + path + } +} + +impl Drop for TestDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +fn path_text(path: &Path) -> &str { + path.to_str().expect("UTF-8 path") +} + +fn run(args: &[&str]) -> Output { + Command::new(env!("CARGO_BIN_EXE_irc-gateway")) + .args(args) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .expect("run irc-gateway") +} + +fn stdout(output: &Output) -> &str { + std::str::from_utf8(&output.stdout).expect("UTF-8 stdout") +} + +fn stderr(output: &Output) -> &str { + std::str::from_utf8(&output.stderr).expect("UTF-8 stderr") +} + +fn assert_exit(output: &Output, code: i32) { + assert_eq!( + output.status.code(), + Some(code), + "stdout:\n{}\nstderr:\n{}", + stdout(output), + stderr(output) + ); +} + +#[test] +fn help_preserves_upstream_arguments_and_documents_controls() { + let output = run(&["--help"]); + assert!(output.status.success()); + let help = stdout(&output); + for marker in [ + "[FIRSTNAME]", + "[LASTNAME]", + "[PASSWORD]", + "[MASTER_UUID]", + "[IRC_HOST]", + "[IRC_PORT]", + "[#CHANNEL]", + "--fake-script", + "--messages-per-second", + "--burst", + "irc-connect", + "grid-chat", + "teleport", + ] { + assert!(help.contains(marker), "help omitted {marker}:\n{help}"); + } + + let output = run(&[]); + assert_exit(&output, EXIT_USAGE); +} + +#[test] +fn scripted_bidirectional_bridge_filters_maps_and_accepts_only_master() { + let directory = TestDir::new("bidirectional"); + let script = directory.write(&format!( + "irc-connect\n\ + grid-login\tconnected to https://grid.invalid/cap/private\n\ + grid-chat\t{OTHER}\tAlice \tnormal\tfull\thello IRC ☃\n\ + grid-chat\t{OTHER}\tAlice \twhisper\tfull\tignored type\n\ + grid-chat\t{OTHER}\tAlice \tnormal\tpartial\tignored range\n\ + grid-chat\t{SELF_ID}\tGateway Resident\tnormal\tfull\tignored self\n\ + irc\t#metacrate\tBob\thello grid ☃\n\ + irc\t#other\tMallory\tignored channel\n\ + irc\t#metacrate\tSLGateway\tignored self\n\ + teleport\t{OTHER}\t{SESSION}\n\ + teleport\t{MASTER}\t{SESSION}\n\ + shutdown\n" + )); + let output = run(&["--fake-script", path_text(&script)]); + assert!(output.status.success(), "{}", stderr(&output)); + assert!(output.stderr.is_empty()); + let text = stdout(&output); + for expected in [ + "CALL irc-join #metacrate", + "CALL irc-privmsg #metacrate hello IRC ☃", + "CALL grid-chat channel=0 type=Normal hello grid ☃", + &format!("CALL teleport-lure-respond {MASTER} {SESSION} accept=true"), + "active-transports=0 active-tasks=0 shutdown=true", + ] { + assert!(text.contains(expected), "missing {expected}:\n{text}"); + } + for rejected in [ + "ignored type", + "ignored range", + "ignored self", + "ignored channel", + &format!("teleport-lure-respond {OTHER}"), + "grid.invalid", + "/cap/private", + ] { + assert!( + !text.contains(rejected), + "leaked/forwarded {rejected}:\n{text}" + ); + } + assert!(text.contains("")); +} + +#[test] +fn reconnect_rate_limit_duplicate_and_loop_scenarios_are_deterministic() { + let directory = TestDir::new("resilience"); + let script = directory.write(&format!( + "irc-connect\n\ + grid-chat\t{OTHER}\tAlice\tnormal\tfull\tone\n\ + grid-chat\t{OTHER}\tAlice\tnormal\tfull\ttwo\n\ + grid-chat\t{OTHER}\tAlice\tnormal\tfull\ttwo\n\ + irc\t#metacrate\tRelay\t one\n\ + irc-disconnect\n\ + grid-chat\t{OTHER}\tAlice\tnormal\tfull\twhile offline\n\ + advance\t1000\n\ + irc-connect\n\ + advance\t1000\n\ + shutdown\n" + )); + let output = run(&[ + "--fake-script", + path_text(&script), + "--messages-per-second", + "1", + "--burst", + "1", + ]); + assert!(output.status.success(), "{}", stderr(&output)); + let text = stdout(&output); + assert_eq!( + text.matches("CALL irc-join #metacrate").count(), + 2, + "{text}" + ); + assert_eq!( + text.matches(" two").count(), + 2, + "one log and one transport call expected:\n{text}" + ); + assert_eq!( + text.matches(" one").count(), + 2, + "one log and one call expected:\n{text}" + ); + assert!( + !text.contains("CALL grid-chat"), + "loop was reflected:\n{text}" + ); + assert!( + text.contains("CALL irc-privmsg #metacrate while offline"), + "{text}" + ); + assert!(text.contains("queued=0"), "{text}"); +} + +#[test] +fn malformed_and_oversized_inputs_fail_without_network_access() { + let directory = TestDir::new("invalid"); + let invalid = directory.write("irc\tmissing fields\n"); + let output = run(&["--fake-script", path_text(&invalid)]); + assert_exit(&output, EXIT_INPUT); + assert!(stderr(&output).contains("line 1")); + + let oversized = directory.write(&format!("grid-login\t{}\n", "x".repeat(4097))); + let output = run(&["--fake-script", path_text(&oversized)]); + assert_exit(&output, EXIT_INPUT); + + let output = run(&[ + "First", + "Last", + "secret", + "not-a-uuid", + "irc.invalid", + "6667", + "#room", + ]); + assert_exit(&output, EXIT_USAGE); + assert!(!stderr(&output).contains("secret")); +}