diff --git a/.gitignore b/.gitignore index 0c6be7b..faa40cd 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ *.local.* /demo/ /bin/ +**pycache** diff --git a/src/archive.rs b/src/archive.rs index 1221f97..930e5b1 100644 --- a/src/archive.rs +++ b/src/archive.rs @@ -465,10 +465,7 @@ pub fn checkpoint_all(root: &Path) -> Result { /// Retire redo before an atomic snapshot replacement. The old native snapshot /// is synced first, so a crash after unlink and before rename retains every /// committed record. After rename there is no old redo to resurrect deletions. -fn prepare_rewrite(p: &Path) -> Result<()> { - if recover_redo(p)?.is_none() { - return Ok(()); - } +fn retire_redo(p: &Path) -> Result<()> { std::fs::File::open(p)?.sync_all()?; std::fs::remove_file(p.with_extension("rredo"))?; std::fs::File::open(p.parent().unwrap())?.sync_all()?; @@ -479,6 +476,34 @@ fn prepare_rewrite(p: &Path) -> Result<()> { .remove(p); Ok(()) } +/// True when the redo journal no longer matches its native baseline after an +/// external in-place write or replacement. In that state the journal is no +/// longer authoritative and must not be replayed over the native file. +fn is_untrusted_redo(error: &anyhow::Error) -> bool { + let text = format!("{error:#}"); + text.contains("archive redo baseline changed in place") + || text.contains("archive shorter than redo baseline") + || text.contains("missing archive redo baseline") + || text.contains("archive redo journal exceeds bound") +} +/// Retire redo before an atomic snapshot replacement or with an explicit offline +/// handoff (storage-checkpoint). A journal whose baseline was externally +/// rewritten in place can no longer be trusted: because the caller is about to +/// replace the native file (rewrite) or to hand control of it to a native +/// writer (checkpoint), the current durably synced native becomes authoritative +/// and the untrusted journal is retired rather than failing the operation. +fn prepare_rewrite(p: &Path) -> Result<()> { + match recover_redo(p) { + Ok(Some(_)) => retire_redo(p), + Ok(None) => Ok(()), + Err(error) if is_untrusted_redo(&error) => { + // The journal no longer matches its baseline. Treat the native file + // as authoritative, sync it, and retire the untrusted journal. + retire_redo(p) + } + Err(error) => Err(error), + } +} fn scan_unlocked Result<()>>( root: &Path, host: &str, @@ -1012,4 +1037,120 @@ mod redo_rewrite_tests { .is_empty() ); } + + /// Build an archive large enough to have rolled its redo journal, so the + /// baseline is non-empty (`base > 0`) and the anchor guard is meaningful. + fn rolled_fixture() -> (tempfile::TempDir, std::path::PathBuf) { + let root = tempfile::tempdir().unwrap(); + // One more than REDO_FRAMES forces a redo rotation at frame 128, which + // snapshots the current native length as the new non-empty baseline. + for i in 0..=REDO_FRAMES as u64 { + append_new( + root.path(), + "example.org", + Some("alice"), + "archive", + Entry { + id: format!("roll-{i}"), + when: i as f64, + peer: String::new(), + xml: "x".into(), + }, + ) + .unwrap(); + } + let path = storage::path(root.path(), "example.org", Some("alice"), "archive") + .with_extension("list"); + // Recovery must see a non-empty baseline (rotation happened). + let data = std::fs::read(path.with_extension("rredo")).unwrap(); + let base = u64::from_le_bytes(data[8..16].try_into().unwrap()); + assert!(base > 0, "fixture should have rolled a non-empty baseline"); + (root, path) + } + + /// An external writer that rewrote the native archive in place over the + /// committed baseline without a handoff leaves the redo journal untrusted. + fn corrupt_baseline(path: &std::path::Path) -> u64 { + let data = std::fs::read(path.with_extension("rredo")).unwrap(); + let base = u64::from_le_bytes(data[8..16].try_into().unwrap()); + let mut native = std::fs::read(path).unwrap(); + // Flip the trailing bytes of the committed baseline so its anchor changes. + let start = (base.saturating_sub(64)) as usize; + for i in start..native.len().min(base as usize) { + native[i] ^= 0x5a; + } + let mut file = std::fs::OpenOptions::new() + .write(true) + .truncate(false) + .open(path) + .unwrap(); + std::io::Write::write_all(&mut file, &native).unwrap(); + file.sync_all().unwrap(); + base + } + + #[test] + fn runtime_append_fails_closed_on_untrusted_redo_baseline() { + let (root, path) = rolled_fixture(); + corrupt_baseline(&path); + // A live append must fail closed (not silently overwrite the native + // file): the journal no longer matches its baseline. + let err = append_new( + root.path(), + "example.org", + Some("alice"), + "archive", + Entry { + id: "post-corruption".into(), + when: 100000., + peer: String::new(), + xml: "new".into(), + }, + ) + .expect_err("live append must fail on an untrusted redo baseline"); + assert!( + is_untrusted_redo(&err), + "expected an untrusted-baseline error, got: {err:#}" + ); + // The journal is preserved for explicit offline handoff. + assert!(path.with_extension("rredo").exists()); + } + + #[test] + fn offline_checkpoint_retires_untrusted_redo_baseline() { + let (root, path) = rolled_fixture(); + let base = corrupt_baseline(&path); + assert!(base > 0); + assert!(path.with_extension("rredo").exists()); + // An external rewrite removed the trusted baseline; an explicit offline + // handoff accepts the current synced native as authoritative and retires + // the untrusted journal so the archive becomes writable again. + let retired = checkpoint_all(root.path()).unwrap(); + assert_eq!(retired, 1); + assert!( + !path.with_extension("rredo").exists(), + "offline checkpoint must retire the untrusted journal" + ); + // The archive accepts new appends: with the journal retired the current + // (now authoritative) native length becomes the fresh baseline. + let id = append_new( + root.path(), + "example.org", + Some("alice"), + "archive", + Entry { + id: "after-handoff".into(), + when: 200000., + peer: String::new(), + xml: "new".into(), + }, + ) + .unwrap(); + assert_eq!(id, "after-handoff"); + // The original baseline tail was externally rewritten, so the native + // holds whatever the external writer put there; checkpoint only retires + // the journal (handoff), it does not try to re-encode damaged records. + // A fresh valid journal is (re)created on the next append. + assert!(path.with_extension("rredo").exists()); + } } diff --git a/src/server.rs b/src/server.rs index c7d240e..0f135fd 100644 --- a/src/server.rs +++ b/src/server.rs @@ -2887,6 +2887,43 @@ fn archive_policy(rt: &Runtime) -> &str { _ => "always", } } +/// Sanitized, payload-free label for a session fault: values come from the +/// error chain only and never contain stanza payload, tokens or passwords. +fn sanitized_label(error: &anyhow::Error) -> String { + error + .chain() + .map(|x| x.to_string()) + .find(|s| s.chars().all(|c| c.is_ascii() && !c.is_control())) + .unwrap_or_default() + .chars() + .take(160) + .collect() +} +/// Best-effort MAM archival so a storage fault (for example an untrusted redo +/// baseline after an external write to the data directory) cannot take down +/// message delivery. History is an attribute of delivery, not a prerequisite: +/// on failure the original stanza is returned so the session stays up and the +/// message is still routed. The fault is logged without payload. +fn archive_message_best_effort( + rt: &Runtime, + ctx: &mut ConnectionLog, + owner: &str, + peer: &str, + xml: &str, + time: f64, +) -> String { + match crate::mam::archive_for_user(&rt.data, &rt.domain, owner, peer, xml, time) { + Ok((_, stamped)) => stamped, + Err(error) => { + ctx.event( + Level::Warn, + "mam-archive-skipped", + &[("err", &sanitized_label(&error))], + ); + xml.to_string() + } + } +} fn routed(n: roxmltree::Node<'_, '_>, from: &str, to: &str) -> String { protocol::address(n, from, to) } @@ -3647,15 +3684,14 @@ async fn authenticated_transport( boolean(&rt.config, &rt.domain, "mam_smart_enable", false), )? { - let (_, stamped) = crate::mam::archive_for_user( - &rt.data, - &rt.domain, + sent_xml = archive_message_best_effort( + &rt, + ctx, &user, protocol::bare(to), &xml, now, - )?; - sent_xml = stamped; + ); } else { sent_xml = xml.clone(); } @@ -3677,10 +3713,7 @@ async fn authenticated_transport( boolean(&rt.config, &rt.domain, "mam_smart_enable", false), )? { - let (_, stamped) = crate::mam::archive_for_user( - &rt.data, &rt.domain, u, &owner, &xml, now, - )?; - xml = stamped; + xml = archive_message_best_effort(&rt, ctx, u, &owner, &xml, now); } message_phases[2] = recipient_mam_started.elapsed(); } @@ -4053,15 +4086,29 @@ async fn incoming_federated(rt: Runtime, stanza: crate::federation::FederatedSta boolean(&rt.config, &rt.domain, "mam_smart_enable", false), )? { - delivery_xml = crate::mam::archive_for_user( + let time = chrono::Utc::now().timestamp_millis() as f64 / 1000.0; + delivery_xml = match crate::mam::archive_for_user( &rt.data, &rt.domain, &user, protocol::bare(from), &delivery_xml, - chrono::Utc::now().timestamp_millis() as f64 / 1000.0, - )? - .1; + time, + ) { + Ok((_, stamped)) => stamped, + Err(error) => { + // A storage fault must not drop an inbound federated message: + // deliver it unarchived so it still reaches the recipient. + rt.logger.event( + Level::Warn, + "s2s", + None, + "mam-archive-skipped", + &[("err", &sanitized_label(&error))], + ); + delivery_xml + } + }; } let targets = rt.hub.message_targets_for_type(to, typ); let mut delivered = 0; diff --git a/tests/mam_broken_baseline_network.py b/tests/mam_broken_baseline_network.py new file mode 100644 index 0000000..654c65a --- /dev/null +++ b/tests/mam_broken_baseline_network.py @@ -0,0 +1,338 @@ +#!/usr/bin/env python3 +"""A damaged MAM redo baseline must not take down the live message session. + +Usage: python3 tests/mam_broken_baseline_network.py target/release/rust-xmpp + +Reproduces the production symptom reported with Monal: the native archive at +`data//archive/.list` no longer matches the anchor recorded in its +`.rredo` journal ("archive redo baseline changed in place"). On every message the +previous build drained the session with `connection-closed reason= +protocol-or-transport-error`, so the client reconnect-looped and its message was +never delivered. + +The archive store is a durable attribute of delivery, not a prerequisite. This +test seeds an archive for `load-0`, corrupts its redo baseline offline to force +exactly that recovery error, then verifies that: + * a fresh message is still routed load-0 -> load-1, + * BOTH sessions survive (ping round-trip), no `connection-closed`, + * the runtime stays fail-closed (the untrusted journal is NOT silently dropped), + * the offline `storage-checkpoint` repair retires the untrusted journal and + restores MAM history reads on the now-authoritative native file. +""" + +import argparse +import hashlib +import json +import socket +import subprocess +import sys +import tempfile +import time +import xml.etree.ElementTree as ET +from pathlib import Path + +from _wire_client import Client, local + +REDO_MAGIC = b"XARRED01" +REDO_HEADER = 96 +BODY = "must-survive-broken-baseline" + + +def start(binary, config, port, log): + server = subprocess.Popen( + [binary, "serve", str(config), "--experimental"], stdout=log, stderr=log + ) + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + assert server.poll() is None, "server stopped during startup" + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.1): + return server + except OSError: + time.sleep(0.05) + server.kill() + server.wait() + raise AssertionError("server did not start") + + +def stop(server): + server.terminate() + try: + code = server.wait(timeout=10) + except subprocess.TimeoutExpired: + server.kill() + server.wait() + raise AssertionError("SIGTERM did not complete orderly shutdown") + assert code == 0, f"SIGTERM shutdown exited {code}" + + +def corrupt_baseline(path): + """Force the `archive redo baseline changed in place` recovery error. + + Rebuilds a rotated-style journal header leaving the native file untouched: + the header records the native length as a non-empty baseline but stores a + wrong anchor. `recover_redo` then rejects it exactly as when an external + writer rewrote the native file in place after the journal rotated. + """ + native = path.read_bytes() + length = len(native) + assert length > 0, "seed produced an empty native MAM archive" + header = bytearray(REDO_HEADER) + header[:8] = REDO_MAGIC + header[8:16] = length.to_bytes(8, "little") + header[16:32] = b"\0" * 16 # dev/ino are only a cache identity, not validated + tail = native[-min(length, 4096) :] + real_anchor = hashlib.sha256(tail).digest() + wrong = bytearray(real_anchor) + wrong[0] ^= 0xFF + header[32:64] = wrong + header[64:96] = hashlib.sha256(header[:64]).digest() + path.with_suffix(".rredo").write_bytes(header) + assert hashlib.sha256(header[:64]).digest() == header[64:96], ( + "rebuilt header checksum" + ) + assert header[32:64] != real_anchor, "anchor did not diverge from native" + + +def run(binary): + with tempfile.TemporaryDirectory(prefix="xmpp-mam-broken-") as tmp: + root = Path(tmp) + subprocess.run( + [ + "openssl", + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-days", + "1", + "-keyout", + str(root / "key.pem"), + "-out", + str(root / "cert.pem"), + "-subj", + "/CN=localhost", + "-addext", + "subjectAltName=DNS:localhost", + ], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + config = root / "config.json" + config.write_text( + json.dumps( + { + "global": { + "data_path": str(root / "data"), + "interfaces": ["127.0.0.1"], + "c2s_ports": [port], + "s2s_ports": [], + "http_ports": [], + "https_ports": [], + "modules_enabled": [ + "tls", + "saslauth", + "smacks", + "ping", + "mam", + "carbons", + ], + "smacks_hibernation_time": 600, + "archive_expires_after": "never", + "default_archive_policy": True, + "ssl": { + "key": str(root / "key.pem"), + "certificate": str(root / "cert.pem"), + }, + }, + "hosts": { + "localhost": { + "options": { + "authentication": "internal_hashed", + "password_hash": "SHA-256", + } + } + }, + } + ) + ) + for number in range(2): + subprocess.run( + [ + binary, + "create-user", + str(config), + f"load-{number}", + "--password-stdin", + ], + input="load-test-password\n", + text=True, + check=True, + capture_output=True, + ) + + with (root / "server.log").open("w+") as log: + server = None + try: + # Seed: leave a real, valid MAM entry for load-0 so its archive + # carries a genuine .list + .rredo pair with a non-zero baseline. + server = start(binary, config, port, log) + alice = Client( + port, root / "cert.pem", "load-0", "load-test-password", "desktop" + ) + bob = Client( + port, root / "cert.pem", "load-1", "load-test-password", "desktop" + ) + alice.available() + bob.available() + alice.send( + f"seed" + ) + bob.body("seed") + alice.sync() + stop(server) + server = None + for client in (alice, bob): + client.close() + finally: + if server is not None and server.poll() is None: + server.kill() + server.wait() + + # The sender localpart is percent-encoded on disk (`load-0` -> `load%2d0`). + sender_archive = root / "data" / "localhost" / "archive" / "load%2d0.list" + assert sender_archive.exists(), "seed did not create load-0 MAM archive" + assert sender_archive.with_suffix(".rredo").exists(), ( + "seed did not attach load-0 redo journal" + ) + corrupt_baseline(sender_archive) + damaged_journal = sender_archive.with_suffix(".rredo").read_bytes() + + with (root / "server.log").open("a") as log: + server = None + try: + server = start(binary, config, port, log) + alice = Client( + port, root / "cert.pem", "load-0", "load-test-password", "desktop" + ) + bob = Client( + port, root / "cert.pem", "load-1", "load-test-password", "desktop" + ) + alice.available() + bob.available() + # A damaged MAM baseline must not drop the dispatch: deliver and + # keep both sessions alive. + alice.send( + f"{BODY}" + ) + bob.body(BODY) + # Ping round-trips prove neither session was closed by the archive fault. + alice.sync() + bob.sync() + alice.send( + f"second" + ) + bob.body("second") + alice.sync() + bob.sync() + finally: + if server is not None and server.poll() is None: + server.kill() + server.wait() + + log_text = (root / "server.log").read_text() + skipped = [l for l in log_text.splitlines() if "mam-archive-skipped" in l] + assert skipped, ( + "server did not log the best-effort archive skip (isolation inactive?)" + ) + assert not any( + "connection-closed" in l and "protocol-or-transport-error" in l + for l in log_text.splitlines() + ), "a session was closed by the archive fault" + assert not any("session-error-label" in l for l in log_text.splitlines()), ( + "unhandled archive fault labelled a session error" + ) + + # Runtime must stay fail-closed: the untrusted journal is preserved. + assert sender_archive.with_suffix(".rredo").read_bytes() == damaged_journal, ( + "runtime dropped the untrusted redo journal instead of failing closed" + ) + + # Offline handoff must repair: checkpoint retires the untrusted journal + # so the now-authoritative native file becomes a clean one-journal base. + result = subprocess.run( + [binary, "storage-checkpoint", str(config)], + capture_output=True, + text=True, + check=True, + ) + assert "Durable native archive handoff:" in result.stdout + assert not sender_archive.with_suffix(".rredo").exists(), ( + "storage-checkpoint did not retire the untrusted redo journal" + ) + + # After repair MAM history is readable again (endpoint reads now succeed). + phase3_offset = len((root / "server.log").read_text().splitlines()) + with (root / "server.log").open("a") as log: + server = None + try: + server = start(binary, config, port, log) + alice = Client( + port, root / "cert.pem", "load-0", "load-test-password", "desktop" + ) + bob = Client( + port, root / "cert.pem", "load-1", "load-test-password", "desktop" + ) + alice.available() + bob.available() + # A MAM query to load-0's own archive must serve history. + # Query the bare account JID (not the full resource). + alice.send( + f"" + f"" + f"" + f"urn:xmpp:mam:2" + ) + result_iq = alice.iq("mam") + assert ( + result_iq.find( + ".//{urn:ietf:params:xml:ns:xmpp-stanzas}internal-server-error" + ) + is None + ), ET.tostring(result_iq) + # After reconnect the sender archive is fresh (journal was retired); + # a new message archives cleanly. + alice.send( + f"post-repair" + ) + bob.body("post-repair") + alice.sync() + finally: + if server is not None and server.poll() is None: + server.kill() + server.wait() + + post_log = (root / "server.log").read_text().splitlines()[phase3_offset:] + fresh_errors = [ + l for l in post_log if "archive redo baseline changed in place" in l + ] + assert not fresh_errors, ( + "archive fault returned after offline repair:\n" + "\n".join(fresh_errors) + ) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--binary", default="target/release/rust-xmpp") + args = parser.parse_args() + run(args.binary) + print("PASS: damaged MAM redo baseline does not drop live message delivery") + + +if __name__ == "__main__": + main()