Chili Palmer
d1e05d1b30
All checks were successful
Native Rust workspace compile / compile (push) Successful in 21m47s
1574 lines
53 KiB
Rust
1574 lines
53 KiB
Rust
//! 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<String>,
|
|
/// Avatar last name. May also be supplied as `GRID_LAST_NAME`.
|
|
#[arg(value_name = "LASTNAME")]
|
|
last_name: Option<String>,
|
|
/// Avatar password. May also be supplied as `GRID_PASSWORD`.
|
|
#[arg(value_name = "PASSWORD")]
|
|
password: Option<String>,
|
|
/// UUID whose teleport lures are accepted. May also be `GRID_MASTER_UUID`.
|
|
#[arg(value_name = "MASTER_UUID")]
|
|
master_uuid: Option<String>,
|
|
/// IRC server host. May also be `IRC_HOST`.
|
|
#[arg(value_name = "IRC_HOST")]
|
|
irc_host: Option<String>,
|
|
/// IRC server port. May also be `IRC_PORT`.
|
|
#[arg(value_name = "IRC_PORT")]
|
|
irc_port: Option<u16>,
|
|
/// IRC channel, including its leading '#'. May also be `IRC_CHANNEL`.
|
|
#[arg(value_name = "#CHANNEL")]
|
|
channel: Option<String>,
|
|
|
|
/// Run a deterministic, bounded offline bridge script.
|
|
#[arg(long, value_name = "FILE")]
|
|
fake_script: Option<PathBuf>,
|
|
/// 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<String>,
|
|
/// 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<String>,
|
|
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<PendingMessage>,
|
|
recent: VecDeque<RecentMessage>,
|
|
inbound_recent: VecDeque<RecentMessage>,
|
|
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<String>,
|
|
}
|
|
|
|
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<GridClient>,
|
|
network: NetworkManager,
|
|
subscriptions: Mutex<Vec<Subscription>>,
|
|
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<T>(&self, action: impl FnOnce(&mut AgentManager) -> T) -> T {
|
|
action(lock(&self.client).self_())
|
|
}
|
|
|
|
fn install(self: &Arc<Self>, sender: &mpsc::Sender<BridgeEvent>, dropped: &Arc<AtomicUsize>) {
|
|
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<LiveGrid>,
|
|
irc: mpsc::Sender<IrcCommand>,
|
|
}
|
|
|
|
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<IrcCommand>,
|
|
events: mpsc::Sender<BridgeEvent>,
|
|
dropped: Arc<AtomicUsize>,
|
|
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<LiveConfig, ProgramError> {
|
|
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<String>,
|
|
env: &str,
|
|
message: &'static str,
|
|
) -> Result<String, ProgramError> {
|
|
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<Option<bool>, 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<LiveGrid>,
|
|
event_tx: mpsc::Sender<BridgeEvent>,
|
|
irc_tx: mpsc::Sender<IrcCommand>,
|
|
cancellation: CancellationTokenSource,
|
|
bridge: tokio::task::JoinHandle<Bridge>,
|
|
irc: tokio::task::JoinHandle<()>,
|
|
dropped: Arc<AtomicUsize>,
|
|
) {
|
|
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<T: BridgeTransport + Send + 'static>(
|
|
mut bridge: Bridge,
|
|
mut transport: T,
|
|
mut events: mpsc::Receiver<BridgeEvent>,
|
|
logger: Arc<dyn BridgeLogger>,
|
|
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<IrcCommand>,
|
|
events: &mpsc::Sender<BridgeEvent>,
|
|
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<R: AsyncBufRead + Unpin>(reader: &mut R) -> io::Result<Option<String>> {
|
|
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<String>,
|
|
command: String,
|
|
params: Vec<String>,
|
|
trailing: Option<String>,
|
|
}
|
|
|
|
fn parse_irc_line(line: &str) -> Option<IrcMessage> {
|
|
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<Vec<String>>,
|
|
}
|
|
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<Vec<ScriptEvent>, 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<ScriptEvent, ProgramError> {
|
|
let fields: Vec<_> = line.split('\t').collect();
|
|
let invalid = || ProgramError::InvalidScript {
|
|
line: number,
|
|
reason: "expected irc-connect, irc-disconnect, irc<TAB>target<TAB>nick<TAB>message, grid-chat<TAB>uuid<TAB>name<TAB>normal|other<TAB>full|other<TAB>message, grid-login<TAB>message, teleport<TAB>source-uuid<TAB>session-uuid, advance<TAB>milliseconds, 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::<Vec<_>>().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::<String>()
|
|
.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 "<redacted>".into();
|
|
}
|
|
value
|
|
.split_whitespace()
|
|
.map(|word| {
|
|
if word.contains("://") {
|
|
"<redacted-url>"
|
|
} else {
|
|
word
|
|
}
|
|
})
|
|
.collect::<Vec<_>>()
|
|
.join(" ")
|
|
}
|
|
|
|
fn elapsed_ms(start: Instant) -> u64 {
|
|
u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX)
|
|
}
|
|
fn try_event(sender: &mpsc::Sender<BridgeEvent>, 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, ProgramError> {
|
|
UUID::new_with_string(value.into()).map_err(|_| ProgramError::InvalidScript { line, reason })
|
|
}
|
|
fn io_error(action: impl Into<String>, source: io::Error) -> ProgramError {
|
|
ProgramError::Io {
|
|
action: action.into(),
|
|
source,
|
|
}
|
|
}
|
|
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
|
|
mutex
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
struct DropProbe(Arc<AtomicBool>);
|
|
|
|
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: "<Alice> 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");
|
|
}
|
|
}
|