From d3ee5ae8e523c5b687a5cf1621add77b6c58ef0c Mon Sep 17 00:00:00 2001 From: Ardalan Amiri Sani Date: Sat, 27 Jun 2026 00:33:48 -0700 Subject: [PATCH 1/3] (1) Use the server to send messages between apps when adding an app. (2) Support notifications for multiple apps in the server. --- app_native/Cargo.toml | 1 + app_native/examples/app.rs | 153 +++++----------- app_native/src/lib.rs | 8 +- client_lib/src/http_client.rs | 66 +++++++ client_lib/src/pairing.rs | 19 ++ server/src/fcm.rs | 131 ++++++++++++++ server/src/main.rs | 285 ++++++++++++++++++++---------- server/src/notification_target.rs | 222 +++++++++++++++++++++++ 8 files changed, 682 insertions(+), 203 deletions(-) diff --git a/app_native/Cargo.toml b/app_native/Cargo.toml index a183847..20d55ee 100644 --- a/app_native/Cargo.toml +++ b/app_native/Cargo.toml @@ -18,6 +18,7 @@ serde = "1.0" serde_derive = "1.0" openmls = "=0.8.1" docopt = "~1.1" +base64-url = "3.0.3" [features] default = [] diff --git a/app_native/examples/app.rs b/app_native/examples/app.rs index e39ce70..82a0e79 100644 --- a/app_native/examples/app.rs +++ b/app_native/examples/app.rs @@ -20,15 +20,12 @@ use docopt::Docopt; use std::env; use std::fs; use std::fs::File; -use std::io::{self, BufRead, BufReader, Write, Read}; +use std::io::{self, BufRead, BufReader, Write}; use std::path::Path; use std::sync::Arc; use std::sync::Mutex; use std::thread; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use std::net::{TcpListener, TcpStream, SocketAddr}; -use std::str::FromStr; -use std::io::ErrorKind; // This is a simple app that pairs with the Secluso camera, receives motion videos, // and launches livestream sessions. @@ -41,7 +38,6 @@ use std::io::ErrorKind; const CAMERA_ADDR: &str = "127.0.0.1"; const CAMERA_NAME: &str = "Camera"; const DATA_DIR: &str = "example_app_data"; -const FIRST_APP_ADDR: &str = "127.0.0.1"; pub const MAX_ALLOWED_MSG_LEN: u64 = 65536; @@ -94,6 +90,11 @@ fn main() -> io::Result<()> { let clients: Arc>>> = Arc::new(Mutex::new(None)); let http_client = HttpClient::new(server_addr, server_username, server_password); + // We assume here that the new secret is shared via + // another channel, e.g., QR code scan. + // Also, a new secret needs to be used for every app added. + let add_app_secret = vec![2u8; NUM_SECRET_BYTES]; + if first_time { if args.flag_reset { panic!("No state to reset!"); @@ -122,25 +123,19 @@ fn main() -> io::Result<()> { ) } else { println!("Sending the add_app request"); - let addr = SocketAddr::from_str(&(FIRST_APP_ADDR.to_owned() + ":12350")) - .map_err(|e| io::Error::other(format!("{e}")))?; - - let mut stream = TcpStream::connect(&addr)?; // get key packages let key_packages_vec = get_key_packages(&mut clients.lock().unwrap())?; - write_varying_len(&mut stream, &key_packages_vec)?; - - let new_app_data_vec = read_varying_len(&mut stream)?; - - // We assume here that the new secret is shared via - // another channel, e.g., QR code scan. - let new_secret = vec![2u8; NUM_SECRET_BYTES]; + println!("About to send add_app request"); + http_client.add_app_request("test_add_app_request_token", key_packages_vec)?; + println!("About to wait for add_app response"); + let new_app_data_vec = http_client.add_app_check("test_add_app_response_token")?; + println!("Received add_app response"); let epochs: [u64; NUM_MLS_CLIENTS] = join_camera_groups( &mut clients.lock().unwrap(), - new_secret.clone(), + add_app_secret.clone(), new_app_data_vec, )?; @@ -166,30 +161,35 @@ fn main() -> io::Result<()> { } } - let add_app_request: Arc>> = Arc::new(Mutex::new(None)); + let add_app_request: Arc>>> = Arc::new(Mutex::new(None)); if !args.flag_secondary_app { let add_app_request_clone = Arc::clone(&add_app_request); + let http_client_clone = http_client.clone(); thread::spawn(move || loop { - let listener = TcpListener::bind("0.0.0.0:12350").unwrap(); - for incoming in listener.incoming() { - match incoming { - Ok(stream) => { - println!("Incoming connection accepted."); - let mut stream_opt = add_app_request_clone.lock().unwrap(); - *stream_opt = Some(stream); - }, - - Err(e) => { - println!("Incoming connection error: {e}"); - } - } - } + println!("About to wait for add_app request"); + match http_client_clone.add_app_check("test_add_app_request_token") { + Ok(data) => { + println!("Received add_app request."); + let mut data_opt = add_app_request_clone.lock().unwrap(); + *data_opt = Some(data); + }, + + Err(e) => { + println!("Error listening for add_app requests: {e}"); + } + } }); } - main_loop(clients, http_client, add_app_request, args.flag_num_iters)?; + main_loop( + clients, + http_client, + add_app_request, + args.flag_num_iters, + add_app_secret, + )?; Ok(()) } @@ -212,8 +212,9 @@ fn deregister_all( fn main_loop( clients: Arc>>>, http_client: HttpClient, - add_app_request: Arc>>, + add_app_request: Arc>>>, num_iters: usize, + add_app_secret: Vec, ) -> io::Result<()> { for iter in 0..num_iters { thread::sleep(Duration::from_secs(1)); @@ -230,15 +231,16 @@ fn main_loop( livestream(Arc::clone(&clients), &http_client, 2)?; } - let mut add_app_stream_opt = add_app_request.lock().unwrap(); - if let Some(add_app_stream) = add_app_stream_opt.as_mut() { + let mut add_app_data_opt = add_app_request.lock().unwrap(); + if let Some(add_app_data) = add_app_data_opt.as_ref() { println!("Add app request detected"); handle_add_app_request( Arc::clone(&clients), &http_client, - add_app_stream, + add_app_data, + add_app_secret.clone(), )?; - *add_app_stream_opt = None; + *add_app_data_opt = None; } } @@ -248,15 +250,15 @@ fn main_loop( fn handle_add_app_request( clients: Arc>>>, http_client: &HttpClient, - stream: &mut TcpStream, + add_app_data: &Vec, + add_app_secret: Vec, ) -> io::Result<()> { println!("handle_add_app_request called"); - let new_secret = vec![2u8; NUM_SECRET_BYTES]; - let new_app_key_packages_vec = read_varying_len(stream)?; + let new_app_key_packages_vec = add_app_data.clone(); let config_msg_enc = - generate_add_app_request_config_command(&mut clients.lock().unwrap(), new_app_key_packages_vec, new_secret.clone())?; + generate_add_app_request_config_command(&mut clients.lock().unwrap(), new_app_key_packages_vec, add_app_secret.clone())?; let config_group_name = get_group_name(&mut clients.lock().unwrap(), "config")?; @@ -286,13 +288,13 @@ fn handle_add_app_request( let new_app_data_vec = process_add_app_config_response( &mut clients.lock().unwrap(), config_response.clone(), - new_secret, + add_app_secret, ).unwrap(); increment_epoch("motion_epoch"); increment_epoch("thumbnail_epoch"); - write_varying_len(stream, &new_app_data_vec)?; + http_client.add_app_request("test_add_app_response_token", new_app_data_vec)?; Ok(()) } @@ -516,66 +518,3 @@ fn fetch_livestream_chunk( format!("Error: could not fetch livestream chunk (timeout)!"), )); } - -// FIXME: copied from camera_hub/src/pairing.rs. -fn write_varying_len(stream: &mut TcpStream, msg: &[u8]) -> io::Result<()> { - // FIXME: is u64 necessary? - let len = msg.len() as u64; - let len_data = len.to_be_bytes(); - - stream.write_all(&len_data)?; - stream.write_all(msg)?; - stream.flush()?; - - Ok(()) -} - -fn read_varying_len(stream: &mut TcpStream) -> io::Result> { - let mut len_data = [0u8; 8]; - - match stream.read_exact(&mut len_data) { - Ok(_) => {} - Err(ref e) if e.kind() == ErrorKind::WouldBlock => { - return Err(io::Error::new( - ErrorKind::WouldBlock, - "Length read would block", - )); - } - Err(e) => return Err(e), - } - - let len = u64::from_be_bytes(len_data); - - if len > MAX_ALLOWED_MSG_LEN { - println!("Communicated message length ({len}) exceeds the allowed length ({MAX_ALLOWED_MSG_LEN})"); - return Err(io::Error::new( - ErrorKind::InvalidInput, - "Intended message length is too large", - )) - } - - let mut msg = vec![0u8; len as usize]; - let mut offset = 0; - - while offset < msg.len() { - match stream.read(&mut msg[offset..]) { - Ok(0) => { - return Err(io::Error::new( - ErrorKind::UnexpectedEof, - "Socket closed during read", - )) - } - Ok(n) => { - offset += n; - } - Err(ref e) if e.kind() == ErrorKind::WouldBlock => { - // retry a few times with a short delay - thread::sleep(Duration::from_millis(10)); - continue; - } - Err(e) => return Err(e), - } - } - - Ok(msg) -} diff --git a/app_native/src/lib.rs b/app_native/src/lib.rs index 2c9f721..7082ef3 100644 --- a/app_native/src/lib.rs +++ b/app_native/src/lib.rs @@ -17,7 +17,7 @@ use secluso_client_lib::mls_clients::{ CONFIG, FCM, LIVESTREAM, MLS_CLIENT_TAGS, MOTION, NUM_MLS_CLIENTS, THUMBNAIL, NUM_COMMON_MLS_CLIENTS, NUM_DEDICATED_MLS_CLIENTS, }; -use secluso_client_lib::pairing::{self, MAX_ALLOWED_MSG_LEN}; +use secluso_client_lib::pairing::{self, MAX_ALLOWED_MSG_LEN, generate_add_app_secret}; use secluso_client_lib::video::{encrypt_video_file, decrypt_video_file, decrypt_thumbnail_file}; use openmls::prelude::KeyPackage; use serde::{Deserialize, Serialize}; @@ -783,6 +783,12 @@ pub fn process_heartbeat_config_response( } } +pub fn get_add_app_secret() -> io::Result { + generate_add_app_secret() + .map_err(|e| io::Error::new(io::ErrorKind::Other, e)) + +} + pub fn get_key_packages(clients: &mut Option>) -> io::Result> { if clients.is_none() { return Err(io::Error::other( diff --git a/client_lib/src/http_client.rs b/client_lib/src/http_client.rs index 8717765..18cd746 100644 --- a/client_lib/src/http_client.rs +++ b/client_lib/src/http_client.rs @@ -22,6 +22,7 @@ const MAX_COMMAND_FILE_SIZE: u64 = 100 * 1024; // 100 kibibytes const MAX_CHECK_RESP_SIZE: u64 = 20 * 1024; // 20 kibibytes const MAX_NOTIFICATION_TARGET_SIZE: u64 = 10 * 1024; // 10 kibibytes const IOS_NOTIFICATION_RESP_MAX_SIZE: u64 = 10 * 1024; // 10 kibibytes +const MAX_ADD_APP_REQUEST_SIZE: u64 = 100 * 1024; // 100 kibibytes #[derive(Clone)] pub struct HttpClient { @@ -901,6 +902,71 @@ impl HttpClient { Ok(response_vec) } + + pub fn add_app_check(&self, op: &str) -> io::Result> { + let max_size = MAX_ADD_APP_REQUEST_SIZE; + + let server_url = format!("{}/add_app_check/{}", self.server_addr, op); + + let client = Client::builder() + .timeout(None) + .build() + .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?; + + let response = self.authorized_headers(client + .get(&server_url)) + .send() + .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?; + + if response.status() == StatusCode::CONFLICT { + Self::give_hint_to_updater(); + } + + if !response.status().is_success() { + return Err(io::Error::new( + io::ErrorKind::Other, + format!("Server error: {}", response.status()), + )); + } + + let mut data = Vec::new(); + let mut limited = response.take(max_size); + limited.read_to_end(&mut data)?; + + if data.len() >= max_size as usize { + return Err(io::Error::new( + io::ErrorKind::Other, + "Add app request exceeded maximum allowed size", + )); + } + + Ok(data) + } + + pub fn add_app_request(&self, op: &str, data: Vec) -> io::Result<()> { + let server_url = format!("{}/add_app_request/{}", self.server_addr, op); + + let client = Client::new(); + let response = self.authorized_headers(client + .post(server_url)) + .header("Content-Type", "application/octet-stream") + .body(data) + .send() + .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?; + + if response.status() == StatusCode::CONFLICT { + Self::give_hint_to_updater(); + } + + if !response.status().is_success() { + return Err(io::Error::new( + io::ErrorKind::Other, + format!("Server error: {}", response.status()), + )); + } + + Ok(()) + } } #[cfg(test)] diff --git a/client_lib/src/pairing.rs b/client_lib/src/pairing.rs index 53dc7f3..22ec189 100644 --- a/client_lib/src/pairing.rs +++ b/client_lib/src/pairing.rs @@ -185,6 +185,25 @@ pub fn generate_raspberry_camera_secret( Ok(()) } +pub fn generate_add_app_secret() -> anyhow::Result { + let crypto = OpenMlsRustCrypto::default(); + let secret = crypto + .crypto() + .random_vec(NUM_SECRET_BYTES) + .context("Failed to generate camera secret bytes")?; + + let add_app_secret = CameraSecret { + version: CAMERA_SECRET_VERSION.to_string(), + secret: base64_url::encode(&secret), + wifi_password: None, + }; + + let qr_content = serde_json::to_string(&add_app_secret) + .context("Failed to serialize add_app secret into JSON")?; + + Ok(qr_content) +} + impl App { pub fn new(key_package: KeyPackage) -> Self { Self { key_package } diff --git a/server/src/fcm.rs b/server/src/fcm.rs index fbc4a89..2532d9f 100644 --- a/server/src/fcm.rs +++ b/server/src/fcm.rs @@ -12,8 +12,18 @@ use reqwest::Url; use secluso_server_backbone::types::ConfigResponse; use serde::{Deserialize, Serialize}; use serde_json::json; +use std::collections::HashSet; use std::error::Error; +use std::io::{self, ErrorKind}; +use std::path::Path; use std::{fs, thread, time}; +use std::sync::OnceLock; +use rocket::tokio::sync::Mutex; + +use rocket::tokio::fs as tokio_fs; +use rocket::tokio::io::AsyncWriteExt; + +use crate::security::check_path_sandboxed; // Fixed bundle id used to locate the Firebase app const BUNDLE_ID: &str = "com.secluso.mobile"; @@ -25,6 +35,10 @@ const OAUTH_TOKEN_ALLOWED_HOSTS: &[&str] = &[ "accounts.google.com", ]; +const LEGACY_FCM_TOKEN_FILE: &str = "fcm_token"; +const FCM_TOKENS_DIR: &str = "fcm_tokens"; +const FCM_TOKEN_FILE_PREFIX: &str = "fcm_token_"; + // In this file we send very sensitive stuff over HTTP requests: a JWT assertion signed with the // Firebase service-account private key, bearer access tokens, and push payloads tied to user // devices. If any endpoint URL is quietly changed to plain http:// or to a lookalike host, @@ -563,3 +577,120 @@ pub fn send_notification(device_token: String, msg: Vec) -> Result<(), Box> = OnceLock::new(); + +// FIXME: we only add fcm_tokens, but never remove stale ones. +pub(crate) async fn store_fcm_token(root: &Path, token: &str) -> io::Result<()> { + // Two overlapping calls to store_fcm_token from one app for the same token can + // result in duplicates. This lock is used to prevent that. + let lock = FCM_TOKEN_STORE_LOCK.get_or_init(|| Mutex::new(())); + let _guard = lock.lock().await; + + tokio_fs::create_dir_all(root).await?; + + let tokens_dir = root.join(FCM_TOKENS_DIR); + check_path_sandboxed(root, &tokens_dir)?; + tokio_fs::create_dir_all(&tokens_dir).await?; + + let mut entries = tokio_fs::read_dir(&tokens_dir).await?; + while let Some(entry) = entries.next_entry().await? { + if !entry.file_type().await?.is_file() { + continue; + } + + let file_name = entry.file_name(); + let Some(file_name) = file_name.to_str() else { + continue; + }; + + if !file_name.starts_with(FCM_TOKEN_FILE_PREFIX) { + continue; + } + + let path = entry.path(); + check_path_sandboxed(root, &path)?; + + let existing = tokio_fs::read_to_string(&path).await?; + if existing.trim() == token { + return Ok(()); + } + } + + for index in 1u64.. { + let token_path = tokens_dir.join(format!("{}{}", FCM_TOKEN_FILE_PREFIX, index)); + check_path_sandboxed(root, &token_path)?; + + let mut file = match tokio_fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&token_path) + .await + { + Ok(file) => file, + Err(e) if e.kind() == ErrorKind::AlreadyExists => continue, + Err(e) => return Err(e), + }; + + file.write_all(token.as_bytes()).await?; + file.sync_all().await?; + return Ok(()); + } + + unreachable!() +} + +pub(crate) async fn load_fcm_tokens(root: &Path) -> io::Result> { + let mut tokens = Vec::new(); + let mut seen = HashSet::new(); + + if !root.exists() { + return Ok(tokens); + } + + let legacy_token_path = root.join(LEGACY_FCM_TOKEN_FILE); + check_path_sandboxed(root, &legacy_token_path)?; + if legacy_token_path.exists() { + let token = tokio_fs::read_to_string(&legacy_token_path).await?; + let token = token.trim().to_string(); + + if !token.is_empty() && seen.insert(token.clone()) { + tokens.push(token); + } + } + + let tokens_dir = root.join(FCM_TOKENS_DIR); + check_path_sandboxed(root, &tokens_dir)?; + + if !tokens_dir.exists() { + return Ok(tokens); + } + + let mut entries = tokio_fs::read_dir(&tokens_dir).await?; + while let Some(entry) = entries.next_entry().await? { + if !entry.file_type().await?.is_file() { + continue; + } + + let file_name = entry.file_name(); + let Some(file_name) = file_name.to_str() else { + continue; + }; + + if !file_name.starts_with(FCM_TOKEN_FILE_PREFIX) { + continue; + } + + let token_path = entry.path(); + check_path_sandboxed(root, &token_path)?; + + let token = tokio_fs::read_to_string(token_path).await?; + let token = token.trim().to_string(); + + if !token.is_empty() && seen.insert(token.clone()) { + tokens.push(token); + } + } + + Ok(tokens) +} \ No newline at end of file diff --git a/server/src/main.rs b/server/src/main.rs index eaa8728..5e2aee9 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -49,7 +49,7 @@ pub mod notification_target; pub mod security; use self::auth::{initialize_users, BasicAuth, FailStore}; -use self::fcm::send_notification; +use self::fcm::{send_notification, store_fcm_token, load_fcm_tokens}; use self::security::{check_path_sandboxed, join_validated_child}; // Store the version of the current crate, which we'll use in all responses. @@ -103,9 +103,17 @@ struct PairingEntry { expired: bool, } +// Add App Structures +struct AddAppEntry { + payload: AsyncMutex>>, + notify: Notify, +} + type PairingSessionKey = (String, String); type SharedPairingState = Arc>>>>; type AllEventState = Arc>; +type AddAppKey = (String, String); +type SharedAddAppState = Arc>>; // Simple rate limiters for the server const MAX_MOTION_FILE_SIZE: usize = 50; // in mebibytes @@ -113,11 +121,13 @@ const MAX_NUM_PENDING_MOTION_FILES: usize = 100; const MAX_LIVESTREAM_FILE_SIZE: usize = 20; // in mebibytes const MAX_NUM_PENDING_LIVESTREAM_FILES: usize = 50; const MAX_COMMAND_FILE_SIZE: usize = 100; // in kibibytes +const MAX_ADD_APP_REQUEST_SIZE: usize = 100; // in kibibytes const MAX_JSON_SIZE: usize = 10; // in kibibytes #[cfg(not(test))] const PAIRING_SESSION_TIMEOUT: Duration = Duration::from_secs(45); #[cfg(test)] const PAIRING_SESSION_TIMEOUT: Duration = Duration::from_millis(250); +const ADD_APP_REQUEST_TIMEOUT: Duration = Duration::from_secs(45); async fn get_num_files(path: &Path) -> io::Result { let mut entries = fs::read_dir(path).await?; @@ -137,38 +147,16 @@ async fn persist_pair_notification_target( target: &NotificationTarget, ) -> io::Result<()> { let root = Path::new("data").join(&auth.username); - let target_path = root.join("notification_target.json"); - check_path_sandboxed(&root, &target_path)?; - - fs::create_dir_all(&root).await?; - let target_json = serde_json::to_vec(target) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; - fs::write(target_path, target_json).await?; - Ok(()) + notification_target::store_notification_target(&root, target).await } async fn load_notification_target( root: &Path, notification_target_policy: ¬ification_target::UnifiedPushPolicy, ) -> io::Result> { - let target_path = root.join("notification_target.json"); - check_path_sandboxed(root, &target_path)?; - - if !target_path.exists() { - return Ok(None); - } - - let raw = fs::read_to_string(target_path).await?; - let parsed = serde_json::from_str::(&raw) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; - if let Err(err) = - notification_target::validate_notification_target(notification_target_policy, &parsed) - { - warn!("Ignoring invalid persisted notification target: {err}"); - return Ok(None); - } - - Ok(Some(parsed)) + let targets = + notification_target::load_notification_targets(root, notification_target_policy).await?; + Ok(targets.into_iter().next()) } #[post("/pair", data = "")] @@ -195,8 +183,8 @@ async fn pair( let token = &data.pairing_token; // Check for disallowed quote characters in the token - if token.contains('"') { - debug!("[PAIR] Invalid token contains quote character: {}", token); + if token.is_empty() || token.contains('"') { + debug!("[PAIR] Invalid token (empty or contains quote character: {})", token); return Json(PairingResponse { status: "invalid_token".into(), notification_target: None, @@ -571,15 +559,20 @@ async fn delete_camera(camera: &str, auth: &BasicAuth) -> io::Result<()> { #[post("/fcm_token", data = "")] async fn upload_fcm_token(data: Data<'_>, auth: &BasicAuth) -> io::Result { let root = Path::new("data").join(&auth.username); - let token_path = root.join("fcm_token"); - check_path_sandboxed(&root, &token_path)?; + + let token_bytes = data.open(5.kibibytes()).into_bytes().await?; + let token = String::from_utf8(token_bytes.to_vec()) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; + let token = token.trim(); - let mut file = fs::File::create(&token_path).await?; - // FIXME: hardcoded max size - let mut stream = data.open(5.kibibytes()); - tokio::io::copy(&mut stream, &mut file).await?; - // Flush the file to disk - file.sync_all().await?; + if token.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "Error: FCM token is empty.", + )); + } + + store_fcm_token(&root, token).await?; Ok("ok".to_string()) } @@ -595,13 +588,7 @@ async fn upload_notification_target( .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?; let root = Path::new("data").join(&auth.username); - let target_path = root.join("notification_target.json"); - check_path_sandboxed(&root, &target_path)?; - - fs::create_dir_all(&root).await?; - let target_json = serde_json::to_vec(&target) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; - fs::write(target_path, target_json).await?; + notification_target::store_notification_target(&root, &target).await?; Ok("ok".to_string()) } @@ -626,78 +613,101 @@ async fn send_fcm_notification( auth: &BasicAuth, ) -> io::Result { let root = Path::new("data").join(&auth.username); - let notification_target = - load_notification_target(&root, notification_target_policy.inner()).await?; + let notification_targets = + notification_target::load_notification_targets(&root, notification_target_policy.inner()) + .await?; let notification_msg = data.open(8.kibibytes()).into_bytes().await?; - if let Some(target) = notification_target.as_ref() { - if target.platform.eq_ignore_ascii_case("android_unified") { - let endpoint_url = match target.unifiedpush_endpoint_url.as_deref() { - Some(value) if !value.trim().is_empty() => value, - _ => { - debug!("Skipping UnifiedPush notification; endpoint URL not available"); - return Ok("ok".to_string()); - } - }; - let pub_key = match target.unifiedpush_pub_key.as_deref() { - Some(value) if !value.trim().is_empty() => value, - _ => { - debug!("Skipping UnifiedPush notification; public key not available"); - return Ok("ok".to_string()); - } - }; - let auth_secret = match target.unifiedpush_auth.as_deref() { - Some(value) if !value.trim().is_empty() => value, - _ => { - debug!("Skipping UnifiedPush notification; auth secret not available"); - return Ok("ok".to_string()); - } - }; + let mut attempted_notification_target = false; + // FIXME: caller won't know if the notification failed to send + let mut notification_target_success_count: usize = 0; - match notification_target::send_notification( - notification_target_policy.inner(), - endpoint_url, - pub_key, - auth_secret, - notification_msg.as_ref(), - ) - .await - { - Ok(_) => { - debug!("UnifiedPush notification sent successfully."); - } - Err(e) => { - debug!("Failed to send UnifiedPush notification: {}", e); - } + for target in notification_targets.iter() { + if !target.platform.eq_ignore_ascii_case("android_unified") { + continue; + } + + attempted_notification_target = true; + + let endpoint_url = match target.unifiedpush_endpoint_url.as_deref() { + Some(value) if !value.trim().is_empty() => value, + _ => { + debug!("Skipping UnifiedPush notification; endpoint URL not available"); + return Ok("ok".to_string()); + } + }; + let pub_key = match target.unifiedpush_pub_key.as_deref() { + Some(value) if !value.trim().is_empty() => value, + _ => { + debug!("Skipping UnifiedPush notification; public key not available"); + return Ok("ok".to_string()); + } + }; + let auth_secret = match target.unifiedpush_auth.as_deref() { + Some(value) if !value.trim().is_empty() => value, + _ => { + debug!("Skipping UnifiedPush notification; auth secret not available"); + return Ok("ok".to_string()); + } + }; + + match notification_target::send_notification( + notification_target_policy.inner(), + endpoint_url, + pub_key, + auth_secret, + notification_msg.as_ref(), + ) + .await + { + Ok(_) => { + notification_target_success_count += 1; + debug!("UnifiedPush notification sent successfully."); + } + Err(e) => { + debug!("Failed to send UnifiedPush notification: {}", e); } - return Ok("ok".to_string()); } } + if attempted_notification_target { + debug!( + "UnifiedPush notification fan-out completed: {notification_target_success_count} successful sends." + ); + } + if fcm_config.inner().is_none() { debug!("Skipping FCM notification; server has no FCM configuration"); return Ok("ok".to_string()); } - let token_path = root.join("fcm_token"); - check_path_sandboxed(&root, &token_path)?; - if !token_path.exists() { + let tokens = load_fcm_tokens(&root).await?; + if tokens.is_empty() { return Err(io::Error::other("Error: FCM token not available.")); } - let token = fs::read_to_string(token_path).await?; + + let notification_payload = notification_msg.to_vec(); task::block_in_place(|| { // FIXME: caller won't know if the notification failed to send + let mut success_count: usize = 0; - match send_notification(token, notification_msg.to_vec()) { - Ok(_) => { - debug!("Notification sent successfully."); - } - Err(e) => { - debug!("Failed to send notification: {}", e); + for token in tokens { + match send_notification(token, notification_payload.clone()) { + Ok(_) => { + success_count += 1; + debug!("Notification sent successfully."); + } + Err(e) => { + debug!("Failed to send notification: {}", e); + } } } + + debug!("FCM notification fan-out completed: {success_count} successful sends."); }); + + Ok("ok".to_string()) } @@ -1208,6 +1218,87 @@ async fn upload_debug_logs(data: Data<'_>, auth: &BasicAuth) -> io::Result")] +async fn add_app_check( + op: &str, + auth: &BasicAuth, + state: &rocket::State, +) -> Result, Custom> { + if op.is_empty() || op.contains('"') { + debug!("[ADD_APP_CHECK] Invalid op (empty or contains quote character: {})", op); + return Err(Custom(Status::BadRequest, "invalid op".to_string())); + } + + let key = (auth.username.clone(), op.to_string()); + let entry = state + .entry(key.clone()) + .or_insert_with(|| { + Arc::new(AddAppEntry { + payload: AsyncMutex::new(None), + notify: Notify::new(), + }) + }) + .clone(); + + let result = timeout(ADD_APP_REQUEST_TIMEOUT, async { + loop { + let notified = entry.notify.notified(); + if let Some(payload) = entry.payload.lock().await.take() { + return payload; + } + notified.await; + } + }) + .await; + + state.remove(&key); + result.map_err(|_| Custom(Status::RequestTimeout, "request timed out".to_string())) +} + +#[post("/add_app_request/", data = "")] +async fn add_app_request( + op: &str, + data: Data<'_>, + auth: &BasicAuth, + state: &rocket::State, +) -> Result> { + if op.is_empty() || op.contains('"') { + debug!("[ADD_APP_REQUEST] Invalid op (empty or contains quote character: {})", op); + return Err(Custom(Status::BadRequest, "invalid op".to_string())); + } + + let key = (auth.username.clone(), op.to_string()); + let entry = state + .get(&key) + .map(|entry| entry.value().clone()) + .ok_or_else(|| Custom(Status::NotFound, "no app is waiting for this op".to_string()))?; + + let payload = data + .open(MAX_ADD_APP_REQUEST_SIZE.kibibytes()) + .into_bytes() + .await + .map_err(|error| Custom(Status::BadRequest, error.to_string()))?; + if !payload.is_complete() { + return Err(Custom( + Status::PayloadTooLarge, + "request payload is too large".to_string(), + )); + } + + let mut pending_payload = entry.payload.lock().await; + if pending_payload.is_some() { + return Err(Custom( + Status::Conflict, + "a request is already pending for this op".to_string(), + )); + } + *pending_payload = Some(payload.into_inner()); + drop(pending_payload); + entry.notify.notify_one(); + + Ok("ok".to_string()) +} + #[launch] fn rocket() -> rocket::Rocket { build_rocket() @@ -1216,6 +1307,7 @@ fn rocket() -> rocket::Rocket { pub fn build_rocket() -> rocket::Rocket { let all_event_state: AllEventState = Arc::new(DashMap::new()); let pairing_state: SharedPairingState = Arc::new(Mutex::new(HashMap::new())); + let add_app_state: SharedAddAppState = Arc::new(DashMap::new()); let failure_store: FailStore = Arc::new(DashMap::new()); let mut network_type: Option = None; @@ -1292,6 +1384,7 @@ pub fn build_rocket() -> rocket::Rocket { .manage(pairing_state) .manage(fcm_config) .manage(notification_target_policy) + .manage(add_app_state) .mount( "/", routes![ @@ -1317,6 +1410,8 @@ pub fn build_rocket() -> rocket::Rocket { upload_debug_logs, retrieve_fcm_data, retrieve_server_status, + add_app_check, + add_app_request, ], ) } diff --git a/server/src/notification_target.rs b/server/src/notification_target.rs index c079483..2e067b8 100644 --- a/server/src/notification_target.rs +++ b/server/src/notification_target.rs @@ -8,8 +8,21 @@ use reqwest::Url; use secluso_server_backbone::types::{IosRelayBinding, NotificationTarget}; use std::{env, time::Duration}; use web_push_native::{p256, Auth, WebPushBuilder}; +use rocket::tokio::fs as tokio_fs; +use rocket::tokio::io::AsyncWriteExt; +use std::collections::HashSet; +use std::io::{self, ErrorKind}; +use std::path::Path; +use std::sync::OnceLock; +use rocket::tokio::sync::Mutex; + +use crate::security::check_path_sandboxed; pub const UNIFIEDPUSH_ALLOWED_HOSTS_ENV: &str = "SECLUSO_UNIFIEDPUSH_ALLOWED_HOSTS"; +const LEGACY_NOTIFICATION_TARGET_FILE: &str = "notification_target.json"; +const NOTIFICATION_TARGETS_DIR: &str = "notification_targets"; +const NOTIFICATION_TARGET_FILE_PREFIX: &str = "notification_target_"; +const NOTIFICATION_TARGET_FILE_SUFFIX: &str = ".json"; // Hosted distributors we intentionally support out of the box. Self-hosted // backends must be added explicitly via SECLUSO_UNIFIEDPUSH_ALLOWED_HOSTS. @@ -302,6 +315,215 @@ pub async fn send_notification( Ok(()) } +fn is_notification_target_file(file_name: &str) -> bool { + file_name.starts_with(NOTIFICATION_TARGET_FILE_PREFIX) + && file_name.ends_with(NOTIFICATION_TARGET_FILE_SUFFIX) +} + +fn notification_target_key(target: &NotificationTarget) -> String { + let platform = target.platform.to_ascii_lowercase(); + + if platform == "android_unified" { + return format!( + "android_unified:{}", + target + .unifiedpush_endpoint_url + .as_deref() + .unwrap_or("") + .trim() + ); + } + + if platform == "ios" { + if let Some(binding) = target.ios_relay_binding.as_ref() { + return format!( + "ios:{}:{}", + binding.hub_id.trim(), + binding.app_install_id.trim() + ); + } + + return "ios:placeholder".to_string(); + } + + serde_json::to_string(target).unwrap_or(platform) +} + +static NOTIFICATION_TARGET_STORE_LOCK: OnceLock> = OnceLock::new(); + +// FIXME: we only add notification_targets, but never remove stale ones. +pub(crate) async fn store_notification_target( + root: &Path, + target: &NotificationTarget, +) -> io::Result<()> { + // Two overlapping calls to store_notification_target from one app can + // result in duplicates. This lock is used to prevent that. + let lock = NOTIFICATION_TARGET_STORE_LOCK.get_or_init(|| Mutex::new(())); + let _guard = lock.lock().await; + + tokio_fs::create_dir_all(root).await?; + + let targets_dir = root.join(NOTIFICATION_TARGETS_DIR); + check_path_sandboxed(root, &targets_dir)?; + tokio_fs::create_dir_all(&targets_dir).await?; + + let target_json = serde_json::to_vec(target) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; + let target_key = notification_target_key(target); + + let mut entries = tokio_fs::read_dir(&targets_dir).await?; + while let Some(entry) = entries.next_entry().await? { + if !entry.file_type().await?.is_file() { + continue; + } + + let file_name = entry.file_name(); + let Some(file_name) = file_name.to_str() else { + continue; + }; + + if !is_notification_target_file(file_name) { + continue; + } + + let path = entry.path(); + check_path_sandboxed(root, &path)?; + + let existing_raw = match tokio_fs::read_to_string(&path).await { + Ok(value) => value, + Err(e) if e.kind() == ErrorKind::NotFound => continue, + Err(e) => return Err(e), + }; + + let existing = match serde_json::from_str::(&existing_raw) { + Ok(value) => value, + Err(_) => continue, + }; + + if notification_target_key(&existing) == target_key { + let mut file = tokio_fs::File::create(&path).await?; + file.write_all(&target_json).await?; + file.sync_all().await?; + return Ok(()); + } + } + + for index in 1u64.. { + let target_path = targets_dir.join(format!( + "{}{}{}", + NOTIFICATION_TARGET_FILE_PREFIX, + index, + NOTIFICATION_TARGET_FILE_SUFFIX + )); + check_path_sandboxed(root, &target_path)?; + + let mut file = match tokio_fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&target_path) + .await + { + Ok(file) => file, + Err(e) if e.kind() == ErrorKind::AlreadyExists => continue, + Err(e) => return Err(e), + }; + + file.write_all(&target_json).await?; + file.sync_all().await?; + return Ok(()); + } + + unreachable!() +} + +async fn read_notification_target_file( + path: &Path, + notification_target_policy: &UnifiedPushPolicy, +) -> io::Result> { + let raw = match tokio_fs::read_to_string(path).await { + Ok(value) => value, + Err(e) if e.kind() == ErrorKind::NotFound => return Ok(None), + Err(e) => return Err(e), + }; + + let parsed = match serde_json::from_str::(&raw) { + Ok(value) => value, + Err(e) => { + warn!("Ignoring invalid notification target JSON: {e}"); + return Ok(None); + } + }; + + if let Err(err) = validate_notification_target(notification_target_policy, &parsed) { + warn!("Ignoring invalid persisted notification target: {err}"); + return Ok(None); + } + + Ok(Some(parsed)) +} + +pub(crate) async fn load_notification_targets( + root: &Path, + notification_target_policy: &UnifiedPushPolicy, +) -> io::Result> { + let mut targets = Vec::new(); + let mut seen = HashSet::new(); + + if !root.exists() { + return Ok(targets); + } + + let legacy_target_path = root.join(LEGACY_NOTIFICATION_TARGET_FILE); + check_path_sandboxed(root, &legacy_target_path)?; + if legacy_target_path.exists() { + if let Some(target) = + read_notification_target_file(&legacy_target_path, notification_target_policy).await? + { + let key = notification_target_key(&target); + if seen.insert(key) { + targets.push(target); + } + } + } + + let targets_dir = root.join(NOTIFICATION_TARGETS_DIR); + check_path_sandboxed(root, &targets_dir)?; + + if !targets_dir.exists() { + return Ok(targets); + } + + let mut entries = tokio_fs::read_dir(&targets_dir).await?; + while let Some(entry) = entries.next_entry().await? { + if !entry.file_type().await?.is_file() { + continue; + } + + let file_name = entry.file_name(); + let Some(file_name) = file_name.to_str() else { + continue; + }; + + if !is_notification_target_file(file_name) { + continue; + } + + let target_path = entry.path(); + check_path_sandboxed(root, &target_path)?; + + if let Some(target) = + read_notification_target_file(&target_path, notification_target_policy).await? + { + let key = notification_target_key(&target); + if seen.insert(key) { + targets.push(target); + } + } + } + + Ok(targets) +} + #[cfg(test)] mod tests { use super::{validate_notification_target, UnifiedPushPolicy}; From eab8289b0ef83e1c38026bf266bf2d122834eae7 Mon Sep 17 00:00:00 2001 From: Ardalan Amiri Sani Date: Tue, 30 Jun 2026 00:47:28 -0700 Subject: [PATCH 2/3] Remove unnecessary dependency from app_native. --- app_native/Cargo.toml | 1 - 1 file changed, 1 deletion(-) diff --git a/app_native/Cargo.toml b/app_native/Cargo.toml index 20d55ee..a183847 100644 --- a/app_native/Cargo.toml +++ b/app_native/Cargo.toml @@ -18,7 +18,6 @@ serde = "1.0" serde_derive = "1.0" openmls = "=0.8.1" docopt = "~1.1" -base64-url = "3.0.3" [features] default = [] From 10bf5bacf27021f9aa78d21c6ad1e2f70b35e50c Mon Sep 17 00:00:00 2001 From: Ardalan Amiri Sani Date: Tue, 30 Jun 2026 11:10:05 -0700 Subject: [PATCH 3/3] Update server_backbone with routes needed for the add-app process. --- server_backbone/src/lib.rs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/server_backbone/src/lib.rs b/server_backbone/src/lib.rs index 9aab380..3d3275b 100644 --- a/server_backbone/src/lib.rs +++ b/server_backbone/src/lib.rs @@ -11,9 +11,11 @@ pub mod routes { const PARAM_NONE: &[&str] = &[]; const PARAM_CAMERA: &[&str] = &["camera"]; const PARAM_CAMERA_FILENAME: &[&str] = &["camera", "filename"]; + const PARAM_CAMERA_FILENAME_COUNTER: &[&str] = &["camera", "filename", "counter"]; + const PARAM_OP: &[&str] = &["op"]; pub const ROUTE_PAIR: &str = "/pair"; - pub const ROUTE_UPLOAD: &str = "//"; + pub const ROUTE_UPLOAD: &str = "///"; pub const ROUTE_BULK_CHECK: &str = "/bulkCheck"; pub const ROUTE_RETRIEVE: &str = "//"; pub const ROUTE_DELETE_FILE: &str = "//"; @@ -33,6 +35,8 @@ pub mod routes { pub const ROUTE_FCM_CONFIG: &str = "/fcm_config"; pub const ROUTE_STATUS: &str = "/status"; pub const ROUTE_DEBUG_LOGS: &str = "/debug_logs"; + pub const ROUTE_ADD_APP_CHECK: &str = "/add_app_check/"; + pub const ROUTE_ADD_APP_REQUEST: &str = "/add_app_request/"; pub const BASE_ROUTES: &[RouteSpec] = &[ RouteSpec { @@ -43,7 +47,7 @@ pub mod routes { RouteSpec { method: HttpMethod::Post, path: ROUTE_UPLOAD, - params: PARAM_CAMERA_FILENAME, + params: PARAM_CAMERA_FILENAME_COUNTER, }, RouteSpec { method: HttpMethod::Post, @@ -145,6 +149,16 @@ pub mod routes { path: ROUTE_DEBUG_LOGS, params: PARAM_NONE, }, + RouteSpec { + method: HttpMethod::Get, + path: ROUTE_ADD_APP_CHECK, + params: PARAM_OP, + }, + RouteSpec { + method: HttpMethod::Post, + path: ROUTE_ADD_APP_REQUEST, + params: PARAM_OP, + }, ]; }