From 3ffaee06951e33a619342cfc11b5228eba13dcde Mon Sep 17 00:00:00 2001 From: John Coffey Date: Sat, 19 Sep 2026 09:42:03 -0700 Subject: [PATCH] SCIM: a suspended account's open sessions end, and no credential it holds still works (SCIM-52) The push router records which account each subscription belongs to, and a new Revoke event drops every subscription the account holds, so its IMAP IDLE, JMAP event streams and WebSockets on this node end. Other users' subscriptions to its shared mailboxes stay. Test 28 now checks an open IDLE is ended, and that the account's password over HTTP and its own API key, both already cached, are refused on the next request. --- crates/common/src/ipc.rs | 5 ++ crates/scim/src/users.rs | 12 ++++ crates/services/src/state_manager/manager.rs | 13 ++++ crates/services/src/state_manager/mod.rs | 2 + tests/src/scim/acceptance.rs | 72 ++++++++++++++++++++ 5 files changed, 104 insertions(+) diff --git a/crates/common/src/ipc.rs b/crates/common/src/ipc.rs index ff8f405..12da9fb 100644 --- a/crates/common/src/ipc.rs +++ b/crates/common/src/ipc.rs @@ -43,6 +43,11 @@ pub enum PushEvent { account_id: u32, broadcast: bool, }, + // inbuxa: SCIM-52: ends the push subscriptions the account itself holds + // (IMAP IDLE, JMAP event streams and WebSockets) on this node + Revoke { + account_id: u32, + }, Stop, } diff --git a/crates/scim/src/users.rs b/crates/scim/src/users.rs index 0d74cde..992bee3 100644 --- a/crates/scim/src/users.rs +++ b/crates/scim/src/users.rs @@ -691,6 +691,18 @@ pub async fn replace( external_id, ); if active_change.is_some() { + // SCIM-52: sessions the account has open are ended + if !active { + let _ = ctx + .server + .inner + .ipc + .push_tx + .send(common::ipc::PushEvent::Revoke { + account_id: id.document_id(), + }) + .await; + } audit( ctx, if active { diff --git a/crates/services/src/state_manager/manager.rs b/crates/services/src/state_manager/manager.rs index 121d993..5b22b2d 100644 --- a/crates/services/src/state_manager/manager.rs +++ b/crates/services/src/state_manager/manager.rs @@ -49,6 +49,7 @@ pub fn spawn_push_router(inner: Arc, mut change_rx: mpsc::Receiver { + let owner = account_ids.first().copied().unwrap_or(u32::MAX); for account_id in account_ids { subscribers .entry(account_id) @@ -57,10 +58,22 @@ pub fn spawn_push_router(inner: Arc, mut change_rx: mpsc::Receiver { + for subscriber_list in subscribers.values_mut() { + subscriber_list + .ipc + .retain(|subscriber| subscriber.owner != account_id); + } + purge_needed = true; + } + PushEvent::PushServerRegister { activate, expired } => { for account_id in activate { subscribers.entry(account_id).or_default().is_push = true; diff --git a/crates/services/src/state_manager/mod.rs b/crates/services/src/state_manager/mod.rs index 851932c..98d274c 100644 --- a/crates/services/src/state_manager/mod.rs +++ b/crates/services/src/state_manager/mod.rs @@ -28,6 +28,8 @@ const SEND_TIMEOUT: Duration = Duration::from_millis(500); struct IpcSubscriber { types: Bitmap, tx: mpsc::Sender, + // inbuxa: SCIM-52: the account whose session subscribed + owner: u32, } #[derive(Debug)] diff --git a/tests/src/scim/acceptance.rs b/tests/src/scim/acceptance.rs index c33fedf..f65c3c3 100644 --- a/tests/src/scim/acceptance.rs +++ b/tests/src/scim/acceptance.rs @@ -1184,11 +1184,78 @@ async fn imap_login(address: &str, ok: bool) { assert_eq!(accepted, ok, "IMAP login of {address}"); } +type IdleLines = tokio::io::Lines>>; + +/// An IMAP session in IDLE. +async fn idle_session(address: &str) -> (IdleLines, tokio::io::WriteHalf) { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + let stream = tokio::net::TcpStream::connect("127.0.0.1:9991") + .await + .unwrap(); + let (reader, mut writer) = tokio::io::split(stream); + let mut lines = BufReader::new(reader).lines(); + lines.next_line().await.unwrap(); + for (tag, command) in [ + ("a", format!("LOGIN \"{address}\" \"{USER_SECRET}\"")), + ("b", "SELECT INBOX".to_string()), + ] { + writer + .write_all(format!("{tag} {command}\r\n").as_bytes()) + .await + .unwrap(); + loop { + let line = lines.next_line().await.unwrap().unwrap(); + if let Some(status) = line.strip_prefix(&format!("{tag} ")) { + assert!(status.starts_with("OK"), "{command}: {line}"); + break; + } + } + } + writer.write_all(b"c IDLE\r\n").await.unwrap(); + let line = lines.next_line().await.unwrap().unwrap(); + assert!(line.starts_with('+'), "IDLE: {line}"); + (lines, writer) +} + +/// Whether the server ends an IDLE session within ten seconds. +async fn idle_ends(idle: &mut (IdleLines, tokio::io::WriteHalf)) -> bool { + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + match tokio::time::timeout_at(deadline, idle.0.next_line()).await { + Ok(Ok(Some(line))) if line.starts_with("* BYE") => return true, + Ok(Ok(Some(_))) => continue, + Ok(Ok(None)) | Ok(Err(_)) => return true, + Err(_) => return false, + } + } +} + /// Test 28 (SCIM-52): suspension stops sign-in, not mail. async fn suspension(test: &TestServer, scim: &ScimTest) { let address = format!("suspended@{SCIM_DOMAIN}"); let id = manual_user(test, scim, "suspended").await; imap_login(&address, true).await; + + // Credentials used, and cached, before the suspension + let user = Account::new("suspended@scim.example.com", USER_SECRET, &[], "", id); + let (_, user_key) = api_key_with_id(&user, json!({"@type": "Inherit"})).await; + let basic = format!( + "Basic {}", + base64::Engine::encode( + &base64::engine::general_purpose::STANDARD, + format!("{address}:{USER_SECRET}") + ) + ); + let bearer = format!("Bearer {user_key}"); + for authorization in [&basic, &bearer] { + assert_eq!( + crate::scim::jmap_session_status(authorization).await, + 200, + "test 28" + ); + } + let mut idle = idle_session(&address).await; + scim.client .patch( &format!("/Users/{id}"), @@ -1197,6 +1264,11 @@ async fn suspension(test: &TestServer, scim: &ScimTest) { .await .assert_status(200); imap_login(&address, false).await; + assert!(idle_ends(&mut idle).await, "test 28: an open IDLE is ended"); + for authorization in [&basic, &bearer] { + let status = crate::scim::jmap_session_status(authorization).await; + assert!(matches!(status, 401 | 403), "test 28: {status}"); + } let mut lmtp = SmtpConnection::connect().await; lmtp.ingest( "sender@remote.example.org",