Sequester with key rotation

.github/workflows/ci-rust.yml

@@ -63,7 +63,7 @@ jobs:
       TESTBED_SERVER: http://localhost:6777
     services:
       parsec-testbed-server:
-        image: ghcr.io/scille/parsec-cloud/parsec-testbed-server:3.2.1-a.0.dev.20078.9f7ecb1
+        image: ghcr.io/scille/parsec-cloud/parsec-testbed-server:3.2.4-a.0.dev.20102.1979660
         ports:
           - 6777:6777
     steps:

.github/workflows/ci-web.yml

@@ -36,7 +36,7 @@ jobs:
     # https://github.com/Scille/parsec-cloud/pkgs/container/parsec-cloud%2Fparsec-testbed-server
     services:
       parsec-testbed-server:
-        image: ghcr.io/scille/parsec-cloud/parsec-testbed-server:3.2.1-a.0.dev.20078.9f7ecb1
+        image: ghcr.io/scille/parsec-cloud/parsec-testbed-server:3.2.4-a.0.dev.20102.1979660
         ports:
           - 6777:6777
     steps:

libparsec/crates/client/src/certif/mod.rs

@@ -461,6 +461,7 @@ impl CertificateOps {
             })?
     }
 
+    #[allow(unused)]
     pub async fn encrypt_for_sequester_services(
         &self,
         data: &[u8],

libparsec/crates/client/src/certif/realm_key_rotation.rs

@@ -3,20 +3,21 @@
 use std::collections::HashMap;
 
 use libparsec_client_connection::ConnectionError;
-use libparsec_platform_storage::certificates::{GetCertificateError, UpTo};
+use libparsec_platform_storage::certificates::{GetCertificateError, PerTopicLastTimestamps, UpTo};
 use libparsec_protocol::authenticated_cmds;
 use libparsec_types::prelude::*;
 
 use super::{
-    store::CertifStoreError, CertificateBasedActionOutcome, CertificateOps, InvalidCertificateError,
+    encrypt::encrypt_for_sequester_services, store::CertifStoreError,
+    CertificateBasedActionOutcome, CertificateOps, InvalidCertificateError,
 };
 use crate::{
     certif::{
         realm_keys_bundle::{self, GenerateNextKeyBundleForRealmError},
         CertifPollServerError,
     },
-    greater_timestamp, EventTooMuchDriftWithServerClock, GreaterTimestampOffset,
-    InvalidKeysBundleError,
+    greater_timestamp, CertifEncryptForSequesterServicesError, EventTooMuchDriftWithServerClock,
+    GreaterTimestampOffset, InvalidKeysBundleError,
 };
 
 #[derive(Debug, thiserror::Error)]
@@ -108,9 +109,12 @@ pub(super) async fn rotate_realm_key_idempotent(
             }),
             Rep::RealmNotFound => Err(CertifRotateRealmKeyError::UnknownRealm),
             Rep::AuthorNotAllowed => Err(CertifRotateRealmKeyError::AuthorNotAllowed),
