From aab64baf869b98c532872a3c817a24e5ea163f75 Mon Sep 17 00:00:00 2001 From: Jean Date: Sun, 12 Jul 2026 13:34:38 +0000 Subject: [PATCH] nickserv: log out the loser of a registration conflict When a peer's registration wins a name a local session was logged into, that session authenticated against a credential that no longer exists. The node now clears its accountname (the ircd emits RPL_LOGGEDOUT) and tells the user why. A new outbound channel lets the engine push these services-initiated actions to the uplink from the gossip path, drained by the link loop alongside incoming traffic. --- modules/nickserv/nickserv.rs | 3 ++ src/engine/db.rs | 26 ++++++++++-- src/engine/mod.rs | 79 +++++++++++++++++++++++++++++++++++- src/engine/service.rs | 5 +++ src/engine/state.rs | 9 ++++ src/link.rs | 69 ++++++++++++++++++------------- src/main.rs | 6 ++- 7 files changed, 164 insertions(+), 33 deletions(-) diff --git a/modules/nickserv/nickserv.rs b/modules/nickserv/nickserv.rs index bb2bdbe..7cbf5b5 100644 --- a/modules/nickserv/nickserv.rs +++ b/modules/nickserv/nickserv.rs @@ -21,6 +21,9 @@ impl Service for NickServ { fn gecos(&self) -> &str { "Nickname Services" } + fn manages_accounts(&self) -> bool { + true + } fn on_command(&mut self, from: &Sender, args: &[&str], ctx: &mut ServiceCtx, _net: &Network, db: &mut Db) { let me = self.uid.as_str(); diff --git a/src/engine/db.rs b/src/engine/db.rs index 36b0f8c..2aad47e 100644 --- a/src/engine/db.rs +++ b/src/engine/db.rs @@ -206,6 +206,13 @@ pub struct LogEntry { event: Event, } +#[cfg(test)] +impl LogEntry { + pub(crate) fn for_test(origin: &str, seq: u64, lamport: u64, event: Event) -> Self { + LogEntry { origin: origin.to_string(), seq, lamport, event } + } +} + // Append-only log, the sole persistent source of truth: `open` replays it, // `append` stamps and writes a locally-authored entry, and `ingest` folds in an // entry authored by another node. The version vector (highest seq applied per @@ -439,12 +446,25 @@ impl Db { } /// Fold an entry authored by another node into the store — the services-side - /// of the gossip seam. Idempotent (re-delivered entries are dropped). - pub fn ingest(&mut self, entry: LogEntry) -> std::io::Result<()> { + /// of the gossip seam. Idempotent (re-delivered entries are dropped). Returns + /// the account name if this ingest changed its owner (a registration conflict + /// resolved against the local claim), so the caller can log out stale sessions. + pub fn ingest(&mut self, entry: LogEntry) -> std::io::Result> { + // Record the current owner of an incoming account before applying, to see + // if this ingest transfers ownership away from us. + let contested = match &entry.event { + Event::AccountRegistered(a) => self.accounts.get(&key(&a.name)).map(|cur| (a.name.clone(), cur.home.clone())), + _ => None, + }; if let Some(event) = self.log.ingest(entry)? { apply(&mut self.accounts, &mut self.channels, event); } - Ok(()) + if let Some((name, prev_home)) = contested { + if self.accounts.get(&key(&name)).is_some_and(|cur| cur.home != prev_home) { + return Ok(Some(name)); // ownership moved to another node + } + } + Ok(None) } /// Rewrite the log to one entry per account and channel, reclaiming churn. diff --git a/src/engine/mod.rs b/src/engine/mod.rs index c4885a7..da40351 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -7,6 +7,7 @@ use std::collections::HashMap; use std::time::{Duration, Instant}; use base64::{engine::general_purpose::STANDARD, Engine as _}; +use tokio::sync::mpsc; use crate::proto::{NetAction, NetEvent, RegReply}; use db::{Db, LogEntry, RegError}; @@ -74,11 +75,14 @@ pub struct Engine { sasl_sessions: HashMap, // client uid -> in-progress exchange reg_limiter: RegLimiter, chan_service: Option, // uid to source channel modes from (ChanServ) + nick_service: Option, // uid of the account service (NickServ), for its notices + irc_out: Option>, // services-initiated actions -> the uplink } impl Engine { pub fn new(services: Vec>, db: Db) -> Self { let chan_service = services.iter().find(|s| s.manages_channels()).map(|s| s.uid().to_string()); + let nick_service = services.iter().find(|s| s.manages_accounts()).map(|s| s.uid().to_string()); Self { services, network: Network::default(), @@ -86,6 +90,21 @@ impl Engine { sasl_sessions: HashMap::new(), reg_limiter: RegLimiter::new(), chan_service, + nick_service, + irc_out: None, + } + } + + // Wire the channel the link layer drains, so the engine can push actions to the + // uplink that no ircd event asked for (here: a forced logout after losing a + // registration conflict). + pub fn set_irc_out(&mut self, tx: mpsc::UnboundedSender) { + self.irc_out = Some(tx); + } + + fn emit_irc(&self, action: NetAction) { + if let Some(tx) = &self.irc_out { + let _ = tx.send(action); // unbounded: never blocks; only fails if the link is down } } @@ -104,7 +123,29 @@ impl Engine { } pub fn gossip_ingest(&mut self, entry: LogEntry) -> std::io::Result<()> { - self.db.ingest(entry) + // If ingesting a peer's registration took an account name away from a + // local session (it lost the conflict), log that session out. + if let Some(account) = self.db.ingest(entry)? { + self.logout_lost_sessions(&account); + } + Ok(()) + } + + // A registration conflict resolved against this node: the name `account` is now + // owned by another node. Any local session logged into it authenticated against + // the old owner's credential, which no longer exists — force them out and say why. + fn logout_lost_sessions(&mut self, account: &str) { + for uid in self.network.uids_logged_into(account) { + self.network.clear_account(&uid); + self.emit_irc(NetAction::Metadata { target: uid.clone(), key: "accountname".to_string(), value: String::new() }); + if let Some(ns) = &self.nick_service { + self.emit_irc(NetAction::Notice { + from: ns.clone(), + to: uid, + text: format!("Your registration of \x02{account}\x02 collided with another network and it now belongs elsewhere. You have been logged out; please register a different name."), + }); + } + } } // Compact the log if it has grown past the live account count. Returns @@ -988,6 +1029,42 @@ mod tests { assert!(out.iter().any(|a| matches!(a, NetAction::Metadata { key, value, .. } if key == "accountname" && value == "realnick")), "must log into the current nick's account: {out:?}"); } + // Losing a registration conflict logs out the local session on that name and + // notifies them over the services-initiated outbound path. + #[test] + fn lost_conflict_logs_out_local_session() { + let mut e = engine_with("lostconf", "alice", "sesame"); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + e.set_irc_out(tx); + // A local user logs into alice. + e.handle(NetEvent::UserConnect { uid: "000AAAAAB".into(), nick: "alice".into(), host: "h".into() }); + e.handle(NetEvent::Privmsg { from: "000AAAAAB".into(), to: "42SAAAAAA".into(), text: "IDENTIFY sesame".into() }); + assert_eq!(e.network.account_of("000AAAAAB"), Some("alice"), "logged in before the conflict"); + + // An earlier claim from another node wins and takes the name over. + let winner = db::Account { + name: "alice".into(), password_hash: "OTHER".into(), email: None, + ts: 0, home: "peer".into(), scram256: None, scram512: None, certfps: vec![], + }; + let entry = LogEntry::for_test("peer", 0, 1, db::Event::AccountRegistered(winner)); + e.gossip_ingest(entry).unwrap(); + + // The local session is logged out... + assert!(e.network.account_of("000AAAAAB").is_none(), "session cleared after losing the name"); + // ...and the outbound path got a logout (empty accountname) plus a notice. + let mut logout = false; + let mut notice = false; + while let Ok(a) = rx.try_recv() { + match a { + NetAction::Metadata { target, key, value } if target == "000AAAAAB" && key == "accountname" && value.is_empty() => logout = true, + NetAction::Notice { to, text, .. } if to == "000AAAAAB" && text.contains("collided") => notice = true, + _ => {} + } + } + assert!(logout, "a logout must be pushed to the uplink"); + assert!(notice, "the user must be told why"); + } + // IDENTIFY accepts the two-arg account+password form: log into a named // account even when the current nick differs; a wrong password is refused. #[test] diff --git a/src/engine/service.rs b/src/engine/service.rs index 8786a50..ab6b5ba 100644 --- a/src/engine/service.rs +++ b/src/engine/service.rs @@ -24,6 +24,11 @@ pub trait Service: Send { fn manages_channels(&self) -> bool { false } + // Whether this is the account service (NickServ), so the engine can source + // account-related notices from it. + fn manages_accounts(&self) -> bool { + false + } fn on_command(&mut self, from: &Sender, args: &[&str], ctx: &mut ServiceCtx, net: &Network, db: &mut Db); } diff --git a/src/engine/state.rs b/src/engine/state.rs index 113efee..0daaf3a 100644 --- a/src/engine/state.rs +++ b/src/engine/state.rs @@ -81,6 +81,15 @@ impl Network { self.accounts.get(uid).map(String::as_str) } + // Uids of every live session currently identified to `account`. + pub fn uids_logged_into(&self, account: &str) -> Vec { + self.accounts + .iter() + .filter(|(_, a)| a.eq_ignore_ascii_case(account)) + .map(|(uid, _)| uid.clone()) + .collect() + } + pub fn set_account(&mut self, uid: &str, account: &str) { self.accounts.insert(uid.to_string(), account.to_string()); } diff --git a/src/link.rs b/src/link.rs index 696fde2..4eca24e 100644 --- a/src/link.rs +++ b/src/link.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use anyhow::Result; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpStream; -use tokio::sync::Mutex; +use tokio::sync::{mpsc, Mutex}; use crate::engine::db::Db; use crate::engine::Engine; @@ -12,7 +12,7 @@ use crate::proto::{NetAction, Protocol}; // One uplink session: connect, handshake + burst, then translate lines forever. // The engine is shared with the gossip layer, so it is locked per operation and // never held across the registration key-stretching await. -pub async fn run(mut proto: Box, engine: Arc>, addr: &str) -> Result<()> { +pub async fn run(mut proto: Box, engine: Arc>, addr: &str, mut irc_rx: mpsc::UnboundedReceiver) -> Result<()> { let stream = TcpStream::connect(addr).await?; let (read, mut write) = stream.into_split(); let mut lines = BufReader::new(read).lines(); @@ -27,35 +27,48 @@ pub async fn run(mut proto: Box, engine: Arc>, addr: } } - while let Some(line) = lines.next_line().await? { - tracing::debug!(dir = "<<", %line); - for event in proto.parse(&line) { - let actions = engine.lock().await.handle(event); - for action in actions { - let outs = match action { - // Registration: derive the password off the reactor so the - // ~1s of key stretching can't stall the link, after a cheap - // gate that rejects taken names and rate-limits floods. - NetAction::DeferRegister { account, password, email, reply } => { - let pre = engine.lock().await.pre_register_check(&account, &reply); - match pre { - Some(rejection) => rejection, - None => { - let iterations = engine.lock().await.scram_iterations(); - let creds = tokio::task::spawn_blocking(move || { - Db::derive_credentials(&password, iterations) - }) - .await?; - engine.lock().await.complete_register(&account, creds, email, reply) + loop { + tokio::select! { + // A line from the uplink: translate it and answer. + line = lines.next_line() => { + let Some(line) = line? else { break }; + tracing::debug!(dir = "<<", %line); + for event in proto.parse(&line) { + let actions = engine.lock().await.handle(event); + for action in actions { + let outs = match action { + // Registration: derive the password off the reactor so the + // ~1s of key stretching can't stall the link, after a cheap + // gate that rejects taken names and rate-limits floods. + NetAction::DeferRegister { account, password, email, reply } => { + let pre = engine.lock().await.pre_register_check(&account, &reply); + match pre { + Some(rejection) => rejection, + None => { + let iterations = engine.lock().await.scram_iterations(); + let creds = tokio::task::spawn_blocking(move || { + Db::derive_credentials(&password, iterations) + }) + .await?; + engine.lock().await.complete_register(&account, creds, email, reply) + } + } + } + action => vec![action], + }; + for act in outs { + for out in proto.serialize(&act) { + send(&mut write, &out).await?; } } } - action => vec![action], - }; - for act in outs { - for out in proto.serialize(&act) { - send(&mut write, &out).await?; - } + } + } + // A services-initiated action (e.g. a forced logout after a lost + // registration conflict), pushed by the engine from the gossip path. + Some(action) = irc_rx.recv() => { + for out in proto.serialize(&action) { + send(&mut write, &out).await?; } } } diff --git a/src/main.rs b/src/main.rs index 216d9aa..81e7bb2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -57,6 +57,10 @@ async fn main() -> Result<()> { db.set_outbound(gossip_tx.clone()); let engine = Arc::new(Mutex::new(Engine::new(services, db))); + // Channel for services-initiated actions to reach the uplink (drained by the link loop). + let (irc_tx, irc_rx) = tokio::sync::mpsc::unbounded_channel(); + engine.lock().await.set_irc_out(irc_tx); + if let Some(gossip) = cfg.gossip.clone() { tracing::info!(peers = cfg.peer.len(), "starting gossip"); tokio::spawn(gossip::run(engine.clone(), gossip, cfg.peer.clone(), cfg.server.sid.clone(), gossip_tx)); @@ -77,5 +81,5 @@ async fn main() -> Result<()> { let addr = format!("{}:{}", cfg.uplink.host, cfg.uplink.port); tracing::info!(server = %cfg.server.name, %addr, "linking to uplink"); - link::run(proto, engine, &addr).await + link::run(proto, engine, &addr, irc_rx).await }