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",