-            Rep::ParticipantMismatch => {
+            Rep::ParticipantMismatch { last_realm_certificate_timestamp } => {
                 // List of participants got updated in our back, refresh and retry
-                ops.poll_server_for_new_certificates(None)
+                let latest_known_timestamps = PerTopicLastTimestamps::new_for_realm(realm_id, last_realm_certificate_timestamp);
+                ops.poll_server_for_new_certificates(
+                    Some(&latest_known_timestamps)
+                )
                     .await
                     .map_err(|e| match e {
                         CertifPollServerError::Offline => CertifRotateRealmKeyError::Offline,
@@ -157,7 +161,34 @@ pub(super) async fn rotate_realm_key_idempotent(
                     ballpark_client_late_offset,
                 })
             }
-            bad_rep @ (Rep::InvalidCertificate { .. } | Rep::UnknownStatus { .. }) => {
+
+            // TODO: provide a dedicated error for this exotic behavior ?
+            Rep::SequesterServiceUnavailable { .. } => Err(CertifRotateRealmKeyError::Offline),
+            // TODO: we should send a dedicated event for this, and return an according error
+            Rep::RejectedBySequesterService { .. } => { todo!() }
+            // Sequester services has changed concurrently, should poll for new certificates and retry
+            Rep::SequesterServiceMismatch { last_sequester_certificate_timestamp } => {
+                let latest_known_timestamps = PerTopicLastTimestamps::new_for_sequester(last_sequester_certificate_timestamp);
+                ops
+                    .poll_server_for_new_certificates(Some(&latest_known_timestamps))
+                    .await
+                    .map_err(|err| match err {
+                        CertifPollServerError::Stopped => CertifRotateRealmKeyError::Stopped,
+                        CertifPollServerError::Offline => CertifRotateRealmKeyError::Offline,
+                        CertifPollServerError::InvalidCertificate(err) => CertifRotateRealmKeyError::InvalidCertificate(err),
+                        CertifPollServerError::Internal(err) => err.context("Cannot poll server for new certificates").into(),
+                    })?;
+                timestamp = ops.device.now();
+                continue;
+            }
+
+            // Unexpected errors :(
+            bad_rep @ (
+                Rep::InvalidCertificate { .. }
+                // We only encrypt for sequester services when the realm is sequestered
+                | Rep::OrganizationNotSequestered
+                | Rep::UnknownStatus { .. }
+            ) => {
                 Err(anyhow::anyhow!("Unexpected server response: {:?}", bad_rep).into())
             }
         };
@@ -279,14 +310,33 @@ async fn generate_realm_rotate_key_req(
                 per_participant_keys_bundle_access
             };
 
-            // 4) Now we have everything we need to build the request object !
+            // 4) Encrypte keys bundle access for each active sequester service
+
+            let per_sequester_service_keys_bundle_access = {
+                let sequester_blob =
+                    encrypt_for_sequester_services(store, &keys_bundle_access_cleartext)
+                        .await
+                        .map_err(|e| match e {
+                            CertifEncryptForSequesterServicesError::Stopped => {
+                                CertifRotateRealmKeyError::Stopped
+                            }
+                            CertifEncryptForSequesterServicesError::Internal(err) => err
+                                .context("Cannot encrypt manifest for sequester services")
+                                .into(),
+                        })?;
+
+                sequester_blob.map(|sequester_blob| HashMap::from_iter(sequester_blob.into_iter()))
+            };
+
+            // 5) Now we have everything we need to build the request object !
 
             use authenticated_cmds::latest::realm_rotate_key::Req;
 
             let req = Req {
                 realm_key_rotation_certificate: certif.into(),
                 keys_bundle: keys_bundle_encrypted,
                 per_participant_keys_bundle_access,
+                per_sequester_service_keys_bundle_access,
             };
 
             Ok((req, key_index))

libparsec/crates/client/src/monitors/user_sync.rs

@@ -94,7 +94,7 @@ fn task_future_factory(user_ops: Arc<UserOps>, event_bus: EventBus) -> impl Futu
                             log::warn!(
                                 "Rejected upload by sequester service {}: {}",
                                 service_id,
-                                reason
+                                reason.as_ref().map_or("<no reason>", |r| r.as_str())
                             );
                         }
                         UserSyncError::InvalidCertificate(error) => {

libparsec/crates/client/src/user/sync.rs

@@ -8,8 +8,7 @@ use libparsec_types::prelude::*;
 use super::{UserOps, UserStoreUpdateError};
 use crate::{
     certif::{
-        CertifEncryptForSequesterServicesError, CertifEnsureRealmCreatedError,
-        CertifPollServerError, CertifValidateManifestError, InvalidCertificateError,
+        CertifEnsureRealmCreatedError, CertifValidateManifestError, InvalidCertificateError,
         InvalidManifestError,
     },
     greater_timestamp, EventUserOpsOutboundSyncDone, GreaterTimestampOffset,
@@ -21,10 +20,10 @@ pub enum UserSyncError {
     Offline,
     #[error("Component has stopped")]
     Stopped,
-    #[error("Sequester service `{service_id}` rejected the data: `{reason}`")]
+    #[error("Sequester service `{service_id}` rejected the data: `{}`", .reason.as_ref().map(|x| x.as_ref()).unwrap_or_else(|| "<no reason>"))]
     RejectedBySequesterService {
         service_id: SequesterServiceID,
-        reason: String,
+        reason: Option<String>,
     },
     #[error(transparent)]
     InvalidCertificate(#[from] Box<InvalidCertificateError>),
@@ -104,16 +103,6 @@ async fn upload_manifest(
 
         let signed = to_sync_um.dump_and_sign(&ops.device.signing_key);
         let ciphered = ops.device.user_realm_key.encrypt(&signed).into();
-        let sequester_blob = ops
-            .certificates_ops
-            .encrypt_for_sequester_services(&signed)
-            .await
-            .map_err(|e| match e {
-                CertifEncryptForSequesterServicesError::Stopped => UserSyncError::Stopped,
-                CertifEncryptForSequesterServicesError::Internal(err) => {
-                    err.context("Cannot encrypt for sequester services").into()
-                }
-            })?;
 
         // Sync the vlob with server
 
@@ -125,7 +114,6 @@ async fn upload_manifest(
                 key_index: 0,
                 timestamp,
                 blob: ciphered,
-                sequester_blob: sequester_blob.map(|v| v.into_iter().collect()),
             };
             let rep = ops.cmds.send(req).await?;
             match rep {
@@ -143,23 +131,7 @@ async fn upload_manifest(
                 }
                 // Timeout is about sequester service webhook not being available, no need
                 // for a custom handling of such an exotic error
-                Rep::SequesterServiceUnavailable => Err(UserSyncError::Offline),
-                Rep::SequesterInconsistency { .. } => {
-                    // Sequester services must have been concurrently modified in our back,
-                    // update and retry
-                    ops.certificates_ops
-                        .poll_server_for_new_certificates(None)
-                        .await
-                        .map_err(|err| match err {
-                            CertifPollServerError::Offline => UserSyncError::Offline,
-                            CertifPollServerError::Stopped => UserSyncError::Stopped,
-                            CertifPollServerError::InvalidCertificate(what) => {
-                                UserSyncError::InvalidCertificate(what)
-                            }
-                            err @ CertifPollServerError::Internal(_) => UserSyncError::Internal(err.into()),
-                        })?;
-                    continue;
-                }
+                Rep::SequesterServiceUnavailable { .. } => Err(UserSyncError::Offline),
                 Rep::RejectedBySequesterService {
                     service_id,
                     reason,
@@ -181,8 +153,6 @@ async fn upload_manifest(
                     Rep::AuthorNotAllowed
                     // `outbound_sync_inner` ensures the realm is created before calling us
                     | Rep::RealmNotFound
-                    // We only encrypt for sequester services when the realm is sequestered
-                    | Rep::OrganizationNotSequestered
                     // The user realm is never supposed to do key rotation
                     | Rep::BadKeyIndex { .. }
                     | Rep::UnknownStatus { .. }
@@ -196,7 +166,6 @@ async fn upload_manifest(
                 key_index: 0,
                 timestamp,
                 blob: ciphered,
-                sequester_blob: sequester_blob.map(|v| v.into_iter().collect()),
             };
             let rep = ops.cmds.send(req).await?;
             match rep {
@@ -214,23 +183,7 @@ async fn upload_manifest(
                 }
                 // Timeout is about sequester service webhook not being available, no need
                 // for a custom handling of such an exotic error
-                Rep::SequesterServiceUnavailable => Err(UserSyncError::Offline),
-                Rep::SequesterInconsistency { .. } => {
-                    // Sequester services must have been concurrently modified in our back,
-                    // update and retry
-                    ops.certificates_ops
-                        .poll_server_for_new_certificates(None)
-                        .await
-                        .map_err(|err| match err {
-                            CertifPollServerError::Offline => UserSyncError::Offline,
-                            CertifPollServerError::Stopped => UserSyncError::Stopped,
-                            CertifPollServerError::InvalidCertificate(what) => {
-                                UserSyncError::InvalidCertificate(what)
-                            }
-                            err @ CertifPollServerError::Internal(_) => UserSyncError::Internal(err.into()),
-                        })?;
-                    continue;
-                }
+                Rep::SequesterServiceUnavailable { .. } => Err(UserSyncError::Offline),
                 Rep::RejectedBySequesterService {
                     service_id,
                     reason,
@@ -252,8 +205,6 @@ async fn upload_manifest(
                     Rep::AuthorNotAllowed
                     // The vlob must exists since we are at version > 1
                     | Rep::VlobNotFound
-                    // We only encrypt for sequester services when the realm is sequestered
-                    | Rep::OrganizationNotSequestered
                     // The user realm is never supposed to do key rotation
                     | Rep::BadKeyIndex { .. }
                     | Rep::UnknownStatus { .. }

libparsec/crates/client/src/workspace/transactions/outbound_sync.rs

@@ -1,6 +1,5 @@
 // Parsec Cloud (https://parsec.cloud) Copyright (c) BUSL-1.1 2016-present Scille SAS
 
-use std::collections::HashMap;
 use std::sync::Arc;
 
 use libparsec_client_connection::protocol::authenticated_cmds;
@@ -10,8 +9,7 @@ use libparsec_types::prelude::*;
 use super::super::WorkspaceOps;
 use super::WorkspaceSyncError;
 use crate::certif::{
-    CertifBootstrapWorkspaceError, CertifEncryptForRealmError,
-    CertifEncryptForSequesterServicesError, CertifPollServerError,
+    CertifBootstrapWorkspaceError, CertifEncryptForRealmError, CertifPollServerError,
 };
 use crate::workspace::store::{
     ForUpdateSyncLocalOnlyError, ReadChunkOrBlockError, WorkspaceStoreOperationError,
@@ -442,18 +440,6 @@ async fn upload_manifest<M: RemoteManifest>(
                 }
             })?;
         let encrypted = encrypted.into();
-        let sequester_blob = ops
-            .certificates_ops
-            .encrypt_for_sequester_services(&signed)
-            .await
-            .map_err(|e| match e {
-                CertifEncryptForSequesterServicesError::Stopped => WorkspaceSyncError::Stopped,
-                CertifEncryptForSequesterServicesError::Internal(err) => err
-                    .context("Cannot encrypt manifest for sequester services")
-                    .into(),
-            })?;
-        let sequester_blob =
-            sequester_blob.map(|sequester_blob| HashMap::from_iter(sequester_blob.into_iter()));
 
         // Sync the vlob with server
 
@@ -465,7 +451,6 @@ async fn upload_manifest<M: RemoteManifest>(
                 vlob_id,
                 timestamp: to_upload.timestamp(),
                 blob: encrypted,
-                sequester_blob,
             };
             let rep = ops.cmds.send(req).await?;
             match rep {
@@ -491,7 +476,7 @@ async fn upload_manifest<M: RemoteManifest>(
                 },
 
                 // TODO: provide a dedicated error for this exotic behavior ?
-                Rep::SequesterServiceUnavailable => Err(WorkspaceSyncError::Offline),
+                Rep::SequesterServiceUnavailable { .. } => Err(WorkspaceSyncError::Offline),
                 // TODO: we should send a dedicated event for this, and return an according error
                 Rep::RejectedBySequesterService { .. } => todo!(),
                 // A key rotation occured concurrently, should poll for new certificates and retry
@@ -508,27 +493,11 @@ async fn upload_manifest<M: RemoteManifest>(
                         })?;
                     continue;
                 }
-                // Sequester services has changed concurrently, should poll for new certificates and retry
-                Rep::SequesterInconsistency { last_common_certificate_timestamp } => {
-                    let latest_known_timestamps = PerTopicLastTimestamps::new_for_common(last_common_certificate_timestamp);
-                    ops.certificates_ops
-                        .poll_server_for_new_certificates(Some(&latest_known_timestamps))
-                        .await
-                        .map_err(|err| match err {
-                            CertifPollServerError::Stopped => WorkspaceSyncError::Stopped,
-                            CertifPollServerError::Offline => WorkspaceSyncError::Offline,
-                            CertifPollServerError::InvalidCertificate(err) => WorkspaceSyncError::InvalidCertificate(err),
-                            CertifPollServerError::Internal(err) => err.context("Cannot poll server for new certificates").into(),
-                        })?;
-                    continue;
-                }
 
                 // Unexpected errors :(
                 bad_rep @ (
-                    // Got sequester info from certificates
-                    Rep::OrganizationNotSequestered
                     // Already checked the realm exists when we called `CertificateOps::encrypt_for_realm`
-                    | Rep::RealmNotFound
+                    Rep::RealmNotFound
                     // Don't know what to do with this status :/
                     | Rep::UnknownStatus { .. }
                 ) => {
@@ -543,7 +512,6 @@ async fn upload_manifest<M: RemoteManifest>(
                 version: to_upload.version(),
                 timestamp: to_upload.timestamp(),
                 blob: encrypted,
-                sequester_blob,
             };
             let rep = ops.cmds.send(req).await?;
             match rep {
@@ -569,7 +537,7 @@ async fn upload_manifest<M: RemoteManifest>(
                 },
 
                 // TODO: provide a dedicated error for this exotic behavior ?
-                Rep::SequesterServiceUnavailable => Err(WorkspaceSyncError::Offline),
+                Rep::SequesterServiceUnavailable { .. } => Err(WorkspaceSyncError::Offline),
                 // TODO: we should send a dedicated event for this, and return an according error
                 Rep::RejectedBySequesterService { .. } => todo!(),
                 // A key rotation occured concurrently, should poll for new certificates and retry
@@ -586,27 +554,11 @@ async fn upload_manifest<M: RemoteManifest>(
                         })?;
                     continue;
                 }
-                // Sequester services has changed concurrently, should poll for new certificates and retry
-                Rep::SequesterInconsistency { last_common_certificate_timestamp }=> {
-                    let latest_known_timestamps = PerTopicLastTimestamps::new_for_common(last_common_certificate_timestamp);
-                    ops.certificates_ops
-                        .poll_server_for_new_certificates(Some(&latest_known_timestamps))
-                        .await
-                        .map_err(|err| match err {
-                            CertifPollServerError::Stopped => WorkspaceSyncError::Stopped,
-                            CertifPollServerError::Offline => WorkspaceSyncError::Offline,
-                            CertifPollServerError::InvalidCertificate(err) => WorkspaceSyncError::InvalidCertificate(err),
-                            CertifPollServerError::Internal(err) => err.context("Cannot poll server for new certificates").into(),
-                        })?;
-                    continue;
-                }
 
                 // Unexpected errors :(
                 bad_rep @ (
-                    // Got sequester info from certificates
-                    Rep::OrganizationNotSequestered
                     // Already checked the vlob exists since the manifest has version > 0
-                    | Rep::VlobNotFound
+                    Rep::VlobNotFound
                     // Don't know what to do with this status :/
                     | Rep::UnknownStatus { .. }
                 ) => {