From 90167dd46f40398f793896c792f329970929ed9a Mon Sep 17 00:00:00 2001 From: Jean Date: Thu, 16 Jul 2026 10:56:22 +0000 Subject: [PATCH] db: persist jupes across restart (event-source them) A fold-parity property test caught it: jupes were plain in-memory Db fields that logged no event, so after any services restart every juped server was silently un-juped and could relink. Moved jupes into NetData (like akills/ forbids/groups) and event-sourced them with JupeAdded/JupeRemoved (Local scope), so a jupe now replays from the log. Adds the round-trip property test (live state == replayed state) as a permanent guard against this bug class. --- src/engine/db/event.rs | 19 ++++++++- src/engine/db/mod.rs | 9 ++--- src/engine/db/network.rs | 17 ++++---- src/engine/db/tests.rs | 83 ++++++++++++++++++++++++++++++++++++++++ src/engine/mod.rs | 2 + src/grpc.rs | 2 + 6 files changed, 119 insertions(+), 13 deletions(-) diff --git a/src/engine/db/event.rs b/src/engine/db/event.rs index 80c5555..591f729 100644 --- a/src/engine/db/event.rs +++ b/src/engine/db/event.rs @@ -132,6 +132,10 @@ pub enum Event { // "NICK", "CHAN", or "EMAIL". ForbidAdded { kind: String, mask: String, setter: String, reason: String, ts: u64 }, ForbidRemoved { kind: String, mask: String }, + // Server jupes (OperServ JUPE). Node-local (each network holds a rogue server + // by minting a fake one on `sid`), so Local scope but still persisted. + JupeAdded { name: String, sid: String, reason: String }, + JupeRemoved { name: String }, } // Whether an event replicates across the federation. Account identity is Global @@ -232,7 +236,9 @@ impl Event { | Event::ChannelUsed { .. } | Event::ChannelNoExpire { .. } | Event::ChannelExpiryWarned { .. } - | Event::ChannelOperNoteSet { .. } => Scope::Local, + | Event::ChannelOperNoteSet { .. } + | Event::JupeAdded { .. } + | Event::JupeRemoved { .. } => Scope::Local, } } } @@ -603,6 +609,17 @@ pub(crate) fn apply(accounts: &mut HashMap, channels: &mut Hash Event::ForbidRemoved { kind, mask } => { net.forbids.retain(|f| !(f.kind == kind && f.mask.eq_ignore_ascii_case(&mask))); } + Event::JupeAdded { name, sid, reason } => { + // Idempotent over a snapshot; keep jupe_seq ahead so a fresh add can't + // reuse a sid. + if !net.jupes.iter().any(|j| j.name.eq_ignore_ascii_case(&name)) { + net.jupes.push(Jupe { name, sid, reason }); + net.jupe_seq += 1; + } + } + Event::JupeRemoved { name } => { + net.jupes.retain(|j| !j.name.eq_ignore_ascii_case(&name)); + } Event::AccountExpiryWarned { account } => { if let Some(a) = accounts.get_mut(&key(&account)) { a.expiry_warned = true; diff --git a/src/engine/db/mod.rs b/src/engine/db/mod.rs index f55bef9..f8ffb61 100644 --- a/src/engine/db/mod.rs +++ b/src/engine/db/mod.rs @@ -301,6 +301,9 @@ pub struct NetData { pub opers: HashMap, // Session-limit exceptions (OperServ SESSION): per-IP-mask allowances. pub sess_exceptions: Vec, + // Server jupes (OperServ JUPE), and the counter for minting their fake SIDs. + pub jupes: Vec, + pub jupe_seq: u32, // The bot auto-assigned to newly registered channels (BotServ AUTOASSIGN), // if any. Casefolded nick; None disables auto-assignment. pub default_bot: Option, @@ -947,10 +950,6 @@ pub struct Db { net: NetData, // Services ignores (OperServ IGNORE), node-local and in-memory. ignores: Vec, - // Juped servers (OperServ JUPE), node-local: the introducing node owns them, - // so they aren't federated (that would collide SIDs across nodes). - jupes: Vec, - jupe_seq: u32, // Network defence level (OperServ DEFCON), 5 = normal down to 1 = lockdown. // Node-local; the derived restrictions are read by the engine and services. defcon: u8, @@ -1026,7 +1025,7 @@ impl Db { apply(&mut accounts, &mut channels, &mut grouped, &mut bots, &mut host_cfg, &mut net, event); } tracing::info!(accounts = accounts.len(), channels = channels.len(), "account store loaded"); - Self { accounts, channels, grouped, log, scram_iterations: scram::DEFAULT_ITERATIONS, email_enabled: false, email_brand: "Network Services".to_string(), email_accent: "#4f46e5".to_string(), email_logo: String::new(), codes: HashMap::new(), auth_fails: HashMap::new(), vhost_req_times: HashMap::new(), report_times: HashMap::new(), bots, host_cfg, net, ignores: Vec::new(), jupes: Vec::new(), jupe_seq: 0, defcon: 5, external_accounts: false } + Self { accounts, channels, grouped, log, scram_iterations: scram::DEFAULT_ITERATIONS, email_enabled: false, email_brand: "Network Services".to_string(), email_accent: "#4f46e5".to_string(), email_logo: String::new(), codes: HashMap::new(), auth_fails: HashMap::new(), vhost_req_times: HashMap::new(), report_times: HashMap::new(), bots, host_cfg, net, ignores: Vec::new(), defcon: 5, external_accounts: false } } /// Fold an entry authored by another node into the store — the services-side diff --git a/src/engine/db/network.rs b/src/engine/db/network.rs index e6ac001..294abc6 100644 --- a/src/engine/db/network.rs +++ b/src/engine/db/network.rs @@ -221,25 +221,28 @@ impl Db { /// Jupe a server name: allocate a fake sid, store it, return the sid to /// introduce (or the existing sid if the name is already juped). pub fn jupe_add(&mut self, name: &str, reason: &str) -> String { - if let Some(j) = self.jupes.iter().find(|j| j.name.eq_ignore_ascii_case(name)) { + if let Some(j) = self.net.jupes.iter().find(|j| j.name.eq_ignore_ascii_case(name)) { return j.sid.clone(); } - let sid = jupe_sid(self.jupe_seq); - self.jupe_seq += 1; - self.jupes.push(Jupe { name: name.to_string(), sid: sid.clone(), reason: reason.to_string() }); + let sid = jupe_sid(self.net.jupe_seq); + // Persisted (Local scope) so a juped server stays juped across a restart. + let _ = self.log.append(Event::JupeAdded { name: name.to_string(), sid: sid.clone(), reason: reason.to_string() }); + self.net.jupe_seq += 1; + self.net.jupes.push(Jupe { name: name.to_string(), sid: sid.clone(), reason: reason.to_string() }); sid } /// Lift a jupe. Returns the sid to squit if it existed. pub fn jupe_del(&mut self, name: &str) -> Option { - let sid = self.jupes.iter().find(|j| j.name.eq_ignore_ascii_case(name)).map(|j| j.sid.clone())?; - self.jupes.retain(|j| !j.name.eq_ignore_ascii_case(name)); + let sid = self.net.jupes.iter().find(|j| j.name.eq_ignore_ascii_case(name)).map(|j| j.sid.clone())?; + let _ = self.log.append(Event::JupeRemoved { name: name.to_string() }); + self.net.jupes.retain(|j| !j.name.eq_ignore_ascii_case(name)); Some(sid) } /// The juped servers, as (name, sid, reason). pub fn jupes(&self) -> Vec<(String, String, String)> { - self.jupes.iter().map(|j| (j.name.clone(), j.sid.clone(), j.reason.clone())).collect() + self.net.jupes.iter().map(|j| (j.name.clone(), j.sid.clone(), j.reason.clone())).collect() } /// File an abuse report, rate-limited per reporter. Returns the new report's diff --git a/src/engine/db/tests.rs b/src/engine/db/tests.rs index 9194b61..9e1a2e5 100644 --- a/src/engine/db/tests.rs +++ b/src/engine/db/tests.rs @@ -462,6 +462,89 @@ assert_eq!(db.channel("#c").unwrap().join_mode("alice"), None, "removal survives reopen"); } + // A jupe must survive a services restart — otherwise every juped server is + // silently un-juped on restart and can relink. + #[test] + fn jupes_survive_a_restart() { + let p = tmp("jupepersist"); + let sid = { + let mut db = Db::open(&p, "N1"); + let sid = db.jupe_add("rogue.example", "abuse"); + assert_eq!(db.jupes().len(), 1, "juped while live"); + sid + }; + let db = Db::open(&p, "N1"); + assert_eq!(db.jupes(), vec![("rogue.example".to_string(), sid, "abuse".to_string())], "the jupe replays from the log"); + } + + // Fold parity: the live state after a broad sequence of writes must equal the + // state rebuilt by replaying the same log — every event's manual live mutation + // has to match its `apply()` arm, or a restart silently changes the data. + #[test] + fn live_state_matches_replay_after_a_broad_sequence() { + let p = tmp("foldparity"); + let snapshot = |db: &Db| -> String { + let mut a: Vec<_> = db.accounts().cloned().collect(); + a.sort_by(|x, y| x.name.cmp(&y.name)); + let mut c: Vec<_> = db.channels().cloned().collect(); + c.sort_by(|x, y| x.name.cmp(&y.name)); + let mut gn = db.groups(); + gn.sort(); + let gv: Vec<_> = gn.iter().filter_map(|n| db.group(n)).collect(); + let mut ak = db.akills(); + ak.sort_by(|x, y| x.mask.cmp(&y.mask)); + let mut fb = db.forbids(); + fb.sort_by(|x, y| x.mask.cmp(&y.mask)); + let mut of = db.vhost_offers().to_vec(); + of.sort(); + let mut jp = db.jupes(); + jp.sort(); + format!("{a:?}\n{c:?}\n{gv:?}\n{ak:?}\n{fb:?}\n{of:?}\n{jp:?}") + }; + + let live = { + let mut db = Db::open(&p, "N1"); + db.scram_iterations = 4096; + db.register("alice", "pw", Some("a@x.io".into())).unwrap(); + db.register("bob", "pw", None).unwrap(); + db.register("carol", "pw", None).unwrap(); + db.register_channel("#chan", "bob").unwrap(); + db.register_channel("#other", "carol").unwrap(); + db.access_add("#chan", "alice", "op").unwrap(); + db.akick_add("#chan", "*!*@bad.host", "spam").unwrap(); + db.set_desc("#chan", "a place").unwrap(); + db.set_url("#chan", "https://x.io").unwrap(); + db.set_channel_email("#chan", "s@x.io").unwrap(); + db.set_mlock("#chan", "nt", "s").unwrap(); + db.set_channel_setting("#chan", ChanSetting::Private, true).unwrap(); + db.set_successor("#other", Some("alice")).unwrap(); + db.set_greet("alice", "hi there").unwrap(); + db.set_account_autoop("alice", false).unwrap(); + db.set_account_kill("alice", false).unwrap(); + db.set_account_hide_status("alice", true).unwrap(); + db.ajoin_add("alice", "#chan", "").unwrap(); + db.set_vhost("alice", "alice.vhost", "oper", None).unwrap(); + db.certfp_add("alice", "aabbccddeeff00112233445566778899").unwrap(); + db.suspend_account("bob", "oper", "bad", None).unwrap(); + db.memo_send("carol", "alice", "hello there", false).unwrap(); + db.group_register("!grp", "alice").unwrap(); + db.group_set_flags("!grp", "carol", "").unwrap(); + db.group_register("!other", "bob").unwrap(); + db.group_set_flags("!other", "alice", "").unwrap(); + db.akill_add("G", "*@evil", "oper", "spam", None).unwrap(); + db.forbid_add("NICK", "bad*", "oper", "squat").unwrap(); + db.vhost_offer_add("cool.vhost").unwrap(); + db.jupe_add("rogue.server", "abuse"); + // The compound one: dropping alice must purge her access, successorship + // and group standing — identically in the live path and on replay. + db.drop_account("alice").unwrap(); + db.drop_channel("#other").unwrap(); + snapshot(&db) + }; + let replayed = snapshot(&Db::open(&p, "N1")); + assert_eq!(live, replayed, "live state must equal the state replayed from the log"); + } + // Dropping an account erases every reference to it — channel access and // successorship, group membership and foundership — so a later re-registration // of the same name can't inherit its standing. Holds live and after replay. diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 4a38251..d87c817 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -1300,6 +1300,8 @@ fn audit_summary(event: &db::Event) -> Option { AkillRemoved { kind, mask } => format!("lifted the {} on \x02{mask}\x02", ban_kind_label(kind)), ForbidAdded { kind, mask, reason, .. } => format!("forbade {} \x02{mask}\x02 ({reason})", kind.to_ascii_lowercase()), ForbidRemoved { kind, mask } => format!("un-forbade {} \x02{mask}\x02", kind.to_ascii_lowercase()), + JupeAdded { name, reason, .. } => format!("juped server \x02{name}\x02 ({reason})"), + JupeRemoved { name } => format!("lifted the jupe on \x02{name}\x02"), AccountOperNoteSet { account, note } => match note { Some(_) => format!("set a staff note on \x02{account}\x02"), None => format!("cleared the staff note on \x02{account}\x02"), diff --git a/src/grpc.rs b/src/grpc.rs index 7c5774a..6de557c 100644 --- a/src/grpc.rs +++ b/src/grpc.rs @@ -179,6 +179,8 @@ fn to_wire(entry: &LogEntry) -> Option { | Event::AkillRemoved { .. } | Event::ForbidAdded { .. } | Event::ForbidRemoved { .. } + | Event::JupeAdded { .. } + | Event::JupeRemoved { .. } | Event::AccountExpiryWarned { .. } | Event::ChannelExpiryWarned { .. } | Event::AccountOperNoteSet { .. }